code: def test_map_operator_bulk(ray_start_regular_shared, use_actors):
      # Create with inputs.
      input_op = InputDataBuffer(
          make_ref_bundles([[np.ones(1024) * i] for i in range(100)])
      )
      compute_strategy = (
          ActorPoolStrategy(size=1)
          if use_actors
          else TaskPoolStrategy()
      )
      op = MapOperator.create(
          _mul2_map_data_prcessor,
          input_op=input_op,
          name="TestMapper",
          compute_strategy=compute_strategy,
      )
  
      # Feed data and block on exec.
      op.start(ExecutionOptions(actor_locality_enabled=False))
      if use_actors:
          # Actor will be pending after starting the operator.
          assert (
              op.progress_str() == "0 actors (1 pending) [locality off]"
          )
      assert op.internal_queue_size() == 0
      i = 0
      while input_op.has_next():
          op.add_input(input_op.get_next(), 0)
          i += 1
          if use_actors:
              assert op.internal_queue_size() == i
          else:
              assert op.internal_queue_size() == 0
      op.all_inputs_done()
  
      tasks = op.get_active_tasks()
      while tasks:
          run_op_tasks_sync(op, only_existing=True)
          tasks = op.get_active_tasks()
          if use_actors and tasks:
              # After actor is ready (first work ref resolved), actor will remain ready
              # while there is work to do.
              assert op.progress_str() == "1 actors [locality off]"
  
      assert op.internal_queue_size() == 0
      if use_actors:
          # After all work is done, actor will have been killed to free up resources..
          assert op.progress_str() == "0 actors [locality off]"
      else:
          assert op.progress_str() == ""
  
      # Check we return transformed bundles in order.
      assert not op.completed()
      assert np.array_equal(
          _take_outputs(op),
          [[np.ones(1024) * i * 2] for i in range(100)],
      )
      assert op.completed()
  
      # Check dataset stats.
      stats = op.get_stats()
      assert "TestMapper" in stats, stats
      assert len(stats["TestMapper"]) == 100, stats
  
      # Check memory stats.
      metrics = op.metrics.as_dict()
      assert metrics["obj_store_mem_alloc"] == pytest.approx(
          832200, 0.5
      ), metrics
      assert metrics["obj_store_mem_peak"] == pytest.approx(
          1688000, 0.5
      ), metrics
      assert metrics["obj_store_mem_freed"] == pytest.approx(
          832200, 0.5
      ), metrics
  
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 7 for calling a global function with example usage: test_map_operator_bulk(ray_start_regular_shared, use_actors) without return types