From 7fe932805add2ac8ceaff1b447b3893a2e13cf05 Mon Sep 17 00:00:00 2001 From: comphead Date: Thu, 17 Sep 2026 12:07:24 -0700 Subject: [PATCH] fix: fall back when a schema repeats a Parquet field id Under `spark.sql.parquet.fieldId.read.enabled` Spark resolves each requested field to the one Parquet field carrying its id, and raises FOUND_DUPLICATE_FIELD_IN_FIELD_ID_LOOKUP_MODE when more than one answers. Comet never looked at field ids, so a requested schema that repeats one was read positionally and returned rows where Spark raises. The ids ride in the requested schema's `StructField.metadata`, so this is decidable from the plan. `DataTypeSupport` gains a `hasDuplicateFieldIds` predicate beside the existing duplicate-name one, and `CometScanTypeChecker` declines a field list that trips it. Both entry points are overridden, because they see different things. `isTypeSupported` is handed each field's data type, so it sees nested structs as the trait's recursion reaches them but can never compare two top-level fields; `isSchemaSupported` is where the schema's own field list is available. Each list is inspected exactly once, so nothing is re-walked at an enclosing level. `ArrowCachedBatchSerializer.supportsType` also accepted a struct with duplicate child names, so caching such a relation stored it in Comet's Arrow format, which Java Arrow cannot import back because it keys struct children by name. One more schema check delegates it to Spark's default cache format, alongside the interval types already excluded there, and the two copies of that predicate in the shuffle gate now call the shared one rather than spelling it out inline. No Parquet decoding changes; see #5786 for that path. Closes #5801. --- .../org/apache/comet/DataTypeSupport.scala | 36 +++++- .../apache/comet/rules/CometScanRule.scala | 28 ++++- .../arrow/ArrowCachedBatchSerializer.scala | 8 +- .../shuffle/CometShuffleExchangeExec.scala | 6 +- .../comet/exec/CometInMemoryCacheSuite.scala | 81 ++++++------- .../comet/exec/CometNativeReaderSuite.scala | 106 ++++++++++++++++++ .../comet/rules/CometScanRuleSuite.scala | 60 +++++++++- 7 files changed, 279 insertions(+), 46 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala b/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala index 5a5b3158d14..dca806a3347 100644 --- a/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala +++ b/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala @@ -21,9 +21,10 @@ package org.apache.comet import scala.collection.mutable.ListBuffer +import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils import org.apache.spark.sql.types._ -import org.apache.comet.DataTypeSupport.{ARRAY_ELEMENT, MAP_KEY, MAP_VALUE} +import org.apache.comet.DataTypeSupport.{hasDuplicateFieldNames, ARRAY_ELEMENT, MAP_KEY, MAP_VALUE} trait DataTypeSupport { @@ -53,7 +54,7 @@ trait DataTypeSupport { BinaryType | StringType | _: DecimalType | DateType | TimestampType | TimestampNTZType | CalendarIntervalType => true - case StructType(fields) if fields.map(_.name).distinct.length != fields.length => + case StructType(fields) if hasDuplicateFieldNames(fields) => // Java Arrow keys struct children by name, so a struct with duplicate field names // cannot cross the JVM Arrow boundary intact fallbackReasons += s"Unsupported ${name}: struct with duplicate field names" @@ -85,6 +86,37 @@ object DataTypeSupport { case _ => false } + /** True when two of `fields` carry byte-identical names. */ + def hasDuplicateFieldNames(fields: Array[StructField]): Boolean = + fields.map(_.name).distinct.length != fields.length + + /** + * True when two of `fields` declare the same Parquet field id. + * + * Deliberately not Spark's check: `ParquetReadSupport.matchIdField` raises when one *requested* + * id is carried by several fields *in the file*. The two coincide only when the requested + * schema equals the file schema, which is exactly when DataFusion's opener skips the expression + * adapter that would have validated the lookup (#5801). So this can decline a read Spark would + * have accepted, costing native execution but not correctness; it cannot report a duplicate + * Spark would not. File-side ambiguity is left to #5786. + * + * Only meaningful under `spark.sql.parquet.fieldId.read.enabled`; callers gate on that. + */ + def hasDuplicateFieldIds(fields: Array[StructField]): Boolean = { + val ids = fields.flatMap(fieldId) + ids.distinct.length != ids.length + } + + private def fieldId(field: StructField): Option[Int] = { + if (!ParquetUtils.hasFieldId(field)) None + else { + // A malformed id is not this check's business to report -- getFieldId raises on one -- so + // treat it as absent and let the reader complain about it. + try Some(ParquetUtils.getFieldId(field)) + catch { case _: IllegalArgumentException => None } + } + } + /** * `dt` with every array/map/struct nullability flag forced to `true` at all nesting levels (map * key fields stay non-null per Arrow's map invariant). Re-derives Spark's `private[spark]` diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index f7e0fe5cb98..c8a800557e9 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -49,7 +49,7 @@ import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, isSpark35Plu import org.apache.comet.DataTypeSupport.isComplexType import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection} import org.apache.comet.objectstore.NativeConfig -import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported} +import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported, readFieldId} import org.apache.comet.serde.operator.{CometIcebergNativeScan, CometNativeScan} import org.apache.comet.shims.{CometTypeShim, ShimCometStreaming, ShimFileFormat, ShimSubqueryBroadcast} @@ -1094,6 +1094,22 @@ case class CometScanRule(session: SparkSession) case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim { + /** + * `isTypeSupported` only ever sees a field's *data type*, so nothing there can compare two + * top-level fields. Spark's field id ambiguity applies at the schema root too, so check it + * here. + */ + override def isSchemaSupported( + schema: StructType, + fallbackReasons: ListBuffer[String]): Boolean = { + if (duplicateFieldIds(schema.fields)) { + fallbackReasons += "duplicate Parquet field ids among top-level fields" + false + } else { + super.isSchemaSupported(schema, fallbackReasons) + } + } + override def isTypeSupported( dt: DataType, name: String, @@ -1118,10 +1134,20 @@ case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim { false case s: StructType if s.fields.isEmpty => false + case StructType(fields) if duplicateFieldIds(fields) => + // Under field id matching Spark resolves each requested field to the one Parquet field + // carrying its id and raises when more than one answers. Comet reads such a struct + // positionally instead, so hand the read back to Spark and let it report the ambiguity. + fallbackReasons += s"Unsupported ${name}: struct with duplicate Parquet field ids" + false case _ => super.isTypeSupported(dt, name, fallbackReasons) } } + + /** True when the session resolves Parquet fields by id and `fields` repeat one. */ + private def duplicateFieldIds(fields: Array[StructField]): Boolean = + readFieldId(SQLConf.get) && DataTypeSupport.hasDuplicateFieldIds(fields) } object CometScanRule extends Logging { diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala index 821f84e0c2b..d0e7d87911a 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala @@ -39,7 +39,7 @@ import org.apache.spark.storage.StorageLevel import org.apache.spark.unsafe.types.UTF8String import org.apache.spark.util.io.ChunkedByteBuffer -import org.apache.comet.CometArrowAllocator +import org.apache.comet.{CometArrowAllocator, DataTypeSupport} /** * Cached batch format used when Comet writes Spark in-memory cache data. @@ -663,6 +663,11 @@ object ArrowCachedBatchSerializer { * This mirrors the vectors `Utils.getFieldVector` accepts. A type missing from that list throws * during cache materialization, so it has to be delegated to Spark's default cache format * instead. Interval types are the notable omission. + * + * A struct with duplicate child names is rejected for a different reason: Java Arrow keys a + * struct vector's children by name, so the batch cannot be imported back across the C data + * interface (#5605) and the native scan over the cache could never read it. Storing it in + * Comet's format would only be work thrown away. */ def supportsType(dt: DataType): Boolean = dt match { case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | @@ -672,6 +677,7 @@ object ArrowCachedBatchSerializer { case _: StringType => true case ArrayType(elementType, _) => supportsType(elementType) case MapType(keyType, valueType, _) => supportsType(keyType) && supportsType(valueType) + case StructType(fields) if DataTypeSupport.hasDuplicateFieldNames(fields) => false case StructType(fields) => fields.forall(f => supportsType(f.dataType)) case _ => false } diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala index 1f5d06a53af..d4fd3d89660 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala @@ -49,7 +49,7 @@ import org.apache.spark.util.random.XORShiftRandom import com.google.common.base.Objects -import org.apache.comet.{CometConf, CometExplainInfo} +import org.apache.comet.{CometConf, CometExplainInfo, DataTypeSupport} import org.apache.comet.CometConf.{COMET_SHUFFLE_ENABLED, COMET_SHUFFLE_MODE} import org.apache.comet.CometSparkSessionExtensions.{cometCelebornShuffleFallbackReason, hasFallbackReason, isCometCelebornShuffleManagerEnabled, isCometShuffleManagerEnabled, isSpark40Plus, withFallbackReasons} import org.apache.comet.serde.{Compatible, OperatorOuterClass, QueryPlanSerde, SupportLevel, Unsupported} @@ -480,7 +480,7 @@ object CometShuffleExchangeExec fields.nonEmpty && fields.forall(f => supportedSerializableDataType(f.dataType)) && // Java Arrow keys struct children by name, so the FFI import of a decoded batch // fails on duplicate field names - fields.map(f => f.name).distinct.length == fields.length + !DataTypeSupport.hasDuplicateFieldNames(fields) case ArrayType(elementType, _) => supportedSerializableDataType(elementType) case MapType(keyType, valueType, _) => @@ -607,7 +607,7 @@ object CometShuffleExchangeExec case StructType(fields) => fields.nonEmpty && fields.forall(f => supportedSerializableDataType(f.dataType)) && // Java Arrow stream reader cannot work on duplicate field name - fields.map(f => f.name).distinct.length == fields.length + !DataTypeSupport.hasDuplicateFieldNames(fields) case ArrayType(elementType, _) => supportedSerializableDataType(elementType) case MapType(keyType, valueType, _) => diff --git a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala index e72f4d10f74..43e950595f2 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala @@ -295,6 +295,19 @@ class CometInMemoryCacheSuite extends CometTestBase { } } + // Column expression and the reason the serializer has to decline it. + private val unsupportedForArrowCache = Seq( + // Interval types have no Arrow vector in Utils.getFieldVector. Without the schema check in + // the serializer, caching this relation fails outright with "Unsupported Arrow Vector for + // serialize: class org.apache.arrow.vector.DurationVector". + "make_dt_interval(0, 0, 0, id) AS payload", + // Java Arrow keys a struct vector's children by name, so the two `a` children collapse into + // one and the batch fails its arity check coming back across the C data interface. Without + // the check, this is cached in Comet's format and the read dies with + // "ArrowArray struct has 2 children (expected 1)". + // See https://github.com/apache/datafusion-comet/issues/5605. + "named_struct('a', id, 'a', id + 1) AS payload") + test("Comet cache serializer delegates unsupported types to Spark's cache format") { withSQLConf( SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", @@ -303,53 +316,45 @@ class CometInMemoryCacheSuite extends CometTestBase { CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true", "spark.comet.sparkToColumnar.enabled" -> "true") { - spark.catalog.clearCache() - - // Interval types have no Arrow vector in Utils.getFieldVector. Without the schema check in - // the serializer, caching this relation fails outright with "Unsupported Arrow Vector for - // serialize: class org.apache.arrow.vector.DurationVector". - spark - .sql(""" - SELECT id AS key, make_dt_interval(0, 0, 0, id) AS dt - FROM range(1000) - """) - .createOrReplaceTempView("default_cached_batch") + for (column <- unsupportedForArrowCache) { + spark.catalog.clearCache() - spark.catalog.cacheTable("default_cached_batch") - spark.table("default_cached_batch").count() + spark + .sql(s"SELECT id AS key, $column FROM range(1000)") + .createOrReplaceTempView("default_cached_batch") - assert( - cachedBatchTypes("default_cached_batch").sameElements( - Array("org.apache.spark.sql.execution.columnar.DefaultCachedBatch"))) - - // Columnar read path, delegated to Spark's serializer. - val columnarDf = spark.sql(""" - SELECT key, dt - FROM default_cached_batch - WHERE key >= 10 AND key < 20 - """) - assert(columnarDf.collect().length == 10) - checkSparkAnswer(columnarDf) + spark.catalog.cacheTable("default_cached_batch") + spark.table("default_cached_batch").count() - val columnarPlan = columnarDf.queryExecution.executedPlan.toString() - assert(!columnarPlan.contains("CometInMemoryTableScan")) + assert( + cachedBatchTypes("default_cached_batch").sameElements( + Array("org.apache.spark.sql.execution.columnar.DefaultCachedBatch")), + s"$column was cached in Comet's format") - // Row read path: disabling the vectorized cache reader makes Spark use - // convertCachedBatchToInternalRow. - withSQLConf(SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "false") { - val rowDf = spark.sql(""" - SELECT dt + // Columnar read path, delegated to Spark's serializer. + val columnarDf = spark.sql(""" + SELECT key, payload FROM default_cached_batch WHERE key >= 10 AND key < 20 """) - assert(rowDf.collect().length == 10) - checkSparkAnswer(rowDf) + checkSparkAnswer(columnarDf) + assert( + !columnarDf.queryExecution.executedPlan.toString().contains("CometInMemoryTableScan")) + + // Row read path: disabling the vectorized cache reader makes Spark use + // convertCachedBatchToInternalRow. + withSQLConf(SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "false") { + val rowDf = spark.sql(""" + SELECT payload + FROM default_cached_batch + WHERE key >= 10 AND key < 20 + """) + checkSparkAnswer(rowDf) + assert(!rowDf.queryExecution.executedPlan.toString().contains("CometInMemoryTableScan")) + } - val rowPlan = rowDf.queryExecution.executedPlan.toString() - assert(!rowPlan.contains("CometInMemoryTableScan")) + spark.catalog.clearCache() } - - spark.catalog.clearCache() } } diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala index 25c0e93002a..93567756d75 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -929,6 +929,112 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + private def withFieldId(name: String, id: Int): StructField = + StructField( + name, + LongType, + nullable = true, + new MetadataBuilder().putLong("parquet.field.id", id.toLong).build()) + + /** + * Write a one-row file and assert Comet hands the read back to Spark under field id matching, + * and keeps it native without. + * + * The file must carry no key-value metadata: arrow-rs folds Spark's metadata into the physical + * schema, so a Spark-written file never compares equal to the requested schema and always + * reaches the expression adapter that would have caught the ambiguity. `writeDirect` is what + * keeps the metadata out. See https://github.com/apache/datafusion-comet/issues/5801. + */ + private def checkDuplicateFieldIdsFallBack( + messageType: String, + writeRecord: RecordConsumer => Unit, + readSchema: StructType): Unit = withTempPath { dir => + writeDirect( + new Path(dir.getCanonicalPath, "duplicate-field-ids.parquet").toString, + messageType, + writeRecord) + + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + val df = spark.read.schema(readSchema).parquet(dir.getCanonicalPath) + val plan = df.queryExecution.executedPlan + assert( + collect(plan) { case scan: CometNativeScanExec => scan }.isEmpty, + s"expected no native scan, got:\n$plan") + val error = intercept[Exception](df.collect()) + assert( + causeChain(error).exists(e => + String.valueOf(e.getMessage).contains("""Found duplicate field(s) "1"""")), + s"unexpected error: $error") + } + + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") { + val df = spark.read.schema(readSchema).parquet(dir.getCanonicalPath) + assert( + collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec => + scan + }.nonEmpty, + "expected a native scan with field id matching off") + checkSparkAnswer(df) + } + } + + test("native scan declines a nested struct whose fields repeat a Parquet field id") { + // Spark resolves each requested field to the one Parquet field carrying its id and raises when + // more than one answers. DataFusion's opener skips the expression adapter when the file's + // physical schema compares equal to the logical schema and no predicate is pushed, so the + // native scan read this positionally and returned rows where Spark raises. + checkDuplicateFieldIdsFallBack( + """message spark_schema { + | optional group s { + | optional int64 x = 1; + | optional int64 y = 1; + | } + |} + """.stripMargin, + { rc: RecordConsumer => + rc.startMessage() + rc.startField("s", 0) + rc.startGroup() + rc.startField("x", 0) + rc.addLong(10L) + rc.endField("x", 0) + rc.startField("y", 1) + rc.addLong(20L) + rc.endField("y", 1) + rc.endGroup() + rc.endField("s", 0) + rc.endMessage() + }, + StructType( + Seq( + StructField( + "s", + StructType(Seq(withFieldId("x", 1), withFieldId("y", 1))), + nullable = true)))) + } + + test("native scan declines top-level fields that repeat a Parquet field id") { + // The same ambiguity one level up. It reaches the guard through a different override, because + // `isTypeSupported` only sees field data types and so never compares two top-level fields. + checkDuplicateFieldIdsFallBack( + """message spark_schema { + | optional int64 x = 1; + | optional int64 y = 1; + |} + """.stripMargin, + { rc: RecordConsumer => + rc.startMessage() + rc.startField("x", 0) + rc.addLong(10L) + rc.endField("x", 0) + rc.startField("y", 1) + rc.addLong(20L) + rc.endField("y", 1) + rc.endMessage() + }, + StructType(Seq(withFieldId("x", 1), withFieldId("y", 1)))) + } + /** Write a Parquet file using a raw RecordConsumer for full schema control. */ private def writeDirect( path: String, diff --git a/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala index b0211edf5cf..6bb4d6254a2 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometScanRuleSuite.scala @@ -19,13 +19,15 @@ package org.apache.comet.rules +import scala.collection.mutable.ListBuffer import scala.util.Random import org.apache.spark.sql._ import org.apache.spark.sql.comet._ import org.apache.spark.sql.execution._ import org.apache.spark.sql.execution.adaptive.QueryStageExec -import org.apache.spark.sql.types.{DataTypes, StructField, StructType} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types._ import org.apache.comet.CometConf import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator} @@ -181,4 +183,60 @@ class CometScanRuleSuite extends CometTestBase { } } + test("CometScanTypeChecker declines a schema that repeats a Parquet field id") { + // Spark resolves a requested field to the one Parquet field carrying its id and raises + // FOUND_DUPLICATE_FIELD_IN_FIELD_ID_LOOKUP_MODE when more than one answers, so Comet must not + // read such a schema natively. See https://github.com/apache/datafusion-comet/issues/5801. + def withId(name: String, id: Int): StructField = + StructField( + name, + LongType, + nullable = true, + new MetadataBuilder().putLong("parquet.field.id", id.toLong).build()) + + val duplicate = StructType(Array(withId("x", 1), withId("y", 1))) + // The root case needs its own coverage: `isTypeSupported` only sees field data types, so the + // nested case goes through a different override. + val declined = Seq[(String, StructType)]( + "root" -> duplicate, + "nested struct" -> StructType(Array(StructField("col", duplicate))), + "map value" -> StructType(Array(StructField("col", MapType(StringType, duplicate))))) + + for ((label, schema) <- declined) { + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + val reasons = ListBuffer.empty[String] + assert(!CometScanTypeChecker().isSchemaSupported(schema, reasons), s"$label: accepted") + assert(reasons.exists(_.contains("duplicate Parquet field ids")), s"$label: $reasons") + } + // With field id matching off the two fields are told apart by name, so the read is + // unambiguous and the schema stays supported. + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") { + assert( + CometScanTypeChecker().isSchemaSupported(schema, ListBuffer.empty), + s"$label: declined with field id matching off") + } + } + + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + val accepted = Seq[(String, StructType)]( + "distinct ids" -> StructType(Array(withId("x", 1), withId("y", 2))), + // Only one side carries an id, so nothing can collide. + "one id absent" -> StructType(Array(withId("x", 1), StructField("y", LongType))), + // getFieldId raises on a non-integral id; reporting it as a duplicate would be the wrong + // error, so the predicate treats it as absent. + "malformed id" -> { + val bad = new MetadataBuilder().putString("parquet.field.id", "nope").build() + StructType( + Array( + StructField("x", LongType, nullable = true, bad), + StructField("y", LongType, nullable = true, bad))) + }) + for ((label, schema) <- accepted) { + assert( + CometScanTypeChecker().isSchemaSupported(schema, ListBuffer.empty), + s"$label: declined") + } + } + } + }