code: def test_configure_spread_e2e(
      ray_start_10_cpus_shared, restore_data_context
  ):
      from ray import remote_function
  
      tasks = []
  
      def _test_hook(fn, args, strategy):
          if "map_task" in str(fn):
              tasks.append(strategy)
  
      remote_function._task_launch_hook = _test_hook
      DataContext.get_current().use_streaming_executor = True
      DataContext.get_current().execution_options.preserve_order = True
      DataContext.get_current().large_args_threshold = 0
  
      # Simple 2-stage pipeline.
      ray.data.range(2, parallelism=2).map(
          lambda x: x, num_cpus=2
      ).take_all()
  
      # Read tasks get SPREAD by default, subsequent ones use default policy.
      tasks = sorted(tasks)
      assert tasks == ["DEFAULT", "DEFAULT", "SPREAD", "SPREAD"]
  
imports: import itertools
  import random
  import threading
  import time
  from typing import Any, List
  import pandas
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
label_desc: Use image node_id 10 for calling a global function with example usage: test_configure_spread_e2e(ray_start_10_cpus_shared, restore_data_context) without return types