viirya commented on code in PR #27191:
URL: https://github.com/apache/datafusion/pull/27191#discussion_r4503011132
##########
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:
This also fires when the requirement comes from a top-level `ORDER BY`,
which `SanityCheckPlan` never checks:
```sql
CREATE EXTERNAL TABLE sorted_t(k VARCHAR, a DATE, c TIMESTAMP)
STORED AS PARQUET LOCATION '...' WITH ORDER (k, a, c);
EXPLAIN SELECT k, c FROM sorted_t WHERE a = CAST(c AS DATE) ORDER BY k, c;
```
`main` plans this with no sort. With this change it plans a `SortExec` over
the whole scan. The TopK form (`... ORDER BY k, a, c LIMIT 10) ORDER BY k, c`),
which the #26762 thread lists as working on `main`, also gets an extra sort.
Could the input record `[k, c]` instead? Among rows tied on `k`, `a = CAST(c
AS DATE)` implies `[c]` by the same argument as `implied_ordering`. With `[k,
c]` in the input's orderings, the projection would keep it after dropping `a`,
and no guard would be needed. If the guard stays, please add a plan test for
this shape.
##########
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
Review Comment:
This one already plans on `main`. If it's there to catch the failure that
the walk change alone would cause, could the comment say so, and could the test
check the plan? What it is guarding is where the sort ends up.
##########
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;
+----
+2026-01-01 2026-01-01T09:00:00 1
+2026-01-02 2026-01-02T01:00:00 2
+2026-01-02 2026-01-02T05:00:00 3
+2026-01-03 2026-01-03T01:00:00 4
+
+# The same with a per-partition constant leading the sort.
Review Comment:
`lower(k)` under `WHERE k = 'a'` is a uniform constant, so the TopK here is
just `[a, c]`, and this repeats the previous query. A `k` that really is
per-partition (for example the `GROUP BY lower(g)` shape above) would test the
case in my comment on `mod.rs`, ideally with an `EXPLAIN`.
##########
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`
Review Comment:
The comment says the result must not depend on which of these calls ran
last, but it still does when a `Heterogeneous` constant leads the ordering.
With `k` such a constant, `CAST(c AS DATE) = a`, and ordering `[k, a, c]`,
`ordering_satisfy([c])` returns:
- equality, ordering, then constant: `true`
- equality, constant, then ordering: `false`
- constant, ordering, then equality: `false`
`add_constants` drops `k` from the normal ordering, but
`normalize_sort_exprs` keeps it, so here `ordering[0]` is `k` and nothing is
implied. One way to fix it is to choose the head the same way `SortKeyWalk`
does, past the leading fixed expressions, both here and in
`discover_new_orderings`. A case like this would also belong in
`implied_ordering_does_not_depend_on_call_order`.
--
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]
