wip

Author: jayshrivastava

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -3147,10 +3147,6 @@ fn test_pushdown_with_empty_group_by() {
 
 #[test]
 fn test_pushdown_through_aggregate_with_reordered_input_columns() {
-    // Test filter pushdown through aggregate when a ProjectionExec below the
-    // aggregate reorders columns, causing input indices to differ from output indices.
-    // This reproduces the bug where grouping_columns were built from input schema
-    // positions rather than output schema positions.
     let scan = TestScanBuilder::new(schema()).with_support(true).build();
 
     // Reorder scan output from (a, b, c) to (c, a, b)

datafusion/physical-plan/src/aggregates/mod.rs

@@ -1479,8 +1479,7 @@ impl ExecutionPlan for AggregateExec {
         // INPUT columns which may have different indices (e.g., when an intermediate
         // ProjectionExec reorders columns). We must compare in the same index space.
         let output_schema = self.schema();
-        let num_grouping_cols = self.group_by.expr().len();
-        let grouping_columns: HashSet<_> = (0..num_grouping_cols)
+        let grouping_columns: HashSet<_> = (0..self.group_by.expr().len())
             .map(|i| Column::new(output_schema.field(i).name(), i))
             .collect();
 
@@ -1503,18 +1502,10 @@ impl ExecutionPlan for AggregateExec {
 
             // For GROUPING SETS, verify this filter's columns appear in all grouping sets
             if self.group_by.groups().len() > 1 {
-                // Map filter columns to their grouping expression index via output position
                 let filter_column_indices: Vec<usize> = filter_columns
                     .iter()
                     .filter_map(|filter_col| {
-                        if filter_col.index() < num_grouping_cols
-                            && output_schema.field(filter_col.index()).name()
-                                == filter_col.name()
-                        {
-                            Some(filter_col.index())
-                        } else {
-                            None
-                        }
+                        grouping_columns.get(filter_col).map(|col| col.index())
                     })
                     .collect();