[Datasets] Always preserve order for the BulkExecutor. (ray-project#3…

…2437)

This PR always preserves order for the bulk executor. We may revisit this in the future, at which point we'd update all of the tests that rely on order preservation.

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

python/ray/data/_internal/execution/bulk_executor.py

@@ -24,6 +24,8 @@ class BulkExecutor(Executor):
     """
 
     def __init__(self, options: ExecutionOptions):
+        # Bulk executor always preserves order.
+        options.preserve_order = True
         super().__init__(options)
         self._stats: Optional[DatasetStats] = DatasetStats(stages={}, parent=None)
         self._executed = False

python/ray/data/_internal/execution/interfaces.py

@@ -156,7 +156,8 @@ class ExecutionOptions:
     # node (node driving the execution).
     locality_with_output: bool = False
 
-    # Set this to preserve the ordering between blocks processed by operators.
+    # Set this to preserve the ordering between blocks processed by operators under the
+    # streaming executor. The bulk executor always preserves order.
     preserve_order: bool = False
 
     # Whether to enable locality-aware task dispatch to actors (on by default).

python/ray/data/tests/test_bulk_executor.py

@@ -31,7 +31,18 @@ def ref_bundles_to_list(bundles: List[RefBundle]) -> List[List[Any]]:
     return output
 
 
-@pytest.mark.parametrize("preserve_order", [False, True])
+@pytest.mark.parametrize(
+    "preserve_order",
+    [
+        True,
+        pytest.param(
+            False,
+            marks=pytest.mark.skip(
+                reason="Bulk executor currently always preserves order"
+            ),
+        ),
+    ],
+)
 def test_multi_stage_execution(ray_start_10_cpus_shared, preserve_order):
     executor = BulkExecutor(ExecutionOptions(preserve_order=preserve_order))
     inputs = make_ref_bundles([[x] for x in range(20)])