diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 0523628ad7e6a..390f71e135bd3 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -177,8 +177,8 @@ use arrow::record_batch::RecordBatch; use arrow_schema::FieldRef; use datafusion_common::stats::Precision; use datafusion_common::{ - Constraint, Constraints, Result, ScalarValue, assert_eq_or_internal_err, - internal_err, not_impl_err, + ColumnStatistics, Constraint, Constraints, Result, ScalarValue, + assert_eq_or_internal_err, internal_err, not_impl_err, }; use datafusion_execution::TaskContext; use datafusion_execution::memory_pool::MemoryLimit; @@ -1510,7 +1510,12 @@ impl AggregateExec { }) } None => { - let num_rows = self.estimate_num_rows(child_statistics); + let num_rows = self.estimate_num_rows(child_statistics, partition); + let column_statistics = self.nullify_group_columns_for_empty_input( + column_statistics, + child_statistics, + &num_rows, + ); let total_byte_size = num_rows .get_value() @@ -1549,13 +1554,79 @@ impl AggregateExec { ) -> Option { let logical_rows = self.logical_rows_without_group_exprs()?; - Some(match (self.mode.output_mode(), partition) { + Some(self.scale_logical_rows(logical_rows, partition)) + } + + /// Scales a logical aggregate row count to the rows this operator emits, + /// which for partial aggregation is once per output partition. + fn scale_logical_rows(&self, logical_rows: usize, partition: Option) -> usize { + match (self.mode.output_mode(), partition) { (AggregateOutputMode::Final, _) => logical_rows, (AggregateOutputMode::Partial, Some(_)) => logical_rows, (AggregateOutputMode::Partial, None) => { logical_rows * self.cache.output_partitioning().partition_count() } - }) + } + } + + /// Number of rows a grouped aggregate emits for an empty input. + /// + /// Grouping expressions yield no groups, so the only rows are the + /// grand-total rows of the empty grouping sets that `GROUPING SETS(())`, + /// `ROLLUP` and `CUBE` introduce alongside the non-empty ones. + fn output_rows_for_empty_input(&self, partition: Option) -> usize { + let empty_grouping_sets = self + .group_by + .groups + .iter() + .filter(|nulls| nulls.iter().all(|is_null| *is_null)) + .count(); + + self.scale_logical_rows(empty_grouping_sets, partition) + } + + /// Reports the grouping columns of an empty input as all NULL. + /// + /// The only rows such an input produces are grand-total rows, which hold + /// NULL in every grouping column, so the values copied from the child do not + /// describe the output. Rules that answer `MIN`/`MAX` from statistics read + /// these values, so an input value here becomes a wrong query result. + /// + /// The bounds are typed nulls rather than [`Precision::Absent`], both + /// because NULL is the `MIN`/`MAX` of such a column and because the data + /// type lets downstream interval analysis keep intersecting intervals of + /// that type, as `FilterExec` does for a column with no rows. + fn nullify_group_columns_for_empty_input( + &self, + mut column_statistics: Vec, + child_statistics: &Statistics, + num_rows: &Precision, + ) -> Vec { + let empty_input = child_statistics.num_rows.get_value() == Some(&0); + let emits_rows = num_rows.get_value().is_some_and(|&rows| rows > 0); + if !empty_input || !emits_rows { + return column_statistics; + } + + let schema = self.schema(); + for (idx, column_stats) in column_statistics + .iter_mut() + .take(self.group_by.expr.len()) + .enumerate() + { + let typed_null = ScalarValue::try_from(schema.field(idx).data_type()) + .unwrap_or(ScalarValue::Null); + let mut null_bound = Precision::Exact(typed_null); + if matches!(num_rows, Precision::Inexact(_)) { + null_bound = null_bound.to_inexact(); + } + column_stats.min_value = null_bound.clone(); + column_stats.max_value = null_bound; + column_stats.distinct_count = num_rows.map(|_| 0); + column_stats.null_count = *num_rows; + } + + column_statistics } /// Exact number of logical aggregate rows for aggregates without group-by @@ -1577,7 +1648,11 @@ impl AggregateExec { /// Estimates the output row count for grouped aggregations, combining NDV, /// input row count, and TopK limit into a single [`Precision`]. - fn estimate_num_rows(&self, child_statistics: &Statistics) -> Precision { + fn estimate_num_rows( + &self, + child_statistics: &Statistics, + partition: Option, + ) -> Precision { let ndv = if !self.group_by.expr.is_empty() { self.compute_group_ndv(child_statistics) } else { @@ -1596,7 +1671,11 @@ impl AggregateExec { } num_rows } else if value == 0 { - child_statistics.num_rows + // The limit bounds groups built from input rows, not the rows + // the empty grouping sets contribute. + child_statistics + .num_rows + .map(|_| self.output_rows_for_empty_input(partition)) } else { let grouping_set_num = self.group_by.groups.len(); let mut num_rows = @@ -3025,6 +3104,7 @@ mod tests { use datafusion_physical_expr::expressions::Literal; use crate::projection::ProjectionExec; + use crate::repartition::RepartitionExec; use datafusion_physical_expr::projection::ProjectionExpr; use futures::{FutureExt, Stream, StreamExt}; use insta::{allow_duplicates, assert_snapshot}; @@ -5889,6 +5969,160 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_aggregate_statistics_empty_input_with_grouping_sets() -> Result<()> { + let schema = empty_grouping_sets_test_schema(); + + // `GROUP BY a` produces no groups for an empty input. + let grouped = build_test_aggregate( + &schema, + empty_input_statistics(), + simple_group_by(&schema, &["a"]), + None, + )?; + let stats = StatisticsContext::new().compute(&grouped, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Exact(0)); + + // `GROUPING SETS((a), ())`, as ROLLUP and CUBE produce, still emits the + // grand-total row of the empty grouping set on an empty input. + let with_empty_set = build_test_aggregate( + &schema, + empty_input_statistics(), + grouping_sets_with_empty(&schema, 1)?, + None, + )?; + let stats = + StatisticsContext::new().compute(&with_empty_set, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Exact(1)); + + // `GROUPING SETS((a), (), ())` emits one grand-total row per empty + // grouping set, because execution gives each duplicate its own ordinal. + let with_duplicate_empty_sets = build_test_aggregate( + &schema, + empty_input_statistics(), + grouping_sets_with_empty(&schema, 2)?, + None, + )?; + let stats = StatisticsContext::new() + .compute(&with_duplicate_empty_sets, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Exact(2)); + + Ok(()) + } + + /// Partial aggregation emits the grand-total row from every output + /// partition, so the whole-plan estimate scales with the partition count + /// while a single-partition request does not. + #[tokio::test] + async fn test_aggregate_statistics_empty_input_partial_mode_scaling() -> Result<()> { + let schema = empty_grouping_sets_test_schema(); + let input = Arc::new(RepartitionExec::try_new( + Arc::new(StatisticsExec::new( + empty_input_statistics(), + (*schema).clone(), + )), + Partitioning::RoundRobinBatch(4), + )?) as Arc; + + let agg = AggregateExec::try_new( + AggregateMode::Partial, + grouping_sets_with_empty(&schema, 1)?, + vec![count_a_aggregate(&schema)?], + vec![None], + input, + Arc::clone(&schema), + )?; + assert_eq!(agg.properties().output_partitioning().partition_count(), 4); + + let context = StatisticsContext::new(); + assert_eq!( + context.compute(&agg, &StatisticsArgs::new())?.num_rows, + Precision::Exact(4) + ); + // Inexact because a repartition only estimates its per-partition row + // count. The grouping column statistics carry that same precision. + let partition_statistics = + context.compute(&agg, &StatisticsArgs::new().with_partition(Some(0)))?; + assert_eq!(partition_statistics.num_rows, Precision::Inexact(1)); + let group_column = &partition_statistics.column_statistics[0]; + let typed_null = Precision::Inexact(ScalarValue::Int32(None)); + assert_eq!(group_column.min_value, typed_null); + assert_eq!(group_column.max_value, typed_null); + assert_eq!(group_column.distinct_count, Precision::Inexact(0)); + assert_eq!(group_column.null_count, Precision::Inexact(1)); + + Ok(()) + } + + /// The input's min, max and distinct values must not reach the output + /// column statistics. See `nullify_group_columns_for_empty_input`. + #[tokio::test] + async fn test_aggregate_statistics_empty_input_nullifies_group_columns() -> Result<()> + { + let schema = empty_grouping_sets_test_schema(); + let mut input_statistics = empty_input_statistics(); + input_statistics.column_statistics[0] = ColumnStatistics { + null_count: Precision::Exact(0), + max_value: Precision::Exact(ScalarValue::Int32(Some(5))), + min_value: Precision::Exact(ScalarValue::Int32(Some(5))), + sum_value: Precision::Absent, + distinct_count: Precision::Exact(1), + byte_size: Precision::Absent, + }; + + let agg = build_test_aggregate( + &schema, + input_statistics, + grouping_sets_with_empty(&schema, 1)?, + None, + )?; + + let stats = StatisticsContext::new().compute(&agg, &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Exact(1)); + let group_column = &stats.column_statistics[0]; + let typed_null = Precision::Exact(ScalarValue::Int32(None)); + assert_eq!(group_column.min_value, typed_null); + assert_eq!(group_column.max_value, typed_null); + assert_eq!(group_column.distinct_count, Precision::Exact(0)); + assert_eq!(group_column.null_count, Precision::Exact(1)); + + Ok(()) + } + + fn empty_grouping_sets_test_schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Float64, false), + ])) + } + + fn empty_input_statistics() -> Statistics { + Statistics { + num_rows: Precision::Exact(0), + total_byte_size: Precision::Exact(0), + column_statistics: vec![ + ColumnStatistics::new_unknown(), + ColumnStatistics::new_unknown(), + ], + } + } + + /// `GROUPING SETS((a), (), ...)` with `empty_sets` empty grouping sets, as + /// `ROLLUP(a)` and `CUBE(a)` produce with one. + fn grouping_sets_with_empty( + schema: &SchemaRef, + empty_sets: usize, + ) -> Result { + let mut groups = vec![vec![false]]; + groups.resize(1 + empty_sets, vec![true]); + Ok(PhysicalGroupBy::new( + vec![(col("a", schema)?, "a".to_string())], + vec![(lit(ScalarValue::Int32(None)), "a".to_string())], + groups, + true, + )) + } + fn build_test_aggregate( schema: &SchemaRef, stats: Statistics, diff --git a/datafusion/sqllogictest/test_files/grouping.slt b/datafusion/sqllogictest/test_files/grouping.slt index 2c05dd851e61a..7d893f94be74a 100644 --- a/datafusion/sqllogictest/test_files/grouping.slt +++ b/datafusion/sqllogictest/test_files/grouping.slt @@ -261,3 +261,90 @@ query II SELECT SUM(v1), COUNT(*) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((), (v1)) ---- NULL 0 + +# rollup_empty_input_outer_count: an outer COUNT(*) over ROLLUP is answered from the inner row-count statistics +query I +SELECT COUNT(*) FROM (SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY ROLLUP(v1)) +---- +1 + +# cube_empty_input_outer_count: an outer COUNT(*) over CUBE is answered from the inner row-count statistics +query I +SELECT COUNT(*) FROM (SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY CUBE(v1)) +---- +1 + +# grouping_sets_empty_input_outer_count: an outer COUNT(*) over GROUPING SETS is answered from the inner row-count statistics +query I +SELECT COUNT(*) FROM (SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((), (v1))) +---- +1 + +# duplicate_empty_grouping_sets_empty_input: each empty grouping set emits its own grand-total row +query I +SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((v1), (), ()) +---- +NULL +NULL + +# duplicate_empty_grouping_sets_empty_input_outer_count: the row-count statistics must match those two rows +query I +SELECT COUNT(*) FROM (SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((v1), (), ())) +---- +2 + +# duplicate_empty_grouping_sets_empty_input_limit: LIMIT applies to the grand-total rows, so one of +# the two is returned +query I +SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((v1), (), ()) LIMIT 1 +---- +NULL + +# duplicate_empty_grouping_sets_empty_input_limit_outer_count: the outer COUNT(*) is answered from +# the inner row-count statistics, which must agree with the rows the limit lets through +query I +SELECT COUNT(*) FROM (SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((v1), (), ()) LIMIT 1) +---- +1 + +# duplicate_empty_grouping_sets_empty_input_limit_above_row_count_outer_count: a limit above the row +# count leaves both grand-total rows +query I +SELECT COUNT(*) FROM (SELECT SUM(v1) FROM generate_series(10) AS t1(v1) WHERE false GROUP BY GROUPING SETS((v1), (), ()) LIMIT 5) +---- +2 + +# An empty Hive-partitioned file has no rows and exact partition-column statistics, the +# combination an outer MIN/MAX needs to be answered from statistics. +statement ok +COPY (SELECT * FROM (VALUES (1)) v(a) WHERE false) +TO 'test_files/scratch/grouping/p=x/empty.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE hive_partitioned_empty (a INT, p VARCHAR) +STORED AS PARQUET PARTITIONED BY (p) +LOCATION 'test_files/scratch/grouping/'; + +# rollup_empty_input_grand_total_row: the single row ROLLUP emits holds NULL in the grouping column +query T +SELECT p FROM hive_partitioned_empty GROUP BY ROLLUP(p) +---- +NULL + +# rollup_empty_input_outer_min_max: the only row is the NULL grand-total row, so MIN/MAX are NULL +query TT +SELECT MIN(p), MAX(p) FROM (SELECT p FROM hive_partitioned_empty GROUP BY ROLLUP(p)) +---- +NULL NULL + +# cube_empty_input_outer_min_max: the only row is the NULL grand-total row, so MIN/MAX are NULL +query TT +SELECT MIN(p), MAX(p) FROM (SELECT p FROM hive_partitioned_empty GROUP BY CUBE(p)) +---- +NULL NULL + +# group_by_empty_input_outer_min_max: a plain GROUP BY emits no rows, so MIN/MAX are NULL +query TT +SELECT MIN(p), MAX(p) FROM (SELECT p FROM hive_partitioned_empty GROUP BY p) +---- +NULL NULL