[GLUTEN-7751][VL] Merge two consecutive aggregates to one in complete…

… mode (#7752)

What changes were proposed in this pull request?
Fix #7751.

In issue #4668, support for the CH backend's merge agg capability was added. We can also support this feature in the velox backend.

How was this patch tested?
Through existing unit tests and GA verification, and additional new unit tests have been added.

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala

@@ -68,6 +68,7 @@ object VeloxRuleApi {
     injector.injectPreTransform(_ => PushDownInputFileExpression.PreOffload)
     injector.injectPreTransform(c => FallbackOnANSIMode.apply(c.session))
     injector.injectPreTransform(c => FallbackMultiCodegens.apply(c.session))
+    injector.injectPreTransform(c => MergeTwoPhasesHashBaseAggregate(c.session))
     injector.injectPreTransform(_ => RewriteSubqueryBroadcast())
     injector.injectPreTransform(c => BloomFilterMightContainJointRewriteRule.apply(c.session))
     injector.injectPreTransform(c => ArrowScanReplaceRule.apply(c.session))

backends-velox/src/main/scala/org/apache/gluten/execution/HashAggregateExecTransformer.scala

@@ -214,7 +214,7 @@ abstract class HashAggregateExecTransformer(
             VeloxIntermediateData.getIntermediateTypeNode(aggregateFunction)
           )
           aggregateNodeList.add(aggFunctionNode)
-        case Final =>
+        case Final | Complete =>
           val aggFunctionNode = ExpressionBuilder.makeAggregateFunction(
             VeloxAggregateFunctionsBuilder.create(args, aggregateFunction, aggregateMode),
             childrenNodeList,
@@ -242,7 +242,7 @@ abstract class HashAggregateExecTransformer(
                 aggregateFunction.inputAggBufferAttributes.head.nullable)
             )
             aggregateNodeList.add(partialNode)
-          case Final =>
+          case Final | Complete =>
             val aggFunctionNode = ExpressionBuilder.makeAggregateFunction(
               VeloxAggregateFunctionsBuilder.create(args, aggregateFunction, aggregateMode),
               childrenNodeList,
@@ -275,7 +275,7 @@ abstract class HashAggregateExecTransformer(
             expression.mode match {
               case Partial | PartialMerge =>
                 typeNodeList.add(VeloxIntermediateData.getIntermediateTypeNode(aggregateFunction))
-              case Final =>
+              case Final | Complete =>
                 typeNodeList.add(
                   ConverterUtils
                     .getTypeNode(aggregateFunction.dataType, aggregateFunction.nullable))
@@ -356,7 +356,7 @@ abstract class HashAggregateExecTransformer(
           // The process of handling the inconsistency in column types and order between
           // Spark and Velox is exactly the opposite of applyExtractStruct.
           aggregateExpression.mode match {
-            case PartialMerge | Final =>
+            case PartialMerge | Final | Complete =>
               val newInputAttributes = new ArrayBuffer[Attribute]()
               val childNodes = new JArrayList[ExpressionNode]()
               val (sparkOrders, sparkTypes) =
@@ -467,7 +467,7 @@ abstract class HashAggregateExecTransformer(
             // by previous projection.
             childrenNodes.add(ExpressionBuilder.makeSelection(colIdx))
             colIdx += 1
-          case Partial =>
+          case Partial | Complete =>
             aggFunc.children.foreach {
               _ =>
                 childrenNodes.add(ExpressionBuilder.makeSelection(colIdx))
@@ -600,7 +600,7 @@ abstract class HashAggregateExecTransformer(
         }
         val aggregateFunc = aggExpr.aggregateFunction
         val childrenNodes = aggExpr.mode match {
-          case Partial =>
+          case Partial | Complete =>
             aggregateFunc.children.toList.map(
               expr => {
                 ExpressionConverter
@@ -784,7 +784,7 @@ case class HashAggregateExecPullOutHelper(
         expr.mode match {
           case Partial | PartialMerge =>
             expr.aggregateFunction.aggBufferAttributes
-          case Final =>
+          case Final | Complete =>
             Seq(aggregateAttributes(index))
           case other =>
             throw new GlutenNotSupportException(s"Unsupported aggregate mode: $other.")