code: def test_optimize_lazy_reuse_base_data(
      ray_start_regular_shared,
      local_path,
      enable_dynamic_splitting,
      with_shuffle,
  ):
      num_blocks = 4
      dfs = [
          pd.DataFrame({"one": list(range(i, i + 4))})
          for i in range(num_blocks)
      ]
      paths = [
          os.path.join(local_path, f"test{i}.csv")
          for i in range(num_blocks)
      ]
      for df, path in zip(dfs, paths):
          df.to_csv(path, index=False)
      counter = Counter.remote()
      source = MySource(paths, counter)
      ds = ray.data.read_datasource(source, parallelism=4)
      num_reads = ray.get(counter.get.remote())
      assert num_reads == 1, num_reads
      ds = ds.map(column_udf("id", lambda x: x))
      if with_shuffle:
          ds = ds.random_shuffle()
      ds.take()
      num_reads = ray.get(counter.get.remote())
      assert num_reads == num_blocks, num_reads
  
imports: import os
  from typing import List
  import numpy
  import pandas
  import pytest
  import ray
  from ray._private.internal_api import memory_summary
  from ray.data import Dataset
  from ray.data.block import BlockMetadata
  from ray.data.context import DataContext
  from ray.data.datasource import Datasource, ReadTask
  from ray.data.datasource.csv_datasource import CSVDatasource
  from ray.data.tests.util import column_udf, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 12 for calling a global function with example usage: test_optimize_lazy_reuse_base_data(ray_start_regular_shared, local_path, enable_dynamic_splitting, with_shuffle) without return types