class_docstr: Fuses linear chains of compatible physical operators.
class_docstr_tok: ["Fuses","linear","chains","of","compatible","physical","operators","."]
docstr: Starting at the given operator, traverses up the DAG of operators
  and recursively fuses compatible MapOperator -> MapOperator pairs.
  Returns the current (root) operator after completing upstream operator fusions.
docstr_tok: ["Starting","at","the","given","operator",",","traverses","up","the","DAG","of","operators","and","recursively","fuses","compatible","MapOperator","-",">","MapOperator","pairs",".","Returns","the","current","(","root",")","operator","after","completing","upstream","operator","fusions","."]
code: def _fuse_map_operators_in_dag(
      self, dag: PhysicalOperator
  ) -> MapOperator:
      """Starting at the given operator, traverses up the DAG of operators
      and recursively fuses compatible MapOperator -> MapOperator pairs.
      Returns the current (root) operator after completing upstream operator fusions.
      """
      upstream_ops = dag.input_dependencies
      while (
          len(upstream_ops) == 1
          and isinstance(dag, MapOperator)
          and isinstance(upstream_ops[0], MapOperator)
          and self._can_fuse(dag, upstream_ops[0])
      ):
          # Fuse operator with its upstream op.
          dag = self._get_fused_map_operator(dag, upstream_ops[0])
          upstream_ops = dag.input_dependencies
  
      # Done fusing back-to-back map operators together here,
      # move up the DAG to find the next map operators to fuse.
      dag._input_dependencies = [
          self._fuse_map_operators_in_dag(upstream_op)
          for upstream_op in upstream_ops
      ]
      return dag
  
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 4 for calling the OperatorFusionRule obj's underlying member method code with example usage: obj._fuse_map_operators_in_dag(dag) and returns: dag