code: def test_pipelined_execution(ray_start_10_cpus_shared):
      executor = StreamingExecutor(
          ExecutionOptions(preserve_order=True)
      )
      inputs = make_ref_bundles([[x] for x in range(20)])
      o1 = InputDataBuffer(inputs)
      o2 = MapOperator.create(
          make_map_transformer(lambda block: [b * -1 for b in block]),
          o1,
      )
      o3 = MapOperator.create(
          make_map_transformer(lambda block: [b * 2 for b in block]), o2
      )
  
      def reverse_sort(inputs: List[RefBundle], ctx):
          reversed_list = inputs[::-1]
          return reversed_list, {}
  
      ctx = DataContext.get_current()
      o4 = AllToAllOperator(reverse_sort, o3, ctx.target_max_block_size)
      it = executor.execute(o4)
      output = ref_bundles_to_list(it)
      expected = [[x * -2] for x in range(20)][::-1]
      assert output == expected, (output, expected)
  
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 4 for calling a global function with example usage: test_pipelined_execution(ray_start_10_cpus_shared) and returns: reversed_list,  dict