[NSE-5][SCALA] Fix ColumnarBroadcastExchange didn't fallback issue w/…

… DPP (#6) * [SCALA] Fix ColumnarBroadcastExchange didn't fallback issue when DPP is enabled Signed-off-by: Chendi Xue <[email protected]> * [SCALA & JAVA] Fix ExpressionEvaluator api change null pointer issue Signed-off-by: Chendi Xue <[email protected]>
oap-project · Jan 5, 2021 · 089ccf9 · 089ccf9
1 parent cea3f04
commit 089ccf9
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 1 deletion.
diff --git a/core/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java b/core/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java
@@ -238,6 +238,7 @@ public void close() {
   }
 
   byte[] getSchemaBytesBuf(Schema schema) throws IOException {
+    if (schema == null) return null; 
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     MessageSerializer.serialize(new WriteChannel(Channels.newChannel(out)), schema);
     return out.toByteArray();

diff --git a/core/src/main/scala/com/intel/oap/ColumnarGuardRule.scala b/core/src/main/scala/com/intel/oap/ColumnarGuardRule.scala
@@ -151,6 +151,10 @@ case class ColumnarGuardRule(conf: SparkConf) extends Rule[SparkPlan] {
    */
   private def insertRowGuardRecursive(plan: SparkPlan): SparkPlan = {
     plan match {
+      case p: ShuffleExchangeExec =>
+        RowGuard(p.withNewChildren(p.children.map(insertRowGuardOrNot)))
+      case p: BroadcastExchangeExec =>
+        RowGuard(p.withNewChildren(p.children.map(insertRowGuardOrNot)))
       case p: ShuffledHashJoinExec =>
         RowGuard(p.withNewChildren(p.children.map(insertRowGuardRecursive)))
       case p if !supportCodegen(p) =>

diff --git a/core/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala b/core/src/main/scala/org/apache/spark/sql/execution/ColumnarBroadcastExchangeExec.scala
@@ -190,7 +190,8 @@ class ColumnarBroadcastExchangeExec(mode: BroadcastMode, child: SparkPlan)
           throw e
       } finally {
         val timeout: Int = SQLConf.get.broadcastTimeout.toInt
-        relation.asInstanceOf[ColumnarHashedRelation].countDownClose(timeout)
+        if (relation != null)
+          relation.asInstanceOf[ColumnarHashedRelation].countDownClose(timeout)
       }
     }
   }