limit the max column num in WSCG (#1165)

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

@@ -226,6 +226,9 @@ class GazellePluginConfig(conf: SQLConf) extends Logging {
     conf
       .getConfString("spark.oap.sql.columnar.shuffleSplitDefaultSize", "8192").toInt
 
+  val maxWholeStageCodegenColumnNum: Int =
+    conf.getConfString("spark.oap.sql.columnar.codegen.maxColumnNum", "30").toInt
+
   val numaBindingInfo: GazelleNumaBindingInfo = {
     val enableNumaBinding: Boolean =
       conf.getConfString("spark.oap.sql.columnar.numaBinding", "false").toBoolean

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

@@ -186,7 +186,7 @@ case class ColumnarConditionProjectExec(
         }
       }
     }
-    true
+    super.supportColumnarCodegen
   }
 
   def getKernelFunction(childTreeNode: TreeNode): TreeNode = {

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

@@ -295,7 +295,7 @@ case class ColumnarBroadcastHashJoinExec(
   override def getChild: SparkPlan = streamedPlan
 
   override def supportColumnarCodegen: Boolean = {
-    this.supportCodegen
+    this.supportCodegen && super.supportColumnarCodegen
   }
 
   val output_skip_alias =

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

@@ -729,7 +729,7 @@ case class ColumnarHashAggregateExec(
         }
       }
     }
-    return true
+    super.supportColumnarCodegen
   }
 
   def getKernelFunction: TreeNode = {

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

@@ -287,7 +287,7 @@ case class ColumnarShuffledHashJoinExec(
   }
 
   override def supportColumnarCodegen: Boolean = {
-    this.supportCodegen
+    this.supportCodegen && super.supportColumnarCodegen
   }
 
   val output_skip_alias =

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

@@ -139,7 +139,7 @@ case class ColumnarSortExec(
       Seq(child.executeColumnar())
   }
 
-  override def supportColumnarCodegen: Boolean = true
+  override def supportColumnarCodegen: Boolean = super.supportColumnarCodegen
 
   override def getBuildPlans: Seq[(SparkPlan, SparkPlan)] = child match {
     case c: ColumnarCodegenSupport if c.supportColumnarCodegen == true =>

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

@@ -269,7 +269,7 @@ case class ColumnarSortMergeJoinExec(
         return false
       }
     }
-    true
+    super.supportColumnarCodegen
   }
 
   val output_skip_alias =

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

@@ -46,7 +46,10 @@ trait ColumnarCodegenSupport extends SparkPlan {
   /**
    * Whether this SparkPlan supports whole stage codegen or not.
    */
-  def supportColumnarCodegen: Boolean = true
+  def supportColumnarCodegen: Boolean = {
+    // TODO: support large num columns in WSCG
+    output.size <= GazellePluginConfig.getSessionConf.maxWholeStageCodegenColumnNum
+  }
 
   /**
    * Returns all the RDDs of ColumnarBatch which generates the input rows.
@@ -91,7 +94,7 @@ case class ColumnarWholeStageCodegenExec(child: SparkPlan)(val codegenStageId: I
 
   // This is not strictly needed because the codegen transformation happens after the columnar
   // transformation but just for consistency
-  override def supportColumnarCodegen: Boolean = true
+  override def supportColumnarCodegen: Boolean = super.supportColumnarCodegen
 
   override def supportsColumnar: Boolean = true