fix: preserve global limit for multi-partition fetch

[discord9]

Which issue does this PR close?

• Closes LimitPushdown can mistake per-partition fetch for a global limit #23799.

Rationale for this change

An operator-level fetch does not necessarily enforce a query-wide limit. On a
plan with multiple output partitions, each partition can apply the fetch
independently:

GlobalLimitExec: skip=0, fetch=5
  TestScan: partitions=2

Replacing the global limit with only a scan fetch is incorrect:

TestScan: partitions=2, fetch=5

The scan can return five rows from each partition, or ten rows in total. Its
fetch is still useful for stopping each partition early, but the plan also
needs one limit across their combined output:

CoalescePartitionsExec: fetch=5
  TestScan: partitions=2, fetch=5

SortPreservingMergeExec is used instead when input ordering must be
preserved.

LimitPushdown removes limit nodes while descending the plan, so a numeric
fetch alone is not enough to describe what remains to be enforced. The same
number can mean:

• a limit on each output partition;
• a limit on the combined output of all partitions;
• an existing operator fetch; or
• an early-stop value copied to descendants after the limit is already
  enforced.

The rule therefore tracks the outstanding scope separately:

enum LimitScope {
    Local,
    Global,
}

pending: Option<LimitScope>

The states mean:

pending	Meaning
None	The required limit is already enforced. A retained fetch is only for early stopping.
Some(LimitScope::Local)	Each output partition still needs its own limit.
Some(LimitScope::Global)	The combined output of all partitions still needs one limit.

A pending global limit is enforced before descending into a multi-partition
plan. The rule merges the partitions with CoalescePartitionsExec or
SortPreservingMergeExec, applies the global limit there, and may continue to
copy the fetch to children only as an early-stop value.

This also makes cloning the state for child plans safe:

• a local limit is intentionally independent for each output partition;
• a global limit is enforced before entering a multi-partition fan-out such as
  UnionExec; and
• after a limit is enforced, pending is None and skip is cleared before
  visiting children, so descendants cannot apply the same OFFSET again.

An existing operator fetch can satisfy the current limit only when there is no
OFFSET and the existing fetch is at least as strict as the requested one. For
example, fetch=10 does not enforce LIMIT 5:

GlobalLimitExec: skip=0, fetch=5
  TestFetchOnlyExec: fetch=10

If the operator supports with_fetch, the rule lowers its fetch to 5. If it
cannot be changed, the explicit limit remains or is enforced between that
operator and its child. An existing fetch never satisfies an OFFSET by itself.

The rule also handles single-output operators that hide multi-partition
children. A local limit on one output partition is equivalent to a global limit
at that point, so the scope is promoted before descending. If recursion later
reaches a multi-partition child, the global limit is enforced before continuing.

What changes are included in this PR?

• Track an outstanding limit as Option<LimitScope>, with separate Local and
  Global scopes.
• Preserve operator fetches as early-stop values without treating a
  per-partition fetch as a global limit.
• Enforce pending global limits by merging multiple output partitions first,
  preserving order when required.
• Let an existing fetch satisfy a limit only when skip == 0 and
  existing_fetch <= required_fetch.
• Lower a wider existing fetch through with_fetch where supported; otherwise
  preserve or insert the explicit limit.
• Keep OFFSET handling separate and clear an already-enforced OFFSET before
  descending into child plans.
• Remove no-op GlobalLimitExec nodes without changing the limit state being
  carried through the plan.
• Implement the required ExecutionPlan::apply_expressions method on the two
  test execution plans used by this regression suite.

Are these changes tested?

Yes. The physical_optimizer::limit_pushdown integration tests cover 37 plan
shapes, including:

• fetch-capable and unfetchable multi-partition inputs;
• LIMIT, OFFSET + LIMIT, and OFFSET-only plans;
• global limits over Union and multi-partition local limits;
• single-output extension operators that hide multi-partition children;
• nested limits and no-op global limit nodes;
• an existing fetch that is stricter than the requested limit;
• a wider existing fetch that cannot be lowered;
• a wider fetch on an operator that allows pushdown but cannot change its own
  fetch;
