[Data] Remove dead execute_to_legacy_block_iterator (#46714)

Signed-off-by: Balaji Veeramani <[email protected]>

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

@@ -17,24 +17,12 @@
 from ray.data._internal.plan import ExecutionPlan
 from ray.data._internal.stats import DatasetStats
 from ray.data._internal.util import unify_block_metadata_schema
-from ray.data.block import Block, BlockMetadata
-from ray.types import ObjectRef
+from ray.data.block import BlockMetadata
 
 # Warn about tasks larger than this.
 TASK_SIZE_WARN_THRESHOLD_BYTES = 100000
 
 
-def execute_to_legacy_block_iterator(
-    executor: Executor,
-    plan: ExecutionPlan,
-) -> Iterator[Tuple[ObjectRef[Block], BlockMetadata]]:
-    """Same as execute_to_legacy_bundle_iterator but returning blocks and metadata."""
-    bundle_iter = execute_to_legacy_bundle_iterator(executor, plan)
-    for bundle in bundle_iter:
-        for block, metadata in bundle.blocks:
-            yield block, metadata
-
-
 def execute_to_legacy_bundle_iterator(
     executor: Executor,
     plan: ExecutionPlan,

python/ray/data/tests/test_execution_optimizer.py

@@ -1498,19 +1498,6 @@ def test_execute_to_legacy_block_list(
     assert ds._plan._snapshot_stats.time_total_s > 0
 
 
-def test_execute_to_legacy_block_iterator(
-    ray_start_regular_shared,
-):
-    ds = ray.data.range(10)
-    assert ds._plan._snapshot_stats is None
-    for batch in ds.iter_batches():
-        assert batch is not None
-
-    assert ds._plan._snapshot_stats is not None
-    assert "ReadRange" in ds._plan._snapshot_stats.metadata
-    assert ds._plan._snapshot_stats.time_total_s > 0
-
-
 def test_streaming_executor(
     ray_start_regular_shared,
 ):