feat: add read array support

Author: comphead

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

@@ -60,14 +60,16 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
     logWarning(s"Comet native execution is disabled due to: $reason")
   }
 
-  def supportedDataType(dt: DataType, allowStruct: Boolean = false): Boolean = dt match {
+  def supportedDataType(dt: DataType, allowComplex: Boolean = false): Boolean = dt match {
     case _: ByteType | _: ShortType | _: IntegerType | _: LongType | _: FloatType |
         _: DoubleType | _: StringType | _: BinaryType | _: TimestampType | _: DecimalType |
         _: DateType | _: BooleanType | _: NullType =>
       true
     case dt if isTimestampNTZType(dt) => true
-    case s: StructType if allowStruct =>
-      s.fields.map(_.dataType).forall(supportedDataType(_, allowStruct))
+    case s: StructType if allowComplex =>
+      s.fields.map(_.dataType).forall(supportedDataType(_, allowComplex))
+    case a: ArrayType if allowComplex =>
+      supportedDataType(a.elementType)
     case dt =>
       emitWarning(s"unsupported Spark data type: $dt")
       false
@@ -726,7 +728,8 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
           binding,
           (builder, binaryExpr) => builder.setLtEq(binaryExpr))
 
-      case Literal(value, dataType) if supportedDataType(dataType, allowStruct = value == null) =>
+      case Literal(value, dataType)
+          if supportedDataType(dataType, allowComplex = value == null) =>
         val exprBuilder = ExprOuterClass.Literal.newBuilder()
 
         if (value == null) {
@@ -2666,7 +2669,9 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
         withInfo(join, "SortMergeJoin is not enabled")
         None
 
-      case op if isCometSink(op) && op.output.forall(a => supportedDataType(a.dataType, true)) =>
+      case op
+          if isCometSink(op) && op.output.forall(a =>
+            supportedDataType(a.dataType, allowComplex = true)) =>
         // These operators are source of Comet native execution chain
         val scanBuilder = OperatorOuterClass.Scan.newBuilder()
         val source = op.simpleStringWithNodeId()

spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala

@@ -486,9 +486,10 @@ object CometScanExec extends DataTypeSupport {
 
   override def isAdditionallySupported(dt: DataType): Boolean = {
     if (CometConf.COMET_NATIVE_SCAN_IMPL.get() == CometConf.SCAN_NATIVE_ICEBERG_COMPAT) {
-      // TODO add array and map
+      // TODO add map
       dt match {
         case s: StructType => s.fields.map(_.dataType).forall(isTypeSupported)
+        case a: ArrayType => isTypeSupported(a.elementType)
         case _ => false
       }
     } else {

spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala

@@ -31,19 +31,20 @@ import org.apache.comet.CometConf
 class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
       pos: Position): Unit = {
-    super.test(testName, testTags: _*) {
-      withSQLConf(
-        CometConf.COMET_EXEC_ENABLED.key -> "true",
-        SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
-        CometConf.COMET_ENABLED.key -> "true",
-        CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false",
-        CometConf.COMET_NATIVE_SCAN_IMPL.key -> CometConf.SCAN_NATIVE_DATAFUSION) {
-        testFun
-      }
-    }
+    Seq(CometConf.SCAN_NATIVE_DATAFUSION, CometConf.SCAN_NATIVE_ICEBERG_COMPAT).foreach(scan =>
+      super.test(s"$testName - $scan", testTags: _*) {
+        withSQLConf(
+          CometConf.COMET_EXEC_ENABLED.key -> "true",
+          SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
+          CometConf.COMET_ENABLED.key -> "true",
+          CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false",
+          CometConf.COMET_NATIVE_SCAN_IMPL.key -> scan) {
+          testFun
+        }
+      })
   }
 
-  test("native reader - read simple struct fields") {
+  test("native reader - read simple STRUCT fields") {
     testSingleLineQuery(
       """
         |select named_struct('firstName', 'John', 'lastName', 'Doe', 'age', 35) as personal_info union all
@@ -52,7 +53,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
       "select personal_info.* from tbl")
   }
 
-  test("native reader - read simple array fields") {
+  test("native reader - read simple ARRAY fields") {
     testSingleLineQuery(
       """
         |select array(1, 2, 3) as arr union all