class_docstr: Rule for pushing down the limit operator.
  
  When a limit operator is present, we apply the limit on the
  most upstream operator that supports it. Notably, we move the
  Limit operator downstream from Read op, any other non-OneToOne operator,
  or any operator which could potentially change the number of output rows.
  
  In addition, we also fuse consecutive Limit operators into a single
  Limit operator, i.e. `Limit[n] -> Limit[m]` becomes `Limit[min(n, m)]`.
class_docstr_tok: ["Rule","for","pushing","down","the","limit","operator",".","When","a","limit","operator","is","present",",","we","apply","the","limit","on","the","most","upstream","operator","that","supports","it",".","Notably",",","we","move","the","Limit","operator","downstream","from","Read","op",",","any","other","non-OneToOne","operator",",","or","any","operator","which","could","potentially","change","the","number","of","output","rows",".","In","addition",",","we","also","fuse","consecutive","Limit","operators","into","a","single","Limit","operator",",","i.e",".","`","Limit","[","n","]","-",">","Limit","[","m","]","`","becomes","`","Limit","[","min","(","n",",","m",")","]","`","."]
docstr: Given a DAG of LogicalOperators, traverse the DAG and fuse all
  back-to-back Limit operators, i.e.
  Limit[n] -> Limit[m] becomes Limit[min(n, m)].
  
  Returns a new LogicalOperator with the Limit operators fusion applied.
docstr_tok: ["Given","a","DAG","of","LogicalOperators",",","traverse","the","DAG","and","fuse","all","back-to-back","Limit","operators",",","i.e",".","Limit","[","n","]","-",">","Limit","[","m","]","becomes","Limit","[","min","(","n",",","m",")","]",".","Returns","a","new","LogicalOperator","with","the","Limit","operators","fusion","applied","."]
code: def _apply_limit_fusion(self, op: LogicalOperator) -> LogicalOperator:
      """Given a DAG of LogicalOperators, traverse the DAG and fuse all
      back-to-back Limit operators, i.e.
      Limit[n] -> Limit[m] becomes Limit[min(n, m)].
  
      Returns a new LogicalOperator with the Limit operators fusion applied.
      """
  
      # Post-order traversal.
      nodes: Iterable[LogicalOperator] = deque()
      for node in op.post_order_iter():
          nodes.appendleft(node)
  
      while len(nodes) > 0:
          current_op = nodes.pop()
  
          # If we encounter two back-to-back Limit operators, fuse them.
          if isinstance(current_op, Limit):
              upstream_op = current_op.input_dependency
              if isinstance(upstream_op, Limit):
                  new_limit = min(current_op._limit, upstream_op._limit)
                  fused_limit_op = Limit(
                      upstream_op.input_dependency, new_limit
                  )
  
                  # Link the fused Limit operator to its input and output ops, i.e.:
                  # `upstream_input -> limit_upstream -> limit_downstream -> downstream_output`  # noqa: E501
                  # becomes `upstream_input -> fused_limit -> downstream_output`
                  fused_limit_op._input_dependencies = (
                      upstream_op.input_dependencies
                  )
                  fused_limit_op._output_dependencies = (
                      current_op.output_dependencies
                  )
  
                  # Replace occurrences of the upstream Limit operator in
                  # output_dependencies with the newly fused Limit operator.
                  upstream_input = upstream_op.input_dependency
                  upstream_input._output_dependencies = [fused_limit_op]
  
                  for current_output in current_op.output_dependencies:
                      current_output._input_dependencies = [
                          fused_limit_op
                      ]
                  nodes.append(fused_limit_op)
      return current_op
  
file_path: ray/python/ray/data/_internal/logical/rules/limit_pushdown.py
imports: import copy
  from collections import deque
  from typing import Iterable, List
  from ray.data._internal.logical.interfaces import LogicalOperator, LogicalPlan, Rule
  from ray.data._internal.logical.operators.one_to_one_operator import AbstractOneToOne, Limit
  from ray.data._internal.logical.operators.read_operator import Read
label_desc: Use image node_id 3 for calling the LimitPushdownRule obj's underlying member method code with example usage: obj._apply_limit_fusion(op) and returns: current_op