code: def test_lazy_block_list(shutdown_only, target_max_block_size):
      # Test 10 tasks, each task returning 10 blocks, each block has 1 row and each
      # row has 1024 bytes.
      num_blocks_per_task = 10
      block_size = 1024
      num_tasks = 10
  
      ds = ray.data.read_datasource(
          RandomBytesDatasource(),
          parallelism=num_tasks,
          num_batches_per_task=num_blocks_per_task,
          row_size=block_size,
      )
      ds.schema()
  
      # Check internal states of LazyBlockList before execution
      block_list = ds._plan._in_blocks
      block_refs = block_list._block_partition_refs
      cached_metadata = block_list._cached_metadata
      metadata = block_list.get_metadata()
  
      assert isinstance(block_list, LazyBlockList)
      assert len(block_refs) == num_tasks
      assert block_refs[0] is not None and all(
          map(lambda ref: ref is None, block_refs[1:])
      )
      assert all(
          map(
              lambda ref: ref is None,
              block_list._block_partition_meta_refs,
          )
      )
      assert len(cached_metadata) == num_tasks
      for i, block_metadata in enumerate(cached_metadata):
          if i == 0:
              assert len(block_metadata) == num_blocks_per_task
              for m in block_metadata:
                  assert m.num_rows == 1
          else:
              assert block_metadata is None
      assert len(metadata) == num_tasks - 1 + num_blocks_per_task
      for i, block_metadata in enumerate(metadata):
          if i < num_blocks_per_task:
              assert block_metadata.num_rows == 1
              assert block_metadata.schema is not None
          else:
              assert block_metadata.num_rows == num_blocks_per_task
              assert block_metadata.schema is None
  
      # Check APIs of LazyBlockList
      new_block_list = block_list.copy()
      new_block_list.clear()
      assert len(block_list._block_partition_refs) == num_tasks
  
      block_lists = block_list.split(2)
      assert len(block_lists) == num_tasks / 2
      assert len(block_lists[0]._block_partition_refs) == 2
      assert len(block_lists[0]._cached_metadata) == 2
  
      block_lists = block_list.split_by_bytes(
          block_size * num_blocks_per_task * 2
      )
      assert len(block_lists) == num_tasks / 2
      assert len(block_lists[0]._block_partition_refs) == 2
      assert len(block_lists[0]._cached_metadata) == 2
  
      new_block_list = block_list.truncate_by_rows(
          num_blocks_per_task * 3
      )
      assert len(new_block_list._block_partition_refs) == 3
      assert len(new_block_list._cached_metadata) == 3
  
      left_block_list, right_block_list = block_list.divide(3)
      assert len(left_block_list._block_partition_refs) == 3
      assert len(left_block_list._cached_metadata) == 3
      assert (
          len(right_block_list._block_partition_refs) == num_tasks - 3
      )
      assert len(right_block_list._cached_metadata) == num_tasks - 3
  
      new_block_list = block_list.randomize_block_order()
      assert len(new_block_list._block_partition_refs) == num_tasks
      assert len(new_block_list._cached_metadata) == num_tasks
  
      output_blocks = block_list.get_blocks_with_metadata()
      assert len(output_blocks) == num_tasks * num_blocks_per_task
      for _, metadata in output_blocks:
          assert metadata.num_rows == 1
      for _, metadata in block_list.iter_blocks_with_metadata():
          assert metadata.num_rows == 1
  
      # Check internal states of LazyBlockList after execution
      ds = ds.materialize()
      metadata = block_list.get_metadata()
  
      assert block_list._num_computed() == num_tasks
      assert len(block_refs) == num_tasks
      assert all(map(lambda ref: ref is not None, block_refs))
      assert all(
          map(
              lambda ref: ref is None,
              block_list._block_partition_meta_refs,
          )
      )
      assert len(cached_metadata) == num_tasks
      for block_metadata in cached_metadata:
          assert len(block_metadata) == num_blocks_per_task
          for m in block_metadata:
              assert m.num_rows == 1
      assert len(metadata) == num_tasks * num_blocks_per_task
      for block_metadata in metadata:
          assert block_metadata.num_rows == 1
          assert block_metadata.schema is not None
  
imports: import os
  import time
  from dataclasses import astuple, dataclass
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data import Dataset
  from ray.data._internal.arrow_block import ArrowBlockAccessor
  from ray.data._internal.lazy_block_list import LazyBlockList
  from ray.data.block import BlockMetadata
  from ray.data.datasource import Datasource
  from ray.data.datasource.csv_datasource import CSVDatasource
  from ray.data.datasource.datasource import Reader, ReadTask
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_blocks_expected_in_plasma, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import *
label_desc: Use image node_id 4 for calling a global function with example usage: test_lazy_block_list(shutdown_only, target_max_block_size) without return types