alamb commented on code in PR #27191:
URL: https://github.com/apache/datafusion/pull/27191#discussion_r4514816149
##########
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
Review Comment:
I found all this talk about head super confusing so I tried to document it
better
- https://github.com/apache/datafusion/pull/27266
##########
datafusion/physical-expr/src/equivalence/properties/walk.rs:
##########
@@ -276,6 +296,10 @@ impl<'a> SortKeyWalk<'a> {
if class.is_some_and(|class| class.constant.is_some()) {
return true;
}
+ // A column is only fixed if it is pinned or constant:
Review Comment:
isn't it more like a varying column can't be fixed (by definition?)
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1519,6 +1543,46 @@ mod tests {
assert!(matches!(remapped, OrderingRequirements::Soft(_)));
}
+ /// The guard compares the alternative that is pushed down, not the first
+ /// one, which may be an alternative that cannot be remapped. Here the
+ /// input is sorted by `c`, and the projection proves `score` (`c`
renamed),
+ /// but not `score % 2`, so the requirement can be pushed down.
+ #[test]
+ fn projection_guard_compares_the_pushed_alternative() -> Result<()> {
Review Comment:
what is this tet comparing? I am pretty confused
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -568,15 +582,35 @@ 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. If a
Review Comment:
It would help me here to write out the examples more clearly: add whitespace
between input and outputs, and explain them in more detail -- for example how I
have done on https://github.com/apache/datafusion/pull/27266
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -568,15 +582,35 @@ 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. If a
+ /// key is a function of the keys after it, as `a` is in `[k, a, c]` with
+ /// `a = CAST(c AS DATE)`, rows tied on the keys before it are sorted by
the
+ /// keys after it, so the ordering without that key holds: `[k, c]`. The
+ /// constants that a normal ordering can keep (those that differ between
+ /// partitions) are left out, because within a partition they have one
+ /// value.
+ fn implied_by(&self, ordering: &LexOrdering) ->
Result<Vec<Vec<PhysicalSortExpr>>> {
Review Comment:
I wonder why this doesn't return `Result<Vec<LexicalOrdering>>` which I
think is what the PhysicaSortExpr represent
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -568,13 +583,32 @@ 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())?);
+ new_orderings.extend(self.implied_by(ordering)?);
}
- if !new_orderings.is_empty() {
- self.add_orderings(new_orderings);
+ self.add_orderings(new_orderings)
+ }
+
+ /// Returns the orderings that the normal ordering `ordering` implies. If a
+ /// key is a function of the keys after it, as `a` is in `[k, a, c]` with
+ /// `a = CAST(c AS DATE)`, rows tied on the keys before it are sorted by
the
+ /// keys after it, so the ordering without that key holds: `[k, c]`. The
+ /// constants that a normal ordering can keep (those that differ between
+ /// partitions) are left out, because within a partition they have one
+ /// value.
+ fn implied_by(&self, ordering: &LexOrdering) ->
Result<Vec<Vec<PhysicalSortExpr>>> {
+ 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())?
Review Comment:
I also vote to return `ok()` and leave a comment explaining why the error is
ignored
This will make it less disruptive to backport
Then we can make a PR to main perhaps to propagate the error
--
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]
