feat: ArrayType literal

Author: comphead

spark/src/main/scala/org/apache/comet/DataTypeSupport.scala

@@ -73,4 +73,9 @@ object DataTypeSupport {
   val ARRAY_ELEMENT = "array element"
   val MAP_KEY = "map key"
   val MAP_VALUE = "map value"
+
+  def isComplexType(dt: DataType): Boolean = dt match {
+    case _: StructType | _: ArrayType | _: MapType => true
+    case _ => false
+  }
 }

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -37,6 +37,7 @@ import org.apache.spark.sql.types._
 import org.apache.comet.{CometConf, CometSparkSessionExtensions, DataTypeSupport}
 import org.apache.comet.CometConf._
 import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, isCometScanEnabled, withInfo, withInfos}
+import org.apache.comet.DataTypeSupport.isComplexType
 import org.apache.comet.parquet.{CometParquetScan, SupportsComet}
 
 /**
@@ -277,11 +278,6 @@ case class CometScanRule(session: SparkSession) extends Rule[SparkPlan] {
     val partitionSchemaSupported =
       typeChecker.isSchemaSupported(partitionSchema, fallbackReasons)
 
-    def isComplexType(dt: DataType): Boolean = dt match {
-      case _: StructType | _: ArrayType | _: MapType => true
-      case _ => false
-    }
-
     def hasMapsContainingStructs(dataType: DataType): Boolean = {
       dataType match {
         case s: StructType => s.exists(field => hasMapsContainingStructs(field.dataType))

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

@@ -52,6 +52,7 @@ import com.google.protobuf.ByteString
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.{isCometScan, withInfo}
+import org.apache.comet.DataTypeSupport.isComplexType
 import org.apache.comet.expressions._
 import org.apache.comet.objectstore.NativeConfig
 import org.apache.comet.serde.ExprOuterClass.{AggExpr, DataType => ProtoDataType, Expr, ScalarFunc}
@@ -846,9 +847,11 @@ object QueryPlanSerde extends Logging with CometExprShim {
             allowComplex = value == null ||
               // Nested literal support for native reader
               // can be tracked https://github.com/apache/datafusion-comet/issues/1937
+              // now supports only Array of primitive
               (Seq(CometConf.SCAN_NATIVE_ICEBERG_COMPAT, CometConf.SCAN_NATIVE_DATAFUSION)
                 .contains(CometConf.COMET_NATIVE_SCAN_IMPL.get()) && dataType
-                .isInstanceOf[ArrayType])) =>
+                .isInstanceOf[ArrayType]) && !isComplexType(
+                dataType.asInstanceOf[ArrayType].elementType)) =>
         val exprBuilder = ExprOuterClass.Literal.newBuilder()
 
         if (value == null) {

spark/src/test/scala/org/apache/comet/CometFuzzTestSuite.scala

@@ -39,6 +39,7 @@ import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
 import org.apache.spark.sql.types._
 
 import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
+import org.apache.comet.DataTypeSupport.isComplexType
 import org.apache.comet.testing.{DataGenOptions, ParquetGenerator}
 
 class CometFuzzTestSuite extends CometTestBase with AdaptiveSparkPlanHelper {

spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala

@@ -44,7 +44,7 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
 import org.apache.spark.sql.internal._
 import org.apache.spark.sql.test._
-import org.apache.spark.sql.types.{ArrayType, DataType, DecimalType, MapType, StructType}
+import org.apache.spark.sql.types.{DecimalType, StructType}
 
 import org.apache.comet._
 import org.apache.comet.shims.ShimCometSparkSessionExtensions
@@ -1128,9 +1128,4 @@ abstract class CometTestBase
     usingDataSourceExec(conf) &&
     !CometConf.COMET_SCAN_ALLOW_INCOMPATIBLE.get(conf)
   }
-
-  def isComplexType(dt: DataType): Boolean = dt match {
-    case _: StructType | _: ArrayType | _: MapType => true
-    case _ => false
-  }
 }