fix: Capture global ORDER BY requirement under ScalarSubqueryExec root#23677
Open
sgrebnov wants to merge 1 commit into
Open
fix: Capture global ORDER BY requirement under ScalarSubqueryExec root#23677sgrebnov wants to merge 1 commit into
sgrebnov wants to merge 1 commit into
Conversation
`OutputRequirements` (add mode) walks the plan via `require_top_ordering_helper` to capture the query's global ORDER BY as an `OutputRequirementExec`. The helper bails on any node with `children.len() != 1`. `ScalarSubqueryExec` is such a node (child 0 is the order-transparent main input, the rest are uncorrelated subquery plans), so when it is the plan root the rule stamps an empty `OutputRequirementExec(order_by=[], dist_by=Unspecified)` at the top and never records the global ordering requirement. `ScalarSubqueryExec` is order-transparent on child 0 (`maintains_input_order()[0] == true`, no required input ordering), so `require_top_ordering_helper` now descends through child 0 to find and wrap the top `SortExec` / `SortPreservingMergeExec`, reattaching the subquery children unchanged. Tested at two levels in `datafusion/core/tests/physical_optimizer/output_requirements.rs`: - `require_top_ordering_descends_through_scalar_subquery` (rule level) - `scalar_subquery_root_preserves_global_ordering_end_to_end` (end to end)
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #23677 +/- ##
=======================================
Coverage ? 80.66%
=======================================
Files ? 1087
Lines ? 367420
Branches ? 367420
=======================================
Hits ? 296368
Misses ? 53395
Partials ? 17657 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
neilconway
reviewed
Jul 18, 2026
neilconway
left a comment
neilconway
left a comment
Contributor
There was a problem hiding this comment.
Thanks for investigating this, and for the well-reasoned and well-tested PR.
Comment on lines
354
to
357
| /// Idempotent: if the plan is already topped by an `OutputRequirementExec`, it | ||
| /// is returned unchanged so that re-running this rule (as adaptive execution | ||
| /// in datafusion-ballista AQE does after every completed stage, see | ||
| /// datafusion-ballista#1359) does not stack wrappers. |
Contributor
There was a problem hiding this comment.
This rule is not actually idempotent, because we only check for an existing wrapper at the root of the plan, but wrappers might get added below the root of the plan, so in some cases we'll add a redundant OutputRequirementExec. Should we update the logic to add a top-level OutputRequirementExec to be more intelligent?
Comment on lines
+385
to
+402
| // `ScalarSubqueryExec` is a multi-child but order-transparent root: child 0 is | ||
| // the main input (it copies that child's `PlanProperties` and reports | ||
| // `maintains_input_order()[0] == true` with no required input ordering), while | ||
| // the remaining children are subquery plans that don't contribute to output | ||
| // ordering. The generic `children.len() != 1` guard below would stop the search | ||
| // at this node and lose the query's global ORDER BY (the top `SortExec` lives | ||
| // below the main input), so descend through child 0 and reattach the rest. | ||
| if plan.downcast_ref::<ScalarSubqueryExec>().is_some() { | ||
| let (new_main, is_changed) = | ||
| require_top_ordering_helper(Arc::clone(children[0]))?; | ||
| if is_changed { | ||
| let mut new_children: Vec<Arc<dyn ExecutionPlan>> = | ||
| children.iter().map(|&c| Arc::clone(c)).collect(); | ||
| new_children[0] = new_main; | ||
| return Ok((plan.with_new_children(new_children)?, true)); | ||
| } | ||
| return Ok((plan, false)); | ||
| } |
Contributor
There was a problem hiding this comment.
What if we added a helper like
fn output_requirement_child(plan: &dyn ExecutionPlan) -> Option<usize> {
let children = plan.children();
if children.len() == 1 {
Some(0)
} else if plan.is::<ScalarSubqueryExec>() {
Some(0)
} else {
None
}
}That might isolate the subquery-specific logic to a single place?
| .as_ref() | ||
| .is_none_or(|o| matches!(o, OrderingRequirements::Soft(_)))) | ||
| { | ||
| // Keep searching for a `SortExec` as long as ordering is maintained, |
Contributor
There was a problem hiding this comment.
Not yours, but this comment should be generalized to SortExec + SortPreservingMergeExec.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
After the DataFusion 53 → 54 upgrade, some queries with a
ScalarSubquerystopped respecting the globalORDER BYand returned unordered results.Rationale for this change
In add mode,
OutputRequirementswalks the plan viarequire_top_ordering_helperto capture the globalORDER BYas anOutputRequirementExec. The helper bails on any node withchildren.len() != 1.ScalarSubqueryExechas multiple children (child 0 = order-transparent main input, the rest = uncorrelated subquery plans),so when it is the root the rule stops immediately and stamps an empty
OutputRequirementExec(order_by=[], dist_by=Unspecified)at the top — losing the ordering requirement.ScalarSubqueryExecis order-transparent on child 0 (maintains_input_order()[0]== true, no required input ordering), so the search should descend through it like any other single-child order-preserving operator.What changes are included in this PR?
require_top_ordering_helpernow special-casesScalarSubqueryExec: it descends through child 0 to find and wrap the topSortExec/SortPreservingMergeExec, reattaching the subquery children unchanged.Are these changes tested?
Yes — two tests in
datafusion/core/tests/physical_optimizer/output_requirements.rs:Rule level (
require_top_ordering_descends_through_scalar_subquery): asserts theOutputRequirementExeccarrying the ordering is placed below theScalarSubqueryExec; without the fix it lands empty at the root.End to end (
scalar_subquery_root_preserves_global_ordering_end_to_end): runs the full default optimizer pipeline over aScalarSubqueryExecroot whose ordering comes from aSortPreservingMergeExecover a two-partition ordered source, then executes the plan and checks the rows. Verified by toggling the fix:1, 3, 5, 7, 2, 4, 6, 8— merge dropped, rows partition-interleaved1, 2, 3, 4, 5, 6, 7, 8— globally orderedExisting
subquery.slt/ TPC-H snapshots are unchanged. I was unable to construct a SQL query that triggers the bug - over built-in sources the globalORDER BYis always preserved regardless of the missing requirement. It only surfaces when a custom physical planner supplies a plan shaped like the end-to-end test: aSortPreservingMergeExecover already-sorted partitions, with noSortExecabove it.Are there any user-facing changes?
No.