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 fuse(self, prev: Stage):
      if not self.can_fuse(prev):
          raise ValueError(
              f"Tried to fuse {prev} with {self}, but these are not fusable."
          )
      name = prev.name + "->" + self.name
      prev_fn = prev.fn
      if isinstance(self.fn, CallableClass) and isinstance(
          prev_fn, CallableClass
      ):
          assert self.fn == prev_fn
          assert (
              prev.fn_constructor_args == self.fn_constructor_args
              and prev.fn_constructor_kwargs
              == self.fn_constructor_kwargs
          )
          # If both UDFs are callable classes, they must be equal and have the same
          # construction args, so we tell the previous stage to reuse the passed
          # (instantiated) callable class UDF that's provided to the block function.
          use_outer_fn = True
          prev_fn = None
      else:
          # Otherwise, we're either fusing two non-callable class UDFs, or a
          # non-callable class UDF with a callable class UDF. In either case, prev
          # will be a non-callable class UDF, so we use it within the block function.
          use_outer_fn = False
  
      # Package args into a flat positional args list.
      fn_args, unpack_args = _pack_args(
          self.fn_args,
          self.fn_kwargs,
          prev.fn_args,
          prev.fn_kwargs,
      )
  
      block_fn1 = prev.block_fn
      block_fn2 = self.block_fn
      if (
          prev.min_rows_per_block is not None
          and self.min_rows_per_block is not None
      ):
          min_rows_per_block = max(
              prev.min_rows_per_block, self.min_rows_per_block
          )
      elif prev.min_rows_per_block is not None:
          min_rows_per_block = prev.min_rows_per_block
      else:
          min_rows_per_block = self.min_rows_per_block
  
      def block_fn(
          blocks: Iterable[Block],
          ctx: TaskContext,
          fn: UserDefinedFunction,
          *fn_args,
          **fn_kwargs,
      ) -> Iterable[Block]:
          assert not fn_kwargs, fn_kwargs
          # Unpack flat position args list into
          (
              self_fn_args,
              self_fn_kwargs,
              prev_fn_args,
              prev_fn_kwargs,
          ) = unpack_args(fn_args)
          self_fn_args = (
              self_fn_args if fn is None else (fn,) + self_fn_args
          )
          if use_outer_fn:
              prev_fn_ = fn
          else:
              prev_fn_ = prev_fn
          prev_fn_args = (
              prev_fn_args
              if prev_fn_ is None
              else (prev_fn_,) + prev_fn_args
          )
          blocks = block_fn1(
              blocks, ctx, *prev_fn_args, **prev_fn_kwargs
          )
          return block_fn2(blocks, ctx, *self_fn_args, **self_fn_kwargs)
  
      return OneToOneStage(
          name,
          block_fn,
          self.compute,
          prev.ray_remote_args,
          min_rows_per_block=min_rows_per_block,
          fn=self.fn,
          fn_args=fn_args,
          fn_kwargs={},
          fn_constructor_args=self.fn_constructor_args,
          fn_constructor_kwargs=self.fn_constructor_kwargs,
      )
  
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 3 for calling the OneToOneStage obj's underlying member method code with example usage: obj.fuse(prev) and returns: OneToOneStage, block_fn2