[Data] Remove LazyBlockList (#46054)

After #45860, LazyBlockList is dead code.

Signed-off-by: Balaji Veeramani <balaji@anyscale.com>

python/ray/data/_internal/block_list.py

@@ -8,9 +8,7 @@
 class BlockList:
  """A list of blocks that may be computed or pending computation.
 
- In the basic version of BlockList, all blocks are known ahead of time. In
- LazyBlockList, blocks are not yet computed, so the number of blocks may
- change after execution due to block splitting.
+ All blocks are known ahead of time
  """
 
  def __init__(
@@ -69,7 +67,6 @@ def get_blocks(self) -> List[ObjectRef[Block]]:
  The length of this iterator is not known until execution.
  """
  self._check_if_cleared()
- # Overriden in LazyBlockList for bulk evaluation.
  return list(self._blocks)
 
  def get_blocks_with_metadata(self) -> List[Tuple[ObjectRef[Block], BlockMetadata]]:
@@ -78,7 +75,7 @@ def get_blocks_with_metadata(self) -> List[Tuple[ObjectRef[Block], BlockMetadata
  Prefer calling this instead of the iter form for performance if you
  don't need lazy evaluation.
  """
- self.get_blocks() # Force bulk evaluation in LazyBlockList.
+ self.get_blocks()
  return list(self.iter_blocks_with_metadata())
 
  def iter_blocks_with_metadata(

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

@@ -11,21 +11,11 @@
  PhysicalOperator,
  RefBundle,
 )
-from ray.data._internal.lazy_block_list import LazyBlockList
-from ray.data._internal.logical.interfaces.logical_plan import LogicalPlan
-from ray.data._internal.logical.operators.read_operator import Read
 from ray.data._internal.logical.optimizers import get_execution_plan
-from ray.data._internal.logical.rules.set_read_parallelism import (
- compute_additional_split_factor,
-)
 from ray.data._internal.logical.util import record_operators_usage
 from ray.data._internal.plan import ExecutionPlan
-from ray.data._internal.planner.plan_read_op import (
- apply_output_blocks_handling_to_read_task,
-)
 from ray.data._internal.stats import DatasetStats
 from ray.data.block import Block, BlockMetadata, List
-from ray.data.context import DataContext
 from ray.types import ObjectRef
 
 # Warn about tasks larger than this.
@@ -111,55 +101,6 @@ def execute_to_legacy_block_list(
  return block_list
 
 
-def get_legacy_lazy_block_list_read_only(
- plan: ExecutionPlan,
-) -> LazyBlockList:
- """For a read-only plan, construct a LazyBlockList with ReadTasks from the
- input Datasource or Reader. Note that the plan and the underlying ReadTasks
- are not executed, only their known metadata is fetched.
-
- Args:
- plan: The legacy plan to execute.
-
- Returns:
- The output as a legacy LazyBlockList.
- """
- assert plan.is_read_only(), "This function only supports read-only plans."
- assert isinstance(plan._logical_plan, LogicalPlan)
- read_logical_op = plan._logical_plan.dag
- assert isinstance(read_logical_op, Read)
-
- # In the full dataset execution, the logic in ApplyAdditionalSplitToOutputBlocks
- # is normally executed as part of the MapOperator created in the
- # LogicalPlan -> PhysicalPlan plan translation. In this case, since we
- # get the ReadTasks directly from the Datasource or Reader,
- # we need to manually apply this logic in order to update the ReadTasks.
- ctx = DataContext.get_current()
- (parallelism, _, estimated_num_blocks, k,) = compute_additional_split_factor(
- read_logical_op._datasource_or_legacy_reader,
- read_logical_op._parallelism,
- read_logical_op._mem_size,
- ctx.target_max_block_size,
- cur_additional_split_factor=None,
- )
- read_tasks = read_logical_op._datasource_or_legacy_reader.get_read_tasks(
- parallelism
- )
- for read_task in read_tasks:
- apply_output_blocks_handling_to_read_task(read_task, k)
-
- block_list = LazyBlockList(
- read_tasks,
- read_logical_op.name,
- ray_remote_args=read_logical_op._ray_remote_args,
- owned_by_consumer=False,
- )
- # Update the estimated number of blocks after applying optimizations
- # and fetching metadata (e.g. SetReadParallelismRule).
- block_list._estimated_num_blocks = estimated_num_blocks
- return block_list
-
-
 def _get_execution_dag(
  executor: Executor,
  plan: ExecutionPlan,