• lowering a SortExec fetch through with_fetch; and
• preserving an explicit limit when OFFSET is present.

Verified on the rebased PR branch with:

cargo fmt --all --check
cargo check -p datafusion --test core_integration
cargo test -p datafusion --test core_integration physical_optimizer::limit_pushdown --no-fail-fast
cargo clippy -p datafusion --test core_integration --no-deps -- -D warnings

The focused integration suite passed with 37 tests and no failures.

Are there any user-facing changes?

Yes. Queries using these physical plan shapes now preserve the requested global
LIMIT/OFFSET instead of returning too many rows or applying a limit or offset
more than once. There are no intended public API changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.83051% with 12 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.17%. Comparing base (0a429a3) to head (ebff67e).
⚠️ Report is 14 commits behind head on main.

Files with missing lines	Patch %	Lines
...atafusion/physical-optimizer/src/limit_pushdown.rs	89.83%	10 Missing and 2 partials ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #23800   +/-   ##
=======================================
  Coverage   81.17%   81.17%           
=======================================
  Files        1109     1109           
  Lines      388038   388116   +78     
  Branches   388038   388116   +78     
=======================================
+ Hits       314992   315057   +65     
- Misses      54507    54513    +6     
- Partials    18539    18546    +7     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[neilconway]

Thanks for your work on this, @discord9! This PR is overall very good: thorough explanation of the problem, very good test coverage, and well-written implementation. I spent a while working through it and talking to Claude and the basic approach makes sense to me and clears up a clear semantic shortcoming of the current representation.

[neilconway]

I found this comment confusing (e.g., whose skip and fetch are we referring to? What is a "semantic cap", and is a "non-semantic cap" a thing?). Calling it PendingScope only makes sense if you know the field in GlobalRequirements is named pending, which isn't obvious.

What about:

/// The scope of a row limit: what the limited row count applies to.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum LimitScope {
    /// The limit caps each output partition independently, as enforced by
    /// [`LocalLimitExec`].
    Local,
    /// The limit caps the combined output of all partitions, as enforced by
    /// [`GlobalLimitExec`] over a single-partition input. A fetch on a
    /// multi-partition operator cannot satisfy this scope; the partitions
    /// must first be merged into one stream.
    Global,
}

[neilconway]

I found this comment confusing. What about

        // A GlobalLimitExec with no skip and no fetch enforces nothing, so
        // remove it and keep the carried state untouched. Falling through to
        // the general limit handling would instead re-derive `pending` from
        // this node, which could reopen an already-satisfied requirement.
        // The local-to-global promotion above is kept as well: it relied only
        // on this node having one output partition, and it remains correct
        // after the node is removed.

[neilconway]

Why if we owe fetch=5 but the fetch on the operator is > 5?

We rebuild the operator below via pushdown_plan.with_fetch(skip_and_fetch) below, but some operators might support fetch but not with_fetch (e.g., PartialSort).

[neilconway]

Is copying the global state into every child correct, for an operator with multiple children? e.g., if we have an operator with n children and a single output partition, wouldn't this result in copying the fetch limit into n different operators, so we'd produce too many rows?

[discord9]

Is copying the global state into every child correct, for an operator with multiple children? e.g., if we have an operator with n children and a single output partition, wouldn't this result in copying the fetch limit into n different operators, so we'd produce too many rows?

The final results is still comes from a single partition output CoalescePartitionsExec/SortPreservingMergeExec so the result will only be like what limit=N needs， as the cloned global state to every child should be like this anyway:

pending = None
skip = 0
fetch = Some(N) -- just a early stop hint for child

[neilconway]

This is true of the operators in the tree but I don't think it's true in general. Suppose there's a custom operator with n children, a single output partition, and supports_limit_pushdown() == true: at that node the materialization branch doesn't fire (one partition) and the blocker branch doesn't run (pushdown supported), so the helper returns with pending == Some(LimitScope::Global) still open, and pushdown_limits clones that open requirement into all of its children.