class_docstr: Fuses linear chains of compatible physical operators.
class_docstr_tok: ["Fuses","linear","chains","of","compatible","physical","operators","."]
code: def _get_fused_map_operator(
      self, down_op: MapOperator, up_op: MapOperator
  ) -> MapOperator:
      assert self._can_fuse(down_op, up_op), (
          "Current rule supports fusing MapOperator->MapOperator, but received: "
          f"{type(up_op).__name__} -> {type(down_op).__name__}"
      )
  
      # Fuse operator names.
      name = up_op.name + "->" + down_op.name
  
      down_logical_op = self._op_map.pop(down_op)
      up_logical_op = self._op_map.pop(up_op)
  
      # Merge minimum block sizes.
      down_min_rows_per_block = (
          down_logical_op._min_rows_per_block
          if isinstance(down_logical_op, AbstractUDFMap)
          else None
      )
      up_min_rows_per_block = (
          up_logical_op._min_rows_per_block
          if isinstance(up_logical_op, AbstractUDFMap)
          else None
      )
      if (
          down_min_rows_per_block is not None
          and up_min_rows_per_block is not None
      ):
          min_rows_per_block = max(
              down_min_rows_per_block, up_min_rows_per_block
          )
      elif up_min_rows_per_block is not None:
          min_rows_per_block = up_min_rows_per_block
      else:
          min_rows_per_block = down_min_rows_per_block
  
      target_max_block_size = self._get_merged_target_max_block_size(
          up_op.target_max_block_size, down_op.target_max_block_size
      )
  
      # We take the downstream op's compute in case we're fusing upstream tasks with a
      # downstream actor pool (e.g. read->map).
      compute = None
      if isinstance(down_logical_op, AbstractUDFMap):
          compute = get_compute(down_logical_op._compute)
      ray_remote_args = up_logical_op._ray_remote_args
      # Make the upstream operator's inputs the new, fused operator's inputs.
      input_deps = up_op.input_dependencies
      assert len(input_deps) == 1
      input_op = input_deps[0]
  
      # Fused physical map operator.
      op = MapOperator.create(
          up_op.get_map_transformer().fuse(
              down_op.get_map_transformer()
          ),
          input_op,
          target_max_block_size=target_max_block_size,
          name=name,
          compute_strategy=compute,
          min_rows_per_bundle=min_rows_per_block,
          ray_remote_args=ray_remote_args,
      )
  
      # Build a map logical operator to be used as a reference for further fusion.
      # TODO(Scott): This is hacky, remove this once we push fusion to be purely based
      # on a lower-level operator spec.
      if isinstance(up_logical_op, AbstractUDFMap):
          input_op = up_logical_op.input_dependency
      else:
          # Bottom out at the source logical op (e.g. Read()).
          input_op = up_logical_op
      if isinstance(down_logical_op, AbstractUDFMap):
          logical_op = AbstractUDFMap(
              name,
              input_op,
              down_logical_op._fn,
              down_logical_op._fn_args,
              down_logical_op._fn_kwargs,
              down_logical_op._fn_constructor_args,
              down_logical_op._fn_constructor_kwargs,
              min_rows_per_block,
              compute,
              ray_remote_args,
          )
      else:
          from ray.data._internal.logical.operators.map_operator import (
              AbstractMap,
          )
  
          # The downstream op is AbstractMap instead of AbstractUDFMap.
          logical_op = AbstractMap(
              name,
              input_op,
              ray_remote_args=ray_remote_args,
          )
      self._op_map[op] = logical_op
      # Return the fused physical operator.
      return op
  
imports: from typing import List, Optional, Tuple
  from ray.data._internal.compute import get_compute, is_task_compute
  from ray.data._internal.execution.interfaces import PhysicalOperator, RefBundle, TaskContext
  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.map_operator import MapOperator
  from ray.data._internal.execution.operators.task_pool_map_operator import TaskPoolMapOperator
  from ray.data._internal.logical.interfaces import PhysicalPlan, Rule
  from ray.data._internal.logical.operators.all_to_all_operator import AbstractAllToAll, RandomShuffle, Repartition
  from ray.data._internal.logical.operators.map_operator import AbstractUDFMap
  from ray.data._internal.stats import StatsDict
  from ray.data.context import DataContext
label_desc: Use image node_id 9 for calling the OperatorFusionRule obj's underlying member method code with example usage: obj._get_fused_map_operator(down_op, up_op) and returns: op