feat: add read array support

native/core/src/execution/serde.rs

@@ -111,7 +111,7 @@ pub fn to_arrow_datatype(dt_value: &DataType) -> ArrowDataType {
         {
             DatatypeStruct::List(info) => {
                 let field = Field::new(
-                    "item",
+                    "element",
                     to_arrow_datatype(info.element_type.as_ref().unwrap()),
                     info.contains_null,
                 );

native/core/src/execution/shuffle/row.rs

@@ -3196,58 +3196,63 @@ fn make_builders(
             // Disable dictionary encoding for array element
             let value_builder =
                 make_builders(field.data_type(), NESTED_TYPE_BUILDER_CAPACITY, 1.0)?;
+
+            let value_field = Arc::clone(field);
+
+            // List builders created using `with_field` to override the ListBuilder default field
+            // name which is `item`
             match field.data_type() {
                 DataType::Boolean => {
                     let builder = downcast_builder!(BooleanBuilder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Int8 => {
                     let builder = downcast_builder!(Int8Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Int16 => {
                     let builder = downcast_builder!(Int16Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Int32 => {
                     let builder = downcast_builder!(Int32Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Int64 => {
                     let builder = downcast_builder!(Int64Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Float32 => {
                     let builder = downcast_builder!(Float32Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Float64 => {
                     let builder = downcast_builder!(Float64Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Decimal128(_, _) => {
                     let builder = downcast_builder!(Decimal128Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Timestamp(TimeUnit::Microsecond, _) => {
                     let builder = downcast_builder!(TimestampMicrosecondBuilder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Date32 => {
                     let builder = downcast_builder!(Date32Builder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Binary => {
                     let builder = downcast_builder!(BinaryBuilder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Utf8 => {
                     let builder = downcast_builder!(StringBuilder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 DataType::Struct(_) => {
                     let builder = downcast_builder!(StructBuilder, value_builder);
-                    Box::new(ListBuilder::new(*builder))
+                    Box::new(ListBuilder::new(*builder).with_field(value_field))
                 }
                 // TODO: nested list is not supported. Due to the design of `ListBuilder`, it has
                 // a `T: ArrayBuilder` as type parameter. It makes hard to construct an arbitrarily

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

@@ -182,7 +182,7 @@ object CometNativeScanExec extends DataTypeSupport {
       case null => null
     }
 
-    val newArgs = mapProductIterator(scanExec, transform(_))
+    val newArgs = mapProductIterator(scanExec, transform)
     val wrapped = scanExec.makeCopy(newArgs).asInstanceOf[FileSourceScanExec]
     val batchScanExec = CometNativeScanExec(
       nativeOp,
@@ -202,9 +202,10 @@ object CometNativeScanExec extends DataTypeSupport {
   }
 
   override def isAdditionallySupported(dt: DataType): Boolean = {
-    // 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
     }
   }

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

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.exec
+
+import org.scalactic.source.Position
+import org.scalatest.Tag
+
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.internal.SQLConf
+
+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
+      }
+    }
+  }
+
+  test("native reader - read simple struct fields") {
+    testSingleLineQuery(
+      """
+        |select named_struct('firstName', 'John', 'lastName', 'Doe', 'age', 35) as personal_info union all
+        |select named_struct('firstName', 'Jane', 'lastName', 'Doe', 'age', 40) as personal_info
+        |""".stripMargin,
+      "select personal_info.* from tbl")
+  }
+
+  test("native reader - read simple array fields") {
+    testSingleLineQuery(
+      """
+        |select array(1, 2, 3) as arr union all
+        |select array(2, 3, 4) as arr
+        |""".stripMargin,
+      "select arr from tbl")
+  }
+}