class_docstr: A stage that transforms blocks independently (e.g., map or filter).
class_docstr_tok: ["A","stage","that","transforms","blocks","independently","(","e.g.",",","map","or","filter",")","."]
code: def __call__(
      self,
      blocks: BlockList,
      clear_input_blocks: bool,
      run_by_consumer: bool,
  ) -> Tuple[BlockList, dict]:
      compute = get_compute(self.compute)
      assert (
          self.fn_constructor_args is None
          and self.fn_constructor_kwargs is None
      ) or isinstance(compute, ActorPoolStrategy)
  
      if blocks._owned_by_consumer:
          assert (
              run_by_consumer
          ), "Blocks owned by consumer can only be consumed by consumer"
  
      blocks = compute._apply(
          self.block_fn,
          self.ray_remote_args,
          blocks,
          clear_input_blocks,
          name=self.name,
          min_rows_per_block=self.min_rows_per_block,
          fn=self.fn,
          fn_args=self.fn_args,
          fn_kwargs=self.fn_kwargs,
          fn_constructor_args=self.fn_constructor_args,
          fn_constructor_kwargs=self.fn_constructor_kwargs,
      )
      assert isinstance(blocks, BlockList), blocks
      blocks._owned_by_consumer = run_by_consumer
      return blocks, {}
  
imports: import copy
  import itertools
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Union
  import ray
  from ray._private.internal_api import get_memory_info_reply, get_state_from_address
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.compute import ActorPoolStrategy, BlockTransform, CallableClass, ComputeStrategy, TaskPoolStrategy, UserDefinedFunction, get_compute, is_task_compute
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.lazy_block_list import LazyBlockList
  from ray.data._internal.logical.interfaces.logical_operator import LogicalOperator
  from ray.data._internal.logical.operators.from_operators import AbstractFrom
  from ray.data._internal.logical.operators.input_data_operator import InputData
  from ray.data._internal.logical.operators.read_operator import Read
  from ray.data._internal.logical.rules.operator_fusion import _are_remote_args_compatible
  from ray.data._internal.stats import DatasetStats, DatasetStatsSummary
  from ray.data._internal.util import create_dataset_tag, unify_block_metadata_schema
  from ray.data.block import Block, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.debug import log_once
label_desc: Use image node_id 4 for calling the OneToOneStage obj's underlying member method code with example usage: obj.__call__(blocks, clear_input_blocks, run_by_consumer) and returns: blocks,  dict