fix: preserve aggregate filter pushdown order (#22926)

## Which issue does this PR close?

- Closes #22925.

## Rationale for this change

`FilterPushdown` maps each child pushdown result back to its original
parent
filter by position. `AggregateExec::gather_filters_for_pushdown`
previously
split parent filters into safe and unsafe buckets and then concatenated
the
results, which changed their order.

For example, in `cnt@2 = 1 AND b@1 = bar`, the grouping-column predicate
on
`b` can cross the aggregate, while the predicate on aggregate output
`cnt`
must remain above it. Reordering the returned results could make the
optimizer
associate the pushed-down `b` result with the `cnt` predicate and
incorrectly
remove the latter.

Aggregate pushdown also needs to account for empty-input semantics. A
global
aggregate, or a grouping-sets aggregate containing `()`, can emit a row
even
when its input is empty. Moving any parent predicate below such an
aggregate —
including a column-free predicate such as `false` — can therefore change
the
result.

## What changes are included in this PR?

- Preserve parent-filter order by constructing each child description
once
  with `ChildFilterDescription::from_child_with_allowed_indices`.
- Allow only grouping-output columns that are present in every grouping
set;
  aggregate-result columns remain above the aggregate.
- Mark all parent filters unsupported for global aggregates and grouping
sets
  containing an empty grouping set, while retaining aggregate-generated
  dynamic filters.
- Validate that every child returns one parent-filter result per input
filter
  before positional remapping.
- Add physical optimizer and SQL regression coverage for mixed filter
order,
global-aggregate name collisions, constant predicates, and grouping
sets.

## Are these changes tested?

Yes:

- `cargo test -p datafusion --test core_integration
physical_optimizer::filter_pushdown`
- `cargo fmt --all -- --check`
- `cargo check -p datafusion-physical-plan -p
datafusion-physical-optimizer`
- `git diff --check upstream/main`
- Full GitHub CI, including Rust tests, clippy, and sqllogictests

## Are there any user-facing changes?

There are no public API changes. Filter pushdown now preserves mixed
aggregate
predicates correctly and avoids moving predicates across aggregates when
doing
so could change empty-input results.

Author: discord9

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -41,8 +41,13 @@ 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_physical_expr::{LexOrdering, PhysicalSortExpr, expressions::col};
+use datafusion_functions_aggregate::{
+    count::count_udaf,
+    min_max::{max_udaf, min_udaf},
+};
+use datafusion_physical_expr::{
+    LexOrdering, PhysicalSortExpr, expressions::col, utils::conjunction,
+};
 use datafusion_physical_expr::{
     Partitioning, ScalarFunctionExpr, aggregate::AggregateExprBuilder,
 };
@@ -738,6 +743,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]
@@ -1753,6 +1817,16 @@ fn col_lit_predicate(
     ))
 }
 
+fn assert_parent_filter_remains_above_aggregate(plan: Arc<dyn ExecutionPlan>) {
+    let mut config = ConfigOptions::default();
+    config.execution.parquet.pushdown_filters = true;
+    let optimized = FilterPushdown::new().optimize(plan, &config).unwrap();
+    assert!(
+        optimized.downcast_ref::<FilterExec>().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)
@@ -1997,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::<FilterExec>().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())
@@ -2012,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]

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);
 

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,70 +2040,35 @@ 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();
-
-        // Analyze each filter separately to determine if it can be pushed down
-        let mut safe_filters = Vec::new();
-        let mut unsafe_filters = Vec::new();
-
-        for filter in parent_filters {
-            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 {
-                unsafe_filters.push(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<usize> = 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 {
-                    unsafe_filters.push(filter);
-                    continue;
-                }
-            }
-
-            // This filter is safe to push down
-            safe_filters.push(filter);
+        // 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<usize> =
+            (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));
         }
 
-        // 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),
-        );
+        // 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

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,