code: def test_lazy_fanout(shutdown_only, local_path):
      ray.init(num_cpus=1)
      map_counter = Counter.remote()
  
      def inc(row):
          map_counter.increment.remote()
          row["one"] += 1
          return row
  
      df = pd.DataFrame({"one": [1, 2, 3], "two": ["a", "b", "c"]})
      path = os.path.join(local_path, "test.csv")
      df.to_csv(path, index=False, storage_options={})
      read_counter = Counter.remote()
      source = MySource(path, read_counter)
  
      # Test that fan-out of a lazy dataset results in re-execution up to the datasource,
      # due to block move semantics.
      ds = ray.data.read_datasource(source, parallelism=1)
      ds1 = ds.map(inc)
      ds2 = ds1.map(inc)
      ds3 = ds1.map(inc)
      # Test content.
      assert ds2.materialize().take() == [
          {"one": 3, "two": "a"},
          {"one": 4, "two": "b"},
          {"one": 5, "two": "c"},
      ]
      assert ds3.materialize().take() == [
          {"one": 3, "two": "a"},
          {"one": 4, "two": "b"},
          {"one": 5, "two": "c"},
      ]
      # Test that data is read twice.
      assert ray.get(read_counter.get.remote()) == 2
      # Test that first map is executed twice.
      assert ray.get(map_counter.get.remote()) == 2 * 3 + 3 + 3
  
      # Test that fan-out of a lazy dataset with a non-lazy datasource results in
      # re-execution up to the datasource, due to block move semantics.
      ray.get(map_counter.reset.remote())
  
      def inc(x):
          map_counter.increment.remote()
          return {"item": x["item"] + 1}
  
      # The source data shouldn't be cleared since it's non-lazy.
      ds = ray.data.from_items(list(range(10)))
      ds1 = ds.map(inc)
      ds2 = ds1.map(inc)
      ds3 = ds1.map(inc)
      # Test content.
      assert extract_values("item", ds2.materialize().take()) == list(
          range(2, 12)
      )
      assert extract_values("item", ds3.materialize().take()) == list(
          range(2, 12)
      )
      # Test that first map is executed twice.
      assert ray.get(map_counter.get.remote()) == 2 * 10 + 10 + 10
  
      ray.get(map_counter.reset.remote())
      # The source data shouldn't be cleared since it's non-lazy.
      ds = ray.data.from_items(list(range(10)))
      # Add extra transformation after being lazy.
      ds = ds.map(inc)
      ds1 = ds.map(inc)
      ds2 = ds.map(inc)
      # Test content.
      assert extract_values("item", ds1.materialize().take()) == list(
          range(2, 12)
      )
      assert extract_values("item", ds2.materialize().take()) == list(
          range(2, 12)
      )
      # Test that first map is executed twice, because ds1.materialize()
      # clears up the previous snapshot blocks, and ds2.materialize()
      # has to re-execute ds.map(inc) again.
      assert ray.get(map_counter.get.remote()) == 2 * 10 + 10 + 10
  
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 6 for calling a global function with example usage: test_lazy_fanout(shutdown_only, local_path) and returns: row, dict