viirya commented on code in PR #27191:
URL: https://github.com/apache/datafusion/pull/27191#discussion_r4504026523
##########
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 still happens on 17c73e6: the query plans a `SortExec` over the whole
scan. The new `implied_by` already stores `[c]` when the equality is at the
head. The same argument gives `[k, c]` when the equality comes after a
non-constant prefix: if `ordering[i] = f(ordering[i+1..])`, then
`[ordering[..i], ordering[i+1..]]` holds. Would storing that cover this shape,
so the guard isn't needed here? Or are there other shapes the guard is still
needed for?
##########
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:
Fixed by 9c7ee55, thanks. `implied_ordering_behind_a_leading_constant`
covers the three orders.
##########
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:
The plan check covers this now. The comment still says per-partition
constant, but `lower(k)` under `k = 'a'` is uniform. Could it be reworded?
`implied_ordering_behind_a_leading_constant` is what tests the per-partition
case.
--
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]
