class_docstr: Fuses linear chains of compatible physical operators.
class_docstr_tok: ["Fuses","linear","chains","of","compatible","physical","operators","."]
code: def apply(self, plan: PhysicalPlan) -> PhysicalPlan:
      self._op_map = plan.op_map.copy()
      # Do DFS fusion on compatible pairwise operators in two passes.
      # In the first pass, only fuse back-to-back map operators together.
      fused_dag = self._fuse_map_operators_in_dag(plan.dag)
  
      # Now that we have fused together all back-to-back map operators,
      # we fuse together MapOperator -> AllToAllOperator pairs.
      fused_dag = self._fuse_all_to_all_operators_in_dag(fused_dag)
  
      # Update output dependencies after fusion.
      # TODO(hchen): Instead of updating the depdencies manually,
      # we need a better abstraction for manipulating the DAG.
      self._remove_output_depes(fused_dag)
      self._update_output_depes(fused_dag)
  
      return PhysicalPlan(fused_dag, self._op_map)
  
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 1 for calling the OperatorFusionRule obj's underlying member method code with example usage: obj.apply(plan) and returns: PhysicalPlan