[WIP][CORE] Add PullOutPreProject rule to decouple substrait pre-project

backends-clickhouse/src/main/scala/io/glutenproject/backendsapi/clickhouse/CHSparkPlanExecApi.scala

@@ -338,15 +338,6 @@ class CHSparkPlanExecApi extends SparkPlanExecApi {
     }
   }
 
-  /**
-   * Generate extended Optimizers.
-   *
-   * @return
-   */
-  override def genExtendedOptimizers(): List[SparkSession => Rule[LogicalPlan]] = {
-    List.empty
-  }
-
   /**
    * Generate extended columnar pre-rules.
    *

backends-clickhouse/src/main/scala/io/glutenproject/execution/CHHashAggregateExecTransformer.scala

@@ -174,18 +174,12 @@ case class CHHashAggregateExecTransformer(
       aggParams: AggregationParams,
       input: RelNode = null,
       validation: Boolean = false): RelNode = {
-    val originalInputAttributes = child.output
-    val aggRel = if (needsPreProjection) {
-      aggParams.preProjectionNeeded = true
-      getAggRelWithPreProjection(context, originalInputAttributes, operatorId, input, validation)
-    } else {
-      getAggRelWithoutPreProjection(
-        context,
-        aggregateResultAttributes,
-        operatorId,
-        input,
-        validation)
-    }
+    val aggRel = getAggRelWithoutPreProjection(
+      context,
+      aggregateResultAttributes,
+      operatorId,
+      input,
+      validation)
     // Will check if post-projection is needed. If yes, a ProjectRel will be added after the
     // AggregateRel.
     val resRel = if (!needsPostProjection(allAggregateResultAttributes)) {

backends-velox/src/main/scala/io/glutenproject/backendsapi/velox/SparkPlanExecApiImpl.scala

@@ -447,8 +447,9 @@ class SparkPlanExecApiImpl extends SparkPlanExecApi {
    *
    * @return
    */
-  override def genExtendedOptimizers(): List[SparkSession => Rule[LogicalPlan]] =
-    List(AggregateFunctionRewriteRule)
+  override def genExtendedOptimizers(): List[SparkSession => Rule[LogicalPlan]] = {
+    super.genExtendedOptimizers ++ List(AggregateFunctionRewriteRule)
+  }
 
   /**
    * Generate extended columnar pre-rules.

backends-velox/src/main/scala/io/glutenproject/execution/HashAggregateExecTransformer.scala

@@ -804,21 +804,11 @@ case class HashAggregateExecTransformer(
       validation: Boolean = false): RelNode = {
     val originalInputAttributes = child.output
 
-    var aggRel = if (needsPreProjection) {
+    var aggRel = if (rowConstructNeeded) {
       aggParams.preProjectionNeeded = true
-      getAggRelWithPreProjection(context, originalInputAttributes, operatorId, input, validation)
+      getAggRelWithRowConstruct(context, originalInputAttributes, operatorId, input, validation)
     } else {
-      if (rowConstructNeeded) {
-        aggParams.preProjectionNeeded = true
-        getAggRelWithRowConstruct(context, originalInputAttributes, operatorId, input, validation)
-      } else {
-        getAggRelWithoutPreProjection(
-          context,
-          originalInputAttributes,
-          operatorId,
-          input,
-          validation)
-      }
+      getAggRelWithoutPreProjection(context, originalInputAttributes, operatorId, input, validation)
     }
 
     if (extractStructNeeded()) {

gluten-core/src/main/scala/io/glutenproject/backendsapi/SparkPlanExecApi.scala

@@ -18,6 +18,7 @@ package io.glutenproject.backendsapi
 
 import io.glutenproject.execution._
 import io.glutenproject.expression._
+import io.glutenproject.extension.PullOutPreProject
 import io.glutenproject.substrait.expression.{ExpressionBuilder, ExpressionNode, WindowFunctionNode}
 
 import org.apache.spark.ShuffleDependency
@@ -200,7 +201,9 @@ trait SparkPlanExecApi {
    *
    * @return
    */
