kumarUjjawal commented on code in PR #27191:
URL: https://github.com/apache/datafusion/pull/27191#discussion_r4533792958
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -391,14 +392,27 @@ impl EquivalenceProperties {
.cloned()
.filter_map(|o| self.normalize_sort_exprs(o))
.collect();
- if !normal_orderings.is_empty() {
- self.oeq_class.extend(orderings);
- // Normalize given orderings to update the cache:
- self.oeq_cache.normal_cls.extend(normal_orderings);
- // TODO: If no ordering is found to be redundant during extension,
we
- // can use a shortcut algorithm to update the leading map.
- self.oeq_cache.update_map();
+ if normal_orderings.is_empty() {
+ return;
}
+ // `add_constants` and `add_equal_conditions` discover implied
orderings
+ // too, and whether an ordering is satisfied must not depend on which
of
+ // them ran last. Deriving one fails only on an expression that does
not
+ // match the schema; that ordering is then not implied, which can cost
a
+ // sort but not a wrong result.
+ let mut implied = vec![];
+ for ordering in &normal_orderings {
+ implied.extend(self.implied_by(ordering).unwrap_or_default());
Review Comment:
Should we clear repartition constants before deriving merge orderings?
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -568,15 +582,74 @@ impl EquivalenceProperties {
let mut new_orderings = vec![];
for idx in ordering_idxs {
let ordering = &self.oeq_cache.normal_cls[*idx];
- new_orderings
- .extend(self.implied_ordering(&ordering[0],
ordering[1..].iter())?);
- }
- if !new_orderings.is_empty() {
- self.add_orderings(new_orderings);
+ new_orderings.extend(self.implied_by(ordering)?);
}
+ self.add_orderings(new_orderings);
Ok(())
}
+ /// Returns the orderings that the normal ordering `ordering` implies
+ /// through known equalities: for each key that is a function of the keys
+ /// after it (see [`Self::implied_ordering`]), `ordering` without that key.
+ ///
+ /// # Example: a key that is a function of the keys after it
+ ///
+ /// * Ordering: `[k ASC, a ASC, c ASC]`
+ /// * Equality: `a = CAST(c AS DATE)`
+ /// * Returns: `[k ASC, c ASC]`
+ ///
+ /// Among the rows that tie on `k`, the rows are sorted by `a`, and the
+ /// rows that also tie on `a` are sorted by `c`. `a` is the date of `c`, so
+ /// a row with an earlier `a` also has an earlier `c`, and the rows that
+ /// tie on `k` are sorted by `c` alone:
+ ///
+ /// ```text
+ /// ┌───┬────────────┬──────────────────┐
+ /// │ k │ a │ c │
+ /// ├───┼────────────┼──────────────────┤
+ /// │ x │ 2026-01-01 │ 2026-01-01 09:00 │
+ /// │ x │ 2026-01-02 │ 2026-01-02 01:00 │
+ /// │ x │ 2026-01-02 │ 2026-01-02 05:00 │
+ /// │ y │ 2026-01-01 │ 2026-01-01 03:00 │
+ /// └───┴────────────┴──────────────────┘
+ /// ```
+ ///
+ /// # Example: a constant that differs between partitions
+ ///
+ /// * Ordering: `[k ASC, a ASC, c ASC]`, where `k` is a constant whose
value
+ /// differs between partitions
+ /// * Equality: `a = CAST(c AS DATE)`
+ /// * Returns: `[c ASC]`
+ ///
+ /// A normal ordering can keep such a constant (see
+ /// `EquivalenceGroup::normalize_sort_exprs`). Within a partition `k` has
+ /// one value, so it is left out of the keys and of the result.
+ ///
+ /// # Example: directions that differ
+ ///
+ /// * Ordering: `[a ASC, c DESC]`
+ /// * Equality: `a = CAST(c AS DATE)`
+ /// * Returns: nothing
+ ///
+ /// `c` descends within a date but ascends across dates, so the rows are
+ /// not sorted by `c` in either direction.
+ fn implied_by(&self, ordering: &LexOrdering) -> Result<Vec<LexOrdering>> {
+ let keys: Vec<_> = ordering
+ .iter()
+ .filter(|sort_expr|
self.eq_group.is_expr_constant(&sort_expr.expr).is_none())
+ .collect();
+ let mut implied = vec![];
+ for (idx, key) in keys.iter().enumerate() {
+ if let Some(rest) =
+ self.implied_ordering(key, keys[idx + 1..].iter().copied())?
+ {
+ let prefix = keys[..idx].iter().copied().cloned();
Review Comment:
IMO the Simpler fix would be: keep the constants in the stored implied
ordering (store `[k, c]`, not `[c]`). `SortKeyWalk::new` already skips a
leading fixed head, so `[c]` still holds within a partition, and clearing `k`
then removes it correctly.
wdyt?
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -568,15 +582,74 @@ impl EquivalenceProperties {
let mut new_orderings = vec![];
for idx in ordering_idxs {
let ordering = &self.oeq_cache.normal_cls[*idx];
- new_orderings
- .extend(self.implied_ordering(&ordering[0],
ordering[1..].iter())?);
- }
- if !new_orderings.is_empty() {
- self.add_orderings(new_orderings);
+ new_orderings.extend(self.implied_by(ordering)?);
Review Comment:
`discover_new_orderings` still passes `implied_by` errors up with `?`, while
`add_orderings` ignores them. `implied_by` now checks every key, not just
`ordering[0]`, so whether the call fails still depends on the order of calls.
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -391,14 +392,27 @@ impl EquivalenceProperties {
.cloned()
.filter_map(|o| self.normalize_sort_exprs(o))
.collect();
- if !normal_orderings.is_empty() {
- self.oeq_class.extend(orderings);
- // Normalize given orderings to update the cache:
- self.oeq_cache.normal_cls.extend(normal_orderings);
- // TODO: If no ordering is found to be redundant during extension,
we
- // can use a shortcut algorithm to update the leading map.
- self.oeq_cache.update_map();
+ if normal_orderings.is_empty() {
+ return;
}
+ // `add_constants` and `add_equal_conditions` discover implied
orderings
+ // too, and whether an ordering is satisfied must not depend on which
of
+ // them ran last. Deriving one fails only on an expression that does
not
+ // match the schema; that ordering is then not implied, which can cost
a
+ // sort but not a wrong result.
+ let mut implied = vec![];
+ for ordering in &normal_orderings {
+ implied.extend(self.implied_by(ordering).unwrap_or_default());
+ }
+ self.oeq_class.extend(orderings);
+ // Normalize given orderings to update the cache:
+ self.oeq_cache.normal_cls.extend(normal_orderings);
+ // TODO: If no ordering is found to be redundant during extension, we
+ // can use a shortcut algorithm to update the leading map.
+ self.oeq_cache.update_map();
+ // Each implied ordering is shorter than the one implying it, so this
+ // terminates:
+ self.add_orderings(implied);
Review Comment:
can we use a deduplicated worklist and avoid expanding already-known
candidates?
##########
datafusion/sqllogictest/test_files/window.slt:
##########
@@ -7834,3 +7834,277 @@ 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
Review Comment:
The new 4-partition case (`SELECT SUM(v) AS k, COUNT(*) AS c ...`) only
checks result rows, with no EXPLAIN. It cannot show whether a sort was avoided
or added.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1318,10 +1318,34 @@ 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. Alternatives that cannot be remapped are not
+ // pushed down, so compare against the first one that can.
+ let pushed = parent_required.iter().find(|required| {
Review Comment:
should we walk the alternatives once and keep the first `(original,
remapped)` pair, or have the remap helper return it.
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
Review Comment:
https://github.com/apache/datafusion/blob/28a60df75026ea4ba57e3827841a05f537340503/datafusion/physical-expr/src/equivalence/properties/mod.rs#L276
The walk does not derive it because nothing is pinned. This is the order
dependence the new comment says must not happen.
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -391,14 +392,27 @@ impl EquivalenceProperties {
.cloned()
.filter_map(|o| self.normalize_sort_exprs(o))
.collect();
- if !normal_orderings.is_empty() {
- self.oeq_class.extend(orderings);
- // Normalize given orderings to update the cache:
- self.oeq_cache.normal_cls.extend(normal_orderings);
- // TODO: If no ordering is found to be redundant during extension,
we
- // can use a shortcut algorithm to update the leading map.
- self.oeq_cache.update_map();
+ if normal_orderings.is_empty() {
+ return;
}
+ // `add_constants` and `add_equal_conditions` discover implied
orderings
+ // too, and whether an ordering is satisfied must not depend on which
of
+ // them ran last. Deriving one fails only on an expression that does
not
+ // match the schema; that ordering is then not implied, which can cost
a
+ // sort but not a wrong result.
+ let mut implied = vec![];
+ for ordering in &normal_orderings {
+ implied.extend(self.implied_by(ordering).unwrap_or_default());
+ }
+ self.oeq_class.extend(orderings);
+ // Normalize given orderings to update the cache:
+ self.oeq_cache.normal_cls.extend(normal_orderings);
+ // TODO: If no ordering is found to be redundant during extension, we
+ // can use a shortcut algorithm to update the leading map.
+ self.oeq_cache.update_map();
+ // Each implied ordering is shorter than the one implying it, so this
+ // terminates:
+ self.add_orderings(implied);
Review Comment:
`add_orderings` passes `implied` back to itself without removing duplicates.
Each level also runs `remove_redundant_entries` (O(n²)) and `update_map` over
the whole class, so the work can grow with the number of key subsets.
##########
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` repeats the head-moving logic in `pin` (lines 177-180:
move `head_idx` forward while `is_fixed`) as a separate `position` scan, so the
rule for picking a head now lives in two places.
--
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]
