[jayzhan211]

Which issue does this PR close?

• No issue. This is a refactor; query results and plans are unchanged.

Rationale for this change

Four optimizer rules move a LIMIT/OFFSET (or a fetch) through the physical plan: LimitPushdown, LimitPushPastWindows, the sort pushdown inside EnsureRequirements, and PushdownSort. Each rule answers the same three questions with its own code:

• How do skip and fetch combine? There are six places that add skip + fetch. Two of them saturate; four do not.

• Can a limit go below this node? There are three different checks:

  • supports_limit_pushdown() alone.
  • The same plus CardinalityEffect::Equal.
  • The same plus Equal | GreaterEqual, with a hard-coded type list.

  They differ because a rule can copy a limit below a node as an early-stop hint while keeping it above, or move it below and drop it above. Moving needs more from the node than copying. Nothing in the code names that difference.

• What does LimitPushdown do at each node? It removes every limit node and carries (fetch, skip, satisfied) down. It decides later whether to re-add the limit, and clears skip before recursing. The decision is spread over about 150 lines of nested ifs.

Several recent fixes came from one of these places disagreeing with another: #24958 (skip ignored when a sort moved below a limit), #22525 (OFFSET applied twice), #21744 and #24809 (fetch lost when an operator was removed).

What changes are included in this PR?

Three commits, each of which builds and passes the tests on its own:

1. RowBound (crate-private, physical-optimizer/src/row_bound.rs). This is OFFSET skip LIMIT fetch as a value: combining two bounds, skip + fetch (saturating), the tighter of two fetches, and the one rule for choosing between a GlobalLimitExec and a LocalLimitExec. The sort pushdown in EnsureRequirements and LimitPushPastWindows use it instead of their own arithmetic.

2. LimitPushdown as one walk. The bound a node receives from its ancestors is a small enum:

   • Nothing.
   • Open(bound): no ancestor enforces it yet, so it must be placed at or above the first node it can't pass.
   • Enforced(fetch): an ancestor enforces it, so nodes below may stop early.

   Each node is one of Limit, Pass, Merge or Stop. The rule is a single match over the two (limit_pushdown.rs, enforce), which replaces the satisfied flag and the skip-clearing step. GlobalRequirements, pushdown_limit_helper and pushdown_limit_helper_with_stats are deprecated; they now run on the new walk.

3. ExecutionPlan::limit_behavior(). Operators declare what a limit on their output means for their input:

   • Stops.
   • Limit { skip, fetch }.
   • CopyToChildren (Union).
   • MoveToInput (Projection).
   • Merge { fetch } (CoalescePartitions, SortPreservingMerge).
   • Window(lookahead).

   The copy/move difference is documented once, on the enum. So is the rule that a merge with its own fetch already selects rows: a limit above it can't move below it. Before, that rule was only implied by the merge's cardinality_effect() returning LowerEqual when it has a fetch. The three rules match on it instead of combining supports_limit_pushdown() with cardinality_effect() and type checks. The window's lookahead (frame end plus LEAD offsets) is now computed by BoundedWindowAggExec rather than by the rule. supports_limit_pushdown is deprecated. The default limit_behavior() is derived from it, so operators outside DataFusion keep their current behavior.

The fetch handling in PushdownSort and the other "a fetched operator is a row-selection boundary" checks in EnsureRequirements are unchanged.

What is the testing strategy for this PR?

The goal is no change in results or plans, so the existing tests are the check. All of these pass with no change to expected output:

• every sqllogictest file
• core_integration physical_optimizer:: (648 tests: limit pushdown, sort pushdown, enforce sorting, window)
• the limit_fuzz, window_fuzz, sort_fuzz, sort_query_fuzz and topk_filter_pushdown fuzz tests
• datafusion-physical-optimizer and datafusion-physical-plan unit tests

New tests:

• row_bound.rs unit tests: combining bounds, saturation at usize::MAX, and the Global/Local choice.
• test_fetched_merge_keeps_stricter_topk_above in core/tests/physical_optimizer/ensure_requirements.rs: a TopK over a SortPreservingMergeExec that has its own fetch keeps its fetch above the merge. It passes on main, and fails if a fetched merge is treated as MoveToInput.

Are there any user-facing changes?

No change to query results or plans.

API (upgrade guide updated):

