fix: disable aggregate dynamic filter when unsupported expressions are present (#24816)

datafusion/core/tests/parquet/dynamic_row_group_pruning.rs

@@ -804,3 +804,67 @@ async fn topk_pushdown_does_not_reread_delivered_row_group() {
         "p4096 emitted more than once — rg_plan/decoder desync; got:\n{formatted}",
     );
 }
+
+/// Regression test for issue #24816: Aggregate dynamic filter must be disabled
+/// when an unsupported aggregate expression (e.g., `MIN(c + 1)`) is present alongside
+/// supported ones (`MIN(a)`, `MAX(a)`, `MAX(b)`).
+/// Otherwise, dynamic filter pushed to input scan prunes row groups required by the
+/// unsupported aggregate expression and produces incorrect results.
+#[tokio::test]
+async fn dynamic_rg_pruning_disabled_when_unsupported_aggregate_present() {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Int64, false),
+        Field::new("b", DataType::Int64, false),
+        Field::new("c", DataType::Int64, false),
+    ]));
+
+    // File with 2 row groups:
+    // RG 0: (1, 12, 100), (8, 4, 70)
+    // RG 1: (1, 6, 90), (8, 12, 110)
+    let batch_0 = RecordBatch::try_new(
+        Arc::clone(&schema),
+        vec![
+            Arc::new(Int64Array::from(vec![1, 8])) as ArrayRef,
+            Arc::new(Int64Array::from(vec![12, 4])) as ArrayRef,
+            Arc::new(Int64Array::from(vec![100, 70])) as ArrayRef,
+        ],
+    )
+    .unwrap();
+
+    let batch_1 = RecordBatch::try_new(
+        Arc::clone(&schema),
+        vec![
+            Arc::new(Int64Array::from(vec![1, 8])) as ArrayRef,
+            Arc::new(Int64Array::from(vec![6, 12])) as ArrayRef,
+            Arc::new(Int64Array::from(vec![90, 110])) as ArrayRef,
+        ],
+    )
+    .unwrap();
+
+    let mut ctx = ContextWithParquet::with_custom_data(
+        Scenario::Int,
+        RowGroup(2),
+        Arc::clone(&schema),
+        vec![batch_0, batch_1],
+    )
+    .await;
+
+    let output = ctx
+        .query("SELECT MIN(a), MAX(a), MAX(b), MIN(c + 1) FROM t")
+        .await;
+
+    assert_eq!(output.result_rows, 1, "query must return 1 aggregate row");
+    let formatted = output.pretty_results();
+    assert!(
+        formatted.contains("| 1 | 8 | 12 | 71 |"),
+        "expected output row '| 1 | 8 | 12 | 71 |'; got:\n{formatted}"
+    );
+
+    let pruned = output.row_groups_pruned_dynamic_filter().unwrap_or(0);
+    assert_eq!(
+        pruned,
+        0,
+        "dynamic filter should be disabled when unsupported aggregate is present; pruned={pruned}\n{}",
+        output.description()
+    );
+}

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

@@ -1852,6 +1852,8 @@ impl AggregateExec {
                     aggr_index: i,
                     shared_bound: Arc::new(Mutex::new(ScalarValue::Null)),
                 });
+            } else {
+                return;
             }
         }
 
@@ -8311,4 +8313,49 @@ mod tests {
         assert!(agg.set_dynamic_filter(df).is_err());
         Ok(())
     }
+
+    #[test]
+    fn test_dynamic_filter_disabled_when_unsupported_expr_present() -> Result<()> {
+        use datafusion_expr::Operator;
+        use datafusion_physical_expr::expressions::binary;
+
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
+        let child = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+
+        let col_a = col("a", &schema)?;
+        let col_a_plus_1 = binary(
+            Arc::clone(&col_a),
+            Operator::Plus,
+            lit(ScalarValue::Int64(Some(1))),
+            &schema,
+        )?;
+
+        // AggregateExec with MIN(a) (supported) and MIN(a + 1) (unsupported expr)
+        let agg = AggregateExec::try_new(
+            AggregateMode::Partial,
+            PhysicalGroupBy::new_single(vec![]),
+            vec![
+                Arc::new(
+                    AggregateExprBuilder::new(min_udaf(), vec![col_a])
+                        .schema(Arc::clone(&schema))
+                        .alias("min_a")
+                        .build()?,
+                ),
+                Arc::new(
+                    AggregateExprBuilder::new(min_udaf(), vec![col_a_plus_1])
+                        .schema(Arc::clone(&schema))
+                        .alias("min_a_plus_1")
+                        .build()?,
+                ),
+            ],
+            vec![None, None],
+            child,
+            Arc::clone(&schema),
+        )?;
+
+        // Dynamic filter must NOT be initialized because of the unsupported MIN(a + 1)
+        assert!(agg.dynamic_expressions_produced().is_empty());
+        assert!(agg.dynamic_filter.is_none());
+        Ok(())
+    }
 }