kumarUjjawal commented on code in PR #26459:
URL: https://github.com/apache/datafusion/pull/26459#discussion_r4307426935
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs:
##########
@@ -232,6 +235,71 @@ pub fn replace_with_partial_sort(
Ok(plan)
}
+/// Adds a [`SortExec`] below every node whose ordering requirement is not
+/// satisfied by its child's equivalence properties.
+///
+/// [`pushdown_sorts`] removes a sort when it can map the requirement onto a
+/// descendant that satisfies it. That mapping is computed separately from the
+/// equivalence properties of the operators in between, and the two can
+/// disagree. The plan may then be correct, but its properties cannot show it,
+/// and `SanityCheckPlan` rejects it. Adding the sort back keeps every plan
+/// produced by `EnsureRequirements` valid by the same check.
+///
+/// A requirement that a new [`SortExec`] still cannot satisfy is left as is,
+/// for `SanityCheckPlan` to report.
+///
+/// [`pushdown_sorts`]: sort_pushdown::pushdown_sorts
+pub fn add_sort_for_unproven_ordering(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let requirements = plan.required_input_ordering();
+ if requirements.iter().all(Option::is_none) {
+ return Ok(Transformed::no(plan));
+ }
+ let mut children: Vec<_> = plan.children().into_iter().cloned().collect();
+ let mut changed = false;
+ for (child, requirement) in children.iter_mut().zip(requirements) {
+ let Some(requirement) = requirement else {
+ continue;
+ };
+ let requirement = requirement.into_single();
+ let eq_properties = child.equivalence_properties();
+ if eq_properties.ordering_satisfy_requirement(requirement.clone())? {
+ continue;
+ }
+ let sort_reqs: Vec<_> = requirement.clone().into();
+ let sort_exprs = sort_reqs
+ .into_iter()
+ .filter(|req| eq_properties.is_expr_constant(&req.expr).is_none())
+ .map(Into::into);
+ let Some(ordering) = LexOrdering::new(sort_exprs) else {
+ continue;
+ };
+ let sort: Arc<dyn ExecutionPlan> = Arc::new(
+ SortExec::new(ordering,
Arc::clone(child)).with_preserve_partitioning(
+ child.output_partitioning().partition_count() > 1,
+ ),
Review Comment:
**Share the existing sort-construction logic**
The constant-key filtering, requirement conversion, and
partition-preservation logic duplicate `add_sort_above` and
`add_sort_above_with_distribution` in `utils.rs`. Adding another implementation
means future fixes to these behaviors must reach three separate paths. Extract
a shared constructor that accepts an execution plan, while retaining this
helper's additional satisfaction check.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs:
##########
@@ -232,6 +235,71 @@ pub fn replace_with_partial_sort(
Ok(plan)
}
+/// Adds a [`SortExec`] below every node whose ordering requirement is not
+/// satisfied by its child's equivalence properties.
+///
+/// [`pushdown_sorts`] removes a sort when it can map the requirement onto a
+/// descendant that satisfies it. That mapping is computed separately from the
+/// equivalence properties of the operators in between, and the two can
+/// disagree. The plan may then be correct, but its properties cannot show it,
+/// and `SanityCheckPlan` rejects it. Adding the sort back keeps every plan
+/// produced by `EnsureRequirements` valid by the same check.
+///
+/// A requirement that a new [`SortExec`] still cannot satisfy is left as is,
+/// for `SanityCheckPlan` to report.
+///
+/// [`pushdown_sorts`]: sort_pushdown::pushdown_sorts
+pub fn add_sort_for_unproven_ordering(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let requirements = plan.required_input_ordering();
+ if requirements.iter().all(Option::is_none) {
+ return Ok(Transformed::no(plan));
+ }
+ let mut children: Vec<_> = plan.children().into_iter().cloned().collect();
+ let mut changed = false;
+ for (child, requirement) in children.iter_mut().zip(requirements) {
+ let Some(requirement) = requirement else {
+ continue;
+ };
+ let requirement = requirement.into_single();
+ let eq_properties = child.equivalence_properties();
+ if eq_properties.ordering_satisfy_requirement(requirement.clone())? {
+ continue;
+ }
+ let sort_reqs: Vec<_> = requirement.clone().into();
+ let sort_exprs = sort_reqs
+ .into_iter()
+ .filter(|req| eq_properties.is_expr_constant(&req.expr).is_none())
+ .map(Into::into);
+ let Some(ordering) = LexOrdering::new(sort_exprs) else {
+ continue;
+ };
+ let sort: Arc<dyn ExecutionPlan> = Arc::new(
+ SortExec::new(ordering,
Arc::clone(child)).with_preserve_partitioning(
+ child.output_partitioning().partition_count() > 1,
+ ),
+ );
+ if !sort
+ .equivalence_properties()
+ .ordering_satisfy_requirement(requirement)?
+ {
+ continue;
Review Comment:
**Test the fallback when a sort cannot prove the requirement**
The added tests exercise successful repair, but none covers this explicit
fallback for contradictory properties. Add a focused regression asserting that
an unsatisfying candidate sort is not installed and that `SanityCheckPlan`
still rejects the unchanged requirement. Otherwise, removing this guard would
not be caught despite changing a documented safety behavior.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs:
##########
@@ -232,6 +235,71 @@ pub fn replace_with_partial_sort(
Ok(plan)
}
+/// Adds a [`SortExec`] below every node whose ordering requirement is not
+/// satisfied by its child's equivalence properties.
+///
+/// [`pushdown_sorts`] removes a sort when it can map the requirement onto a
+/// descendant that satisfies it. That mapping is computed separately from the
+/// equivalence properties of the operators in between, and the two can
+/// disagree. The plan may then be correct, but its properties cannot show it,
+/// and `SanityCheckPlan` rejects it. Adding the sort back keeps every plan
+/// produced by `EnsureRequirements` valid by the same check.
+///
+/// A requirement that a new [`SortExec`] still cannot satisfy is left as is,
+/// for `SanityCheckPlan` to report.
+///
+/// [`pushdown_sorts`]: sort_pushdown::pushdown_sorts
+pub fn add_sort_for_unproven_ordering(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let requirements = plan.required_input_ordering();
+ if requirements.iter().all(Option::is_none) {
+ return Ok(Transformed::no(plan));
+ }
+ let mut children: Vec<_> = plan.children().into_iter().cloned().collect();
+ let mut changed = false;
+ for (child, requirement) in children.iter_mut().zip(requirements) {
+ let Some(requirement) = requirement else {
+ continue;
+ };
+ let requirement = requirement.into_single();
+ let eq_properties = child.equivalence_properties();
+ if eq_properties.ordering_satisfy_requirement(requirement.clone())? {
+ continue;
+ }
+ let sort_reqs: Vec<_> = requirement.clone().into();
Review Comment:
This re-implements add_sort_above / add_sort_above_with_check from
physical-optimizer/src/utils.rs:38-128: the same constant-key filter,
LexOrdering::new None check, and preserve_partitioning-if-multi-partition logic.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs:
##########
@@ -232,6 +235,71 @@ pub fn replace_with_partial_sort(
Ok(plan)
}
+/// Adds a [`SortExec`] below every node whose ordering requirement is not
+/// satisfied by its child's equivalence properties.
+///
+/// [`pushdown_sorts`] removes a sort when it can map the requirement onto a
+/// descendant that satisfies it. That mapping is computed separately from the
+/// equivalence properties of the operators in between, and the two can
+/// disagree. The plan may then be correct, but its properties cannot show it,
+/// and `SanityCheckPlan` rejects it. Adding the sort back keeps every plan
+/// produced by `EnsureRequirements` valid by the same check.
+///
+/// A requirement that a new [`SortExec`] still cannot satisfy is left as is,
+/// for `SanityCheckPlan` to report.
+///
+/// [`pushdown_sorts`]: sort_pushdown::pushdown_sorts
+pub fn add_sort_for_unproven_ordering(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let requirements = plan.required_input_ordering();
+ if requirements.iter().all(Option::is_none) {
+ return Ok(Transformed::no(plan));
+ }
+ let mut children: Vec<_> = plan.children().into_iter().cloned().collect();
+ let mut changed = false;
+ for (child, requirement) in children.iter_mut().zip(requirements) {
+ let Some(requirement) = requirement else {
+ continue;
+ };
+ let requirement = requirement.into_single();
+ let eq_properties = child.equivalence_properties();
+ if eq_properties.ordering_satisfy_requirement(requirement.clone())? {
+ continue;
+ }
+ let sort_reqs: Vec<_> = requirement.clone().into();
+ let sort_exprs = sort_reqs
+ .into_iter()
+ .filter(|req| eq_properties.is_expr_constant(&req.expr).is_none())
+ .map(Into::into);
+ let Some(ordering) = LexOrdering::new(sort_exprs) else {
+ continue;
+ };
+ let sort: Arc<dyn ExecutionPlan> = Arc::new(
+ SortExec::new(ordering,
Arc::clone(child)).with_preserve_partitioning(
+ child.output_partitioning().partition_count() > 1,
+ ),
+ );
+ if !sort
+ .equivalence_properties()
+ .ordering_satisfy_requirement(requirement)?
+ {
+ continue;
+ }
+ *child = sort;
+ changed = true;
+ }
+ if changed {
+ plan.replace_children(
Review Comment:
After replace_children(Recompute), nodes such as AggregateExec and
BoundedWindowAggExec recompute required_input_ordering
(ordered_partition_by_indices, the soft aggregate requirement) from the new
child, but the new node's requirement is never checked again.
##########
datafusion/physical-optimizer/src/ensure_requirements/mod.rs:
##########
@@ -267,10 +279,15 @@ impl PhysicalOptimizerRule for EnsureRequirements {
assign_initial_requirements(&mut sort_pushdown);
let adjusted = pushdown_sorts(sort_pushdown)?;
- // 3d: Partial sort
- adjusted
+ // 3d: Sort where an ordering requirement cannot be proven
+ use super::enforce_sorting::add_sort_for_unproven_ordering;
+ let plan = adjusted
.plan
Review Comment:
add_sort_for_unproven_ordering runs after parallelize_sorts (3a) and
replace_with_order_preserving_variants (3b), so a sort it adds never gets those
rewrites.
##########
datafusion/physical-optimizer/src/ensure_requirements/mod.rs:
##########
@@ -267,10 +279,15 @@ impl PhysicalOptimizerRule for EnsureRequirements {
assign_initial_requirements(&mut sort_pushdown);
let adjusted = pushdown_sorts(sort_pushdown)?;
- // 3d: Partial sort
- adjusted
+ // 3d: Sort where an ordering requirement cannot be proven
Review Comment:
The fix patches the symptom in EnsureRequirements, not the cause:
EquivalenceClass::try_with_offset (physical-expr/src/equivalence/class.rs:249)
starts from Self::default() and never copies `constant`, so
EquivalenceGroup::join drops every probe-side (right) constant, and
is_trivial() then deletes the single-member class.
##########
datafusion/physical-optimizer/src/ensure_requirements/mod.rs:
##########
@@ -69,6 +70,17 @@
//! <https://github.com/apache/datafusion/issues/21973>, where the old
//! two-rule pipeline could regress a parallel sort plan into a serial one
//! on pass 2.
+//! - **Requirements are provable**: Every ordering requirement in the
+//! output is satisfied according to the children's
+//! [`EquivalenceProperties`], the same check `SanityCheckPlan` applies.
+//! `pushdown_sorts` removes a sort when it can map the requirement onto a
+//! descendant that satisfies it, but the operators in between can report
+//! weaker properties than that mapping assumed (for example, a hash join
+//! that forgets a probe-side column is constant). Phase 3 then adds a sort
+//! where the requirement cannot be proven, so the rule never emits a plan
+//! that it cannot show to be valid.
Review Comment:
Qualify the guarantee that all ordering requirements are proven
For contradictory equivalence properties, `add_sort_for_unproven_ordering`
deliberately leaves an unsatisfied requirement unchanged when even the
candidate sort cannot satisfy it. This contradicts the new claim that the rule
never emits an unprovable plan. Document this exception here, consistent with
the helper's documentation, so callers do not mistake this pass for successful
validation.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs:
##########
@@ -232,6 +235,71 @@ pub fn replace_with_partial_sort(
Ok(plan)
}
+/// Adds a [`SortExec`] below every node whose ordering requirement is not
+/// satisfied by its child's equivalence properties.
+///
+/// [`pushdown_sorts`] removes a sort when it can map the requirement onto a
+/// descendant that satisfies it. That mapping is computed separately from the
+/// equivalence properties of the operators in between, and the two can
+/// disagree. The plan may then be correct, but its properties cannot show it,
+/// and `SanityCheckPlan` rejects it. Adding the sort back keeps every plan
+/// produced by `EnsureRequirements` valid by the same check.
+///
+/// A requirement that a new [`SortExec`] still cannot satisfy is left as is,
+/// for `SanityCheckPlan` to report.
+///
+/// [`pushdown_sorts`]: sort_pushdown::pushdown_sorts
+pub fn add_sort_for_unproven_ordering(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let requirements = plan.required_input_ordering();
+ if requirements.iter().all(Option::is_none) {
+ return Ok(Transformed::no(plan));
+ }
+ let mut children: Vec<_> = plan.children().into_iter().cloned().collect();
+ let mut changed = false;
+ for (child, requirement) in children.iter_mut().zip(requirements) {
+ let Some(requirement) = requirement else {
+ continue;
+ };
+ let requirement = requirement.into_single();
+ let eq_properties = child.equivalence_properties();
+ if eq_properties.ordering_satisfy_requirement(requirement.clone())? {
+ continue;
+ }
+ let sort_reqs: Vec<_> = requirement.clone().into();
+ let sort_exprs = sort_reqs
+ .into_iter()
+ .filter(|req| eq_properties.is_expr_constant(&req.expr).is_none())
+ .map(Into::into);
+ let Some(ordering) = LexOrdering::new(sort_exprs) else {
+ continue;
+ };
+ let sort: Arc<dyn ExecutionPlan> = Arc::new(
Review Comment:
The new SortExec is always stacked on top of the child, even when the child
is already a SortExec (without fetch) whose ordering is wrong or too weak, and
the misplaced sort left lower in the plan by pushdown_sorts is never removed.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs:
##########
@@ -232,6 +235,71 @@ pub fn replace_with_partial_sort(
Ok(plan)
}
+/// Adds a [`SortExec`] below every node whose ordering requirement is not
+/// satisfied by its child's equivalence properties.
+///
+/// [`pushdown_sorts`] removes a sort when it can map the requirement onto a
+/// descendant that satisfies it. That mapping is computed separately from the
+/// equivalence properties of the operators in between, and the two can
+/// disagree. The plan may then be correct, but its properties cannot show it,
+/// and `SanityCheckPlan` rejects it. Adding the sort back keeps every plan
+/// produced by `EnsureRequirements` valid by the same check.
+///
+/// A requirement that a new [`SortExec`] still cannot satisfy is left as is,
+/// for `SanityCheckPlan` to report.
+///
+/// [`pushdown_sorts`]: sort_pushdown::pushdown_sorts
+pub fn add_sort_for_unproven_ordering(
+ plan: Arc<dyn ExecutionPlan>,
+) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
+ let requirements = plan.required_input_ordering();
+ if requirements.iter().all(Option::is_none) {
+ return Ok(Transformed::no(plan));
+ }
+ let mut children: Vec<_> = plan.children().into_iter().cloned().collect();
+ let mut changed = false;
+ for (child, requirement) in children.iter_mut().zip(requirements) {
+ let Some(requirement) = requirement else {
+ continue;
+ };
+ let requirement = requirement.into_single();
+ let eq_properties = child.equivalence_properties();
+ if eq_properties.ordering_satisfy_requirement(requirement.clone())? {
+ continue;
+ }
+ let sort_reqs: Vec<_> = requirement.clone().into();
+ let sort_exprs = sort_reqs
+ .into_iter()
+ .filter(|req| eq_properties.is_expr_constant(&req.expr).is_none())
+ .map(Into::into);
+ let Some(ordering) = LexOrdering::new(sort_exprs) else {
+ continue;
+ };
+ let sort: Arc<dyn ExecutionPlan> = Arc::new(
+ SortExec::new(ordering,
Arc::clone(child)).with_preserve_partitioning(
Review Comment:
The added SortExec never carries a fetch, even when the requiring parent
(SortPreservingMergeExec / OutputRequirementExec) has one or the sort that
pushdown_sorts removed was a TopK.
##########
datafusion/sqllogictest/test_files/order.slt:
##########
@@ -2276,3 +2276,79 @@ SELECT i, -i AS n FROM (
-128 -128
-128 -128
NULL NULL
+
+# EnsureRequirements removes the sort on `k` because `k` is constant on the
+# probe side of the join, but the join output does not report `k` as constant.
+# The sort must then be kept so the plan can be checked and run.
+# https://github.com/apache/datafusion/issues/26283
+statement ok
+CREATE TABLE probe_constant_t(a INT, b INT) AS VALUES (1, 1), (2, 2), (3, 3),
(4, 4);
+
+statement ok
+CREATE TABLE probe_constant_u(a INT) AS VALUES (1);
+
+statement ok
+CREATE TABLE probe_constant_v(x INT) AS VALUES (1), (2);
+
+query II
+SELECT s.k, s.a
+FROM (
+ SELECT probe_constant_t.a, -2 AS k
+ FROM probe_constant_t JOIN probe_constant_v ON probe_constant_t.b <>
probe_constant_v.x
+) s
+JOIN probe_constant_u ON s.a = probe_constant_u.a
+ORDER BY s.k DESC;
+----
+-2 1
+
+query TT
+EXPLAIN SELECT s.k, s.a
+FROM (
+ SELECT probe_constant_t.a, -2 AS k
+ FROM probe_constant_t JOIN probe_constant_v ON probe_constant_t.b <>
probe_constant_v.x
+) s
+JOIN probe_constant_u ON s.a = probe_constant_u.a
+ORDER BY s.k DESC;
+----
+logical_plan
+01)Sort: s.k DESC NULLS FIRST
+02)--Projection: s.k, s.a
+03)----Inner Join: s.a = probe_constant_u.a
+04)------SubqueryAlias: s
+05)--------Projection: probe_constant_t.a, Int64(-2) AS k
+06)----------Inner Join: Filter: probe_constant_t.b != probe_constant_v.x
+07)------------TableScan: probe_constant_t projection=[a, b]
+08)------------TableScan: probe_constant_v projection=[x]
+09)------TableScan: probe_constant_u projection=[a]
+physical_plan
+01)SortPreservingMergeExec: [k@0 DESC]
+02)--SortExec: expr=[k@0 DESC], preserve_partitioning=[true]
+03)----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)],
projection=[k@2, a@1]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+06)--------ProjectionExec: expr=[a@0 as a, -2 as k]
+07)----------NestedLoopJoinExec: join_type=Inner, filter=b@0 != x@1,
projection=[a@1]
+08)------------DataSourceExec: partitions=1, partition_sizes=[1]
+09)------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+query ITI
+SELECT s.a, s.k, row_number() OVER (PARTITION BY s.a ORDER BY s.k) AS r
Review Comment:
The second regression query (window over a probe-side constant at
target_partitions=1) only checks results, with no EXPLAIN. The Rust test only
checks idempotency up to EnsureRequirements and never runs SanityCheckPlan or
the later rules.
--
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]
