From 1f9efe043a0e808df9e3eb89ba04582b5a38c22c Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 12 Jun 2026 16:09:59 +0800 Subject: [PATCH 1/6] fix: preserve aggregate filter pushdown order --- .../physical_optimizer/filter_pushdown.rs | 63 ++++++++++++++++++- .../physical-plan/src/aggregates/mod.rs | 32 ++++------ 2 files changed, 74 insertions(+), 21 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 909b80cadaae3..f49ed685117b7 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -42,7 +42,9 @@ use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_expr::ScalarUDF; use datafusion_functions::math::random::RandomFunc; use datafusion_functions_aggregate::{count::count_udaf, min_max::min_udaf}; -use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr, expressions::col}; +use datafusion_physical_expr::{ + LexOrdering, PhysicalSortExpr, expressions::col, utils::conjunction, +}; use datafusion_physical_expr::{ Partitioning, ScalarFunctionExpr, aggregate::AggregateExprBuilder, }; @@ -738,6 +740,65 @@ fn test_pushdown_through_aggregates_on_grouping_columns() { ); } +#[test] +fn test_pushdown_through_aggregates_preserves_parent_filter_order() { + // AggregateExec may push filters on grouping columns to its input, but must + // keep filters on aggregate outputs above itself. The parent-filter result + // order must match the incoming filter order, otherwise an unsupported + // aggregate-output filter can be reported as pushed down and removed. + let scan = TestScanBuilder::new(schema()).with_support(true).build(); + + let aggregate_expr = vec![ + AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema()).unwrap()]) + .schema(schema()) + .alias("cnt") + .build() + .map(Arc::new) + .unwrap(), + ]; + let group_by = PhysicalGroupBy::new_single(vec![ + (col("a", &schema()).unwrap(), "a".to_string()), + (col("b", &schema()).unwrap(), "b".to_string()), + ]); + let aggregate = Arc::new( + AggregateExec::try_new( + AggregateMode::Final, + group_by, + aggregate_expr, + vec![None], + scan, + schema(), + ) + .unwrap(), + ); + + let aggregate_schema = aggregate.schema(); + let aggregate_output_filter = col_lit_predicate( + "cnt", + ScalarValue::Int64(Some(1)), + aggregate_schema.as_ref(), + ); + let grouping_key_filter = col_lit_predicate("b", "bar", aggregate_schema.as_ref()); + let predicate = conjunction(vec![aggregate_output_filter, grouping_key_filter]); + let plan = Arc::new(FilterExec::try_new(predicate, aggregate).unwrap()); + + insta::assert_snapshot!( + OptimizationTest::new(plan, FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: cnt@2 = 1 AND b@1 = bar + - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + output: + Ok: + - FilterExec: cnt@2 = 1 + - AggregateExec: mode=Final, gby=[a@0 as a, b@1 as b], aggr=[cnt], ordering_mode=PartiallySorted([1]) + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=b@1 = bar + " + ); +} + /// Test various combinations of handling of child pushdown results /// in an ExecutionPlan in combination with support/not support in a DataSource. #[test] diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 787cd4f03ff6c..49b5dfafa949c 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -2052,11 +2052,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 + .into_iter() + .zip(child_desc.parent_filters.iter_mut()) + { let filter_columns: HashSet<_> = collect_columns(&filter).into_iter().collect(); @@ -2065,7 +2071,7 @@ impl ExecutionPlan for AggregateExec { && !filter_columns.is_subset(&grouping_columns); if references_non_grouping { - unsafe_filters.push(filter); + *child_parent_filter = PushedDownPredicate::unsupported(filter); continue; } @@ -2086,26 +2092,12 @@ impl ExecutionPlan for AggregateExec { }); if has_missing_column { - unsafe_filters.push(filter); + *child_parent_filter = PushedDownPredicate::unsupported(filter); continue; } } - - // This filter is safe to push down - safe_filters.push(filter); } - // Build child filter description with both safe and unsafe filters - let child = self.children()[0]; - let mut child_desc = ChildFilterDescription::from_child(&safe_filters, child)?; - - // Add unsafe filters as unsupported - child_desc.parent_filters.extend( - unsafe_filters - .into_iter() - .map(PushedDownPredicate::unsupported), - ); - // Include self dynamic filter when it's possible if phase == FilterPushdownPhase::Post && config.optimizer.enable_aggregate_dynamic_filter_pushdown From 474e8c5cd74378b49a9cc4538ffa9861a4b3a593 Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 12 Jun 2026 16:35:37 +0800 Subject: [PATCH 2/6] test: add aggregate filter pushdown order slt --- .../push_down_filter_regression.slt | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt index 7ab5e7c79d2ba..57509fd0395b9 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt @@ -515,6 +515,44 @@ physical_plan 05)--------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[count(agg_filter_pushdown.b)] 06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_filter_pushdown.parquet]]}, projection=[a, b], file_type=parquet +# Mixed filters on an aggregate output and a grouping column must preserve their +# parent filter result order. The grouping-column filter can push below the +# aggregate, but the aggregate-output filter must remain above it. +# Disable logical optimizer passes for this regression so the logical filter +# pushdown rule does not split the mixed predicate before the physical +# `AggregateExec::gather_filters_for_pushdown` path sees it. +statement ok +set datafusion.optimizer.max_passes = 0; + +query TT +EXPLAIN SELECT a, b, cnt FROM ( + SELECT a, b, count(b) AS cnt + FROM agg_filter_pushdown + GROUP BY a, b +) q WHERE cnt = 2 AND b = 'foo'; +---- +physical_plan +01)FilterExec: cnt@2 = 2 +02)--ProjectionExec: expr=[a@0 as a, b@1 as b, count(agg_filter_pushdown.b)@2 as cnt] +03)----AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b], aggr=[count(agg_filter_pushdown.b)] +04)------RepartitionExec: partitioning=Hash([a@0, b@1], 4), input_partitions=4 +05)--------AggregateExec: mode=Partial, gby=[a@0 as a, b@1 as b], aggr=[count(agg_filter_pushdown.b)] +06)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +07)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_filter_pushdown.parquet]]}, projection=[a, b], file_type=parquet, predicate=b@1 = CAST(foo AS Utf8View), pruning_predicate=b_null_count@2 != row_count@3 AND b_min@0 <= foo AND foo <= b_max@1, required_guarantees=[] + +# If the aggregate-output filter is incorrectly removed, this query returns 1. +query I +SELECT count(*) FROM ( + SELECT a, b, count(b) AS cnt + FROM agg_filter_pushdown + GROUP BY a, b +) q WHERE cnt = 2 AND b = 'foo'; +---- +0 + +statement ok +reset datafusion.optimizer.max_passes; + statement ok drop table agg_filter_pushdown; From 553a89da4daa11b24cf98f88bccb80230ae90ea5 Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 12 Jun 2026 19:13:11 +0800 Subject: [PATCH 3/6] docs: clarify filter pushdown parent order --- datafusion/physical-plan/src/execution_plan.rs | 6 ++++++ datafusion/physical-plan/src/filter_pushdown.rs | 3 +++ 2 files changed, 9 insertions(+) diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs index 3b9d5d258a838..11a8d69a37669 100644 --- a/datafusion/physical-plan/src/execution_plan.rs +++ b/datafusion/physical-plan/src/execution_plan.rs @@ -664,6 +664,12 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync { /// There are two different phases in filter pushdown, which some operators may handle the same and some differently. /// Depending on the phase the operator may or may not be allowed to modify the plan. /// See [`FilterPushdownPhase`] for more details. + /// + /// Implementations must preserve the order of `parent_filters` in the + /// returned child [`FilterDescription`]: each child parent-filter result is + /// matched back to the corresponding input parent filter by position. + /// Unsupported filters should therefore be marked unsupported in place, + /// rather than removed or appended after supported filters. fn gather_filters_for_pushdown( &self, _phase: FilterPushdownPhase, diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index 810f9ffcbcdb1..382967c7ee1ef 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -302,6 +302,9 @@ pub struct ChildFilterDescription { /// Description of which parent filters can be pushed down into this node. /// Since we need to transmit filter pushdown results back to this node's parent /// we need to track each parent filter for each child, even those that are unsupported / won't be pushed down. + /// The entries must stay in the same order as the input parent filters: the + /// filter pushdown optimizer maps child results back to parent filters by + /// position. pub(crate) parent_filters: Vec, /// Description of which filters this node is pushing down to its children. /// Since this is not transmitted back to the parents we can have variable sized inner arrays From 48cf05c4d89934e5ad8b6bd5d70b767e6ed752cd Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 29 Jul 2026 11:32:04 +0800 Subject: [PATCH 4/6] fix: validate aggregate filter pushdown descriptors --- .../physical-optimizer/src/filter_pushdown.rs | 8 ++ .../physical-plan/src/aggregates/mod.rs | 74 +++++-------------- 2 files changed, 27 insertions(+), 55 deletions(-) diff --git a/datafusion/physical-optimizer/src/filter_pushdown.rs b/datafusion/physical-optimizer/src/filter_pushdown.rs index 28f8155002a50..06aa632a9d3f3 100644 --- a/datafusion/physical-optimizer/src/filter_pushdown.rs +++ b/datafusion/physical-optimizer/src/filter_pushdown.rs @@ -486,6 +486,14 @@ 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); diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 49b5dfafa949c..b981b832794d2 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -159,7 +159,7 @@ use crate::aggregates::{ use crate::execution_plan::{CardinalityEffect, EmissionType}; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, - FilterPushdownPropagation, PushedDownPredicate, + FilterPushdownPropagation, }; use crate::metrics::{ExecutionPlanMetricsSet, MetricsSet}; use crate::statistics::{ChildStats, StatisticsArgs}; @@ -168,7 +168,6 @@ use crate::{ InputOrderMode, SendableRecordBatchStream, Statistics, check_if_same_properties, }; use datafusion_common::config::ConfigOptions; -use datafusion_physical_expr::utils::collect_columns; use parking_lot::Mutex; use std::collections::{HashMap, HashSet}; @@ -2041,62 +2040,27 @@ impl ExecutionPlan for AggregateExec { // This optimization is NOT safe for filters on aggregated columns (like filtering on // the result of SUM or COUNT), as those require computing all groups first. - // Build grouping columns using output indices because parent filters reference the - // AggregateExec's output schema where grouping columns in the output schema. The - // grouping expressions reference input columns which may not match the output schema. - // - // It is safe to assume that the output_schema contains group by columns in the same order - // as the group by expression. See [`create_schema`] and [`AggregateExec`]. - let output_schema = self.schema(); - let grouping_columns: HashSet<_> = (0..self.group_by.expr().len()) - .map(|i| Column::new(output_schema.field(i).name(), i)) - .collect(); + // Grouping columns are output before aggregate columns, in the same order + // as the grouping expressions. A grouping-set null mask marks grouping + // columns that are not available in that set. + let mut allowed_indices: HashSet = + (0..self.group_by.expr().len()).collect(); + for null_mask in self.group_by.groups() { + allowed_indices.retain(|idx| null_mask.get(*idx) != Some(&true)); + } - // 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, child_parent_filter) in parent_filters - .into_iter() - .zip(child_desc.parent_filters.iter_mut()) - { - let filter_columns: HashSet<_> = - collect_columns(&filter).into_iter().collect(); - - // Check if this filter references non-grouping columns - let references_non_grouping = !grouping_columns.is_empty() - && !filter_columns.is_subset(&grouping_columns); - - if references_non_grouping { - *child_parent_filter = PushedDownPredicate::unsupported(filter); - continue; - } - - // For GROUPING SETS, verify this filter's columns appear in all grouping sets - if self.group_by.groups().len() > 1 { - let filter_column_indices: Vec = filter_columns - .iter() - .filter_map(|filter_col| { - grouping_columns.get(filter_col).map(|col| col.index()) - }) - .collect(); - - // Check if any of this filter's columns are missing from any grouping set - let has_missing_column = self.group_by.groups().iter().any(|null_mask| { - filter_column_indices - .iter() - .any(|&idx| null_mask.get(idx) == Some(&true)) - }); - - if has_missing_column { - *child_parent_filter = PushedDownPredicate::unsupported(filter); - continue; - } - } + if self.group_by.expr().is_empty() { + // Preserve the existing global-aggregate behavior: filters that + // reference input columns may still be pushed down when their + // column names can be remapped to the child schema. + allowed_indices.extend(0..child.schema().fields().len()); } + let mut child_desc = ChildFilterDescription::from_child_with_allowed_indices( + &parent_filters, + allowed_indices, + child, + )?; // Include self dynamic filter when it's possible if phase == FilterPushdownPhase::Post From 4b0e6c6c352b88fc079b42937a4ecd3531852ae2 Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 29 Jul 2026 11:53:09 +0800 Subject: [PATCH 5/6] fix: prevent filter pushdown through empty aggregates --- .../physical_optimizer/filter_pushdown.rs | 136 ++++++++++++++---- .../physical-plan/src/aggregates/mod.rs | 30 ++-- 2 files changed, 128 insertions(+), 38 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index f49ed685117b7..80b3f37b74c7f 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -41,7 +41,10 @@ use datafusion_datasource::{ use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_expr::ScalarUDF; use datafusion_functions::math::random::RandomFunc; -use datafusion_functions_aggregate::{count::count_udaf, min_max::min_udaf}; +use datafusion_functions_aggregate::{ + count::count_udaf, + min_max::{max_udaf, min_udaf}, +}; use datafusion_physical_expr::{ LexOrdering, PhysicalSortExpr, expressions::col, utils::conjunction, }; @@ -1814,6 +1817,16 @@ fn col_lit_predicate( )) } +fn assert_parent_filter_remains_above_aggregate(plan: Arc) { + let optimized = FilterPushdown::new() + .optimize(plan, &ConfigOptions::default()) + .unwrap(); + assert!( + optimized.downcast_ref::().is_some(), + "parent filter must remain above aggregate" + ); +} + // ==== Aggregate Dynamic Filter tests ==== // // The end-to-end min/max dynamic filter cases (simple/min/max/mixed/all-nulls) @@ -2058,13 +2071,65 @@ fn test_pushdown_grouping_sets_filter_on_common_column() { ); } +#[tokio::test] +async fn test_no_pushdown_through_global_aggregate_with_name_collision() { + let input_schema = + Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)])); + let scan = TestScanBuilder::new(Arc::clone(&input_schema)) + .with_support(true) + .with_batches(vec![record_batch!(("a", Int64, [1, 20])).unwrap()]) + .build(); + let aggregate_expr = vec![ + AggregateExprBuilder::new(max_udaf(), vec![col("a", &input_schema).unwrap()]) + .schema(Arc::clone(&input_schema)) + .alias("a") + .build() + .map(Arc::new) + .unwrap(), + ]; + let aggregate = Arc::new( + AggregateExec::try_new( + AggregateMode::Single, + PhysicalGroupBy::new_single(vec![]), + aggregate_expr, + vec![None], + scan, + input_schema, + ) + .unwrap(), + ); + + // This is a physical filter above the aggregate, not a SQL WHERE clause. + // Pushing it through would evaluate input `a` instead of MAX(a). + let predicate = Arc::new(BinaryExpr::new( + col("a", aggregate.schema().as_ref()).unwrap(), + Operator::Lt, + Arc::new(Literal::new(ScalarValue::Int64(Some(10)))), + )); + let plan = Arc::new(FilterExec::try_new(predicate, aggregate).unwrap()); + + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + let optimized = FilterPushdown::new().optimize(plan, &config).unwrap(); + assert!(optimized.downcast_ref::().is_some()); + + let session_ctx = SessionContext::new(); + session_ctx.register_object_store( + ObjectStoreUrl::parse("test://").unwrap().as_ref(), + Arc::new(InMemory::new()), + ); + let batches = collect(optimized, session_ctx.state().task_ctx()) + .await + .unwrap(); + assert!( + batches.is_empty(), + "MAX(a) = 20 must be filtered out instead of applying a < 10 to input rows" + ); +} + #[test] -fn test_pushdown_with_empty_group_by() { - // Test that filters can be pushed down when GROUP BY is empty (no grouping columns) - // SELECT count(*) as cnt FROM table WHERE a = 'foo' - // There are no grouping columns, so the filter should still push down +fn test_no_pushdown_constant_false_through_global_aggregate() { let scan = TestScanBuilder::new(schema()).with_support(true).build(); - let aggregate_expr = vec![ AggregateExprBuilder::new(count_udaf(), vec![col("c", &schema()).unwrap()]) .schema(schema()) @@ -2073,41 +2138,58 @@ fn test_pushdown_with_empty_group_by() { .map(Arc::new) .unwrap(), ]; + let aggregate = Arc::new( + AggregateExec::try_new( + AggregateMode::Final, + PhysicalGroupBy::new_single(vec![]), + aggregate_expr, + vec![None], + scan, + schema(), + ) + .unwrap(), + ); + let predicate = Arc::new(Literal::new(ScalarValue::Boolean(Some(false)))); + let plan = Arc::new(FilterExec::try_new(predicate, aggregate).unwrap()); - // Empty GROUP BY - no grouping columns - let group_by = PhysicalGroupBy::new_single(vec![]); + assert_parent_filter_remains_above_aggregate(plan); +} +#[test] +fn test_no_pushdown_constant_false_through_empty_grouping_set() { + let scan = TestScanBuilder::new(schema()).with_support(true).build(); + let group_by = PhysicalGroupBy::new( + vec![(col("a", &schema()).unwrap(), "a".to_string())], + vec![( + Arc::new(Literal::new(ScalarValue::Utf8(None))), + "a".to_string(), + )], + vec![vec![true]], + true, + ); + let aggregate_expr = vec![ + AggregateExprBuilder::new(count_udaf(), vec![col("c", &schema()).unwrap()]) + .schema(schema()) + .alias("cnt") + .build() + .map(Arc::new) + .unwrap(), + ]; let aggregate = Arc::new( AggregateExec::try_new( AggregateMode::Final, group_by, - aggregate_expr.clone(), + aggregate_expr, vec![None], scan, schema(), ) .unwrap(), ); - - // Filter on 'a' - let predicate = col_lit_predicate("a", "foo", &schema()); + let predicate = Arc::new(Literal::new(ScalarValue::Boolean(Some(false)))); let plan = Arc::new(FilterExec::try_new(predicate, aggregate).unwrap()); - // The filter should be pushed down even with empty GROUP BY - insta::assert_snapshot!( - OptimizationTest::new(plan, FilterPushdown::new(), true), - @r" - OptimizationTest: - input: - - FilterExec: a@0 = foo - - AggregateExec: mode=Final, gby=[], aggr=[cnt] - - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true - output: - Ok: - - AggregateExec: mode=Final, gby=[], aggr=[cnt] - - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo - " - ); + assert_parent_filter_remains_above_aggregate(plan); } #[test] diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index b981b832794d2..370c75961608c 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -2050,17 +2050,25 @@ impl ExecutionPlan for AggregateExec { } let child = self.children()[0]; - if self.group_by.expr().is_empty() { - // Preserve the existing global-aggregate behavior: filters that - // reference input columns may still be pushed down when their - // column names can be remapped to the child schema. - allowed_indices.extend(0..child.schema().fields().len()); - } - let mut child_desc = ChildFilterDescription::from_child_with_allowed_indices( - &parent_filters, - allowed_indices, - child, - )?; + // Global aggregates and grouping sets containing an empty grouping set + // emit a row even when their input is empty. Parent filters therefore + // cannot be pushed below them, including filters without column + // references. + let may_emit_on_empty_input = self.group_by.is_true_no_grouping() + || self + .group_by + .groups() + .iter() + .any(|null_mask| null_mask.iter().all(|is_null| *is_null)); + let mut child_desc = if may_emit_on_empty_input { + ChildFilterDescription::all_unsupported(&parent_filters) + } else { + ChildFilterDescription::from_child_with_allowed_indices( + &parent_filters, + allowed_indices, + child, + )? + }; // Include self dynamic filter when it's possible if phase == FilterPushdownPhase::Post From 4e9f319aadd92e2d8771dcbf96db31567b787931 Mon Sep 17 00:00:00 2001 From: discord9 Date: Thu, 30 Jul 2026 02:11:44 +0800 Subject: [PATCH 6/6] test: enable aggregate filter pushdown regressions --- datafusion/core/tests/physical_optimizer/filter_pushdown.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 80b3f37b74c7f..7593fe351548e 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -1818,9 +1818,9 @@ fn col_lit_predicate( } fn assert_parent_filter_remains_above_aggregate(plan: Arc) { - let optimized = FilterPushdown::new() - .optimize(plan, &ConfigOptions::default()) - .unwrap(); + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + let optimized = FilterPushdown::new().optimize(plan, &config).unwrap(); assert!( optimized.downcast_ref::().is_some(), "parent filter must remain above aggregate"