discord9 opened a new pull request, #25065: URL: https://github.com/apache/datafusion/pull/25065
## 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: ```text 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
