chore: extract CreateArray from QueryPlanSerde (apache#1991)

Author: tglanz

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -79,6 +79,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[ArrayRepeat] -> CometArrayRepeat,
     classOf[ArraysOverlap] -> CometArraysOverlap,
     classOf[ArrayUnion] -> CometArrayUnion,
+    classOf[CreateArray] -> CometCreateArray,
     classOf[Ascii] -> CometAscii,
     classOf[ConcatWs] -> CometConcatWs,
     classOf[Chr] -> CometChr,
@@ -1855,16 +1856,6 @@ object QueryPlanSerde extends Logging with CometExprShim {
             .build()
         }
 
-      case CreateArray(children, _) =>
-        val childExprs = children.map(exprToProtoInternal(_, inputs, binding))
-
-        if (childExprs.forall(_.isDefined)) {
-          scalarFunctionExprToProto("make_array", childExprs: _*)
-        } else {
-          withInfo(expr, "unsupported arguments for CreateArray", children: _*)
-          None
-        }
-
       case GetArrayItem(child, ordinal, failOnError) =>
         val childExpr = exprToProtoInternal(child, inputs, binding)
         val ordinalExpr = exprToProtoInternal(ordinal, inputs, binding)

spark/src/main/scala/org/apache/comet/serde/arrays.scala

@@ -361,3 +361,20 @@ object CometArrayUnion extends CometExpressionSerde with IncompatExpr {
     optExprWithInfo(arraysUnionScalarExpr, expr, expr.children: _*)
   }
 }
+
+object CometCreateArray extends CometExpressionSerde {
+  override def convert(
+      expr: Expression,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[ExprOuterClass.Expr] = {
+    val children = expr.children
+    val childExprs = children.map(exprToProtoInternal(_, inputs, binding))
+
+    if (childExprs.forall(_.isDefined)) {
+      scalarFunctionExprToProto("make_array", childExprs: _*)
+    } else {
+      withInfo(expr, "unsupported arguments for CreateArray", children: _*)
+      None
+    }
+  }
+}