class_docstr: A MapOperator implementation that executes tasks on a task pool.
code: def _add_bundled_input(self, bundle: RefBundle):
      # Submit the task as a normal Ray task.
      map_task = cached_remote_fn(_map_task, num_returns="streaming")
      input_blocks = [block for block, _ in bundle.blocks]
      ctx = TaskContext(
          task_idx=self._next_data_task_idx,
          target_max_block_size=self.actual_target_max_block_size,
      )
      data_context = DataContext.get_current()
      ray_remote_args = self._get_runtime_ray_remote_args(
          input_bundle=bundle
      )
      ray_remote_args["name"] = self.name
      ray_remote_args.update(
          data_context._task_pool_data_task_remote_args
      )
  
      gen = map_task.options(**ray_remote_args).remote(
          self._map_transformer_ref,
          data_context,
          ctx,
          *input_blocks,
      )
      self._submit_data_task(gen, bundle)
  
file_path: ray/python/ray/data/_internal/execution/operators/task_pool_map_operator.py
imports: from typing import Any, Dict, Optional
  import ray
  from ray.data._internal.execution.interfaces import ExecutionResources, PhysicalOperator, RefBundle, TaskContext
  from ray.data._internal.execution.operators.map_operator import MapOperator, _map_task
  from ray.data._internal.execution.operators.map_transformer import MapTransformer
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data.context import DataContext
label_desc: Use image node_id 2 for calling the TaskPoolMapOperator obj's underlying member method code with example usage: obj._add_bundled_input(bundle) without return types