You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by pa...@apache.org on 2025/06/13 22:53:34 UTC
(datafusion-comet) branch main updated: fix: correctly handle schemas with nested array of struct (native_iceberg_compat) (#1883)
This is an automated email from the ASF dual-hosted git repository.
parthc pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 6361bcd68 fix: correctly handle schemas with nested array of struct (native_iceberg_compat) (#1883)
6361bcd68 is described below
commit 6361bcd682c36a22ce38acdf198ab5ba24a13a3e
Author: Parth Chandra <pa...@apache.org>
AuthorDate: Fri Jun 13 15:53:31 2025 -0700
fix: correctly handle schemas with nested array of struct (native_iceberg_compat) (#1883)
* fix: correctly handle schemas with nested array of struct
---
.../apache/comet/parquet/NativeBatchReader.java | 12 +++-
.../apache/comet/parquet/ParquetReadSuite.scala | 72 ++++++++++++++++++++++
2 files changed, 81 insertions(+), 3 deletions(-)
diff --git a/common/src/main/java/org/apache/comet/parquet/NativeBatchReader.java b/common/src/main/java/org/apache/comet/parquet/NativeBatchReader.java
index 4b4efd1f2..b57746c17 100644
--- a/common/src/main/java/org/apache/comet/parquet/NativeBatchReader.java
+++ b/common/src/main/java/org/apache/comet/parquet/NativeBatchReader.java
@@ -533,13 +533,20 @@ public class NativeBatchReader extends RecordReader<Void, ColumnarBatch> impleme
return newSchema;
}
+ private static boolean isPrimitiveCatalystType(DataType dataType) {
+ return !(dataType instanceof ArrayType)
+ && !(dataType instanceof MapType)
+ && !(dataType instanceof StructType);
+ }
+
private DataType getSparkTypeByFieldId(
DataType dataType, Type parquetType, boolean caseSensitive) {
DataType newDataType;
if (dataType instanceof StructType) {
newDataType =
getSparkSchemaByFieldId((StructType) dataType, parquetType.asGroupType(), caseSensitive);
- } else if (dataType instanceof ArrayType) {
+ } else if (dataType instanceof ArrayType
+ && !isPrimitiveCatalystType(((ArrayType) dataType).elementType())) {
newDataType =
getSparkArrayTypeByFieldId(
@@ -575,11 +582,10 @@ public class NativeBatchReader extends RecordReader<Void, ColumnarBatch> impleme
}
private DataType getSparkArrayTypeByFieldId(
- ArrayType arrayType, GroupType parquetType, boolean caseSensitive) {
+ ArrayType arrayType, GroupType parquetList, boolean caseSensitive) {
DataType newDataType;
DataType elementType = arrayType.elementType();
DataType newElementType;
- Type parquetList = parquetType.getFields().get(0);
Type parquetElementType;
if (parquetList.getLogicalTypeAnnotation() == null
&& parquetList.isRepetition(Type.Repetition.REPEATED)) {
diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
index f08d5b6ab..69a60d92d 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
@@ -41,6 +41,7 @@ import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.comet.{CometBatchScanExec, CometNativeScanExec, CometScanExec}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -1745,6 +1746,77 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
}
+
+ private def withId(id: Int) =
+ new MetadataBuilder().putLong(ParquetUtils.FIELD_ID_METADATA_KEY, id).build()
+
+ // Based on Spark ParquetIOSuite.test("vectorized reader: array of nested struct")
+ test("array of nested struct with and without field id") {
+ val nestedSchema = StructType(
+ Seq(StructField(
+ "_1",
+ StructType(Seq(
+ StructField("_1", StringType, nullable = true, withId(1)), // Field ID 1
+ StructField(
+ "_2",
+ ArrayType(StructType(Seq(
+ StructField("_1", StringType, nullable = true, withId(2)), // Field ID 2
+ StructField("_2", StringType, nullable = true, withId(3)) // Field ID 3
+ ))),
+ nullable = true))),
+ nullable = true)))
+ val nestedSchemaNoId = StructType(
+ Seq(StructField(
+ "_1",
+ StructType(Seq(
+ StructField("_1", StringType, nullable = true),
+ StructField(
+ "_2",
+ ArrayType(StructType(Seq(
+ StructField("_1", StringType, nullable = true),
+ StructField("_2", StringType, nullable = true)))),
+ nullable = true))),
+ nullable = true)))
+ // data matching the schema
+ val data = Seq(
+ Row(Row("a", null)),
+ Row(Row("b", Seq(Row("c", "d")))),
+ Row(null),
+ Row(Row("e", Seq(Row("f", null), Row(null, "g")))),
+ Row(Row(null, null)),
+ Row(Row(null, Seq(null))),
+ Row(Row(null, Seq(Row(null, null), Row("h", null), null))),
+ Row(Row("i", Seq())),
+ Row(null))
+ val answer =
+ Row(Row("a", null)) ::
+ Row(Row("b", Seq(Row("c", "d")))) ::
+ Row(null) ::
+ Row(Row("e", Seq(Row("f", null), Row(null, "g")))) ::
+ Row(Row(null, null)) ::
+ Row(Row(null, Seq(null))) ::
+ Row(Row(null, Seq(Row(null, null), Row("h", null), null))) ::
+ Row(Row("i", Seq())) ::
+ Row(null) ::
+ Nil
+
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") {
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(data), nestedSchema)
+ withTempPath { path =>
+ df.write.parquet(path.getCanonicalPath)
+ readParquetFile(path.getCanonicalPath) { df =>
+ checkAnswer(df, answer)
+ }
+ }
+ val df2 = spark.createDataFrame(spark.sparkContext.parallelize(data), nestedSchemaNoId)
+ withTempPath { path =>
+ df2.write.parquet(path.getCanonicalPath)
+ readParquetFile(path.getCanonicalPath) { df =>
+ checkAnswer(df, answer)
+ }
+ }
+ }
+ }
}
class ParquetReadV1Suite extends ParquetReadSuite with AdaptiveSparkPlanHelper {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org