jayzhan211 commented on code in PR #24698:
URL: https://github.com/apache/datafusion/pull/24698#discussion_r4400910694
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -348,6 +366,13 @@ impl EquivalenceProperties {
self.oeq_cache.clear();
}
+ /// Clears grouping information invalidated by an operation that changes
Review Comment:
@xavlee thanks for the updates, both earlier points look good now. One
remaining gap: an operator that reorders rows and calls only
`clear_orderings()` still forwards its input's groupings. That's the same thing
`AggregateExec` hit in the previous revision, and it's the pattern custom
operators downstream already follow. Could `clear_orderings()` also clear
groupings? Then every operator that already drops orderings stays correct
without having to know about groupings:
```diff
pub fn clear_orderings(&mut self) {
self.oeq_class.clear();
self.oeq_cache.clear();
+ self.clear_groupings();
}
```
```diff
eq_properties.clear_orderings();
- if input_partitions > 1 {
- eq_properties.clear_groupings();
- }
eq_properties.clear_per_partition_constants();
```
I ran this locally. The only test that changes is the single-partition half
of `grouping_is_cleared_only_when_partitions_are_merged`, which now expects the
grouping to be cleared, the same as orderings. With this change
`aggregate_clears_explicit_grouping` no longer exercises the explicit clear in
`AggregateExec`, so I'd add a clustered case too. It passes with the explicit
clear and fails without it:
```rs
#[test]
fn clustered_aggregate_clears_explicit_grouping() -> Result<()> {
let schema = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, true),
Field::new("c", DataType::Int32, true),
Field::new("value", DataType::Int64, true),
]));
let a = col("a", &schema)?;
let c = col("c", &schema)?;
let input = TestMemoryExec::try_new(&[vec![]], Arc::clone(&schema),
None)?
.try_with_sort_information(vec![
[PhysicalSortExpr::new_default(Arc::clone(&a))].into(),
])?
.try_with_grouping_information(vec![vec![Arc::clone(&c)]])?;
let aggregate = AggregateExec::try_new(
AggregateMode::Single,
PhysicalGroupBy::new_single(vec![
(a, "group_a".to_string()),
(c, "group_c".to_string()),
]),
vec![Arc::new(
AggregateExprBuilder::new(sum_udaf(), vec![col("value",
&schema)?])
.schema(Arc::clone(&schema))
.alias("SUM(value)")
.build()?,
)],
vec![None],
Arc::new(input),
schema,
)?;
assert!(matches!(
aggregate.group_clustering_mode(),
GroupClusteringMode::Partial(_)
));
let properties = aggregate.properties().equivalence_properties();
assert!(!properties.oeq_class().is_empty());
assert!(properties.geq_class().is_empty());
assert!(!properties.grouping_satisfy([col("group_c",
&aggregate.schema())?])?);
Ok(())
}
```
If you'd rather keep `clear_orderings()` as it is, could you note in its doc
comment and in `add_groupings` that operators which reorder rows must also call
`clear_groupings()`? A follow-up PR is fine for either option.
--
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]