• New ExecutionPlan::limit_behavior(), LimitBehavior and WindowLookahead.
• ExecutionPlan::supports_limit_pushdown() is deprecated, and DataFusion no longer calls it. DataFusion's own operators keep returning the same values, so a wrapper that forwards it to its input keeps working.
• limit_pushdown::{GlobalRequirements, pushdown_limit_helper, pushdown_limit_helper_with_stats} are deprecated. They now run on the new walk. A caller that drives them one node at a time no longer applies an OFFSET twice when a limit node with only an OFFSET sits above another limit.
• One difference for custom operators that still implement supports_limit_pushdown(): a single-input operator that returns true and whose cardinality_effect() is Unknown or LowerEqual can now have a window's grown limit pushed below it. LimitPushdown already pushes plain limits through such operators.
• skip + fetch now saturates at usize::MAX in LimitPushdown. SQL can't reach the overflow there, because LIMIT and OFFSET are each at most i64::MAX. The unchecked sums in LimitPushPastWindows are left as they are; LimitPushPastWindows panics with "attempt to add with overflow" for a large ROWS ... FOLLOWING bound #26658 tracks them.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning 6a280f15786f59b7cca9fcef1a0ddb3c3e931a07
    Building datafusion v55.1.0 (current)
       Built [  46.084s] (current)
     Parsing datafusion v55.1.0 (current)
      Parsed [   0.027s] (current)
    Building datafusion v55.1.0 (baseline)
       Built [  15.146s] (baseline)
     Parsing datafusion v55.1.0 (baseline)
      Parsed [   0.027s] (baseline)
    Checking datafusion v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.468s] 229 checks: 229 pass, 31 skip
     Summary no semver update required
    Finished [  63.604s] datafusion
    Building datafusion-physical-optimizer v55.1.0 (current)
       Built [  15.515s] (current)
     Parsing datafusion-physical-optimizer v55.1.0 (current)
      Parsed [   0.016s] (current)
    Building datafusion-physical-optimizer v55.1.0 (baseline)
       Built [  11.965s] (baseline)
     Parsing datafusion-physical-optimizer v55.1.0 (baseline)
      Parsed [   0.015s] (baseline)
    Checking datafusion-physical-optimizer v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.094s] 229 checks: 227 pass, 2 fail, 0 warn, 31 skip

--- failure function_marked_deprecated: function #[deprecated] added ---

Description:
A function is now #[deprecated]. Downstream crates will get a compiler warning when using this function.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.51.0/src/lints/function_marked_deprecated.ron

Failed in:
  function datafusion_physical_optimizer::limit_pushdown::pushdown_limit_helper_with_stats in /home/runner/work/datafusion/datafusion/datafusion/physical-optimizer/src/limit_pushdown.rs:358

--- failure type_marked_deprecated: #[deprecated] added on type ---

Description:
A type is now #[deprecated]. Downstream crates will get a compiler warning when using this type.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.51.0/src/lints/type_marked_deprecated.ron

Failed in:
  Struct GlobalRequirements in /home/runner/work/datafusion/datafusion/datafusion/physical-optimizer/src/limit_pushdown.rs:291

     Summary semver requires new minor version: 0 major and 2 minor checks failed
    Finished [  28.576s] datafusion-physical-optimizer
    Building datafusion-physical-plan v55.1.0 (current)
       Built [  10.585s] (current)
     Parsing datafusion-physical-plan v55.1.0 (current)
      Parsed [   0.167s] (current)
    Building datafusion-physical-plan v55.1.0 (baseline)
       Built [  10.010s] (baseline)
     Parsing datafusion-physical-plan v55.1.0 (baseline)
      Parsed [   0.171s] (baseline)
    Checking datafusion-physical-plan v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.536s] 229 checks: 228 pass, 1 fail, 0 warn, 31 skip

--- failure trait_method_marked_deprecated: trait method #[deprecated] added ---

Description:
A trait method is now #[deprecated]. Downstream crates will get a compiler warning when using this method.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.51.0/src/lints/trait_method_marked_deprecated.ron

Failed in:
  method supports_limit_pushdown in trait datafusion_physical_plan::execution_plan::ExecutionPlan in /home/runner/work/datafusion/datafusion/datafusion/physical-plan/src/execution_plan.rs:102
  method supports_limit_pushdown in trait datafusion_physical_plan::ExecutionPlan in /home/runner/work/datafusion/datafusion/datafusion/physical-plan/src/execution_plan.rs:102

     Summary semver requires new minor version: 0 major and 1 minor checks failed
    Finished [  22.605s] datafusion-physical-plan