fix: fall back when a struct repeats a Parquet field id

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]`

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 {

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
   }

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, _) =>

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()
     }
   }
 

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,