neilconway commented on code in PR #22926:
URL: https://github.com/apache/datafusion/pull/22926#discussion_r3669054411
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -2067,11 +2067,17 @@ impl ExecutionPlan for AggregateExec {
.map(|i| Column::new(output_schema.field(i).name(), i))
.collect();
- // Analyze each filter separately to determine if it can be pushed down
- let mut safe_filters = Vec::new();
- let mut unsafe_filters = Vec::new();
+ // Analyze each filter separately to determine if it can be pushed
down.
+ // The order of `parent_filters` must be preserved: the filter pushdown
+ // optimizer maps child results back to the original parent filters by
+ // position.
+ let child = self.children()[0];
+ let mut child_desc =
ChildFilterDescription::from_child(&parent_filters, child)?;
- for filter in parent_filters {
+ for (filter, child_parent_filter) in parent_filters
Review Comment:
Do you think it's worth testing `debug_assert_eq!(parent_filters.len(),
child_desc.parent_filters.len());` before we do the zip loop? Or more broadly,
I wonder if there are places we can add asserts to make the invariants this
code expects more clearly manifest.
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -2067,11 +2067,17 @@ impl ExecutionPlan for AggregateExec {
.map(|i| Column::new(output_schema.field(i).name(), i))
.collect();
- // Analyze each filter separately to determine if it can be pushed down
- let mut safe_filters = Vec::new();
- let mut unsafe_filters = Vec::new();
+ // Analyze each filter separately to determine if it can be pushed
down.
+ // The order of `parent_filters` must be preserved: the filter pushdown
+ // optimizer maps child results back to the original parent filters by
+ // position.
+ let child = self.children()[0];
+ let mut child_desc =
ChildFilterDescription::from_child(&parent_filters, child)?;
- for filter in parent_filters {
+ for (filter, child_parent_filter) in parent_filters
Review Comment:
e.g., this could also be useful
```
--- a/datafusion/physical-optimizer/src/filter_pushdown.rs
+++ b/datafusion/physical-optimizer/src/filter_pushdown.rs
@@ -486,6 +486,18 @@ fn push_down_filters(
// currently. `self_filters` are the predicates which are provided
by the current node,
// and tried to be pushed down over the child similarly.
+ assert_eq_or_internal_err!(
+ parent_filters.len(),
+ parent_filtered.len(),
+ "Filter pushdown expected {} to return one parent filter result
per input filter for child {}",
+ node.name(),
+ child_idx
+ );
+
// Filter out self_filters that contain volatile expressions and
track indices
let self_filtered = FilteredVec::new(&self_filters,
allow_pushdown_for_expr);
```
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -2067,11 +2067,17 @@ impl ExecutionPlan for AggregateExec {
.map(|i| Column::new(output_schema.field(i).name(), i))
.collect();
- // Analyze each filter separately to determine if it can be pushed down
- let mut safe_filters = Vec::new();
- let mut unsafe_filters = Vec::new();
+ // Analyze each filter separately to determine if it can be pushed
down.
+ // The order of `parent_filters` must be preserved: the filter pushdown
+ // optimizer maps child results back to the original parent filters by
+ // position.
+ let child = self.children()[0];
+ let mut child_desc =
ChildFilterDescription::from_child(&parent_filters, child)?;
Review Comment:
What do you think about doing `from_child_with_allowed_indices` instead?
That would avoid the need to mutate in-place the output of `from_child` below.
--
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]
