Fix/11748 agg emit flag

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<ProjectionExpr> = 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<T, R>(value: (Result<T>, Result<R>)) -> 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<dyn PhysicalOptimizerRule + Send + Sync>,
 }
@@ -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![(

datafusion/core/tests/physical_optimizer/aggregate_statistics.rs

@@ -637,3 +637,110 @@ 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<Schema>,
+    num_rows: Precision<usize>,
+) -> Arc<dyn ExecutionPlan> {
+    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<dyn ExecutionPlan>,
+    agg: &TestAggregate,
+) -> Result<AggregateExec> {
+    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_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),
+    ]));
+    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::<AggregateExec>(),
+        "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);
+    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::<AggregateExec>(),
+        "expected AggregateExec (not optimized), got {}",
+        optimized.name()
+    );
+    Ok(())
+}

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
         "#)
     }
 

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();

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)