From edc502d2aa5db6be13973130e060c22bd1ea170d Mon Sep 17 00:00:00 2001 From: HairstonE Date: Thu, 23 Jul 2026 19:50:53 -0400 Subject: [PATCH 1/2] fix: add flag that marks a global agg standing in for an all constant GROUP BY so we can emit no rows when the input is empty --- datafusion/core/src/physical_planner.rs | 143 ++++++++++--- .../aggregate_statistics.rs | 136 ++++++++++++ .../src/eliminate_group_by_constant.rs | 13 +- .../src/aggregate_statistics.rs | 13 ++ .../src/combine_partial_final_agg.rs | 6 +- .../src/aggregates/aggregate_stream.rs | 13 ++ .../physical-plan/src/aggregates/mod.rs | 200 ++++++++++++++++++ .../proto-models/proto/datafusion.proto | 2 + .../proto-models/src/generated/pbjson.rs | 18 ++ .../proto-models/src/generated/prost.rs | 3 + datafusion/proto/src/physical_plan/mod.rs | 4 + .../tests/cases/roundtrip_physical_plan.rs | 40 ++++ .../sqllogictest/test_files/aggregate.slt | 32 +++ .../optimizer_group_by_constant.slt | 60 +++++- 14 files changed, 640 insertions(+), 43 deletions(-) diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index aef8036c749a8..27b195bdee3ae 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -99,7 +99,7 @@ use datafusion_expr::{ use datafusion_physical_expr::aggregate::{ AggregateFunctionExpr, LoweredAggregate, LoweredAggregateBuilder, }; -use datafusion_physical_expr::expressions::Literal; +use datafusion_physical_expr::expressions::{Column as PhysicalColumn, Literal}; use datafusion_physical_expr::{ LexOrdering, PhysicalSortExpr, create_physical_sort_exprs, }; @@ -1049,6 +1049,7 @@ impl DefaultPhysicalPlanner { input, group_expr, aggr_expr, + schema, .. }) => { let options = session_state.config().options(); @@ -1121,13 +1122,22 @@ impl DefaultPhysicalPlanner { ); } - let groups = self.create_grouping_physical_expr( - group_expr, - logical_input_schema, - &physical_input_schema, - execution_props, - planning_ctx, - )?; + // No-aggregate all-constant GROUP BY keeps the grouped path, + // mirroring EliminateGroupByConstant's guard. + let emit_no_rows_on_empty_input = + is_all_literal_group_by(group_expr) && !aggr_expr.is_empty(); + + let groups = if emit_no_rows_on_empty_input { + PhysicalGroupBy::default() + } else { + self.create_grouping_physical_expr( + group_expr, + logical_input_schema, + &physical_input_schema, + execution_props, + planning_ctx, + )? + }; let agg_filter = aggr_expr .iter() @@ -1192,14 +1202,17 @@ impl DefaultPhysicalPlanner { input_exec }; - let initial_aggr = Arc::new(AggregateExec::try_new( - AggregateMode::Partial, - groups.clone(), - aggregates, - filters.clone(), - input_exec, - Arc::clone(&physical_input_schema), - )?); + let initial_aggr = Arc::new( + AggregateExec::try_new( + AggregateMode::Partial, + groups.clone(), + aggregates, + filters.clone(), + input_exec, + Arc::clone(&physical_input_schema), + )? + .with_emit_no_rows_on_empty_input(emit_no_rows_on_empty_input), + ); let can_repartition = !groups.is_empty() && session_state.config().target_partitions() > 1 @@ -1223,14 +1236,59 @@ impl DefaultPhysicalPlanner { let final_grouping_set = initial_aggr.group_expr().as_final(); - Arc::new(AggregateExec::try_new( - next_partition_mode, - final_grouping_set, - updated_aggregates, - filters, - initial_aggr, - Arc::clone(&physical_input_schema), - )?) + let final_aggr = Arc::new( + AggregateExec::try_new( + next_partition_mode, + final_grouping_set, + updated_aggregates, + filters, + initial_aggr, + Arc::clone(&physical_input_schema), + )? + .with_emit_no_rows_on_empty_input(emit_no_rows_on_empty_input), + ); + if emit_no_rows_on_empty_input { + let final_schema = final_aggr.schema(); + let mut projections: Vec = Vec::with_capacity( + group_expr.len() + final_schema.fields().len(), + ); + for (i, gexpr) in group_expr.iter().enumerate() { + let physical_literal = create_physical_expr( + gexpr, + logical_input_schema, + execution_props, + planning_ctx, + )?; + let output_field = schema.field(i); + let literal_type = + physical_literal.data_type(final_schema.as_ref())?; + + if &literal_type != output_field.data_type() { + return internal_err!( + "GROUP BY constant literal type {literal_type} does not \ + match aggregate output field '{}' type {}", + output_field.name(), + output_field.data_type() + ); + } + projections.push(ProjectionExpr { + expr: physical_literal, + alias: output_field.name().to_string(), + }); + } + + for (j, agg_field) in final_schema.fields().iter().enumerate() { + let output_field = schema.field(group_expr.len() + j); + projections.push(ProjectionExpr { + expr: Arc::new(PhysicalColumn::new(agg_field.name(), j)), + alias: output_field.name().to_string(), + }); + } + + Arc::new(ProjectionExec::try_new(projections, final_aggr)?) + } else { + final_aggr + } } LogicalPlan::Projection(Projection { input, expr, .. }) => self .create_project_physical_exec_with_props( @@ -3232,6 +3290,19 @@ fn tuple_err(value: (Result, Result)) -> Result<(T, R)> { } } +// All group exprs are (possibly aliased) literals; used to flag the +// global-aggregate rewrite in the `LogicalPlan::Aggregate` arm. +fn is_all_literal_group_by(group_expr: &[Expr]) -> bool { + fn is_literal(e: &Expr) -> bool { + match e { + Expr::Alias(a) => is_literal(&a.expr), + Expr::Literal(..) => true, + _ => false, + } + } + !group_expr.is_empty() && group_expr.iter().all(is_literal) +} + struct OptimizationInvariantChecker<'a> { rule: &'a Arc, } @@ -3427,6 +3498,30 @@ mod tests { Field::new(name, DataType::Int64, nullable) } + #[tokio::test] + async fn test_all_constant_group_by_plans_flagged_global_aggregate() -> Result<()> { + let schema = Schema::new(vec![int64_field("c", true)]); + let logical_plan = scan_empty(None, &schema, None)? + .aggregate(vec![lit(1_i64)], vec![count_all()])? + .build()?; + + // create_initial_plan skips physical optimizer passes, so + // AggregateStatistics can't collapse the flagged aggregate first. + let planner = DefaultPhysicalPlanner::default(); + let physical_plan = planner + .create_initial_plan(&logical_plan, &make_session_state()) + .await?; + let explain = displayable(physical_plan.as_ref()).indent(true).to_string(); + + // Flagged global aggregate (empty gby) under a ProjectionExec. + assert_contains!(&explain, "emit_no_rows_on_empty_input=true"); + assert_contains!(&explain, "AggregateExec: mode="); + assert_contains!(&explain, ", gby=[]"); + assert_contains!(&explain, "ProjectionExec"); + + Ok(()) + } + #[tokio::test] async fn logical_range_repartition_plans_output_partitioning() -> Result<()> { let batch = RecordBatch::try_from_iter(vec![( diff --git a/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs b/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs index 0fa60ae20d2be..90b5f831ab0ff 100644 --- a/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs +++ b/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs @@ -49,7 +49,9 @@ use datafusion_physical_plan::aggregates::AggregateMode; use datafusion_physical_plan::aggregates::PhysicalGroupBy; use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion_physical_plan::common; +use datafusion_physical_plan::statistics::{StatisticsArgs, StatisticsContext}; use datafusion_physical_plan::displayable; +use datafusion_physical_plan::empty::EmptyExec; use datafusion_physical_plan::filter::FilterExec; use datafusion_physical_plan::projection::ProjectionExec; @@ -637,3 +639,137 @@ async fn topk_distinct_preserves_nulls() -> Result<()> { Ok(()) } + +/// A two-column Parquet source reporting the given `num_rows` statistic. +fn parquet_source_with_num_rows( + schema: &Arc, + num_rows: Precision, +) -> Arc { + let statistics = Statistics { + num_rows, + total_byte_size: Precision::Absent, + column_statistics: (0..schema.fields().len()) + .map(|_| ColumnStatistics::default()) + .collect(), + }; + let config = FileScanConfigBuilder::new( + ObjectStoreUrl::parse("test:///").unwrap(), + Arc::new(ParquetSource::new(Arc::clone(schema))), + ) + .with_file(PartitionedFile::new("x".to_string(), 100)) + .with_statistics(statistics) + .build(); + DataSourceExec::from_data_source(config) +} + +/// Build a flagged global Partial+Final `count(*)` aggregate over `source`. +fn flagged_count_star_agg( + source: Arc, + agg: &TestAggregate, +) -> Result { + let schema = source.schema(); + let partial_agg = AggregateExec::try_new( + AggregateMode::Partial, + PhysicalGroupBy::default(), + vec![Arc::new(agg.count_expr(&schema))], + vec![None], + source, + Arc::clone(&schema), + )? + .with_emit_no_rows_on_empty_input(true); + + AggregateExec::try_new( + AggregateMode::Final, + PhysicalGroupBy::default(), + vec![Arc::new(agg.count_expr(&schema))], + vec![None], + Arc::new(partial_agg), + Arc::clone(&schema), + ) + .map(|final_agg| final_agg.with_emit_no_rows_on_empty_input(true)) +} + +#[tokio::test] +async fn test_flagged_count_exact_nonempty_is_optimized() -> Result<()> { + // Exact(3): provably non-empty, so the one-row stats rewrite is safe. + let source = mock_data()?; + let agg = TestAggregate::new_count_star(); + let final_agg = flagged_count_star_agg(source, &agg)?; + assert_count_optim_success(final_agg, agg).await?; + Ok(()) +} + +#[tokio::test] +async fn test_flagged_count_exact_empty_becomes_empty_exec() -> Result<()> { + // Exact(0): provably empty, so the subtree becomes an EmptyExec. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + ])); + let source = MemorySourceConfig::try_new_exec(&[vec![]], Arc::clone(&schema), None)?; + let agg = TestAggregate::new_count_star(); + let final_agg = flagged_count_star_agg(source, &agg)?; + + let conf = ConfigOptions::new(); + let optimized = AggregateStatistics::new().optimize(Arc::new(final_agg), &conf)?; + assert!( + optimized.is::(), + "expected EmptyExec, got {}", + optimized.name() + ); + + let task_ctx = Arc::new(TaskContext::default()); + let result = common::collect(optimized.execute(0, task_ctx)?).await?; + let rows: usize = result.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, 0, "EmptyExec must produce zero rows"); + Ok(()) +} + +#[tokio::test] +async fn test_flagged_count_inexact_is_not_optimized() -> Result<()> { + // Inexact: may be empty, so the runtime aggregate is kept (it honors the flag). + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + ])); + let source = parquet_source_with_num_rows(&schema, Precision::Inexact(100)); + let agg = TestAggregate::new_count_star(); + let final_agg = flagged_count_star_agg(source, &agg)?; + + let conf = ConfigOptions::new(); + let optimized = AggregateStatistics::new().optimize(Arc::new(final_agg), &conf)?; + assert!( + optimized.is::(), + "expected AggregateExec (not optimized), got {}", + optimized.name() + ); + Ok(()) +} + +#[test] +fn test_flagged_agg_statistics_num_rows_precision() -> Result<()> { + // statistics_inner grades the flagged aggregate's row-count claim by input exactness. + let agg = TestAggregate::new_count_star(); + + // Exact(3) input -> Exact(1) output rows. + let final_agg = flagged_count_star_agg(mock_data()?, &agg)?; + let stats = StatisticsContext::new().compute(&final_agg, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Exact(1)); + + // Exact(0) input -> Exact(0) output rows. + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + ])); + let empty = MemorySourceConfig::try_new_exec(&[vec![]], Arc::clone(&schema), None)?; + let final_agg = flagged_count_star_agg(empty, &agg)?; + let stats = StatisticsContext::new().compute(&final_agg, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Exact(0)); + + // Inexact input -> Inexact(1). + let source = parquet_source_with_num_rows(&schema, Precision::Inexact(100)); + let final_agg = flagged_count_star_agg(source, &agg)?; + let stats = StatisticsContext::new().compute(&final_agg, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Inexact(1)); + Ok(()) +} diff --git a/datafusion/optimizer/src/eliminate_group_by_constant.rs b/datafusion/optimizer/src/eliminate_group_by_constant.rs index e21241ba7d993..5817410cf183c 100644 --- a/datafusion/optimizer/src/eliminate_group_by_constant.rs +++ b/datafusion/optimizer/src/eliminate_group_by_constant.rs @@ -65,9 +65,7 @@ impl OptimizerRule for EliminateGroupByConstant { .iter() .partition(|expr| is_redundant_group_expr(expr, &group_by_columns)); - if redundant.is_empty() - || (required.is_empty() && aggregate.aggr_expr.is_empty()) - { + if redundant.is_empty() || required.is_empty() { return Ok(Transformed::no(LogicalPlan::Aggregate(aggregate))); } @@ -221,16 +219,17 @@ mod tests { } #[test] - fn test_eliminate_constant() -> Result<()> { + fn test_no_eliminate_all_constant_group_exprs() -> Result<()> { + // Collapsing an all-constant GROUP BY to a global aggregate breaks + // empty-input cardinality (issue #11748); the rule must decline. let scan = test_table_scan()?; let plan = LogicalPlanBuilder::from(scan) .aggregate(vec![lit("test"), lit(123u32)], vec![count(col("c"))])? .build()?; assert_optimized_plan_equal!(plan, @r#" - Projection: Utf8("test"), UInt32(123), count(test.c) - Aggregate: groupBy=[[]], aggr=[[count(test.c)]] - TableScan: test + Aggregate: groupBy=[[Utf8("test"), UInt32(123)]], aggr=[[count(test.c)]] + TableScan: test "#) } diff --git a/datafusion/physical-optimizer/src/aggregate_statistics.rs b/datafusion/physical-optimizer/src/aggregate_statistics.rs index 43b1abb4b68a9..6b650cba11919 100644 --- a/datafusion/physical-optimizer/src/aggregate_statistics.rs +++ b/datafusion/physical-optimizer/src/aggregate_statistics.rs @@ -19,6 +19,7 @@ use datafusion_common::Result; use datafusion_common::config::ConfigOptions; use datafusion_common::scalar::ScalarValue; +use datafusion_common::stats::Precision; use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; use datafusion_physical_plan::aggregates::{ AggregateExec, AggregateInputMode, AggregateMode, @@ -60,6 +61,18 @@ impl PhysicalOptimizerRule for AggregateStatistics { .expect("take_optimizable() ensures that this is a AggregateExec"); let stats = StatisticsContext::new() .compute(partial_agg_exec.input().as_ref(), &StatisticsArgs::new())?; + // The single-row rewrite is only safe for a flagged aggregate when + // the input is provably non-empty; otherwise the runtime aggregate + // decides (it honors the flag). + if partial_agg_exec.emit_no_rows_on_empty_input() + && !matches!(stats.num_rows, Precision::Exact(n) if n > 0) + { + return plan + .map_children(|child| { + self.optimize(child, config).map(Transformed::yes) + }) + .data(); + } let mut projections = vec![]; for expr in partial_agg_exec.aggr_expr() { let field = expr.field(); diff --git a/datafusion/physical-optimizer/src/combine_partial_final_agg.rs b/datafusion/physical-optimizer/src/combine_partial_final_agg.rs index 297a92c45a16d..a792d37a43e21 100644 --- a/datafusion/physical-optimizer/src/combine_partial_final_agg.rs +++ b/datafusion/physical-optimizer/src/combine_partial_final_agg.rs @@ -99,7 +99,11 @@ impl PhysicalOptimizerRule for CombinePartialFinalAggregate { input_agg_exec.input_schema(), ) .map(|combined_agg| { - combined_agg.with_limit_options(agg_exec.limit_options()) + combined_agg + .with_limit_options(agg_exec.limit_options()) + .with_emit_no_rows_on_empty_input( + agg_exec.emit_no_rows_on_empty_input(), + ) }) .ok() .map(Arc::new) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs index ac7727b459300..02f38bb53085e 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs @@ -70,6 +70,8 @@ struct AggregateStreamInner { // None if the dynamic filter is not applicable. See details in `AggrDynFilter`. agg_dyn_filter_state: Option>, finished: bool, + emit_no_rows_on_empty_input: bool, + saw_input_rows: bool, // ==== Execution Resources ==== baseline_metrics: BaselineMetrics, @@ -331,6 +333,8 @@ impl AggregateStream { reservation, finished: false, agg_dyn_filter_state: maybe_dynamic_filter, + emit_no_rows_on_empty_input: agg.emit_no_rows_on_empty_input, + saw_input_rows: false, }; let stream = futures::stream::unfold(inner, |mut this| async move { @@ -341,6 +345,10 @@ impl AggregateStream { loop { let result = match this.input.next().await { Some(Ok(batch)) => { + if batch.num_rows() > 0 { + this.saw_input_rows = true; + } + let result = { let elapsed_compute = this.baseline_metrics.elapsed_compute(); let _timer = elapsed_compute.timer(); // Stops on drop @@ -374,6 +382,11 @@ impl AggregateStream { // Release the input pipeline's resources before finalization. let input_schema = this.input.schema(); this.input = Box::pin(EmptyRecordBatchStream::new(input_schema)); + // Zero input rows means zero groups: end without the + // default row (issue #11748). + if this.emit_no_rows_on_empty_input && !this.saw_input_rows { + return None; + } let timer = this.baseline_metrics.elapsed_compute().timer(); let result = finalize_aggregation(&mut this.accumulators, &this.mode) diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 732da32ab0391..a9ecab095bb26 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -864,6 +864,11 @@ pub struct AggregateExec { /// it remains `Some(..)` to enable dynamic filtering during aggregate execution; /// otherwise, it is cleared to `None`. dynamic_filter: Option>, + + /// True only when this global aggregate is the physical-planner rewrite of + /// a grouping aggregate whose GROUP BY keys were all constants; it must + /// emit zero rows on empty input. Never true for a genuine global aggregate. + emit_no_rows_on_empty_input: bool, } impl AggregateExec { @@ -889,6 +894,7 @@ impl AggregateExec { schema: Arc::clone(&self.schema), input_schema: Arc::clone(&self.input_schema), dynamic_filter: self.dynamic_filter.clone(), + emit_no_rows_on_empty_input: self.emit_no_rows_on_empty_input, } } @@ -909,6 +915,7 @@ impl AggregateExec { schema: Arc::clone(&self.schema), input_schema: Arc::clone(&self.input_schema), dynamic_filter: self.dynamic_filter.clone(), + emit_no_rows_on_empty_input: self.emit_no_rows_on_empty_input, } } @@ -1044,6 +1051,7 @@ impl AggregateExec { input_order_mode, cache: Arc::new(cache), dynamic_filter: None, + emit_no_rows_on_empty_input: false, }; exec.init_dynamic_filter(); @@ -1128,6 +1136,17 @@ impl AggregateExec { Ok(self) } + /// Sets `emit_no_rows_on_empty_input`; see the field doc for the invariant. + pub fn with_emit_no_rows_on_empty_input(mut self, value: bool) -> Self { + self.emit_no_rows_on_empty_input = value; + self + } + + /// Whether this global aggregate must emit zero rows on empty input. + pub fn emit_no_rows_on_empty_input(&self) -> bool { + self.emit_no_rows_on_empty_input + } + /// Input plan pub fn input(&self) -> &Arc { &self.input @@ -1834,6 +1853,9 @@ impl DisplayAs for AggregateExec { if self.input_order_mode != InputOrderMode::Linear { write!(f, ", ordering_mode={:?}", self.input_order_mode)?; } + if self.emit_no_rows_on_empty_input { + write!(f, ", emit_no_rows_on_empty_input=true")?; + } } DisplayFormatType::TreeRender => { let format_expr_with_alias = @@ -1993,6 +2015,7 @@ impl ExecutionPlan for AggregateExec { )?; me.limit_options = self.limit_options; me.dynamic_filter.clone_from(&self.dynamic_filter); + me.emit_no_rows_on_empty_input = self.emit_no_rows_on_empty_input; Ok(Arc::new(me)) } @@ -2723,6 +2746,183 @@ mod tests { use futures::{FutureExt, Stream, StreamExt}; use insta::{allow_duplicates, assert_snapshot}; + fn v_schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, true)])) + } + + fn v_batch(schema: &SchemaRef, values: Vec) -> RecordBatch { + RecordBatch::try_new(Arc::clone(schema), vec![Arc::new(Int64Array::from(values))]) + .unwrap() + } + + /// The Int64 count in the single output row. + fn count_value(batches: &[RecordBatch]) -> i64 { + batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0) + } + + /// Build a global (no grouping) Single-mode `count(v)` aggregate over the + /// given input batches, optionally flagged and optionally FILTERed. + fn global_count_agg( + schema: &SchemaRef, + batches: Vec, + filter: Option>, + emit_no_rows_on_empty_input: bool, + ) -> Result> { + let count_expr = AggregateExprBuilder::new(count_udaf(), vec![col("v", schema)?]) + .schema(Arc::clone(schema)) + .alias("count(v)") + .build() + .map(Arc::new)?; + let input = TestMemoryExec::try_new_exec(&[batches], Arc::clone(schema), None)?; + let agg = AggregateExec::try_new( + AggregateMode::Single, + PhysicalGroupBy::default(), + vec![count_expr], + vec![filter], + input, + Arc::clone(schema), + )? + .with_emit_no_rows_on_empty_input(emit_no_rows_on_empty_input); + Ok(Arc::new(agg)) + } + + #[tokio::test] + async fn flagged_global_agg_empty_input_emits_no_rows() -> Result<()> { + let task_ctx = Arc::new(TaskContext::default()); + let schema = v_schema(); + let agg = global_count_agg(&schema, vec![], None, true)?; + let batches = collect(agg.execute(0, task_ctx)?).await?; + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert!(batches.is_empty(), "expected no batches, got {batches:?}"); + assert_eq!(rows, 0); + Ok(()) + } + + #[tokio::test] + async fn unflagged_global_agg_empty_input_emits_one_row() -> Result<()> { + // Guards the default: a plain global aggregate still emits one row. + let task_ctx = Arc::new(TaskContext::default()); + let schema = v_schema(); + let agg = global_count_agg(&schema, vec![], None, false)?; + let batches = collect(agg.execute(0, task_ctx)?).await?; + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, 1); + assert_eq!(count_value(&batches), 0); + Ok(()) + } + + #[tokio::test] + async fn flagged_global_agg_nonempty_input_emits_one_row() -> Result<()> { + let task_ctx = Arc::new(TaskContext::default()); + let schema = v_schema(); + let batch = v_batch(&schema, vec![1, 2, 3]); + let agg = global_count_agg(&schema, vec![batch], None, true)?; + let batches = collect(agg.execute(0, task_ctx)?).await?; + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, 1); + assert_eq!(count_value(&batches), 3); + Ok(()) + } + + #[test] + fn flagged_global_agg_rebuilds_preserve_flag() -> Result<()> { + // Hand-written clone-constructors must copy the flag. + let schema = v_schema(); + let agg = global_count_agg(&schema, vec![], None, true)?; + assert!( + agg.with_new_aggr_exprs(agg.aggr_expr().to_vec()) + .emit_no_rows_on_empty_input() + ); + assert!( + agg.with_new_limit_options(None) + .emit_no_rows_on_empty_input() + ); + Ok(()) + } + + /// Flagged Partial per partition -> CoalescePartitions -> flagged Final. + fn flagged_partial_final( + schema: &SchemaRef, + partitions: Vec>, + ) -> Result> { + let count_expr = AggregateExprBuilder::new(count_udaf(), vec![col("v", schema)?]) + .schema(Arc::clone(schema)) + .alias("count(v)") + .build() + .map(Arc::new)?; + let input = TestMemoryExec::try_new_exec(&partitions, Arc::clone(schema), None)?; + let partial = Arc::new( + AggregateExec::try_new( + AggregateMode::Partial, + PhysicalGroupBy::default(), + vec![Arc::clone(&count_expr)], + vec![None], + input, + Arc::clone(schema), + )? + .with_emit_no_rows_on_empty_input(true), + ); + let coalesce = Arc::new(CoalescePartitionsExec::new(partial)); + AggregateExec::try_new( + AggregateMode::Final, + PhysicalGroupBy::default(), + vec![count_expr], + vec![None], + coalesce, + Arc::clone(schema), + ) + .map(|f| Arc::new(f.with_emit_no_rows_on_empty_input(true))) + } + + #[tokio::test] + async fn flagged_partial_final_mixed_empty_partitions() -> Result<()> { + // An empty partition's Partial emits nothing; the Final merges the rest. + let task_ctx = Arc::new(TaskContext::default()); + let schema = v_schema(); + let agg = flagged_partial_final( + &schema, + vec![vec![v_batch(&schema, vec![1, 2, 3])], vec![]], + )?; + let batches = collect(agg.execute(0, task_ctx)?).await?; + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, 1); + assert_eq!(count_value(&batches), 3); + Ok(()) + } + + #[tokio::test] + async fn flagged_partial_final_all_empty_partitions() -> Result<()> { + // All Partials silent -> the Final sees zero input rows and emits nothing. + let task_ctx = Arc::new(TaskContext::default()); + let schema = v_schema(); + let agg = flagged_partial_final(&schema, vec![vec![], vec![]])?; + let batches = collect(agg.execute(0, task_ctx)?).await?; + assert!(batches.is_empty(), "expected no batches, got {batches:?}"); + Ok(()) + } + + #[tokio::test] + async fn flagged_global_agg_all_filtered_nonempty_emits_one_row() -> Result<()> { + // FILTER nuance: raw input rows are counted BEFORE the per-aggregate + // FILTER, so a flagged aggregate whose FILTER rejects every row still + // emits one row (with count 0) because input rows existed. + let task_ctx = Arc::new(TaskContext::default()); + let schema = v_schema(); + let batch = v_batch(&schema, vec![1, 2, 3]); + let filter = lit(false); + let agg = global_count_agg(&schema, vec![batch], Some(filter), true)?; + let batches = collect(agg.execute(0, task_ctx)?).await?; + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, 1); + assert_eq!(count_value(&batches), 0); + Ok(()) + } + // Generate a schema which consists of 5 columns (a, b, c, d, e) fn create_test_schema() -> Result { let a = Field::new("a", DataType::Int32, true); diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index 205cf89abed1b..0731b05126c64 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1444,6 +1444,8 @@ message AggregateExecNode { bool has_grouping_set = 12; // Optional dynamic filter expression for pushing down to the child. PhysicalExprNode dynamic_filter = 13; + // True when this global aggregate (all-constant GROUP BY rewrite) must emit zero rows on empty input. + bool emit_no_rows_on_empty_input = 14; } message GlobalLimitExecNode { diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index d23f8eee5fd2c..4028abe23bacf 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -157,6 +157,9 @@ impl serde::Serialize for AggregateExecNode { if self.dynamic_filter.is_some() { len += 1; } + if self.emit_no_rows_on_empty_input { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.AggregateExecNode", len)?; if !self.group_expr.is_empty() { struct_ser.serialize_field("groupExpr", &self.group_expr)?; @@ -199,6 +202,9 @@ impl serde::Serialize for AggregateExecNode { if let Some(v) = self.dynamic_filter.as_ref() { struct_ser.serialize_field("dynamicFilter", v)?; } + if self.emit_no_rows_on_empty_input { + struct_ser.serialize_field("emitNoRowsOnEmptyInput", &self.emit_no_rows_on_empty_input)?; + } struct_ser.end() } } @@ -231,6 +237,8 @@ impl<'de> serde::Deserialize<'de> for AggregateExecNode { "hasGroupingSet", "dynamic_filter", "dynamicFilter", + "emit_no_rows_on_empty_input", + "emitNoRowsOnEmptyInput", ]; #[allow(clippy::enum_variant_names)] @@ -248,6 +256,7 @@ impl<'de> serde::Deserialize<'de> for AggregateExecNode { Limit, HasGroupingSet, DynamicFilter, + EmitNoRowsOnEmptyInput, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -282,6 +291,7 @@ impl<'de> serde::Deserialize<'de> for AggregateExecNode { "limit" => Ok(GeneratedField::Limit), "hasGroupingSet" | "has_grouping_set" => Ok(GeneratedField::HasGroupingSet), "dynamicFilter" | "dynamic_filter" => Ok(GeneratedField::DynamicFilter), + "emitNoRowsOnEmptyInput" | "emit_no_rows_on_empty_input" => Ok(GeneratedField::EmitNoRowsOnEmptyInput), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -314,6 +324,7 @@ impl<'de> serde::Deserialize<'de> for AggregateExecNode { let mut limit__ = None; let mut has_grouping_set__ = None; let mut dynamic_filter__ = None; + let mut emit_no_rows_on_empty_input__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::GroupExpr => { @@ -394,6 +405,12 @@ impl<'de> serde::Deserialize<'de> for AggregateExecNode { } dynamic_filter__ = map_.next_value()?; } + GeneratedField::EmitNoRowsOnEmptyInput => { + if emit_no_rows_on_empty_input__.is_some() { + return Err(serde::de::Error::duplicate_field("emitNoRowsOnEmptyInput")); + } + emit_no_rows_on_empty_input__ = Some(map_.next_value()?); + } } } Ok(AggregateExecNode { @@ -410,6 +427,7 @@ impl<'de> serde::Deserialize<'de> for AggregateExecNode { limit: limit__, has_grouping_set: has_grouping_set__.unwrap_or_default(), dynamic_filter: dynamic_filter__, + emit_no_rows_on_empty_input: emit_no_rows_on_empty_input__.unwrap_or_default(), }) } } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 6baabbf37a41c..4405828738c8c 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -2173,6 +2173,9 @@ pub struct AggregateExecNode { /// Optional dynamic filter expression for pushing down to the child. #[prost(message, optional, tag = "13")] pub dynamic_filter: ::core::option::Option, + /// True when this global aggregate (all-constant GROUP BY rewrite) must emit zero rows on empty input. + #[prost(bool, tag = "14")] + pub emit_no_rows_on_empty_input: bool, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct GlobalLimitExecNode { diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index cea334e42aace..94dd5258e389f 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -1796,6 +1796,9 @@ pub trait PhysicalPlanNodeExt: Sized { agg }; + let agg = + agg.with_emit_no_rows_on_empty_input(hash_agg.emit_no_rows_on_empty_input); + Ok(Arc::new(agg)) } @@ -3262,6 +3265,7 @@ pub trait PhysicalPlanNodeExt: Sized { proto_converter.physical_expr_to_proto(&df_expr, codec) }) .transpose()?, + emit_no_rows_on_empty_input: exec.emit_no_rows_on_empty_input(), }, ))), }) diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index 2671f4d0152f7..42e7ad2fb93ab 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -859,6 +859,46 @@ fn roundtrip_aggregate_with_limit() -> Result<()> { roundtrip_test(Arc::new(agg)) } +#[test] +fn roundtrip_aggregate_emit_no_rows_on_empty_input() -> Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, true)])); + + let aggregates = vec![ + AggregateExprBuilder::new(count_udaf(), vec![col("v", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("count(v)") + .build() + .map(Arc::new)?, + ]; + + // Global aggregate (empty group by) carrying the flag. + let agg = AggregateExec::try_new( + AggregateMode::Single, + PhysicalGroupBy::default(), + aggregates, + vec![None], + Arc::new(EmptyExec::new(Arc::clone(&schema))), + schema, + )? + .with_emit_no_rows_on_empty_input(true); + + let ctx = SessionContext::new(); + let codec = DefaultPhysicalExtensionCodec {}; + let proto_converter = DefaultPhysicalProtoConverter {}; + let result = + roundtrip_test_and_return(Arc::new(agg), &ctx, &codec, &proto_converter)?; + + let result_agg = result + .downcast_ref::() + .expect("expected AggregateExec after roundtrip"); + assert!( + result_agg.emit_no_rows_on_empty_input(), + "flag must survive proto roundtrip" + ); + + Ok(()) +} + #[test] fn roundtrip_aggregate_with_approx_pencentile_cont() -> Result<()> { let field_a = Field::new("a", DataType::Int64, false); diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index cbb9c5d0317dc..67647a98bce40 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/datafusion/sqllogictest/test_files/aggregate.slt @@ -8451,6 +8451,38 @@ CREATE TABLE t1(v1 int); statement error DataFusion error: Error during planning: Aggregate functions are not allowed in the WHERE clause. Consider using HAVING instead SELECT v1 FROM t1 WHERE ((count(v1) % 1) << 1) > 0; +# issue #11748: all-constant GROUP BY over empty input must return zero rows, not one NULL row +query R +SELECT AVG(v1) FROM t1 GROUP BY false HAVING false; +---- + +query R +SELECT AVG(v1) FROM t1 GROUP BY false; +---- + +query IIR +SELECT 1, 2, AVG(v1) FROM t1 GROUP BY 1, 2; +---- + +statement ok +INSERT INTO t1 VALUES (1), (2), (3); + +# On non-empty input the single constant group emits one row. +query R +SELECT AVG(v1) FROM t1 GROUP BY false; +---- +2 + +# HAVING false removes that single row. +query R +SELECT AVG(v1) FROM t1 GROUP BY false HAVING false; +---- + +query IIR +SELECT 1, 2, AVG(v1) FROM t1 GROUP BY 1, 2; +---- +1 2 2 + statement ok DROP TABLE t1; diff --git a/datafusion/sqllogictest/test_files/optimizer_group_by_constant.slt b/datafusion/sqllogictest/test_files/optimizer_group_by_constant.slt index da1e7de22bb7a..e89aa0b5c7e95 100644 --- a/datafusion/sqllogictest/test_files/optimizer_group_by_constant.slt +++ b/datafusion/sqllogictest/test_files/optimizer_group_by_constant.slt @@ -60,10 +60,9 @@ FROM test_table t group by 1, 2, 3 ---- logical_plan -01)Projection: Int64(123), Int64(456), Int64(789), count(Int64(1)), avg(t.c12) -02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1)), avg(t.c12)]] -03)----SubqueryAlias: t -04)------TableScan: test_table projection=[c12] +01)Aggregate: groupBy=[[Int64(123), Int64(456), Int64(789)]], aggr=[[count(Int64(1)), avg(t.c12)]] +02)--SubqueryAlias: t +03)----TableScan: test_table projection=[c12] query TT EXPLAIN @@ -72,8 +71,8 @@ FROM test_table t GROUP BY 1, 2 ---- logical_plan -01)Projection: Date32("2023-05-04") AS dt, Boolean(true) AS today_filter, count(Int64(1)) -02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] +01)Projection: to_date(Utf8("2023-05-04")) AS dt, date_part(Utf8("DAY"),now()) < Int64(1000) AS today_filter, count(Int64(1)) +02)--Aggregate: groupBy=[[Date32("2023-05-04") AS to_date(Utf8("2023-05-04")), Boolean(true) AS date_part(Utf8("DAY"),now()) < Int64(1000)]], aggr=[[count(Int64(1))]] 03)----SubqueryAlias: t 04)------TableScan: test_table projection=[] @@ -90,10 +89,9 @@ FROM test_table t GROUP BY 1 ---- logical_plan -01)Projection: Boolean(true) AS NOT date_part(Utf8("MONTH"),now()) BETWEEN Int64(50) AND Int64(60), count(Int64(1)) -02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]] -03)----SubqueryAlias: t -04)------TableScan: test_table projection=[] +01)Aggregate: groupBy=[[Boolean(true) AS NOT date_part(Utf8("MONTH"),now()) BETWEEN Int64(50) AND Int64(60)]], aggr=[[count(Int64(1))]] +02)--SubqueryAlias: t +03)----TableScan: test_table projection=[] query TT EXPLAIN @@ -117,9 +115,49 @@ logical_plan 02)--SubqueryAlias: t 03)----TableScan: test_table projection=[] +# issue #11748: the physical planner recovers the optimization as a flagged +# global AggregateExec + ProjectionExec (see the logical-plan declines above) +statement ok +set datafusion.explain.logical_plan_only = false; + +query TT +EXPLAIN +SELECT 123, count(1), avg(c12) +FROM test_table t +GROUP BY 1 +---- +logical_plan +01)Aggregate: groupBy=[[Int64(123)]], aggr=[[count(Int64(1)), avg(t.c12)]] +02)--SubqueryAlias: t +03)----TableScan: test_table projection=[c12] +physical_plan +01)ProjectionExec: expr=[123 as Int64(123), count(Int64(1))@0 as count(Int64(1)), avg(t.c12)@1 as avg(t.c12)] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1)), avg(t.c12)], emit_no_rows_on_empty_input=true +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c12], file_type=csv, has_header=true + +# Aliased literal keys unwrap to the same flagged rewrite. +query TT +EXPLAIN +SELECT 123 AS x, count(1) +FROM test_table t +GROUP BY 1 +---- +logical_plan +01)Projection: Int64(123) AS x, count(Int64(1)) +02)--Aggregate: groupBy=[[Int64(123)]], aggr=[[count(Int64(1))]] +03)----SubqueryAlias: t +04)------TableScan: test_table projection=[] +physical_plan +01)ProjectionExec: expr=[123 as x, count(Int64(1))@0 as count(Int64(1))] +02)--AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))], emit_no_rows_on_empty_input=true +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, file_type=csv, has_header=true + +statement ok +set datafusion.explain.logical_plan_only = true; + # Config reset -# The SLT runner sets `target_partitions` to 4 instead of using the default, so +# The SLT runner sets `target_partitions` to 4 instead of using the default, so # reset it explicitly. statement ok set datafusion.execution.target_partitions = 4; From 8ae1f1d3eab4ae16090028806536798a2d17e863 Mon Sep 17 00:00:00 2001 From: HairstonE Date: Thu, 23 Jul 2026 19:50:54 -0400 Subject: [PATCH 2/2] add tests --- .../aggregate_statistics.rs | 41 +++---------------- 1 file changed, 6 insertions(+), 35 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs b/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs index 90b5f831ab0ff..1b4ebbc1f7043 100644 --- a/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs +++ b/datafusion/core/tests/physical_optimizer/aggregate_statistics.rs @@ -49,9 +49,7 @@ use datafusion_physical_plan::aggregates::AggregateMode; use datafusion_physical_plan::aggregates::PhysicalGroupBy; use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion_physical_plan::common; -use datafusion_physical_plan::statistics::{StatisticsArgs, StatisticsContext}; use datafusion_physical_plan::displayable; -use datafusion_physical_plan::empty::EmptyExec; use datafusion_physical_plan::filter::FilterExec; use datafusion_physical_plan::projection::ProjectionExec; @@ -700,8 +698,8 @@ async fn test_flagged_count_exact_nonempty_is_optimized() -> Result<()> { } #[tokio::test] -async fn test_flagged_count_exact_empty_becomes_empty_exec() -> Result<()> { - // Exact(0): provably empty, so the subtree becomes an EmptyExec. +async fn test_flagged_count_exact_empty_is_not_optimized() -> Result<()> { + // Exact(0): not provably non-empty, so the runtime aggregate is kept. let schema = Arc::new(Schema::new(vec![ Field::new("a", DataType::Int32, true), Field::new("b", DataType::Int32, true), @@ -713,15 +711,16 @@ async fn test_flagged_count_exact_empty_becomes_empty_exec() -> Result<()> { let conf = ConfigOptions::new(); let optimized = AggregateStatistics::new().optimize(Arc::new(final_agg), &conf)?; assert!( - optimized.is::(), - "expected EmptyExec, got {}", + optimized.is::(), + "expected AggregateExec (not optimized), got {}", optimized.name() ); + // The kept runtime aggregate honors the flag: zero rows out. let task_ctx = Arc::new(TaskContext::default()); let result = common::collect(optimized.execute(0, task_ctx)?).await?; let rows: usize = result.iter().map(|b| b.num_rows()).sum(); - assert_eq!(rows, 0, "EmptyExec must produce zero rows"); + assert_eq!(rows, 0); Ok(()) } @@ -745,31 +744,3 @@ async fn test_flagged_count_inexact_is_not_optimized() -> Result<()> { ); Ok(()) } - -#[test] -fn test_flagged_agg_statistics_num_rows_precision() -> Result<()> { - // statistics_inner grades the flagged aggregate's row-count claim by input exactness. - let agg = TestAggregate::new_count_star(); - - // Exact(3) input -> Exact(1) output rows. - let final_agg = flagged_count_star_agg(mock_data()?, &agg)?; - let stats = StatisticsContext::new().compute(&final_agg, &StatisticsArgs::new())?; - assert_eq!(stats.num_rows, Precision::Exact(1)); - - // Exact(0) input -> Exact(0) output rows. - let schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Int32, true), - Field::new("b", DataType::Int32, true), - ])); - let empty = MemorySourceConfig::try_new_exec(&[vec![]], Arc::clone(&schema), None)?; - let final_agg = flagged_count_star_agg(empty, &agg)?; - let stats = StatisticsContext::new().compute(&final_agg, &StatisticsArgs::new())?; - assert_eq!(stats.num_rows, Precision::Exact(0)); - - // Inexact input -> Inexact(1). - let source = parquet_source_with_num_rows(&schema, Precision::Inexact(100)); - let final_agg = flagged_count_star_agg(source, &agg)?; - let stats = StatisticsContext::new().compute(&final_agg, &StatisticsArgs::new())?; - assert_eq!(stats.num_rows, Precision::Inexact(1)); - Ok(()) -}