code: def test_context_saved_when_dataset_created(
      ray_start_regular_shared,
      restore_data_context,
  ):
      ctx = DataContext.get_current()
      ctx.set_config("foo", 1)
      d1 = ray.data.range(10)
      d2 = ray.data.range(10)
      assert ctx.get_config("foo") == 1
      assert d1.context.get_config("foo") == 1
      assert d2.context.get_config("foo") == 1
  
      # Changing `d1.context` should not affect `d2.context` or the global context.
      d1.context.set_config("foo", 2)
      assert d1.context.get_config("foo") == 2
      assert d2.context.get_config("foo") == 1
      assert ctx.get_config("foo") == 1
  
      # Changed value can be propagated to remote tasks.
      @ray.remote(num_cpus=0)
      def check(d1, d2):
          assert d1.context.get_config("foo") == 2
          assert d2.context.get_config("foo") == 1
  
      ray.get(check.remote(d1, d2))
  
      # Changing the global context should not affect `d1.context` or `d2.context`.
      ctx.set_config("foo", 3)
      assert ctx.get_config("foo") == 3
      assert d1.context.get_config("foo") == 2
      assert d2.context.get_config("foo") == 1
  
      # The global context has changed.
      # Applying new operators to the existing datasets should use the new
      # global context.
      d2 = d2.map_batches(lambda batch: batch)
      assert d2.context.get_config("foo") == 1
  
file_path: ray/python/ray/data/tests/test_context_propagation.py
imports: import pandas
  import pytest
  import ray
  from ray._private.test_utils import run_string_as_driver
  from ray.data.block import BlockMetadata
  from ray.data.context import DataContext
  from ray.data.datasource import Datasource, ReadTask
  from ray.tests.conftest import *
label_desc: Use image node_id 1 for calling a global function with example usage: test_context_saved_when_dataset_created(ray_start_regular_shared, restore_data_context) without return types