-  def genExtendedOptimizers(): List[SparkSession => Rule[LogicalPlan]]
+  def genExtendedOptimizers(): List[SparkSession => Rule[LogicalPlan]] = {
+    List(PullOutPreProject)
+  }
 
   /**
    * Generate extended Strategies

gluten-core/src/main/scala/io/glutenproject/execution/HashAggregateExecBaseTransformer.scala

@@ -154,29 +154,6 @@ abstract class HashAggregateExecBaseTransformer(
   // Members declared in org.apache.spark.sql.execution.AliasAwareOutputPartitioning
   override protected def outputExpressions: Seq[NamedExpression] = resultExpressions
 
-  // Check if Pre-Projection is needed before the Aggregation.
-  protected def needsPreProjection: Boolean = {
-    groupingExpressions.exists {
-      case _: Attribute => false
-      case _ => true
-    } || aggregateExpressions.exists {
-      expr =>
-        expr.filter match {
-          case None | Some(_: Attribute) | Some(_: Literal) =>
-          case _ => return true
-        }
-        expr.mode match {
-          case Partial =>
-            expr.aggregateFunction.children.exists {
-              case _: Attribute | _: Literal => false
-              case _ => true
-            }
-          // No need to consider pre-projection for PartialMerge and Final Agg.
-          case _ => false
-        }
-    }
-  }
-
   // Check if Post-Projection is needed after the Aggregation.
   protected def needsPostProjection(aggOutAttributes: List[Attribute]): Boolean = {
     // If the result expressions has different size with output attribute,
@@ -195,148 +172,6 @@ abstract class HashAggregateExecBaseTransformer(
     }
   }
 
-  protected def getAggRelWithPreProjection(
-      context: SubstraitContext,
-      originalInputAttributes: Seq[Attribute],
-      operatorId: Long,
-      input: RelNode = null,
-      validation: Boolean): RelNode = {
-    val args = context.registeredFunction
-    // Will add a Projection before Aggregate.
-    // Logic was added to prevent selecting the same column for more than once,
-    // so the expression in preExpressions will be unique.
-    var preExpressions: Seq[Expression] = Seq()
-    var selections: Seq[Int] = Seq()
-    // Indices of filter used columns.
-    var filterSelections: Seq[Int] = Seq()
-
-    def appendIfNotFound(expression: Expression): Unit = {
-      val foundExpr = preExpressions.find(e => e.semanticEquals(expression)).orNull
-      if (foundExpr != null) {
-        // If found, no need to add it to preExpressions again.
-        // The selecting index will be found.
-        selections = selections :+ preExpressions.indexOf(foundExpr)
-      } else {
-        // If not found, add this expression into preExpressions.
-        // A new selecting index will be created.
-        preExpressions = preExpressions :+ expression.clone()
-        selections = selections :+ (preExpressions.size - 1)
-      }
-    }
-
-    // Get the needed expressions from grouping expressions.
-    groupingExpressions.foreach(expression => appendIfNotFound(expression))
-
-    // Get the needed expressions from aggregation expressions.
-    aggregateExpressions.foreach(
-      aggExpr => {
-        val aggregateFunc = aggExpr.aggregateFunction
-        aggExpr.mode match {
-          case Partial =>
-            aggregateFunc.children.foreach(expression => appendIfNotFound(expression))
-          case other =>
-            throw new UnsupportedOperationException(s"$other not supported.")
-        }
-      })
-
-    // Handle expressions used in Aggregate filter.
-    aggregateExpressions.foreach(
-      aggExpr => {
-        if (aggExpr.filter.isDefined) {
-          appendIfNotFound(aggExpr.filter.orNull)
-          filterSelections = filterSelections :+ selections.last
-        }
-      })
-
-    // Create the expression nodes needed by Project node.
-    val preExprNodes = preExpressions
-      .map(
-        ExpressionConverter
-          .replaceWithExpressionTransformer(_, originalInputAttributes)
-          .doTransform(args))
-      .asJava
-    val emitStartIndex = originalInputAttributes.size
-    val inputRel = if (!validation) {
-      RelBuilder.makeProjectRel(input, preExprNodes, context, operatorId, emitStartIndex)
-    } else {
-      // Use a extension node to send the input types through Substrait plan for a validation.
-      val inputTypeNodeList = originalInputAttributes
-        .map(attr => ConverterUtils.getTypeNode(attr.dataType, attr.nullable))
-        .asJava
-      val extensionNode = ExtensionBuilder.makeAdvancedExtension(
-        Any.pack(TypeBuilder.makeStruct(false, inputTypeNodeList).toProtobuf))
-      RelBuilder.makeProjectRel(
-        input,
-        preExprNodes,
-        extensionNode,
-        context,
-        operatorId,
-        emitStartIndex)
-    }
-
-    // Handle the pure Aggregate after Projection. Both grouping and Aggregate expressions are
-    // selections.
-    getAggRelAfterProject(context, selections, filterSelections, inputRel, operatorId)
-  }
-
-  protected def getAggRelAfterProject(
-      context: SubstraitContext,
-      selections: Seq[Int],
-      filterSelections: Seq[Int],
-      inputRel: RelNode,
-      operatorId: Long): RelNode = {
-    val groupingList = new JArrayList[ExpressionNode]()
-    var colIdx = 0
-    while (colIdx < groupingExpressions.size) {
-      val groupingExpr: ExpressionNode = ExpressionBuilder.makeSelection(selections(colIdx))
-      groupingList.add(groupingExpr)
-      colIdx += 1
-    }
-
-    // Create Aggregation functions.
-    val aggregateFunctionList = new JArrayList[AggregateFunctionNode]()
-    aggregateExpressions.foreach(
-      aggExpr => {
-        val aggregateFunc = aggExpr.aggregateFunction
-        val childrenNodeList = new JArrayList[ExpressionNode]()
-        val childrenNodes = aggregateFunc.children.toList.map(
-          _ => {
-            val aggExpr = ExpressionBuilder.makeSelection(selections(colIdx))
-            colIdx += 1
-            aggExpr
-          })
-        for (node <- childrenNodes) {
-          childrenNodeList.add(node)
-        }
-        addFunctionNode(
-          context.registeredFunction,
-          aggregateFunc,
-          childrenNodeList,
-          aggExpr.mode,
-          aggregateFunctionList)
-      })
-
-    val aggFilterList = new JArrayList[ExpressionNode]()
-    aggregateExpressions.foreach(
-      aggExpr => {
-        if (aggExpr.filter.isDefined) {
-          aggFilterList.add(ExpressionBuilder.makeSelection(selections(colIdx)))
-          colIdx += 1
-        } else {
-          // The number of filters should be aligned with that of aggregate functions.
-          aggFilterList.add(null)
-        }
-      })
-
-    RelBuilder.makeAggregateRel(
-      inputRel,
-      groupingList,
-      aggregateFunctionList,
-      aggFilterList,
-      context,
-      operatorId)
-  }
-
   protected def addFunctionNode(
       args: java.lang.Object,
       aggregateFunction: AggregateFunction,