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

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();

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) =>

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)
       }
     }
   }