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") + } + } + } + }