Repository navigation
fix: preserve required ordering in limit pushdown - #24231
Himanshu-2005-code wants to merge 5 commits into
Conversation
|
Local validation completed for the change:
The implementation now preserves |
|
I’ve finished the fix and regression coverage for the order-sensitive limit issue. Summary: preserved required_ordering when limits are rebuilt during limit pushdown cargo check -p datafusion-physical-optimizer — passed |
|
@Himanshu-2005-code Is there a GitHub issue for this bug? Please follow the PR template, and include more details about the user-visible behavior that is being changed and why. Thank you! |
|
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 |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #24231 +/- ##
==========================================
- Coverage 81.01% 81.01% -0.01%
==========================================
Files 1106 1106
Lines 384104 384129 +25
Branches 384104 384129 +25
==========================================
+ Hits 311194 311209 +15
- Misses 54566 54573 +7
- Partials 18344 18347 +3 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Thanks for pointing that out. I’ve updated the PR description to link #24215 and added the rationale, user-visible behavior, changes, and testing details. |
bvolpato
left a comment
bvolpato
left a comment
There was a problem hiding this comment.
Other than these inline comments, this looks good to me.
| let inner = if let Some(fetch) = sort_child.fetch() { | ||
| inner.with_fetch(Some(fetch)).unwrap_or_else(|| { | ||
| Arc::new(LocalLimitExec::new(inner, fetch)) | ||
| let mut limit = |
There was a problem hiding this comment.
Could this remain a LocalLimitExec while attaching required_ordering? In this SPM path, inner can still have multiple partitions, while GlobalLimitExec requires a single input partition and this rule runs after distribution enforcement. Would that make the resulting plan fail sanity checking or execution for an exact-pushdown source that does not support with_fetch()?
| skip: usize, | ||
| satisfied: bool, | ||
| preserve_order: bool, | ||
| required_ordering: Option<LexOrdering>, |
There was a problem hiding this comment.
GlobalRequirements is public, and replacing the bool with Option<LexOrdering> removes its UnwindSafe and RefUnwindSafe auto-trait implementations according to cargo-semver-checks. Is this compatibility break a concern here, or should ordering state be carried without changing those public auto traits?
|
Is it possible to find such query that shows ordering is loss but shouldn't and add it in sqllogictest? |
- pushdown_sort.rs: keep LocalLimitExec (not GlobalLimitExec) on the SPM fallback-limit path, since the input can still have multiple partitions at that point (this rule runs after distribution enforcement, and GlobalLimitExec requires a single input partition). - limit_pushdown.rs: scope GlobalRequirements and pushdown_limit_helper to pub(crate), since nothing outside this module uses them. This avoids the UnwindSafe/RefUnwindSafe auto-trait removal that cargo-semver-checks flagged when required_ordering was added to the struct. - Add a regression test proving a lost required_ordering leaves FileScanConfig work-stealing enabled on the scan it reaches, which is the concrete mechanism by which a multi-file scan can read files out of order under a limit.
|
@bvolpato - Reverted the SPM fallback-limit path back to LocalLimitExec — you were right that GlobalLimitExec would break the single-partition invariant there, since this rule runs after distribution enforcement.
|
|
@jayzhan211 — the actual reordering only manifests via FileScanConfig's work-stealing between sibling partition streams when preserve_order=false, which is timing-dependent and not reachable from a single-pass .slt query (LimitPushdown only runs once per query). Added required_ordering_disables_scan_work_stealing, a deterministic test that ties a lost required_ordering directly to work-stealing staying enabled on the scan — verified it fails if I revert the fix. |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
Thanks @Himanshu-2005-code! I left a few comments.
| /// [`LimitPushdown`]: crate::limit_pushdown::LimitPushdown | ||
| #[derive(Default, Clone, Debug)] | ||
| pub struct GlobalRequirements { | ||
| pub(crate) struct GlobalRequirements { |
There was a problem hiding this comment.
GlobalRequirements and pushdown_limit_helper went from pub to pub(crate). That is a larger break than the UnwindSafe loss it was meant to avoid, and the PR body does not declare it. An external caller that compiles on main now fails:
error[E0603]: struct `GlobalRequirements` is private
error[E0603]: function `pushdown_limit_helper` is private
Fix: keep both public. If the removal is intended, declare it under "user-facing changes" and request the api change label.
-pub(crate) struct GlobalRequirements {
+pub struct GlobalRequirements {
@@
-pub(crate) fn pushdown_limit_helper(
+pub fn pushdown_limit_helper(| let inner = if let Some(fetch) = sort_child.fetch() { | ||
| inner.with_fetch(Some(fetch)).unwrap_or_else(|| { | ||
| Arc::new(LocalLimitExec::new(inner, fetch)) | ||
| let mut limit = LocalLimitExec::new(inner, fetch); |
There was a problem hiding this comment.
The two new set_required_ordering calls are untested. Reverting this file to main leaves every pushdown_sort and limit_pushdown test green, because the snapshots do not display required_ordering.
Fix: assert on the node in test_sort_pushdown_exact_preserves_fetch_with_global_limit. Add the same check for the SortPreservingMergeExec → LocalLimitExec path.
let plan = sort_exec_with_fetch(ordering.clone(), Some(10), source);
let optimized = PushdownSort::new()
.optimize(plan, &ConfigOptions::new())
.unwrap();
let limit = optimized.downcast_ref::<GlobalLimitExec>().unwrap();
assert_eq!(limit.required_ordering().as_ref(), Some(&ordering));5 participants
Which issue does this PR close?
required_ordering, losing order-sensitivity for later passes #24215Rationale for this change
Queries using
ORDER BY ... LIMITcan return incorrect results after the physical plan is rewritten by limit or sort pushdown.When an order-sensitive limit is recreated during optimization, the information that the limit depends on a particular ordering can be lost. If the plan is optimized again later, the limit may then be pushed down without preserving that ordering, allowing the underlying scan to return rows in a different order.
This can cause
ORDER BY ... LIMITqueries to return different or incorrect rows.What changes are included in this PR?
required_orderingwhenGlobalLimitExecandLocalLimitExecnodes are recreated during limit pushdown.Are these changes tested?
Yes.
cargo check -p datafusion-physical-optimizercargo clippy -p datafusion-physical-optimizer --all-targets -- -D warningscargo fmt --all -- --checkcargo test -p datafusion-physical-optimizerlimit_pushdownregression testsAll local tests and checks pass.
Are there any user-facing changes?
Yes.
This fixes incorrect query results for order-sensitive
LIMIToperations after physical-plan rewrites or repeated optimization.Queries such as
ORDER BY ... LIMITwill retain the required ordering information when limits are pushed down and recreated, preventing later optimization passes from treating the limit as order-insensitive.