class_docstr: Fuses linear chains of compatible physical operators.
class_docstr_tok: ["Fuses","linear","chains","of","compatible","physical","operators","."]
docstr: Returns whether the provided downstream operator can be fused with the given
  upstream operator.
  
  We currently support fusing two operators if the following are all true:
      * We are fusing either MapOperator -> MapOperator or
        MapOperator -> AllToAllOperator.
      * They either use the same compute configuration, or the upstream operator
        uses a task pool while the downstream operator uses an actor pool.
      * If both operators involve callable classes, the callable classes are
        the same class AND constructor args are the same for both.
      * They have compatible remote arguments.
docstr_tok: ["Returns","whether","the","provided","downstream","operator","can","be","fused","with","the","given","upstream","operator",".","We","currently","support","fusing","two","operators","if","the","following","are","all","true",":","*","We","are","fusing","either","MapOperator","-",">","MapOperator","or","MapOperator","-",">","AllToAllOperator",".","*","They","either","use","the","same","compute","configuration",",","or","the","upstream","operator","uses","a","task","pool","while","the","downstream","operator","uses","an","actor","pool",".","*","If","both","operators","involve","callable","classes",",","the","callable","classes","are","the","same","class","AND","constructor","args","are","the","same","for","both",".","*","They","have","compatible","remote","arguments","."]
code: def _can_fuse(
      self, down_op: PhysicalOperator, up_op: PhysicalOperator
  ) -> bool:
      """Returns whether the provided downstream operator can be fused with the given
      upstream operator.
  
      We currently support fusing two operators if the following are all true:
          * We are fusing either MapOperator -> MapOperator or
            MapOperator -> AllToAllOperator.
          * They either use the same compute configuration, or the upstream operator
            uses a task pool while the downstream operator uses an actor pool.
          * If both operators involve callable classes, the callable classes are
            the same class AND constructor args are the same for both.
          * They have compatible remote arguments.
      """
      from ray.data._internal.logical.operators.map_operator import (
          AbstractMap,
          AbstractUDFMap,
      )
  
      # We currently only support fusing for the following cases:
      # - TaskPoolMapOperator -> TaskPoolMapOperator/ActorPoolMapOperator
      # - TaskPoolMapOperator -> AllToAllOperator
      # (only RandomShuffle and Repartition LogicalOperators are currently supported)
      if not (
          (
              isinstance(up_op, TaskPoolMapOperator)
              and isinstance(
                  down_op, (TaskPoolMapOperator, ActorPoolMapOperator)
              )
          )
          or (
              isinstance(up_op, TaskPoolMapOperator)
              and isinstance(down_op, AllToAllOperator)
          )
      ):
          return False
  
      down_logical_op = self._op_map[down_op]
      up_logical_op = self._op_map[up_op]
  
      if up_op.get_additional_split_factor() > 1:
          return False
  
      # If the downstream operator takes no input, it cannot be fused with
      # the upstream operator.
      if not down_logical_op._input_dependencies:
          return False
  
      # We currently only support fusing for the following cases:
      # - AbstractMap -> AbstractMap
      # - AbstractMap -> RandomShuffle
      # - AbstractMap -> Repartition (shuffle=True)
      if not (
          (
              isinstance(up_logical_op, AbstractMap)
              and isinstance(down_logical_op, AbstractMap)
          )
          or (
              isinstance(up_logical_op, AbstractMap)
              and isinstance(down_logical_op, RandomShuffle)
          )
          or (
              isinstance(up_logical_op, AbstractMap)
              and isinstance(down_logical_op, Repartition)
          )
      ):
          return False
  
      # Do not fuse Repartition operator if shuffle is disabled
      # (i.e. using split shuffle).
      if (
          isinstance(down_logical_op, Repartition)
          and not down_logical_op._shuffle
      ):
          return False
  
      if isinstance(down_logical_op, AbstractUDFMap) and isinstance(
          up_logical_op, AbstractUDFMap
      ):
          # Allow fusing tasks->actors if the resources are compatible (read->map),
          # but not the other way around. The latter (downstream op) will be used as
          # the compute if fused.
          if is_task_compute(down_logical_op._compute) and get_compute(
              up_logical_op._compute
          ) != get_compute(down_logical_op._compute):
              return False
  
      # Only fuse if the ops' remote arguments are compatible.
      if not _are_remote_args_compatible(
          getattr(up_logical_op, "_ray_remote_args", {}),
          getattr(down_logical_op, "_ray_remote_args", {}),
      ):
          return False
  
      if not self._can_merge_target_max_block_size(
          up_op.target_max_block_size, down_op.target_max_block_size
      ):
          return False
  
      # Otherwise, ops are compatible for fusion.
      return True
  
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 6 for calling the OperatorFusionRule obj's underlying member method code with example usage: obj._can_fuse(down_op, up_op) and returns: True, False, False, False, False, False, False, False, False