[NSE-782] backport master changes to 1.3.1 branch

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

@@ -272,6 +272,10 @@ case class ColumnarHashAggregateExec(
               case t: BooleanType =>
                 vectors(idx)
                   .put(0, res.asInstanceOf[Boolean].booleanValue())
+              case t: DecimalType =>
+                // count() does not care the real value
+                vectors(idx)
+                  .put(0, res.asInstanceOf[Number].intValue())
               case other =>
                 throw new UnsupportedOperationException(s"$other is not supported.")
             }

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

@@ -385,7 +385,8 @@ case class ColumnarPostOverrides() extends Rule[SparkPlan] {
         s" See above override in BroadcastQueryStageExec")
       val localBroadcastXchg = broadcastQueryStageExec.plan.asInstanceOf[BroadcastExchangeExec]
       val dataToArrowColumnar = localBroadcastXchg.child.asInstanceOf[DataToArrowColumnarExec]
-      ColumnarBroadcastExchangeExec(localBroadcastXchg.mode, dataToArrowColumnar)
+      //ColumnarBroadcastExchangeExec(localBroadcastXchg.mode, dataToArrowColumnar)
+      dataToArrowColumnar.child
     case plan: RowToColumnarExec =>
       val child = replaceWithColumnarPlan(plan.child)
       if (columnarConf.enableArrowRowToColumnar) {