diff --git a/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs b/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs index 678aad9f5b1fd..e4f9cf3fa8297 100644 --- a/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs +++ b/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() + ); +} diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 375e392e6a089..568f69a24d076 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/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(()) + } }