code: def test_map_operator_output_unbundling(
      ray_start_regular_shared, use_actors, preserve_order
  ):
      # Tests that the MapOperator's output queue unbundles the bundles returned from
      # tasks; this facilitates features such as dynamic block splitting.
      def noop(block_iter: Iterable[Block], ctx) -> Iterable[Block]:
          for block in block_iter:
              yield block
  
      # Create with inputs.
      input_op = InputDataBuffer(
          make_ref_bundles([[i] for i in range(10)])
      )
      compute_strategy = (
          ActorPoolStrategy() if use_actors else TaskPoolStrategy()
      )
      op = MapOperator.create(
          create_map_transformer_from_block_fn(noop),
          input_op=input_op,
          name="TestMapper",
          compute_strategy=compute_strategy,
          # Send the everything in a single bundle of 10 blocks.
          min_rows_per_bundle=10,
      )
  
      # Feed data and block on exec.
      op.start(ExecutionOptions(preserve_order=preserve_order))
      inputs = []
      while input_op.has_next():
          inputs.append(input_op.get_next())
      # Sanity check: the op will get 10 input bundles.
      assert len(inputs) == 10
      for input_ in inputs:
          op.add_input(input_, 0)
      op.all_inputs_done()
      run_op_tasks_sync(op)
  
      # Check that bundles are unbundled in the output queue.
      outputs = []
      while op.has_next():
          outputs.append(op.get_next())
      assert len(outputs) == 10
      assert op.completed()
  
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 14 for calling a global function with example usage: test_map_operator_output_unbundling(ray_start_regular_shared, use_actors, preserve_order) without return types