fix: reject duplicate Parquet field names before decoding

docs/source/user-guide/latest/compatibility/scans.md

@@ -62,6 +62,11 @@ The following limitation may produce incorrect results without falling back to S
 
 The following limitations raise an error at scan time rather than falling back to Spark:
 
+- Byte-identical sibling field names, including inside structs, arrays, and maps. Comet rejects
+  the entire file before decoding, even when the duplicate fields are not projected or the read
+  schema uses field IDs. This prevents row multiplication and decoder synchronization errors.
+  The check applies in both case-sensitivity modes; names in separate groups do not collide.
+  Disable Comet for the query to use Spark's duplicate-name resolution with an explicit read schema.
 - Invalid UTF-8 bytes in `STRING` columns. Spark permits arbitrary byte sequences in a `STRING`
   column (for example from `CAST(X'C1' AS STRING)`), but Comet's native execution path is built on
   Arrow, whose string type is strictly UTF-8. Reading a Parquet file whose `STRING` column contains

native/core/src/parquet/eager_page_index_reader_factory.rs

@@ -44,7 +44,8 @@
 //! the caller's requested policy, unchanged from stock behavior.
 //!
 //! Filed upstream as apache/datafusion#23978. Revert this once the opener merges its deferred
-//! page-index load back into `FileMetadataCache` instead of bypassing it.
+//! page-index load back into `FileMetadataCache` instead of bypassing it. Preserve the
+//! duplicate-field validation when replacing this factory.
 
 use bytes::Bytes;
 use datafusion::common::Result as DFResult;
@@ -62,6 +63,8 @@ use parquet::arrow::arrow_reader::ArrowReaderOptions;
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::errors::ParquetError;
 use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData};
+use parquet::schema::types::Type;
+use std::collections::HashSet;
 use std::fmt::Debug;
 use std::ops::Range;
 use std::sync::Arc;
@@ -116,6 +119,27 @@ struct EagerPageIndexReader {
     metadata_size_hint: Option<usize>,
 }
 
+// Duplicate sibling names can make the decoder combine distinct leaves into one column,
+// multiplying rows before schema adaptation can reject or resolve the duplicate (#5783).
+// Reject the entire file, including unprojected fields, until the decoder can safely
+// resolve duplicate siblings. Names in separate groups do not collide.
+fn validate_field_names(schema: &Type) -> parquet::errors::Result<()> {
+    if let Type::GroupType { fields, .. } = schema {
+        let mut names = HashSet::with_capacity(fields.len());
+        for field in fields {
+            if !names.insert(field.name()) {
+                return Err(ParquetError::General(format!(
+                    "Comet native scan does not support duplicate Parquet field name '{}' in group '{}'",
+                    field.name(),
+                    schema.name()
+                )));
+            }
+            validate_field_names(field)?;
+        }
+    }
+    Ok(())
+}
+
 impl AsyncFileReader for EagerPageIndexReader {
     fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
         let bytes_scanned = range.end - range.start;
@@ -164,7 +188,7 @@ impl AsyncFileReader for EagerPageIndexReader {
                 options.map(|o| o.column_index_policy())
             };
 
-            DFParquetMetadata::new(store.as_ref(), &object_meta)
+            let metadata = DFParquetMetadata::new(store.as_ref(), &object_meta)
                 .with_decryption_properties(file_decryption_properties)
                 .with_file_metadata_cache(Some(metadata_cache))
                 .with_metadata_size_hint(metadata_size_hint)
@@ -176,7 +200,10 @@ impl AsyncFileReader for EagerPageIndexReader {
                         "Failed to fetch metadata for file {}: {e}",
                         object_meta.location,
                     ))
-                })
+                })?;
+            // Validate cache hits too, before Arrow constructs a decoder for any projection.
+            validate_field_names(metadata.file_metadata().schema_descr().root_schema())?;
+            Ok(metadata)
         }
         .boxed()
     }

native/core/src/parquet/parquet_exec.rs

