[NSE-705] Fallback R2C on unsupported cases (#706)

* fallback on unsupported case * add back ArrayType
oap-project · Jan 18, 2022 · ccb1e71 · ccb1e71
1 parent b571919
commit ccb1e71
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 22 deletions.
diff --git a/native-sql-engine/core/src/main/scala/com/intel/oap/execution/ArrowRowToColumnarExec.scala b/native-sql-engine/core/src/main/scala/com/intel/oap/execution/ArrowRowToColumnarExec.scala
@@ -63,9 +63,10 @@ class ArrowRowToColumnarExec(child: SparkPlan) extends RowToColumnarExec(child =
         case d: DecimalType =>
         case d: TimestampType =>
         case d: BinaryType =>
+        case d: ArrayType =>
         case _ =>
           throw new UnsupportedOperationException(s"${field.dataType} " +
-            s"is not supported in ArrowColumnarToRowExec.")
+            s"is not supported in ArrowRowToColumnarExec.")
       }
     }
   }

diff --git a/native-sql-engine/core/src/main/scala/com/intel/oap/extension/ColumnarOverrides.scala b/native-sql-engine/core/src/main/scala/com/intel/oap/extension/ColumnarOverrides.scala
@@ -338,12 +338,17 @@ case class ColumnarPostOverrides() extends Rule[SparkPlan] {
 
   def replaceWithColumnarPlan(plan: SparkPlan): SparkPlan = plan match {
     case plan: RowToColumnarExec =>
+      val child = replaceWithColumnarPlan(plan.child)
       if (columnarConf.enableArrowRowToColumnar) {
-        val child = replaceWithColumnarPlan(plan.child)
         logDebug(s"ColumnarPostOverrides ArrowRowToColumnarExec(${child.getClass})")
-        new ArrowRowToColumnarExec(child)
+        try {
+          new ArrowRowToColumnarExec(child)
+        } catch {
+          case _: Throwable =>
+            logInfo("ArrowRowToColumnar: Falling back to RowToColumnar...")
+            RowToArrowColumnarExec(child)
+        }
       } else {
-        val child = replaceWithColumnarPlan(plan.child)
         logDebug(s"ColumnarPostOverrides RowToArrowColumnarExec(${child.getClass})")
         RowToArrowColumnarExec(child)
       }
@@ -365,25 +370,25 @@ case class ColumnarPostOverrides() extends Rule[SparkPlan] {
     case r: SparkPlan
         if !r.isInstanceOf[QueryStageExec] && !r.supportsColumnar && r.children.exists(c =>
           c.isInstanceOf[ColumnarToRowExec]) =>
-      // This is a fix for when DPP and AQE both enabled, ColumnarExchange maybe child as a Row SparkPlan
-      val children = r.children.map(c =>
-        c match {
-          case c: ColumnarToRowExec =>
-            if (columnarConf.enableArrowColumnarToRow) {
-              try {
-                val child = replaceWithColumnarPlan(c.child)
-                new ArrowColumnarToRowExec(child)
-              } catch {
-                case _: Throwable =>
-                  logInfo("ArrowColumnarToRow : Falling back to ColumnarToRow...")
-                  c.withNewChildren(c.children.map(replaceWithColumnarPlan))
-              }
-            } else {
-              c.withNewChildren(c.children.map(replaceWithColumnarPlan))
+      // This is a fix for when DPP and AQE both enabled,
+      // ColumnarExchange maybe child as a Row SparkPlan.
+      val children = r.children.map {
+        case c: ColumnarToRowExec =>
+          if (columnarConf.enableArrowColumnarToRow) {
+            try {
+              val child = replaceWithColumnarPlan(c.child)
+              new ArrowColumnarToRowExec(child)
+            } catch {
+              case _: Throwable =>
+                logInfo("ArrowColumnarToRow : Falling back to ColumnarToRow...")
+                c.withNewChildren(c.children.map(replaceWithColumnarPlan))
             }
-          case other =>
-            replaceWithColumnarPlan(other)
-        })
+          } else {
+            c.withNewChildren(c.children.map(replaceWithColumnarPlan))
+          }
+        case other =>
+          replaceWithColumnarPlan(other)
+      }
       r.withNewChildren(children)
     case p =>
       val children = p.children.map(replaceWithColumnarPlan)