jayzhan211 commented on code in PR #27191:
URL: https://github.com/apache/datafusion/pull/27191#discussion_r4538436740
##########
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:
Each round's implied orderings are now deduplicated before they are
expanded. I didn't skip candidates that are already in the class:
`remove_redundant_entries` drops the repeats anyway, and skipping them would
rely on every earlier call having derived their implied 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:
I'd keep `[c]` in this PR. As viirya says, the `RepartitionExec` call order
needs fixing either way, and that fix is now in the PR. With it,
`SortPreservingMergeExec`, `RepartitionExec` and `CoalescePartitionsExec` all
drop `[c]` when they merge partitions.
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
Review Comment:
I couldn't get an order dependence from `new_with_orderings`. It starts with
no equalities, and `add_equal_conditions` and `add_constants` both re-derive
the implied orderings of every stored ordering. With `k` a per-partition
constant and `a = CAST(c AS DATE)`, `ordering_satisfy([c])` is true for `[a,
c]` and for `[k, a, c]`, whichever of the equality and the constant is added
first, and the same as with `add_ordering`. Which sequence of calls do you have
in mind?
--
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]
