Streaming executor fixes ray-project#4

python/ray/data/tests/test_bulk_executor.py

@@ -127,7 +127,11 @@ def test_new_execution_backend_invocation(ray_start_10_cpus_shared):
     DatasetContext.get_current().new_execution_backend = True
     # Read-only: will use legacy executor for now.
     ds = ray.data.range(10)
-    assert ds.take_all() == list(range(10))
+    if DatasetContext.get_current().use_streaming_executor:
+        # Streaming executor doesn't preserve order by default.
+        assert set(ds.take_all()) == set(list(range(10)))
+    else:
+        assert ds.take_all() == list(range(10))
     # read->randomize_block_order: will use new executor, although it's also
     # a read-equivalent once fused.
     ds = ray.data.range(10).randomize_block_order()