kumarUjjawal commented on code in PR #27191:
URL: https://github.com/apache/datafusion/pull/27191#discussion_r4500748348
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -392,12 +392,27 @@ impl EquivalenceProperties {
.filter_map(|o| self.normalize_sort_exprs(o))
.collect();
if !normal_orderings.is_empty() {
+ // What the new orderings imply through known equalities.
`add_constants`
+ // and `add_equal_conditions` discover this too, and whether an
ordering
+ // is satisfied must not depend on which of them ran last. If it
cannot
+ // be derived, nothing is implied.
+ let implied: Vec<_> = normal_orderings
+ .iter()
+ .filter_map(|ordering| {
+ self.implied_ordering(&ordering[0], ordering[1..].iter())
Review Comment:
`add_orderings` looks for an implied ordering only through `ordering[0]`.
Normal orderings keep heterogeneous (per-partition) constants, so when such a
constant leads the ordering the equality behind it is never used, and the
order-dependence this commit removes is still there.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1318,10 +1318,28 @@ fn handle_projection_pushdown(
// (expressed in the projection's output schema) must be remapped into the
// child schema before being pushed down. Forwarding `score@1` unchanged
// could refer to a different input column and sort on the wrong key.
- Ok(
+ let Some(requirements) =
remap_requirement_through_projection(projection_exec, parent_required)
- .map(|requirements| vec![Some(requirements)]),
- )
+ else {
+ return Ok(None);
+ };
+ // If the input already has this ordering, no sort will be added below the
+ // projection. The parent still needs the projection to prove it, which it
+ // may not be able to do: the proof can depend on a column that it drops,
+ // such as a leading per-partition constant. Keep the sort above the
+ // projection in that case.
+ if projection_exec
+ .input()
+ .equivalence_properties()
+ .ordering_satisfy_requirement(requirements.first().clone())?
Review Comment:
The new guard compares `requirements.first()` with
`parent_required.first()`, but these can be different alternatives:
`remap_requirement_through_projection` uses `filter_map`, which silently drops
alternatives it cannot remap.
##########
datafusion/physical-expr/src/equivalence/properties/walk.rs:
##########
@@ -43,19 +43,39 @@ pub(super) struct SortKeyWalk<'a> {
/// found by [`EquivalenceProperties::implied_ordering`] as the walk goes.
implied: Vec<LexOrdering>,
/// For each ordering (the normal orderings of `props`, then `implied`),
- /// the index of its head. Empty until the first pin, so that a walk with
- /// nothing pinned allocates nothing; until then every head is at index 0.
+ /// the index of its head. Empty until a head moves, so that a walk that
+ /// moves no head allocates nothing; until then every head is at index 0.
head_idxs: Vec<usize>,
}
impl<'a> SortKeyWalk<'a> {
pub(super) fn new(props: &'a EquivalenceProperties) -> Self {
- Self {
+ let mut walk = Self {
props,
pinned: HashSet::new(),
implied: vec![],
head_idxs: vec![],
+ };
+ // A normal ordering keeps a constant whose value differs between
+ // partitions (see `EquivalenceGroup::normalize_sort_exprs`), and such
+ // a constant may lead it. Within a partition it has one value, so it
+ // is fixed and the head is the first expression after it.
+ let normal = &props.oeq_cache.normal_cls;
+ if normal
Review Comment:
Every `SortKeyWalk::new` (i.e. every `ordering_satisfy` /
`ordering_satisfy_requirement` call) now scans all normal orderings and runs
`is_fixed` (hash lookups, equivalence-class lookups, recursion into children)
to find head offsets. Those offsets depend only on `props`.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1318,10 +1318,28 @@ fn handle_projection_pushdown(
// (expressed in the projection's output schema) must be remapped into the
// child schema before being pushed down. Forwarding `score@1` unchanged
// could refer to a different input column and sort on the wrong key.
- Ok(
+ let Some(requirements) =
remap_requirement_through_projection(projection_exec, parent_required)
- .map(|requirements| vec![Some(requirements)]),
- )
+ else {
+ return Ok(None);
+ };
+ // If the input already has this ordering, no sort will be added below the
+ // projection. The parent still needs the projection to prove it, which it
+ // may not be able to do: the proof can depend on a column that it drops,
+ // such as a leading per-partition constant. Keep the sort above the
+ // projection in that case.
+ if projection_exec
+ .input()
+ .equivalence_properties()
+ .ordering_satisfy_requirement(requirements.first().clone())?
+ && !projection_exec
Review Comment:
The guard recomputes
`projection_exec.equivalence_properties().ordering_satisfy_requirement(parent_required.first())`.
The caller `pushdown_sorts_helper` has already computed this as
`satisfy_parent` (line 264) for the same plan and requirement.
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
Review Comment:
https://github.com/apache/datafusion/blob/c777864f7c89f8c54a7a50bc480a785df17185c7/datafusion/physical-expr/src/equivalence/properties/mod.rs#L590
`discover_new_orderings` now duplicates work: it computes `implied_ordering`
for each leading ordering and passes the results to `add_orderings`, which
immediately recomputes `implied_ordering` on those results and recurses again,
each level re-normalizing and calling `update_map`.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1318,10 +1318,28 @@ fn handle_projection_pushdown(
// (expressed in the projection's output schema) must be remapped into the
// child schema before being pushed down. Forwarding `score@1` unchanged
// could refer to a different input column and sort on the wrong key.
- Ok(
+ let Some(requirements) =
remap_requirement_through_projection(projection_exec, parent_required)
- .map(|requirements| vec![Some(requirements)]),
- )
+ else {
+ return Ok(None);
+ };
+ // If the input already has this ordering, no sort will be added below the
+ // projection. The parent still needs the projection to prove it, which it
+ // may not be able to do: the proof can depend on a column that it drops,
+ // such as a leading per-partition constant. Keep the sort above the
+ // projection in that case.
+ if projection_exec
Review Comment:
Altitude: the "input has the ordering but the node cannot prove it" guard is
added only for ProjectionExec. HashJoinExec with an embedded `projection`
(`handle_hash_join`) also drops columns before pushing a remapped requirement
to its right child, so it can fail the same #26747 way.
##########
datafusion/physical-expr/src/equivalence/properties/walk.rs:
##########
@@ -43,19 +43,39 @@ pub(super) struct SortKeyWalk<'a> {
/// found by [`EquivalenceProperties::implied_ordering`] as the walk goes.
implied: Vec<LexOrdering>,
/// For each ordering (the normal orderings of `props`, then `implied`),
- /// the index of its head. Empty until the first pin, so that a walk with
- /// nothing pinned allocates nothing; until then every head is at index 0.
+ /// the index of its head. Empty until a head moves, so that a walk that
+ /// moves no head allocates nothing; until then every head is at index 0.
head_idxs: Vec<usize>,
}
impl<'a> SortKeyWalk<'a> {
pub(super) fn new(props: &'a EquivalenceProperties) -> Self {
- Self {
+ let mut walk = Self {
props,
pinned: HashSet::new(),
implied: vec![],
head_idxs: vec![],
+ };
+ // A normal ordering keeps a constant whose value differs between
+ // partitions (see `EquivalenceGroup::normalize_sort_exprs`), and such
+ // a constant may lead it. Within a partition it has one value, so it
+ // is fixed and the head is the first expression after it.
+ let normal = &props.oeq_cache.normal_cls;
+ if normal
+ .iter()
+ .any(|ordering| walk.is_fixed(&ordering[0].expr))
Review Comment:
Cache initial head offsets instead of rescanning each walk
##########
datafusion/physical-expr/src/equivalence/properties/walk.rs:
##########
@@ -43,19 +43,39 @@ pub(super) struct SortKeyWalk<'a> {
/// found by [`EquivalenceProperties::implied_ordering`] as the walk goes.
implied: Vec<LexOrdering>,
/// For each ordering (the normal orderings of `props`, then `implied`),
- /// the index of its head. Empty until the first pin, so that a walk with
- /// nothing pinned allocates nothing; until then every head is at index 0.
+ /// the index of its head. Empty until a head moves, so that a walk that
+ /// moves no head allocates nothing; until then every head is at index 0.
head_idxs: Vec<usize>,
}
impl<'a> SortKeyWalk<'a> {
pub(super) fn new(props: &'a EquivalenceProperties) -> Self {
- Self {
+ let mut walk = Self {
props,
pinned: HashSet::new(),
implied: vec![],
head_idxs: vec![],
+ };
+ // A normal ordering keeps a constant whose value differs between
+ // partitions (see `EquivalenceGroup::normalize_sort_exprs`), and such
+ // a constant may lead it. Within a partition it has one value, so it
+ // is fixed and the head is the first expression after it.
+ let normal = &props.oeq_cache.normal_cls;
+ if normal
+ .iter()
+ .any(|ordering| walk.is_fixed(&ordering[0].expr))
+ {
+ walk.head_idxs = normal
Review Comment:
`SortKeyWalk::new` moves heads past leading fixed expressions but never runs
the implied-ordering discovery that `pin` runs for moved heads. A head exposed
this way implies nothing until some key has been pinned.
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -392,12 +392,27 @@ impl EquivalenceProperties {
.filter_map(|o| self.normalize_sort_exprs(o))
.collect();
if !normal_orderings.is_empty() {
+ // What the new orderings imply through known equalities.
`add_constants`
+ // and `add_equal_conditions` discover this too, and whether an
ordering
+ // is satisfied must not depend on which of them ran last. If it
cannot
+ // be derived, nothing is implied.
+ let implied: Vec<_> = normal_orderings
+ .iter()
+ .filter_map(|ordering| {
+ self.implied_ordering(&ordering[0], ordering[1..].iter())
+ .ok()
+ .flatten()
Review Comment:
`.ok().flatten()` silently discards errors from `implied_ordering` (e.g.
`data_type` or `Interval::make_unbounded` failures). The equivalent call in
`discover_new_orderings` propagates them with `?`.
##########
datafusion/sqllogictest/test_files/window.slt:
##########
@@ -7834,3 +7834,146 @@ DROP TABLE range_null_peers;
statement ok
DROP TABLE range_nan_peers;
+
+# A window ordered by `(total DESC, k)` over an input sorted by `[k, total]`,
+# where `k` is constant within each partition (`lower(g)` under `WHERE g =
'a'`).
+# Within a partition `k` has one value, so the input ordering satisfies the
+# window's requirement and no sort is added.
+# https://github.com/apache/datafusion/issues/26762
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+statement ok
+CREATE TABLE per_partition_constant(g VARCHAR, v BIGINT) AS VALUES ('a', 1),
('b', 2);
+
+query TI
+SELECT k, DENSE_RANK() OVER (ORDER BY total DESC, k) AS r
+FROM (
+ SELECT k, SUM(s) OVER (PARTITION BY k) AS total
+ FROM (SELECT lower(g) AS k, SUM(v) AS s FROM per_partition_constant WHERE g
= 'a' GROUP BY lower(g))
+);
+----
+a 1
+
+statement ok
+set datafusion.explain.physical_plan_only = true;
+
+query TT
+EXPLAIN SELECT k, DENSE_RANK() OVER (ORDER BY total DESC, k) AS r
+FROM (
+ SELECT k, SUM(s) OVER (PARTITION BY k) AS total
+ FROM (SELECT lower(g) AS k, SUM(v) AS s FROM per_partition_constant WHERE g
= 'a' GROUP BY lower(g))
+);
+----
+physical_plan
+01)ProjectionExec: expr=[k@0 as k, dense_rank() ORDER BY [total DESC NULLS
FIRST, k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as
r]
+02)--BoundedWindowAggExec: wdw=[dense_rank() ORDER BY [total DESC NULLS FIRST,
k ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field {
"dense_rank() ORDER BY [total DESC NULLS FIRST, k ASC NULLS LAST] RANGE BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED
PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----ProjectionExec: expr=[k@0 as k, sum(s) PARTITION BY [k] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@2 as total]
+04)------WindowAggExec: wdw=[sum(s) PARTITION BY [k] ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(s) PARTITION BY [k]
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64,
nullable: true }), frame: WindowFrame { units: Rows, start_bound:
Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }]
+05)--------ProjectionExec: expr=[lower(per_partition_constant.g)@0 as k,
sum(per_partition_constant.v)@1 as s]
+06)----------AggregateExec: mode=Single, gby=[lower(g@0) as
lower(per_partition_constant.g)], aggr=[sum(per_partition_constant.v)]
+07)------------FilterExec: g@0 = a
+08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+reset datafusion.explain.physical_plan_only;
+
+# Same, with `k` a per-partition constant that is not derived from a filter
+# (one row per partition), at more than one target partition.
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+query IIII
+SELECT k, c, total, DENSE_RANK() OVER (ORDER BY total DESC, k) AS r
+FROM (
+ SELECT k, c, SUM(c) OVER (PARTITION BY k) AS total
+ FROM (SELECT SUM(v) AS k, COUNT(*) AS c FROM per_partition_constant)
+);
+----
+3 2 2 1
+
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+# When a projection drops the leading constant, it can no longer prove the
+# ordering that its input has, so the sort stays above the projection.
+query II
+SELECT total, DENSE_RANK() OVER (ORDER BY total DESC) AS r
+FROM (
+ SELECT SUM(s) OVER (PARTITION BY k) AS total
+ FROM (SELECT lower(g) AS k, SUM(v) AS s FROM per_partition_constant WHERE g
= 'a' GROUP BY lower(g))
+);
+----
+1 1
+
+# The same when the proof depends on any other dropped column, without a
+# constant: `[k, a, c]` with `a = CAST(c AS DATE)` has `[k, c]`, but not after
+# `a` is dropped.
+statement ok
+CREATE TABLE ordered_through_cast(k VARCHAR, a DATE, c TIMESTAMP) AS VALUES
+ ('a', DATE '2026-01-02', TIMESTAMP '2026-01-02 05:00:00'),
+ ('a', DATE '2026-01-01', TIMESTAMP '2026-01-01 09:00:00'),
+ ('a', DATE '2026-01-02', TIMESTAMP '2026-01-02 01:00:00'),
+ ('b', DATE '2026-01-03', TIMESTAMP '2026-01-03 01:00:00');
+
+query TPI
+SELECT k, c, ROW_NUMBER() OVER (ORDER BY k, c) AS r
+FROM (SELECT k, a, c FROM ordered_through_cast WHERE a = CAST(c AS DATE) ORDER
BY k, a, c LIMIT 10)
+ORDER BY r;
+----
+a 2026-01-01T09:00:00 1
+a 2026-01-02T01:00:00 2
+a 2026-01-02T05:00:00 3
+b 2026-01-03T01:00:00 4
+
+# An ordering that a sort adds implies other orderings through known
equalities,
+# as it does when the equality comes after the sort: `[a, c]` with
+# `a = CAST(c AS DATE)` has `[c]`.
+query DPI
+SELECT a, c, ROW_NUMBER() OVER (ORDER BY c) AS r
+FROM (SELECT a, c FROM ordered_through_cast WHERE a = CAST(c AS DATE) ORDER BY
a, c LIMIT 10)
+ORDER BY r;
Review Comment:
The new test "The same with a per-partition constant leading the sort" (and
the merge variant) checks only result rows, not the plan, so it cannot tell
whether `[c]` was derived or whether a sort was just added.
--
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]
