code: def test_bulk_lazy_eval_split_mode(
      shutdown_only, block_split, tmp_path
  ):
      # Defensively shutdown Ray for the first test here to make sure there
      # is no existing Ray cluster.
      ray.shutdown()
  
      ray.init(num_cpus=8)
      ctx = ray.data.context.DataContext.get_current()
  
      ray.data.range(8, parallelism=8).write_csv(str(tmp_path))
      if not block_split:
          # Setting infinite block size effectively disables block splitting.
          ctx.target_max_block_size = float("inf")
      ds = ray.data.read_datasource(
          SlowCSVDatasource(str(tmp_path)), parallelism=8
      )
  
      start = time.time()
      ds.map(lambda x: x)
      delta = time.time() - start
  
      print("full read time", delta)
      # Should run in ~3 seconds. It takes >9 seconds if bulk read is broken.
      assert delta < 8, delta
  
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 1 for calling a global function with example usage: test_bulk_lazy_eval_split_mode(shutdown_only, block_split, tmp_path) without return types