fix: drop the redundant Limit above a Sort in the same PushDownLimit visit

[adriangb]

Which issue does this PR close?

• Closes PushDownLimit keeps a redundant Limit above a Sort it gave a fetch, so every ORDER BY ... LIMIT n query runs one extra optimizer pass #25671.

Rationale for this change

For Limit(skip=0, fetch=n) over a Sort without a fetch, PushDownLimit sets Sort.fetch = n but keeps the Limit. It removes the Limit only on its next visit, in the next optimizer pass. Thus every ORDER BY ... LIMIT n query runs 3 logical optimizer passes where 2 are enough. The final plan does not change. Only planning time does.

A query engine that wraps every query in a row-cap LIMIT over an ORDER BY hits this on nearly every query.

Reproduction, datafusion-cli on main (95bb0a0):

CREATE TABLE t (a INT, ts TIMESTAMP) AS VALUES (1, TIMESTAMP '2026-01-01T00:00:00');

EXPLAIN VERBOSE SELECT a FROM t ORDER BY ts DESC LIMIT 10;

Rules that change the plan, per pass (the logical_plan after <rule> rows that are not SAME TEXT AS ABOVE):

main:
  pass 0: push_down_limit, optimize_projections
  pass 1: push_down_limit          <- removes the Limit it kept in pass 0
  pass 2: (no change)

this PR:
  pass 0: push_down_limit, optimize_projections
  pass 1: (no change)

The final logical_plan is identical on both:

Projection: t.a
  Sort: t.ts DESC NULLS FIRST, fetch=10
    TableScan: t projection=[a, ts]

What changes are included in this PR?

In rewrite_limit, LogicalPlan::Sort branch, when the rule gives the Sort a new fetch and skip == 0, it now returns the Sort and drops the Limit in the same visit. With skip > 0 the Limit stays, as before. When the Sort already has an equal or smaller fetch and skip == 0, main already dropped the Limit. Both skip == 0 cases now share one branch.

One detail needs attention. The optimizer applies PushDownLimit top-down and then recurses into the children of the node that the rule returns. Before this change, the Sort was a child of the returned Limit, so the same pass also visited it and applied the TopK-through-join pushdown (push_topk_through_join). Now the Sort is the returned node, so the traversal does not visit it again in this pass. The skip == 0 branch therefore calls push_topk_through_join on the Sort directly. Without that call, Limit -> Sort -> Left Join would move the extra pass from the Limit removal to the TopK pushdown instead of removing it. The same was already true on main for a Sort that already had a fetch, so that case also gets the TopK pushdown in the same pass now.

The branch always returns Transformed::yes, also when push_topk_through_join returns Transformed::no, because the Limit is removed in every case.

What is the testing strategy for this PR?

• push_down_limit::test::limit_push_down_sort: snapshot updated. After one pass the plan is Sort: ..., fetch=10 without the Limit above it.

• New push_down_limit::test::limit_push_down_sort_settles_in_first_pass: runs PushDownLimit with max_passes = 3 and asserts that the optimizer stops after 2 passes (pass 1 leaves the plan unchanged). On main it runs 3.

• topk_through_join::test::topk_pushed_through_limit_then_sort_with_two_passes is renamed to topk_pushed_through_limit_then_sort_in_one_pass and now uses max_passes = 1, with the same expected plan. It fails if the new branch does not call push_topk_through_join.

• New topk_through_join::test::topk_pushed_through_limit_over_sort_with_{same,smaller}_fetch_in_one_pass: Limit over a Sort that already has fetch=3, over a Left Join, with max_passes = 1. The TopK moves below the join in the one pass. Both fail on the earlier code.

• New push_down_limit::test::limit_push_down_sort_reports_transformed: calls PushDownLimit::rewrite on Limit -> Sort -> TableScan (no join) and asserts transformed == true. It fails if the branch returns the Transformed from push_topk_through_join as it is.

• user_defined_plan.rs: the example TopKOptimizerRule also matches a Sort with a fetch (see below). The topk_plan and topk_invariants* tests failed without that change.

I checked that the optimizer tests fail without the fix. Results:

• cargo test --profile ci -p datafusion-optimizer: 901 + 26 + 5 passed, 0 failed.
• cargo test --profile ci -p datafusion-sqllogictest --test sqllogictests: 522 of 522 files pass. No expected plan changes, because the final plans are the same.
• cargo fmt --all -- --check and cargo clippy --profile ci -p datafusion-optimizer --all-targets -- -D warnings: clean.

Are there any user-facing changes?

Final plans are the same. Planning an ORDER BY ... LIMIT query without an OFFSET takes one less optimizer pass.

A custom optimizer rule that runs after PushDownLimit and matches Limit over Sort no longer sees that pattern in the first pass. It sees a Sort with a fetch instead. On main such a rule also saw only the Sort with a fetch from the second pass on. The TopKOptimizerRule example in datafusion/core/tests/user_defined/user_defined_plan.rs had this dependency, so it now also matches a Sort with a fetch.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 73.07692% with 21 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.48%. Comparing base (95bb0a0) to head (684da9c).
⚠️ Report is 18 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/optimizer/src/push_down_limit.rs	67.56%	0 Missing and 12 partials ⚠️
...optimizer/src/push_down_limit/topk_through_join.rs	78.04%	1 Missing and 8 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25672      +/-   ##
==========================================
- Coverage   82.48%   82.48%   -0.01%     
==========================================
  Files        1140     1140              
  Lines      437614   438847    +1233     
  Branches   437614   438847    +1233     
==========================================
+ Hits       360986   361998    +1012     
- Misses      54834    54945     +111     
- Partials    21794    21904     +110     

☔ 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.

[martin-g]

Suggested change

	Ok(Transformed::yes(push_topk_through_join(sort)?.data))
	push_topk_through_join(sort)

?
push_topk_through_join() may return Transformed::no()

[adriangb]

It is intentional. At this point the Limit is already removed, so the plan changed even when push_topk_through_join returns Transformed::no. Passing that no through would report an unchanged plan. I made this explicit in the code (684da9c) and added limit_push_down_sort_reports_transformed, which fails with the suggested change.

[martin-g]

Would that benefit of using push_topk_through_join(sort) too ?!

[adriangb]

Yes, good catch. That branch has the same skipped visit, so the TopK pushdown waited one more pass there too. In 684da9c both skip == 0 branches share one path that calls push_topk_through_join, and topk_pushed_through_limit_over_sort_with_{same,smaller}_fetch_in_one_pass cover it with max_passes = 1.