code: def test_all_to_all_operator():
      def dummy_all_transform(bundles: List[RefBundle], ctx):
          assert len(ctx.sub_progress_bar_dict) == 2
          assert list(ctx.sub_progress_bar_dict.keys()) == [
              "Test1",
              "Test2",
          ]
          return make_ref_bundles([[1, 2], [3, 4]]), {"FooStats": []}
  
      input_op = InputDataBuffer(
          make_ref_bundles([[i] for i in range(100)])
      )
      op = AllToAllOperator(
          dummy_all_transform,
          input_op=input_op,
          target_max_block_size=DataContext.get_current().target_max_block_size,
          num_outputs=2,
          sub_progress_bar_names=["Test1", "Test2"],
          name="TestAll",
      )
  
      # Initialize progress bar.
      num_bars = op.initialize_sub_progress_bars(0)
      assert num_bars == 2, num_bars
  
      # Feed data.
      op.start(ExecutionOptions())
      while input_op.has_next():
          op.add_input(input_op.get_next(), 0)
      op.all_inputs_done()
  
      # Check we return transformed bundles.
      assert not op.completed()
      assert _take_outputs(op) == [[1, 2], [3, 4]]
      stats = op.get_stats()
      assert "FooStats" in stats
      assert op.completed()
      op.close_sub_progress_bars()
  
imports: import collections
  import random
  import time
  from typing import Any, Iterable, List
  from unittest.mock import MagicMock
  import numpy
  import pandas
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.compute import ActorPoolStrategy, TaskPoolStrategy
  from ray.data._internal.execution.interfaces import ExecutionOptions, PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.actor_pool_map_operator import ActorPoolMapOperator
  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.limit_operator import LimitOperator
  from ray.data._internal.execution.operators.map_operator import MapOperator, _BlockRefBundler
  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.operators.task_pool_map_operator import TaskPoolMapOperator
  from ray.data._internal.execution.operators.union_operator import UnionOperator
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.tests.util import run_one_op_task, run_op_tasks_sync
  from ray.tests.conftest import *
label_desc: Use image node_id 5 for calling a global function with example usage: test_all_to_all_operator() and returns: make_ref_bundles,  dict