@@ -161,7 +161,8 @@ pub(crate) fn init_datasource_exec(
     // `store_sales`), the page index is re-fetched, uncached, on every open (comet#3978).
     // `EagerPageIndexReaderFactory` forces the page index to load on the first fetch and be
     // cached with the footer, at the cost of losing the skip's benefit when it would have
-    // applied. Filed upstream as apache/datafusion#23978; revert this once that's fixed.
+    // applied. Filed upstream as apache/datafusion#23978; when replacing this factory, preserve
+    // its duplicate-field validation (#5783).
     //
     // TODO: metadata I/O is invisible in metrics. `fetch_metadata` reads via `ObjectStore::get_ranges`,
     // bypassing the `get_bytes` path where `bytes_scanned` is counted. A byte-counting ObjectStore

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

@@ -54,6 +54,109 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
     }
   }
 
+  Seq(
+    ("two children", "named_struct('dup', id, 'dup', id + 100)", "struct<dup: bigint>"),
+    (
+      "three children",
+      "named_struct('dup', id, 'dup', id + 100, 'dup', id + 200)",
+      "struct<dup: bigint>"),
+    (
+      "distinct sibling",
+      "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)",
+      "struct<dup: bigint, other: bigint>"),
+    (
+      "array element",
+      "array(named_struct('dup', id, 'dup', id + 100))",
+      "array<struct<dup: bigint>>"),
+    (
+      "map value",
+      "map('key', named_struct('dup', id, 'dup', id + 100))",
+      "map<string, struct<dup: bigint>>")).foreach { case (shape, expression, readType) =>
+    Seq(1, 4096).foreach { batchSize =>
+      test(s"duplicate Parquet field names fail before decoding - $shape - batch $batchSize") {
+        withSQLConf(
+          SQLConf.CASE_SENSITIVE.key -> "true",
+          CometConf.COMET_BATCH_SIZE.key -> batchSize.toString) {
+          withTempPath { path =>
+            withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+              // Keep all rows in one file so batch size 1 exercises a multi-batch read.
+              spark
+                .range(3)
+                .coalesce(1)
+                .selectExpr(s"$expression as s")
+                .write
+                .parquet(path.toString)
+              // The file is readable by Spark with an explicit schema.
+              assert(
+                spark.read.schema(s"s $readType").parquet(path.toString).collect().length == 3)
+            }
+            val df = spark.read.schema(s"s $readType").parquet(path.toString)
+            assert(
+              find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined)
+            val error = intercept[Exception](df.collect())
+            val messages = Iterator
+              .iterate[Throwable](error)(_.getCause)
+              .takeWhile(_ != null)
+              .map(_.getMessage)
+              .mkString("\n")
+            assert(messages.contains("duplicate Parquet field name 'dup'"), messages)
+          }
+        }
+      }
+    }
+  }
+
+  test("duplicate Parquet field names - unprojected fields and repeated reads") {
+    withTempPath { path =>
+      withSQLConf(CometConf.COMET_ENABLED.key -> "false", SQLConf.CASE_SENSITIVE.key -> "true") {
+        spark
+          .range(3)
+          .selectExpr("id", "named_struct('dup', id, 'dup', id + 100) as s")
+          .write
+          .parquet(path.toString)
+      }
+      Seq(true, false).foreach { caseSensitive =>
+        withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) {
+          val df = spark.read.schema("id bigint").parquet(path.toString)
+          assert(
+            find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined)
+          (1 to 2).foreach { _ =>
+            val error = intercept[Exception](df.collect())
+            val messages = Iterator
+              .iterate[Throwable](error)(_.getCause)
+              .takeWhile(_ != null)
+              .map(_.getMessage)
+              .mkString("\n")
+            assert(messages.contains("duplicate Parquet field name 'dup'"), messages)
+          }
+        }
+      }
+    }
+  }
+
+  test(
+    "duplicate Parquet field names - distinct siblings and repeated names in separate groups") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      withTempPath { path =>
+        withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+          spark
+            .range(3)
+            .selectExpr(
+              "named_struct('dup', id, 'Dup', id + 100) as s",
+              "named_struct('dup', id + 200) as t")
+            .write
+            .parquet(path.toString)
+        }
+        def read = spark.read
+          .schema("s struct<dup: bigint, Dup: bigint>, t struct<dup: bigint>")
+          .parquet(path.toString)
+        assert(
+          find(read.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined)
+        checkSparkAnswer(read)
+      }
+    }
+  }
+
   test("native reader case sensitivity") {
     withTempPath { path =>
       spark.range(10).toDF("a").write.parquet(path.toString)