Fix panic for GROUPING SETS(()) and handle empty-grouping aggregates (#19252)

## Which issue does this PR close?

* Closes #18974.

## Rationale for this change

The DataFusion CLI currently panics with an "index out of bounds" error
when executing queries that use `GROUP BY GROUPING SETS(())`, such as:

```sql
SELECT SUM(v1) FROM generate_series(10) AS t1(v1) GROUP BY GROUPING SETS(())
```

This panic originates in the physical aggregation code, which assumes
that an empty list of grouping expressions always corresponds to "no
grouping". That assumption breaks down in the presence of `GROUPING
SETS`, where an empty set is a valid grouping set that should still
produce a result row (and `__grouping_id`) rather than crashing.

This PR fixes the panic by explicitly distinguishing:

* true "no GROUP BY" aggregations, and
* `GROUPING SETS`/`CUBE`/`ROLLUP` plans that may have empty grouping
expressions but still require grouping-set semantics and a valid
`__grouping_id`.

The change restores robustness of the CLI and ensures
standards-compliant behavior for grouping sets with empty sets.

## What changes are included in this PR?

Summary of the main changes:

* **Track grouping-set usage explicitly in `PhysicalGroupBy`:**

  * Add a `has_grouping_set: bool` field to `PhysicalGroupBy`.
  * Extend `PhysicalGroupBy::new` to accept the `has_grouping_set` flag.
  * Add helper methods:

    * `has_grouping_set(&self) -> bool` to expose the flag, and
* `is_true_no_grouping(&self) -> bool` to represent the case of
genuinely no grouping (no GROUP BY and no grouping sets).

* **Correct group state construction for empty grouping with grouping
sets:**

* Update `PhysicalGroupBy::from_pre_group` so that it only treats
`expr.is_empty()` as "no groups" when `has_grouping_set` is `false`.
* For `GROUPING SETS(())`, we now build at least one group, avoiding the
previous out-of-bounds access on `groups[0]`.

* **Clarify when `__grouping_id` should be present:**

* Replace the previous `is_single` logic with a clearer distinction
based on `has_grouping_set`.
* `num_output_exprs`, `output_exprs`, `num_group_exprs`, and
`group_schema` now add the `__grouping_id` column only when
`has_grouping_set` is `true`.
* `is_single` is redefined as "simple GROUP BY" (no grouping sets), i.e.
`!self.has_grouping_set`.

* **Integrate the new semantics into `AggregateExec`:**

* Use `group_by.is_true_no_grouping()` instead of
`group_by.expr.is_empty()` when choosing between the specialized
no-grouping aggregation path and grouped aggregation.
* Ensure that `is_unordered_unfiltered_group_by_distinct` only treats
plans as grouped when there are grouping expressions **and** no grouping
sets (`!has_grouping_set`).
* Preserve existing behavior for regular `GROUP BY` while correctly
handling `GROUPING SETS` and related constructs.

* **Support `__grouping_id` with the no-grouping aggregation stream:**

* Extend `AggregateStreamInner` with an optional `grouping_id:
Option<ScalarValue>` field.
  * Change `AggregateStream::new` to accept a `grouping_id` argument.
* Introduce `prepend_grouping_id_column` to prepend a `__grouping_id`
column to the finalized accumulator output when needed.
* Wire this up so that no-grouping aggregations can still match a schema
that includes `__grouping_id` in grouping-set scenarios.

* **Planner and execution wiring updates:**

* Update all `PhysicalGroupBy::new` call sites to pass the correct
`has_grouping_set` value:

    * `false` for:

      * ordinary `GROUP BY` or truly no-grouping aggregates.
    * `true` for:

      * `GROUPING SETS`,
      * `CUBE`, and
      * `ROLLUP` physical planning paths.
* Ensure `merge_grouping_set_physical_expr`,
`create_cube_physical_expr`, and `create_rollup_physical_expr` correctly
mark grouping-set plans.

* **Protobuf / physical plan round-trip support:**

* Extend `AggregateExecNode` in `datafusion.proto` with a new `bool
has_grouping_set = 12;` field.
* Update the generated `pbjson` and `prost` code to serialize and
deserialize the new field.
* When constructing `AggregateExec` from protobuf, pass the decoded
`has_grouping_set` into `PhysicalGroupBy::new`.
* When serializing an `AggregateExec` back to protobuf, set
`has_grouping_set` based on `exec.group_expr().has_grouping_set()`.
* Update round-trip physical plan tests to include the new field in
their expectations.

* **Tests and SQL logic coverage:**

* Add sqllogictests for the previously failing cases in `grouping.slt`:

    * `SELECT COUNT(*) FROM test GROUP BY GROUPING SETS (());`
* `SELECT SUM(v1) FROM generate_series(10) AS t1(v1) GROUP BY GROUPING
SETS(())` (the original panic case).
* Extend or adjust unit tests in `aggregates`, `physical_planner`,
`filter_pushdown`, and `coop` modules to account for the
`has_grouping_set` flag in `PhysicalGroupBy` and expected debug output.
* Update proto round-trip tests to validate `has_grouping_set` is
preserved.

## Are these changes tested?

Yes.

* New sqllogictests covering `GROUPING SETS(())` for both a regular
table and `generate_series(10)`:

* `grouping.slt` now asserts the expected scalar results (e.g. `2` and
`55`), preventing regressions on this edge case.
* Updated and existing Rust unit tests:

* `physical-plan/src/aggregates` tests updated to include
`has_grouping_set` in `PhysicalGroupBy` expectations.
* Planner and optimizer tests (e.g. `physical_planner.rs`,
`filter_pushdown`) updated to construct `PhysicalGroupBy` with the new
flag.
* Execution tests in `core/tests/execution/coop.rs` updated to reflect
the new constructor and continue to exercise the no-grouping aggregation
path.
* Protobuf round-trip tests extended to verify that `has_grouping_set`
is correctly serialized and deserialized.

These tests collectively ensure that:

* the panic is fixed,
* the aggregation semantics for `GROUPING SETS(())` are correct, and
* existing aggregate behavior remains unchanged for non-grouping-set
queries.

## Are there any user-facing changes?

Yes, but they are bug fixes and behavior clarifications rather than
breaking changes:

* Queries using `GROUP BY GROUPING SETS(())` no longer cause a runtime
panic in the DataFusion CLI.

* Instead, they return the expected single aggregate row (e.g.
`COUNT(*)` or `SUM(v1)`), consistent with SQL semantics.
* For plans using `GROUPING SETS`, `CUBE`, or `ROLLUP`, the internal
`__grouping_id` column is now present consistently whenever grouping
sets are in use, even when the grouping expressions are empty.
* For ordinary `GROUP BY` queries that do not use grouping sets,
behavior is unchanged: no unexpected `__grouping_id` column is added.

No API signatures were changed in a breaking way for downstream users;
the additions are internal flags and protobuf fields to accurately
represent the physical plan.

## LLM-generated code disclosure

This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed and tested.

Author: kosiew

datafusion/core/src/physical_planner.rs

@@ -1582,7 +1582,8 @@ impl DefaultPhysicalPlanner {
             }
         } else if group_expr.is_empty() {
             // No GROUP BY clause - create empty PhysicalGroupBy
-            Ok(PhysicalGroupBy::new(vec![], vec![], vec![]))
+            // no expressions, no null expressions and no grouping expressions
+            Ok(PhysicalGroupBy::new(vec![], vec![], vec![], false))
         } else {
             Ok(PhysicalGroupBy::new_single(
                 group_expr
@@ -1654,6 +1655,7 @@ fn merge_grouping_set_physical_expr(
         grouping_set_expr,
         null_exprs,
         merged_sets,
+        true,
     ))
 }
 
@@ -1696,7 +1698,7 @@ fn create_cube_physical_expr(
         }
     }
 
-    Ok(PhysicalGroupBy::new(all_exprs, null_exprs, groups))
+    Ok(PhysicalGroupBy::new(all_exprs, null_exprs, groups, true))
 }
 
 /// Expand and align a ROLLUP expression. This is a special case of GROUPING SETS
@@ -1741,7 +1743,7 @@ fn create_rollup_physical_expr(
         groups.push(group)
     }
 
-    Ok(PhysicalGroupBy::new(all_exprs, null_exprs, groups))
+    Ok(PhysicalGroupBy::new(all_exprs, null_exprs, groups, true))
 }
 
 /// For a given logical expr, get a properly typed NULL ScalarValue physical expression
@@ -2832,6 +2834,7 @@ mod tests {
                         true,
                     ],
                 ],
+                has_grouping_set: true,
             },
         )
         "#);
@@ -2942,6 +2945,7 @@ mod tests {
                         false,
                     ],
                 ],
+                has_grouping_set: true,
             },
         )
         "#);

datafusion/core/tests/execution/coop.rs

@@ -170,7 +170,7 @@ async fn agg_no_grouping_yields(
     let inf = Arc::new(make_lazy_exec("value", pretend_infinite));
     let aggr = Arc::new(AggregateExec::try_new(
         AggregateMode::Single,
-        PhysicalGroupBy::new(vec![], vec![], vec![]),
+        PhysicalGroupBy::new(vec![], vec![], vec![], false),
         vec![Arc::new(
             AggregateExprBuilder::new(
                 sum::sum_udaf(),
@@ -204,7 +204,7 @@ async fn agg_grouping_yields(
 
     let aggr = Arc::new(AggregateExec::try_new(
         AggregateMode::Single,
-        PhysicalGroupBy::new(vec![(group, "group".to_string())], vec![], vec![]),
+        PhysicalGroupBy::new(vec![(group, "group".to_string())], vec![], vec![], false),
         vec![Arc::new(
             AggregateExprBuilder::new(sum::sum_udaf(), vec![value_col.clone()])
                 .schema(inf.schema())
@@ -240,6 +240,7 @@ async fn agg_grouped_topk_yields(
                 vec![(group, "group".to_string())],
                 vec![],
                 vec![vec![false]],
+                false,
             ),
             vec![Arc::new(
                 AggregateExprBuilder::new(min_max::max_udaf(), vec![value_col.clone()])
@@ -545,6 +546,7 @@ async fn interleave_then_aggregate_yields(
             vec![], // no GROUP BY columns
             vec![], // no GROUP BY expressions
             vec![], // no GROUP BY physical expressions
+            false,
         ),
         vec![Arc::new(aggregate_expr)],
         vec![None], // no “distinct” flags
@@ -676,7 +678,7 @@ async fn join_agg_yields(
 
     let aggr = Arc::new(AggregateExec::try_new(
         AggregateMode::Single,
-        PhysicalGroupBy::new(vec![], vec![], vec![]),
+        PhysicalGroupBy::new(vec![], vec![], vec![], false),
         vec![Arc::new(aggregate_expr)],
         vec![None],
         projection,

datafusion/core/tests/physical_optimizer/filter_pushdown/mod.rs

@@ -2545,6 +2545,7 @@ fn test_no_pushdown_grouping_sets_filter_on_missing_column() {
             vec![false, false], // (a, b) - both present
             vec![true, false],  // (b) - a is NULL, b present
         ],
+        true,
     );
 
     let aggregate = Arc::new(
@@ -2615,6 +2616,7 @@ fn test_pushdown_grouping_sets_filter_on_common_column() {
             vec![false, false], // (a, b) - both present
             vec![true, false],  // (b) - a is NULL, b present
         ],
+        true,
     );
 
     let aggregate = Arc::new(

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

@@ -177,6 +177,9 @@ pub struct PhysicalGroupBy {
     /// expression in null_expr. If `groups[i][j]` is true, then the
     /// j-th expression in the i-th group is NULL, otherwise it is `expr[j]`.
     groups: Vec<Vec<bool>>,
+    /// True when GROUPING SETS/CUBE/ROLLUP are used so `__grouping_id` should
+    /// be included in the output schema.
+    has_grouping_set: bool,
 }
 
 impl PhysicalGroupBy {
@@ -185,11 +188,13 @@ impl PhysicalGroupBy {
         expr: Vec<(Arc<dyn PhysicalExpr>, String)>,
         null_expr: Vec<(Arc<dyn PhysicalExpr>, String)>,
         groups: Vec<Vec<bool>>,
+        has_grouping_set: bool,
     ) -> Self {
         Self {
             expr,
             null_expr,
             groups,
+            has_grouping_set,
         }
     }
 
@@ -201,6 +206,7 @@ impl PhysicalGroupBy {
             expr,
             null_expr: vec![],
             groups: vec![vec![false; num_exprs]],
+            has_grouping_set: false,
         }
     }
 
@@ -217,6 +223,11 @@ impl PhysicalGroupBy {
         exprs_nullable
     }
 
+    /// Returns true if this has no grouping at all (including no GROUPING SETS)
+    pub fn is_true_no_grouping(&self) -> bool {
+        self.is_empty() && !self.has_grouping_set
+    }
+
     /// Returns the group expressions
     pub fn expr(&self) -> &[(Arc<dyn PhysicalExpr>, String)] {
         &self.expr
@@ -232,14 +243,20 @@ impl PhysicalGroupBy {
         &self.groups
     }
 
+    /// Returns true if this grouping uses GROUPING SETS, CUBE or ROLLUP.
+    pub fn has_grouping_set(&self) -> bool {
+        self.has_grouping_set
+    }
+
     /// Returns true if this `PhysicalGroupBy` has no group expressions
     pub fn is_empty(&self) -> bool {
         self.expr.is_empty()
     }
 
-    /// Check whether grouping set is single group
+    /// Returns true if this is a "simple" GROUP BY (not using GROUPING SETS/CUBE/ROLLUP).
+    /// This determines whether the `__grouping_id` column is included in the output schema.
     pub fn is_single(&self) -> bool {
-        self.null_expr.is_empty()
+        !self.has_grouping_set
     }
 
     /// Calculate GROUP BY expressions according to input schema.
@@ -253,7 +270,7 @@ impl PhysicalGroupBy {
     /// The number of expressions in the output schema.
     fn num_output_exprs(&self) -> usize {
         let mut num_exprs = self.expr.len();
-        if !self.is_single() {
+        if self.has_grouping_set {
             num_exprs += 1
         }
         num_exprs
@@ -270,7 +287,7 @@ impl PhysicalGroupBy {
                 .take(num_output_exprs)
                 .map(|(index, (_, name))| Arc::new(Column::new(name, index)) as _),
         );
-        if !self.is_single() {
+        if self.has_grouping_set {
             output_exprs.push(Arc::new(Column::new(
                 Aggregate::INTERNAL_GROUPING_ID,
                 self.expr.len(),
@@ -281,11 +298,7 @@ impl PhysicalGroupBy {
 
     /// Returns the number expression as grouping keys.
     pub fn num_group_exprs(&self) -> usize {
-        if self.is_single() {
-            self.expr.len()
-        } else {
-            self.expr.len() + 1
-        }
+        self.expr.len() + usize::from(self.has_grouping_set)
     }
 
     pub fn group_schema(&self, schema: &Schema) -> Result<SchemaRef> {
@@ -308,7 +321,7 @@ impl PhysicalGroupBy {
                 .into(),
             );
         }
-        if !self.is_single() {
+        if self.has_grouping_set {
             fields.push(
                 Field::new(
                     Aggregate::INTERNAL_GROUPING_ID,
@@ -344,17 +357,17 @@ impl PhysicalGroupBy {
                 )
                 .collect();
         let num_exprs = expr.len();
-        let groups = if self.expr.is_empty() {
+        let groups = if self.expr.is_empty() && !self.has_grouping_set {
             // No GROUP BY expressions - should have no groups
             vec![]
         } else {
-            // Has GROUP BY expressions - create a single group
             vec![vec![false; num_exprs]]
         };
         Self {
             expr,
             null_expr: vec![],
             groups,
+            has_grouping_set: false,
         }
     }
 }
@@ -374,6 +387,7 @@ impl PartialEq for PhysicalGroupBy {
                 .zip(other.null_expr.iter())
                 .all(|((expr1, name1), (expr2, name2))| expr1.eq(expr2) && name1 == name2)
             && self.groups == other.groups
+            && self.has_grouping_set == other.has_grouping_set
     }
 }
 
@@ -723,8 +737,7 @@ impl AggregateExec {
         partition: usize,
         context: &Arc<TaskContext>,
     ) -> Result<StreamType> {
-        // no group by at all
-        if self.group_by.expr.is_empty() {
+        if self.group_by.is_true_no_grouping() {
             return Ok(StreamType::AggregateStream(AggregateStream::new(
                 self, context, partition,
             )?));
@@ -757,7 +770,7 @@ impl AggregateExec {
     /// on an AggregateExec.
     pub fn is_unordered_unfiltered_group_by_distinct(&self) -> bool {
         // ensure there is a group by
-        if self.group_expr().is_empty() {
+        if self.group_expr().is_empty() && !self.group_expr().has_grouping_set() {
             return false;
         }
         // ensure there are no aggregate expressions
@@ -1954,6 +1967,7 @@ mod tests {
                 vec![true, false],  // (NULL, b)
                 vec![false, false], // (a,b)
             ],
+            true,
         );
 
         let aggregates = vec![Arc::new(
@@ -2103,6 +2117,7 @@ mod tests {
             vec![(col("a", &input_schema)?, "a".to_string())],
             vec![],
             vec![vec![false]],
+            false,
         );
 
         let aggregates: Vec<Arc<AggregateFunctionExpr>> = vec![Arc::new(
@@ -2448,6 +2463,7 @@ mod tests {
             vec![(col("a", &input_schema)?, "a".to_string())],
             vec![],
             vec![vec![false]],
+            false,
         );
 
         // something that allocates within the aggregator
@@ -2892,6 +2908,7 @@ mod tests {
                 vec![true, false, true],
                 vec![true, true, false],
             ],
+            true,
         );
 
         let aggregates: Vec<Arc<AggregateFunctionExpr>> = vec![
@@ -3251,6 +3268,7 @@ mod tests {
                 vec![false, true],  // (a, NULL)
                 vec![false, false], // (a,b)
             ],
+            true,
         );
         let aggr_schema = create_schema(
             &input_schema,
@@ -3302,6 +3320,7 @@ mod tests {
             vec![(col("a", &schema)?, "a".to_string())],
             vec![],
             vec![vec![false]],
+            false,
         );
 
         // Test with MIN for simple intermediate state (min) and AVG for multiple intermediate states (partial sum, partial count).

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

@@ -251,6 +251,23 @@ fn scalar_cmp_null_short_circuit(
     }
 }
 
+/// Prepend the grouping ID column to the output columns if present.
+///
+/// For GROUPING SETS with no GROUP BY expressions, the schema includes a `__grouping_id`
+/// column that must be present in the output. This function inserts it at the beginning
+/// of the columns array to maintain schema alignment.
+fn prepend_grouping_id_column(
+    mut columns: Vec<Arc<dyn arrow::array::Array>>,
+    grouping_id: Option<&ScalarValue>,
+) -> Result<Vec<Arc<dyn arrow::array::Array>>> {
+    if let Some(id) = grouping_id {
+        let num_rows = columns.first().map(|array| array.len()).unwrap_or(1);
+        let grouping_ids = id.to_array_of_size(num_rows)?;
+        columns.insert(0, grouping_ids);
+    }
+    Ok(columns)
+}
+
 impl AggregateStream {
     /// Create a new AggregateStream
     pub fn new(
@@ -350,6 +367,9 @@ impl AggregateStream {
                         let timer = this.baseline_metrics.elapsed_compute().timer();
                         let result =
                             finalize_aggregation(&mut this.accumulators, &this.mode)
+                                .and_then(|columns| {
+                                    prepend_grouping_id_column(columns, None)
+                                })
                                 .and_then(|columns| {
                                     RecordBatch::try_new(
                                         Arc::clone(&this.schema),

datafusion/proto/proto/datafusion.proto

@@ -1223,6 +1223,7 @@ message AggregateExecNode {
   repeated bool groups = 9;
   repeated MaybeFilter filter_expr = 10;
   AggLimit limit = 11;
+  bool has_grouping_set = 12;
 }
 
 message GlobalLimitExecNode {