[SQL] Rewrite UNION ALL of aggregates as ROLLUP

Add an optimizer rule, RewriteUnionAggregateAsRollup, that detects a UNION ALL
of N aggregates whose GROUP BY clauses form a complete prefix-shrinking
hierarchy ([c1..cN], [c1..cN-1], ..., [c1], []) and rewrites it into a single
Aggregate with ROLLUP. Machine-generated SQL (BI tools, hand-translated
queries) commonly emits this hand-expanded shape; collapsing it removes the
redundant repeated scans and aggregations.

Two patterns are recognized:

- Inner-aggregate fold (CTE pattern): the branches fold a shared inner
  aggregate -- sum-of-sum, max-of-max, min-of-min, or sum-of-count. The rewrite
  rolls up OVER the inner aggregate, so the inner level's decimal width and
  overflow behavior are reproduced exactly.

- Source-aggregate (direct pattern): each branch independently aggregates the
  same source, so any deterministic aggregate is allowed (SUM, MAX, MIN, COUNT,
  AVG, COUNT(DISTINCT), COLLECT_SET, FILTERed), provided every branch's
  per-position measure signature is identical.

Correctness is preserved by reconstructing every rollup level from a single
reference branch and binding to the Union output by position, with conservative
guards that REJECT (never mis-apply) any branch the rewrite cannot reproduce
exactly: per-position measure identity (matched by inner-output name, stable
across CTE-copy ExprIds and coercion casts), grouping-key passthrough names,
per-position role layout, eval-mode/timezone/FILTER/DISTINCT signatures, and
value-changing projections above a branch aggregate. The grand-total
(GROUP BY ()) level is unioned back verbatim so empty-input semantics (a global
aggregate emits one row, a grouped ROLLUP zero) are preserved.

Gated by spark.sql.optimizer.unionAggregateToRollup.enabled (internal, default
true).

Adds catalyst unit tests and end-to-end equivalence tests (each firing query is
checked rule-on == rule-off) and regenerates the affected TPCDS v2_7 golden
files (q5a, q14a, q18a, q67a, q77a, q80a).

Author: LuciferYang

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -224,7 +224,8 @@ abstract class Optimizer(catalogManager: CatalogManager)
       ReplaceIntersectWithSemiJoin,
       ReplaceExceptWithFilter,
       ReplaceExceptWithAntiJoin,
-      ReplaceDistinctWithAggregate),
+      ReplaceDistinctWithAggregate,
+      RewriteUnionAggregateAsRollup),
     Batch("Aggregate", fixedPoint,
       RemoveLiteralFromGroupExpressions,
       RemoveRepetitionFromGroupExpressions),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteUnionAggregateAsRollup.scala

@@ -183,6 +183,7 @@ object RuleIdCollection {
       "org.apache.spark.sql.catalyst.optimizer.RewriteNearestByJoin" ::
       "org.apache.spark.sql.catalyst.optimizer.RewriteExceptAll" ::
       "org.apache.spark.sql.catalyst.optimizer.RewriteIntersectAll" ::
+      "org.apache.spark.sql.catalyst.optimizer.RewriteUnionAggregateAsRollup" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyBinaryComparison" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyCaseConversionExpressions" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyCasts" ::

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala

@@ -595,6 +595,18 @@ object SQLConf {
         "for using switch statements in InSet must be non-negative and less than or equal to 600")
       .createWithDefault(400)
 
+  val UNION_AGGREGATE_TO_ROLLUP_ENABLED =
+    buildConf("spark.sql.optimizer.unionAggregateToRollup.enabled")
+      .internal()
+      .doc("When true, the optimizer rewrites a UNION ALL of N aggregates whose GROUP BYs " +
+        "form a strict prefix-shrinking hierarchy into a single Aggregate with ROLLUP. " +
+        "Handles both the inner-aggregate fold pattern (sum-of-sum, max-of-max, etc.) and " +
+        "the source-aggregate pattern (each branch independently aggregates the same source).")
+      .version("4.3.0")
+      .withBindingPolicy(ConfigBindingPolicy.SESSION)
+      .booleanConf
+      .createWithDefault(true)
+
   private val VALID_LOG_LEVELS: Array[String] = Level.values.map(_.toString)
 
   val PLAN_CHANGE_LOG_LEVEL = buildConf("spark.sql.planChangeLog.level")