MaxGekk commented on code in PR #56891:
URL: https://github.com/apache/spark/pull/56891#discussion_r3500534995
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -132,6 +132,17 @@ class DataFrameAggregateSuite extends SharedSparkSession
)
}
+ test("cube()/rollup() with no grouping columns return one grand-total row
over empty input") {
+ // With no grouping columns, cube()/rollup() lower to a global aggregate
(the grand total),
+ // which returns one row even over empty input -- like an aggregation with
no GROUP BY clause.
+ // This is the DataFrame-API surface for the empty CUBE/ROLLUP case (not
expressible in SQL).
+ checkAnswer(spark.range(0).cube().count(), Row(0L))
Review Comment:
Good call covering empty `cube()`/`rollup()` here — it's the only reachable
surface, since the grammar requires ≥1 grouping set inside
`CUBE(...)`/`ROLLUP(...)`, so empty `CUBE()`/`ROLLUP()` isn't SQL-expressible.
Optional: a Spark Connect DataFrame case would additionally exercise the
Connect analyzed-plan path, which differs from classic.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/GroupingAnalyticsTransformer.scala:
##########
@@ -81,34 +88,117 @@ object GroupingAnalyticsTransformer extends SQLConfHelper
with AliasHelper {
child: LogicalPlan,
aggregationExpressions: Seq[NamedExpression]): Aggregate = {
- val groupByAliases = constructGroupByAlias(newAlias, groupByExpressions)
+ if (shouldLowerToGrandTotalAggregate(groupByExpressions,
selectedGroupByExpressions)) {
+ constructGrandTotalAggregate(newAlias, aggregationExpressions, child)
+ } else {
+ val groupByAliases = constructGroupByAlias(newAlias, groupByExpressions)
- val gid = AttributeReference(VirtualColumn.groupingIdName,
GroupingID.dataType, false)()
- val expand = constructExpand(
- selectedGroupByExpressions = selectedGroupByExpressions,
- child = child,
- groupByAliases = groupByAliases,
- gid = gid,
- childOutput = childOutput
- )
- val groupingAttributes = expand.output.drop(childOutput.length)
+ val gid = AttributeReference(VirtualColumn.groupingIdName,
GroupingID.dataType, false)()
+ val expand = constructExpand(
+ selectedGroupByExpressions = selectedGroupByExpressions,
+ child = child,
+ groupByAliases = groupByAliases,
+ gid = gid,
+ childOutput = childOutput
+ )
+ val groupingAttributes = expand.output.drop(childOutput.length)
- val aggregations = constructAggregateExpressions(
- newAlias = newAlias,
- groupByExpressions = groupByExpressions,
- aggregations = aggregationExpressions,
- groupByAliases = groupByAliases,
- groupingAttributes = groupingAttributes,
- gid = gid
- )
+ val aggregations = constructAggregateExpressions(
+ newAlias = newAlias,
+ groupByExpressions = groupByExpressions,
+ aggregations = aggregationExpressions,
+ groupByAliases = groupByAliases,
+ groupingAttributes = groupingAttributes,
+ gid = gid
+ )
+
+ Aggregate(
+ groupingExpressions = groupingAttributes,
+ aggregateExpressions = aggregations,
+ child = expand
+ )
+ }
+ }
- val aggregate = Aggregate(
- groupingExpressions = groupingAttributes,
+ /**
+ * Whether a grouping-set spec is the grand-total-only case `GROUP BY
GROUPING SETS (())` (and
+ * the equivalent empty `CUBE()`/`ROLLUP()`) that [[apply]] lowers to a
global [[Aggregate]]:
+ * no leading group-by expressions and a single empty grouping set, with
+ * [[SQLConf.LOWER_EMPTY_GROUPING_SET_TO_GLOBAL_AGGREGATE]] enabled. This is
the
+ * pre-lowering decision; [[isLoweredToGrandTotalAggregate]] detects the
same case post-lowering.
+ */
+ def shouldLowerToGrandTotalAggregate(
+ groupByExpressions: Seq[Expression],
+ selectedGroupByExpressions: Seq[Seq[Expression]]): Boolean = {
+ conf.getConf(SQLConf.LOWER_EMPTY_GROUPING_SET_TO_GLOBAL_AGGREGATE) &&
+ groupByExpressions.isEmpty &&
+ selectedGroupByExpressions.length == 1 &&
+ selectedGroupByExpressions.head.isEmpty
+ }
+
+ /**
+ * Whether a lowered [[Aggregate]] is the grand total produced by
+ * [[shouldLowerToGrandTotalAggregate]]: its resolved grouping expressions
(without the
+ * `spark_grouping_id` key, as returned by [[collectGroupingExpressions]])
are empty, with the
+ * flag enabled. The flag is part of the check because with it off the same
grand total is
+ * lowered via [[Expand]], whose `spark_grouping_id` key must still be
referenced.
+ *
+ * This keys purely on the collected grouping expressions being empty, so it
also matches a
+ * value-equivalent all-empty multi-set [[Expand]] aggregate (e.g. `GROUP BY
GROUPING SETS ((),
Review Comment:
This `((),())` value-equivalence is the right call (two empty sets stay on
the Expand path, but every Expand row has `spark_grouping_id = 0`, so folding
`grouping_id()` to `0` matches — and both analyzers fold identically via the
shared helpers, so no divergence). It's just untested: worth a golden case for
`GROUP BY GROUPING SETS ((), ())` with `grouping_id()` in SELECT/HAVING/ORDER
BY, flag-on and flag-off, to lock the documented invariant down.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/GroupingAnalyticsTransformer.scala:
##########
@@ -153,19 +243,27 @@ object GroupingAnalyticsTransformer extends SQLConfHelper
with AliasHelper {
/**
* Collect the last grouping expression since the provided [[Aggregate]]
should have grouping id
* as the last grouping key.
+ *
+ * A grand-total `GROUP BY GROUPING SETS (())` is lowered to a global
[[Aggregate]] with no
+ * grouping expressions (see [[apply]]), so there is no `spark_grouping_id`
key and no grouping
+ * columns; return an empty sequence for it.
*/
def collectGroupingExpressions(aggregate: Aggregate): Seq[Expression] = {
- val gid = aggregate.groupingExpressions.last
- gid match {
- case attributeReference: AttributeReference =>
- if (attributeReference.name != VirtualColumn.groupingIdName) {
+ if (aggregate.groupingExpressions.isEmpty) {
Review Comment:
With this guard returning `Seq.empty`, and `isLoweredToGrandTotalAggregate`
keying purely on emptiness, a lowered grand total becomes indistinguishable
from a plain no-`GROUP BY` aggregate (both have empty `groupingExpressions`).
The `Filter`/`Sort` grouping-function cases in `Analyzer` fire on
`hasGroupingFunction(cond)` with no child-type guard, so
`grouping_id()`/`grouping()` in HAVING/ORDER BY over a plain aggregate — e.g.
`SELECT count(*) FROM t HAVING grouping_id() = 0` — appears to fold to
`Literal(0)` and succeed here, whereas the SELECT-list form errors
`UNSUPPORTED_GROUPING_EXPRESSION` via `CheckAnalysis` (and pre-PR this path hit
`Seq.last` on empty). Could you confirm the behavior? A negative golden case (a
grouping function in HAVING and in ORDER BY with no grouping set) would pin it
down; if it does fold, a child-type guard so a plain aggregate still rejects it
would keep parity with the SELECT-list path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
