Skip to content

fix: preserve fetched merges during distribution enforcement - #25065

Draft
discord9 wants to merge 1 commit into
apache:mainfrom
discord9:fix/preserve-distribution-fetch
Draft

fix: preserve fetched merges during distribution enforcement#25065
discord9 wants to merge 1 commit into
apache:mainfrom
discord9:fix/preserve-distribution-fetch

Conversation

@discord9

@discord9 discord9 commented Sep 8, 2026

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Related to #14150 and #23800. This is independently reproducible on main and does not depend on #23800.

Rationale for this change

Running the default physical optimizer a second time can change query results. On main 8a9228164fddeaa03c9ee510b641990e403f2eb5, a global LIMIT over a partitioned aggregate returns one row after the first optimization and three after the second: distribution enforcement removes the CoalescePartitionsExec(fetch=1) that limits the combined output.

Fetched sort-preserving merges have an additional requirement: their ordering determines which rows survive. Replacing SortPreservingMergeExec(fetch=N) with CoalescePartitionsExec(fetch=N) can preserve the row count while selecting the wrong rows, even when the parent does not require ordered output.

What changes are included in this PR?

  • Stop stripping distribution operators at the first operator carrying a fetch, including fetch=0. Preserve the actual global row-selection boundary rather than moving its numeric fetch elsewhere.
  • Stop order-preserving-variant downgrades at fetched sort-preserving merges before recursing into their ordered inputs.
  • Remove the now-unreachable collection and reapplication of fetch from stripped operators. Fetch-free distribution optimization remains unchanged.
  • Correct the existing single-partition fetched-SPM snapshot to retain a local limit and satisfy the merge's input ordering.

What is the testing strategy for this PR?

Native execution regressions cover:

  • One- and three-partition fetched coalesces, including zero fetch.
  • LIMIT over a partitioned aggregate through two complete default optimizer passes.
  • TopK beneath a genuinely order-insensitive aggregate, asserting selected values rather than merely row counts.
  • OFFSET + LIMIT and OFFSET-only, including exact ordered output values.
  • A genuine per-partition LocalLimit that must still return one row from each partition.
  • Fetched and fetch-free SPM downgrade behavior, including order-preserving repartition descendants.

Baseline execution on unmodified main (8a9228164f) and the latest #23800 head (a403530d99) demonstrated the global LIMIT regression through two complete default physical optimizer passes: one row on the first pass, three on the second.

Independent guard-removal experiments on #23800 also verified both protections:

Variant Executed result
Original production code TopK returns [10,20,98,99] instead of [10,20].
Remove the fetched-node removal guard LIMIT returns one row after the first full pass and three after the second.
Remove the fetched-SPM downgrade guard Both passes return two rows, but select [10,99] instead of [10,20].
Both guards present All 87 distribution tests pass.

On the main-based branch, the following checks pass:

cargo fmt --all --check
cargo test -p datafusion --test core_integration
# 1125 passed
cargo test -p datafusion-physical-optimizer --lib
# 37 passed
cargo clippy --all-targets --all-features -- -D warnings
git diff --check

The fix combined with #23800 additionally passes 87 distribution tests, 42 LimitPushdown tests, 30 EnsureRequirements tests, and strict all-target/all-feature Clippy for datafusion and datafusion-physical-optimizer.

Are there any user-facing changes?

Queries retain their global LIMIT/OFFSET and ordered TopK selection when physical plans are optimized again. No public API or configuration changes.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
@codecov-commenter

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 81.74%. Comparing base (8a92281) to head (b50c877).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files
@@            Coverage Diff             @@
##             main   #25065      +/-   ##
==========================================
+ Coverage   81.72%   81.74%   +0.02%     
==========================================
  Files        1127     1128       +1     
  Lines      416519   416633     +114     
  Branches   416519   416633     +114     
==========================================
+ Hits       340401   340587     +186     
+ Misses      56115    55988     -127     
- Partials    20003    20058      +55     

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Core DataFusion crate optimizer Optimizer rules

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants