[NSE-1118] adding more checks for SMJ codegen (#1119)

* adding more checks for SMJ codegen Signed-off-by: Yuan Zhou <yuan.zhou@intel.com> * disable codegen on more cases Signed-off-by: Yuan Zhou <yuan.zhou@intel.com> Signed-off-by: Yuan Zhou <yuan.zhou@intel.com>
oap-project · Sep 28, 2022 · 7ba0c4c · 7ba0c4c
1 parent 2a9a03c
commit 7ba0c4c
Show file tree

Hide file tree

Showing 3 changed files with 45 additions and 1 deletion.
diff --git a/...ve-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarHashAggregateExec.scala b/...ve-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarHashAggregateExec.scala
@@ -698,6 +698,9 @@ case class ColumnarHashAggregateExec(
       }
       val internalExpressionList = expr.aggregateFunction.children
       for (expr <- internalExpressionList) {
+        if (expr.isInstanceOf[Literal]) {
+          return false
+        }
         val colExpr = ColumnarExpressionConverter.replaceWithColumnarExpression(expr)
         if (!colExpr.asInstanceOf[ColumnarExpression].supportColumnarCodegen(
           Lists.newArrayList())) {

diff --git a/...ve-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarSortMergeJoinExec.scala b/...ve-sql-engine/core/src/main/scala/com/intel/oap/execution/ColumnarSortMergeJoinExec.scala
@@ -236,7 +236,41 @@ case class ColumnarSortMergeJoinExec(
   }
 
   // Only has codegen implementation.
-  override def supportColumnarCodegen: Boolean = true
+  override def supportColumnarCodegen: Boolean = {
+    // build check for expr
+    if (leftKeys != null) {
+      for (expr <- leftKeys) {
+        val columnarExpr = ColumnarExpressionConverter.replaceWithColumnarExpression(expr)
+        val supportCodegen =
+          columnarExpr.asInstanceOf[ColumnarExpression].supportColumnarCodegen(null)
+        if (!supportCodegen) {
+          return false
+        }
+      }
+    }
+    if (rightKeys != null) {
+      for (expr <- rightKeys) {
+        val columnarExpr = ColumnarExpressionConverter.replaceWithColumnarExpression(expr)
+        val supportCodegen =
+          columnarExpr.asInstanceOf[ColumnarExpression].supportColumnarCodegen(null)
+        if (!supportCodegen) {
+          return false
+        }
+      }
+    }
+    val conditionExpr: Expression = condition.orNull
+    if (conditionExpr != null) {
+      val columnarConditionExpr =
+        ColumnarExpressionConverter.replaceWithColumnarExpression(conditionExpr)
+      val supportCodegen =
+        columnarConditionExpr.asInstanceOf[ColumnarExpression].supportColumnarCodegen(null)
+      // Columnar SMJ only has codegen version of implementation.
+      if (!supportCodegen) {
+        return false
+      }
+    }
+    true
+  }
 
   val output_skip_alias =
     if (projectList == null || projectList.isEmpty) output

diff --git a/...ne/core/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseCodegenStages.scala b/...ne/core/src/main/scala/org/apache/spark/sql/execution/ColumnarCollapseCodegenStages.scala
@@ -239,6 +239,13 @@ case class ColumnarCollapseCodegenStages(
     */
       case p: ColumnarSortMergeJoinExec if p.left.isInstanceOf[ColumnarLocalLimitExec] =>
         true
+    /**
+    * To filter Inner SMJ and its right child is Projection.
+    */
+      case p: ColumnarSortMergeJoinExec if p.right.isInstanceOf[ColumnarConditionProjectExec] 
+        && p.joinType == Inner
+        && (p.right.asInstanceOf[ColumnarConditionProjectExec].condition != null) =>
+        true
       case _ =>
         false
     }