[NSE-981] Add a test suite for projection codegen

native-sql-engine/core/src/main/scala/com/intel/oap/GazellePluginConfig.scala

@@ -243,6 +243,12 @@ class GazellePluginConfig(conf: SQLConf) extends Logging {
   val enableUDFKey: String = "spark.oap.sql.columnar.enable.udf"
   val enableUDF: Boolean =
     conf.getConfString(enableUDFKey, "false").toBoolean
+
+  // To enable the codegen for projection without the requirement for join existence.
+  // This config is just for test use.
+  val enableProjectionCodegenKey: String = "spark.oap.sql.columnar.projection.codegen"
+  val enableProjectionCodegen: Boolean =
+    conf.getConfString(enableProjectionCodegenKey, "false").toBoolean
 }
 
 object GazellePluginConfig {

native-sql-engine/core/src/main/scala/com/intel/oap/expression/ColumnarExpression.scala

@@ -55,9 +55,13 @@ trait ColumnarExpression {
     "castDATE",
     "castDECIMAL",
     "castDECIMALNullOnOverflow",
+    "castINT",
     "castINTOrNull",
+    "castBIGINT",
     "castBIGINTOrNull",
+    "castFLOAT4",
     "castFLOAT4OrNull",
+    "castFLOAT8",
     "castFLOAT8OrNull",
     "rescaleDECIMAL",
     "extractYear",

native-sql-engine/core/src/main/scala/com/intel/oap/expression/ColumnarUnaryOperator.scala

@@ -423,7 +423,7 @@ class ColumnarLower(child: Expression, original: Expression)
   val gName = "lower"
 
   override def supportColumnarCodegen(args: java.lang.Object): Boolean = {
-    codegenFuncList.contains(gName) && 
+    codegenFuncList.contains(gName) &&
     child.asInstanceOf[ColumnarExpression].supportColumnarCodegen(args)
   }
 
@@ -527,7 +527,30 @@ class ColumnarCast(
   val gName = "Cast"
 
   override def supportColumnarCodegen(args: java.lang.Object): Boolean = {
-    true && 
+    // Casting data to TimestampType/BinaryType is not supported in codegen.
+    if (dataType.isInstanceOf[TimestampType] || dataType == BinaryType) {
+      return false
+    }
+    if (dataType == DateType) {
+      child.dataType match {
+        case TimestampType =>
+          return false
+        case StringType =>
+          return false
+        case _ =>
+      }
+    }
+    if (dataType == StringType) {
+      child.dataType match {
+        case TimestampType =>
+          return false
+        case DateType =>
+          return false
+        case _ =>
+      }
+    }
+
+    true &&
     child.asInstanceOf[ColumnarExpression].supportColumnarCodegen(args)
   }
 

native-sql-engine/core/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseCodegenStages.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution
 
 import java.util.concurrent.atomic.AtomicInteger
 
+import com.intel.oap.GazellePluginConfig
 import com.intel.oap.execution._
 import com.intel.oap.expression.ColumnarExpressionConverter
 import org.apache.spark._
@@ -118,6 +119,8 @@ case class ColumnarCollapseCodegenStages(
     codegenStageCounter: AtomicInteger = new AtomicInteger(0))
     extends Rule[SparkPlan] {
 
+  val enableProjectionCodegen = GazellePluginConfig.getSessionConf.enableProjectionCodegen
+
   private def supportCodegen(plan: SparkPlan): Boolean = plan match {
     case plan: ColumnarCodegenSupport =>
       plan.supportColumnarCodegen
@@ -315,6 +318,10 @@ case class ColumnarCollapseCodegenStages(
       case s: ColumnarSortExec =>
         /*If ColumnarSort is not ahead of ColumnarSMJ, we should not do wscg for it*/
         s.withNewChildren(s.children.map(insertWholeStageCodegen))
+      // For testing projection codegen.
+      case plan: ColumnarCodegenSupport if enableProjectionCodegen && supportCodegen(plan) =>
+        ColumnarWholeStageCodegenExec(insertInputAdapter(plan))(
+          codegenStageCounter.incrementAndGet())
       case plan: ColumnarCodegenSupport if supportCodegen(plan) && existsJoins(plan) =>
         ColumnarWholeStageCodegenExec(insertInputAdapter(plan))(
           codegenStageCounter.incrementAndGet())