From 79abecd829ffd613cdbd8edb60ff58ea6bb5a41f Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Wed, 9 Sep 2026 00:15:39 -0300 Subject: [PATCH 1/5] fix: reject duplicate Parquet field names before decoding --- .../user-guide/latest/compatibility/scans.md | 6 + .../eager_page_index_reader_factory.rs | 29 ++++- native/core/src/parquet/parquet_exec.rs | 3 +- .../comet/exec/CometNativeReaderSuite.scala | 103 ++++++++++++++++++ 4 files changed, 138 insertions(+), 3 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index d5ab79b4a96..6c4ce1271df 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -62,6 +62,12 @@ 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 + ([#5783](https://github.com/apache/datafusion-comet/issues/5783)). 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 diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/eager_page_index_reader_factory.rs index d89a1772835..124216eed24 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/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 arrow::datatypes::{DataType, FieldRef, Schema}; use async_trait::async_trait; @@ -77,7 +78,8 @@ use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaDataBuilder}; use parquet::file::metadata::{ FooterTail, PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader, }; -use parquet::schema::types::{ColumnDescPtr, SchemaDescriptor}; +use parquet::schema::types::{ColumnDescPtr, SchemaDescriptor, Type}; +use std::collections::HashSet; use std::fmt::{Debug, Display, Formatter}; use std::ops::Range; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -372,6 +374,27 @@ fn with_spark_arrow_schema(metadata: Arc) -> ParquetResult 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 { /// Reads a metadata range, counting its requested size before I/O and its returned /// bytes only on success. The returned future borrows this reader; store errors retain @@ -498,6 +521,8 @@ impl AsyncFileReader for EagerPageIndexReader { } let metadata = metadata?; + // Validate cache hits too, before Arrow constructs a decoder for any projection. + validate_field_names(metadata.file_metadata().schema_descr().root_schema())?; if spark_variant_schema { with_spark_arrow_schema(metadata) } else { diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 60a1027f974..d276582a57a 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -176,7 +176,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). // // Preserve bytes_scanned's existing requested data/Bloom-filter range accounting. Footer // and page-index reads through get_metadata bypass it, and coalescing may fetch extra bytes. 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..0e678f9a1f5 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/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"), + ( + "three children", + "named_struct('dup', id, 'dup', id + 100, 'dup', id + 200)", + "struct"), + ( + "distinct sibling", + "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)", + "struct"), + ( + "array element", + "array(named_struct('dup', id, 'dup', id + 100))", + "array>"), + ( + "map value", + "map('key', named_struct('dup', id, 'dup', id + 100))", + "map>")).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, t struct") + .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) From fc6b3d91685e5000693c437f4e9b24383d8a793e Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Wed, 9 Sep 2026 00:28:49 -0300 Subject: [PATCH 2/5] docs: remove issue reference from duplicate-field limitation --- docs/source/user-guide/latest/compatibility/scans.md | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index 6c4ce1271df..23c67ebff6d 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -64,10 +64,9 @@ The following limitations raise an error at scan time rather than falling back t - 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 - ([#5783](https://github.com/apache/datafusion-comet/issues/5783)). 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. + 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 From b696cc31951c223d9d68a0768fb3958970d77753 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Sat, 12 Sep 2026 14:55:18 -0300 Subject: [PATCH 3/5] fix: scope duplicate checks to projected roots --- .../user-guide/latest/compatibility/scans.md | 12 +- .../eager_page_index_reader_factory.rs | 131 +++++++++++++++++- native/core/src/parquet/parquet_exec.rs | 3 +- .../comet/exec/CometNativeReaderSuite.scala | 117 +++++++++++----- 4 files changed, 215 insertions(+), 48 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index 23c67ebff6d..c60ada51ac7 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -62,11 +62,13 @@ 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. +- Byte-identical sibling field names in selected top-level columns, including inside structs, + arrays, and maps. Comet rejects these before decoding to prevent row multiplication and decoder + synchronization errors. Unselected top-level columns are skipped, except for field-ID reads, + which validate the entire file schema. 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. Spark-compatible resolution is tracked in + [#5884](https://github.com/apache/datafusion-comet/issues/5884). - 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 diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/eager_page_index_reader_factory.rs index 124216eed24..dd497e74722 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -47,7 +47,8 @@ //! page-index load back into `FileMetadataCache` instead of bypassing it. Preserve the //! duplicate-field validation when replacing this factory. -use arrow::datatypes::{DataType, FieldRef, Schema}; +use crate::parquet::name_fold::{fold_name, fold_schema_names}; +use arrow::datatypes::{DataType, FieldRef, Schema, SchemaRef}; use async_trait::async_trait; use bytes::Bytes; use datafusion::common::Result as DFResult; @@ -165,6 +166,8 @@ pub struct EagerPageIndexReaderFactory { // Enable the footer workaround only for scans that project Variant. // https://github.com/apache/datafusion-comet/issues/5477 spark_variant_schema: bool, + projected_fields: Option>>, + case_sensitive: bool, } impl EagerPageIndexReaderFactory { @@ -193,9 +196,29 @@ impl EagerPageIndexReaderFactory { metadata_cache, scan_io_metrics, spark_variant_schema: false, + projected_fields: None, + case_sensitive: true, } } + pub(crate) fn with_required_schema( + mut self, + schema: &SchemaRef, + case_sensitive: bool, + use_field_id: bool, + ) -> Self { + // Field-ID projections can rename columns, so names cannot safely restrict the walk. + self.projected_fields = (!use_field_id).then(|| { + Arc::new( + fold_schema_names(schema, case_sensitive) + .into_iter() + .collect(), + ) + }); + self.case_sensitive = case_sensitive; + self + } + pub fn with_spark_variant_schema(mut self, enabled: bool) -> Self { self.spark_variant_schema = enabled; self @@ -227,6 +250,8 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, spark_variant_schema: self.spark_variant_schema, + projected_fields: self.projected_fields.clone(), + case_sensitive: self.case_sensitive, })) } } @@ -242,6 +267,8 @@ struct EagerPageIndexReader { metadata_cache: Arc, metadata_size_hint: Option, spark_variant_schema: bool, + projected_fields: Option>>, + case_sensitive: bool, } // Arrow infers ENUM as Binary, losing the distinction from raw binary that Spark needs. @@ -376,12 +403,21 @@ fn with_spark_arrow_schema(metadata: Arc) -> ParquetResult parquet::errors::Result<()> { +// Only selected top-level subtrees can reach the decoder. Recurse fully within each selected +// subtree because nested projection does not safely separate duplicate leaves (#5884). +fn validate_field_names( + schema: &Type, + projected_fields: Option<&HashSet>, + case_sensitive: bool, +) -> parquet::errors::Result<()> { if let Type::GroupType { fields, .. } = schema { let mut names = HashSet::with_capacity(fields.len()); for field in fields { + if projected_fields.is_some_and(|projected| { + !projected.contains(&fold_name(field.name(), case_sensitive)) + }) { + continue; + } if !names.insert(field.name()) { return Err(ParquetError::General(format!( "Comet native scan does not support duplicate Parquet field name '{}' in group '{}'", @@ -389,7 +425,7 @@ fn validate_field_names(schema: &Type) -> parquet::errors::Result<()> { schema.name() ))); } - validate_field_names(field)?; + validate_field_names(field, None, case_sensitive)?; } } Ok(()) @@ -462,6 +498,8 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata_size_hint = self.metadata_size_hint; let scan_io_metrics = Arc::clone(&self.scan_io_metrics); let spark_variant_schema = self.spark_variant_schema; + let projected_fields = self.projected_fields.clone(); + let case_sensitive = self.case_sensitive; async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -522,7 +560,11 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata = metadata?; // Validate cache hits too, before Arrow constructs a decoder for any projection. - validate_field_names(metadata.file_metadata().schema_descr().root_schema())?; + validate_field_names( + metadata.file_metadata().schema_descr().root_schema(), + projected_fields.as_deref(), + case_sensitive, + )?; if spark_variant_schema { with_spark_arrow_schema(metadata) } else { @@ -866,6 +908,83 @@ mod tests { }, }; + #[test] + fn projected_fields_skip_unselected_roots() { + let schema = parquet::schema::parser::parse_message_type( + "message root { optional int64 a; optional int64 a; optional int64 b; }", + ) + .unwrap(); + let selected = HashSet::from(["b".to_string()]); + validate_field_names(&schema, Some(&selected), true).unwrap(); + validate_field_names(&schema, Some(&HashSet::new()), true).unwrap(); + let selected = HashSet::from(["a".to_string()]); + assert!(validate_field_names(&schema, Some(&selected), true).is_err()); + } + + #[test] + fn projected_fields_match_case_insensitively_and_recurse_fully() { + let schema = parquet::schema::parser::parse_message_type( + "message root { optional group Selected { + optional int64 valid; optional int64 dup; optional int64 dup; + } optional int64 unrelated; }", + ) + .unwrap(); + let selected = HashSet::from(["selected".to_string()]); + assert!(validate_field_names(&schema, Some(&selected), false).is_err()); + let selected = HashSet::from(["unrelated".to_string()]); + validate_field_names(&schema, Some(&selected), false).unwrap(); + } + + #[test] + fn duplicate_names_in_list_element_are_rejected() { + let schema = parquet::schema::parser::parse_message_type( + "message root { optional group a (LIST) { repeated group list { + optional group element { optional int64 dup; optional int64 dup; } + } } }", + ) + .unwrap(); + assert!(validate_field_names(&schema, None, true) + .unwrap_err() + .to_string() + .contains("group 'element'")); + } + + #[test] + fn duplicate_names_in_map_key_value_are_rejected() { + let schema = parquet::schema::parser::parse_message_type( + "message root { optional group a (MAP) { repeated group key_value { + required binary key (UTF8); optional int64 value; optional int64 value; + } } }", + ) + .unwrap(); + assert!(validate_field_names(&schema, None, true) + .unwrap_err() + .to_string() + .contains("group 'key_value'")); + } + + #[test] + fn repeated_names_in_separate_groups_are_valid() { + let schema = parquet::schema::parser::parse_message_type( + "message root { optional group a { optional int64 same; } + optional group b { optional int64 same; } }", + ) + .unwrap(); + validate_field_names(&schema, None, true).unwrap(); + } + + #[test] + fn duplicate_root_names_are_rejected() { + let schema = parquet::schema::parser::parse_message_type( + "message root { optional int64 a; optional int64 a; optional int64 b; }", + ) + .unwrap(); + assert!(validate_field_names(&schema, None, true) + .unwrap_err() + .to_string() + .contains("group 'root'")); + } + #[derive(Debug)] struct RecordingRangeStore { inner: InMemory, diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index d276582a57a..7c13d9732cc 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -194,7 +194,8 @@ pub(crate) fn init_datasource_exec( scan_io_source, parquet_source.metrics(), ) - .with_spark_variant_schema(projects_variant), + .with_spark_variant_schema(projects_variant) + .with_required_schema(&required_schema, case_sensitive, use_field_id), ); parquet_source = parquet_source.with_parquet_file_reader_factory(reader_factory); 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 0e678f9a1f5..989301b0f64 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -72,35 +72,29 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper "map value", "map('key', named_struct('dup', id, 'dup', id + 100))", "map>")).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(s"duplicate Parquet field names fail before decoding - $shape") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + 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) } } } @@ -117,23 +111,74 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } Seq(true, false).foreach { caseSensitive => withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) { - val df = spark.read.schema("id bigint").parquet(path.toString) + val name = if (caseSensitive) "id" else "ID" + val df = spark.read.schema(s"$name 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) + checkAnswer(df, Seq(Row(0L), Row(1L), Row(2L))) + checkAnswer(df.where("id > 1000"), Seq.empty) + checkAnswer(df.selectExpr("count(*)"), Seq(Row(3L))) } } } } } + test("duplicate Parquet field names - root group and unprojected root duplicates") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + withTempPath { path => + writeDirect( + path.toString, + "message spark_schema { optional int64 a = 1; optional int64 a = 2; optional int64 b = 3; }", + { rc => + rc.startMessage() + Seq(("a", 0, 1L), ("a", 1, 2L), ("b", 2, 3L)).foreach { case (name, index, value) => + rc.startField(name, index) + rc.addLong(value) + rc.endField(name, index) + } + rc.endMessage() + }) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + checkAnswer(spark.read.schema("a bigint").parquet(path.toString), Seq(Row(1L))) + } + val selected = spark.read.schema("a bigint").parquet(path.toString) + assert( + find(selected.queryExecution.executedPlan)( + _.isInstanceOf[CometNativeScanExec]).isDefined) + val error = intercept[Exception](selected.collect()) + val messages = Iterator + .iterate[Throwable](error)(_.getCause) + .takeWhile(_ != null) + .map(_.getMessage) + .mkString("\n") + assert(messages.contains("duplicate Parquet field name 'a'"), messages) + val valid = spark.read.schema("b bigint").parquet(path.toString) + assert( + find(valid.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) + checkAnswer(valid, Seq(Row(3L))) + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + val schema = new StructType().add( + "renamed_b", + LongType, + nullable = true, + new MetadataBuilder().putLong("parquet.field.id", 3L).build()) + val byId = spark.read.schema(schema).parquet(path.toString) + assert( + find(byId.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) + val fieldIdError = intercept[Exception](byId.collect()) + val fieldIdMessages = Iterator + .iterate[Throwable](fieldIdError)(_.getCause) + .takeWhile(_ != null) + .map(_.getMessage) + .mkString("\n") + assert(fieldIdMessages.contains("duplicate Parquet field name 'a'"), fieldIdMessages) + } + } + } + } + test( "duplicate Parquet field names - distinct siblings and repeated names in separate groups") { withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { From 07dc72b2c425fcfc1935cb3508f20450dbcbd0ef Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Sun, 13 Sep 2026 15:29:20 -0300 Subject: [PATCH 4/5] test: cover pruned duplicate Parquet fields --- .../comet/exec/CometNativeReaderSuite.scala | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) 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 989301b0f64..8babf51432a 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -397,6 +397,40 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + test("duplicate Parquet field names outside a nested projection remain readable") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(3) + .coalesce(1) + .selectExpr("named_struct('dup', id, 'dup', id + 100, 'other', id + 900) as s") + .write + .parquet(path.toString) + } + Seq(true, false).foreach { caseSensitive => + withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) { + val name = "other" + val df = spark.read.schema(s"s struct<$name: bigint>").parquet(path.toString) + assert( + find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) + checkSparkAnswerAndOperator(df) + checkAnswer(df, Seq(Row(Row(900L)), Row(Row(901L)), Row(Row(902L)))) + // Missing fields require Comet's cast, which decodes the complete physical struct. + val unpruned = spark.read + .schema(s"s struct<$name: bigint, missing: bigint>") + .parquet(path.toString) + val error = intercept[Exception](unpruned.collect()) + val messages = Iterator + .iterate[Throwable](error)(_.getCause) + .takeWhile(_ != null) + .map(_.getMessage) + .mkString("\n") + assert(messages.contains("duplicate Parquet field name 'dup'"), messages) + } + } + } + } + test("native reader - read simple STRUCT fields") { testSingleLineQuery( """ From 38b8fd162d4a21424ca9fcc22df90ae0b2baa18e Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Sun, 13 Sep 2026 15:29:47 -0300 Subject: [PATCH 5/5] fix: validate only safely decoded Parquet fields Reuse structural narrowing before pruning duplicate siblings. Keep full subtree validation when casts or schema hints change what the decoder reads. --- .../user-guide/latest/compatibility/scans.md | 6 +- .../eager_page_index_reader_factory.rs | 283 +++++++++++++++--- native/core/src/parquet/parquet_exec.rs | 2 +- native/core/src/parquet/schema_adapter.rs | 2 +- 4 files changed, 251 insertions(+), 42 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index c60ada51ac7..0dc88124afe 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -64,8 +64,10 @@ The following limitations raise an error at scan time rather than falling back t - Byte-identical sibling field names in selected top-level columns, including inside structs, arrays, and maps. Comet rejects these before decoding to prevent row multiplication and decoder - synchronization errors. Unselected top-level columns are skipped, except for field-ID reads, - which validate the entire file schema. The check applies in both case-sensitivity modes; + synchronization errors. Unselected columns and safely pruned nested fields are skipped. + Reads requiring a full-subtree cast still validate that subtree. Files with embedded Arrow + schema hints and Variant scans conservatively validate selected subtrees in full; field-ID + reads validate the entire file schema. 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. Spark-compatible resolution is tracked in [#5884](https://github.com/apache/datafusion-comet/issues/5884). diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/eager_page_index_reader_factory.rs index dd497e74722..c7de3581446 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -47,8 +47,10 @@ //! page-index load back into `FileMetadataCache` instead of bypassing it. Preserve the //! duplicate-field validation when replacing this factory. -use crate::parquet::name_fold::{fold_name, fold_schema_names}; -use arrow::datatypes::{DataType, FieldRef, Schema, SchemaRef}; +use crate::parquet::name_fold::fold_name; +use crate::parquet::parquet_support::SparkParquetOptions; +use crate::parquet::schema_adapter::is_pure_structural_narrowing; +use arrow::datatypes::{DataType, FieldRef, Fields, Schema, SchemaRef}; use async_trait::async_trait; use bytes::Bytes; use datafusion::common::Result as DFResult; @@ -166,8 +168,7 @@ pub struct EagerPageIndexReaderFactory { // Enable the footer workaround only for scans that project Variant. // https://github.com/apache/datafusion-comet/issues/5477 spark_variant_schema: bool, - projected_fields: Option>>, - case_sensitive: bool, + projection: Option<(SchemaRef, SparkParquetOptions)>, } impl EagerPageIndexReaderFactory { @@ -196,26 +197,17 @@ impl EagerPageIndexReaderFactory { metadata_cache, scan_io_metrics, spark_variant_schema: false, - projected_fields: None, - case_sensitive: true, + projection: None, } } pub(crate) fn with_required_schema( mut self, schema: &SchemaRef, - case_sensitive: bool, - use_field_id: bool, + options: &SparkParquetOptions, ) -> Self { // Field-ID projections can rename columns, so names cannot safely restrict the walk. - self.projected_fields = (!use_field_id).then(|| { - Arc::new( - fold_schema_names(schema, case_sensitive) - .into_iter() - .collect(), - ) - }); - self.case_sensitive = case_sensitive; + self.projection = (!options.use_field_id).then(|| (Arc::clone(schema), options.clone())); self } @@ -250,8 +242,7 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, spark_variant_schema: self.spark_variant_schema, - projected_fields: self.projected_fields.clone(), - case_sensitive: self.case_sensitive, + projection: self.projection.clone(), })) } } @@ -267,8 +258,7 @@ struct EagerPageIndexReader { metadata_cache: Arc, metadata_size_hint: Option, spark_variant_schema: bool, - projected_fields: Option>>, - case_sensitive: bool, + projection: Option<(SchemaRef, SparkParquetOptions)>, } // Arrow infers ENUM as Binary, losing the distinction from raw binary that Spark needs. @@ -403,34 +393,78 @@ fn with_spark_arrow_schema(metadata: Arc) -> ParquetResult>, + projected_fields: Option<&Fields>, case_sensitive: bool, ) -> parquet::errors::Result<()> { if let Type::GroupType { fields, .. } = schema { let mut names = HashSet::with_capacity(fields.len()); for field in fields { - if projected_fields.is_some_and(|projected| { - !projected.contains(&fold_name(field.name(), case_sensitive)) - }) { + let projected = projected_fields.and_then(|projected| { + projected.iter().find(|candidate| { + fold_name(candidate.name(), case_sensitive) + == fold_name(field.name(), case_sensitive) + }) + }); + if projected_fields.is_some() && projected.is_none() { continue; } 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() + field.name(), schema.name() ))); } - validate_field_names(field, None, case_sensitive)?; + validate_field_type(field, projected.map(|f| f.data_type()), case_sensitive)?; } } Ok(()) } +fn validate_field_type( + schema: &Type, + projected: Option<&DataType>, + case_sensitive: bool, +) -> ParquetResult<()> { + match projected { + Some(DataType::Struct(fields)) => { + validate_field_names(schema, Some(fields), case_sensitive) + } + Some( + DataType::List(element) + | DataType::LargeList(element) + | DataType::FixedSizeList(element, _), + ) if schema.is_group() && schema.get_fields().len() == 1 => { + let wrapper = &schema.get_fields()[0]; + // Standard three-level LIST. Legacy layouts retain full validation. + if wrapper.is_group() + && wrapper.get_fields().len() == 1 + && wrapper.name() != "array" + && wrapper.name() != format!("{}_tuple", schema.name()) + { + validate_field_type( + &wrapper.get_fields()[0], + Some(element.data_type()), + case_sensitive, + ) + } else { + validate_field_names(schema, None, case_sensitive) + } + } + Some(DataType::Map(entries, _)) if schema.is_group() && schema.get_fields().len() == 1 => { + validate_field_type( + &schema.get_fields()[0], + Some(entries.data_type()), + case_sensitive, + ) + } + _ => validate_field_names(schema, None, case_sensitive), + } +} + impl AsyncFileReader for EagerPageIndexReader { /// Reads a metadata range, counting its requested size before I/O and its returned /// bytes only on success. The returned future borrows this reader; store errors retain @@ -498,8 +532,7 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata_size_hint = self.metadata_size_hint; let scan_io_metrics = Arc::clone(&self.scan_io_metrics); let spark_variant_schema = self.spark_variant_schema; - let projected_fields = self.projected_fields.clone(); - let case_sensitive = self.case_sensitive; + let projection = self.projection.clone(); async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -560,10 +593,69 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata = metadata?; // Validate cache hits too, before Arrow constructs a decoder for any projection. + // ponytail: schema hints can change the later cast; validate full subtrees until + // this guard can share the opener's final schema (including Variant rewriting). + let schema_hints = spark_variant_schema + || metadata + .file_metadata() + .key_value_metadata() + .is_some_and(|entries| { + entries + .iter() + .any(|entry| entry.key == ARROW_SCHEMA_META_KEY) + }); + let physical_schema = if projection.is_some() { + Some(parquet_to_arrow_schema( + metadata.file_metadata().schema_descr(), + None, + )?) + } else { + None + }; + let selected = projection.as_ref().zip(physical_schema.as_ref()).map( + |((required, options), physical)| { + required + .fields() + .iter() + .map(|field| { + physical + .fields() + .iter() + .find(|source| { + fold_name(source.name(), options.case_sensitive) + == fold_name(field.name(), options.case_sensitive) + }) + .map_or_else( + || Arc::clone(field), + |source| { + if !schema_hints + && is_pure_structural_narrowing( + source.data_type(), + field.data_type(), + options, + ) + { + Arc::clone(field) + } else { + Arc::new( + field + .as_ref() + .clone() + .with_data_type(source.data_type().clone()), + ) + } + }, + ) + }) + .collect::() + }, + ); validate_field_names( metadata.file_metadata().schema_descr().root_schema(), - projected_fields.as_deref(), - case_sensitive, + selected.as_ref(), + projection + .as_ref() + .is_none_or(|(_, options)| options.case_sensitive), )?; if spark_variant_schema { with_spark_arrow_schema(metadata) @@ -908,16 +1000,89 @@ mod tests { }, }; + #[tokio::test] + async fn projected_fields_with_arrow_hints_validate_full_subtree() { + use arrow::datatypes::Field; + let fields = Fields::from(vec![ + Field::new("dup", DataType::Int64, true), + Field::new("dup", DataType::Int64, true), + Field::new( + "other", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Int64)), + true, + ), + ]); + let schema = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(fields), + true, + )])); + let mut bytes = Vec::new(); + ArrowWriter::try_new(&mut bytes, schema, None) + .unwrap() + .close() + .unwrap(); + let size = bytes.len() as u64; + let store = Arc::new(InMemory::new()); + let location = Path::from("arrow-hints.parquet"); + store + .put(&location, Bytes::from(bytes).into()) + .await + .unwrap(); + let runtime = datafusion::execution::runtime_env::RuntimeEnv::default(); + let metrics = ExecutionPlanMetricsSet::new(); + let required = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(Fields::from(vec![Field::new( + "other", + DataType::Int64, + true, + )])), + true, + )])); + let options = SparkParquetOptions::new_without_timezone( + datafusion_comet_spark_expr::EvalMode::Legacy, + false, + ); + let factory = EagerPageIndexReaderFactory::new( + store, + runtime.cache_manager.get_file_metadata_cache(), + ScanIoSource::ObjectStore, + &metrics, + ) + .with_required_schema(&required, &options); + let mut reader = factory + .create_reader( + 0, + PartitionedFile::new(location.to_string(), size), + None, + &metrics, + ) + .unwrap(); + let error = reader.get_metadata(None).await.unwrap_err(); + assert!(error + .to_string() + .contains("duplicate Parquet field name 'dup'")); + } + #[test] fn projected_fields_skip_unselected_roots() { let schema = parquet::schema::parser::parse_message_type( "message root { optional int64 a; optional int64 a; optional int64 b; }", ) .unwrap(); - let selected = HashSet::from(["b".to_string()]); + let selected = Fields::from(vec![arrow::datatypes::Field::new( + "b", + DataType::Int64, + true, + )]); validate_field_names(&schema, Some(&selected), true).unwrap(); - validate_field_names(&schema, Some(&HashSet::new()), true).unwrap(); - let selected = HashSet::from(["a".to_string()]); + validate_field_names(&schema, Some(&Fields::empty()), true).unwrap(); + let selected = Fields::from(vec![arrow::datatypes::Field::new( + "a", + DataType::Int64, + true, + )]); assert!(validate_field_names(&schema, Some(&selected), true).is_err()); } @@ -929,12 +1094,54 @@ mod tests { } optional int64 unrelated; }", ) .unwrap(); - let selected = HashSet::from(["selected".to_string()]); + let selected = Fields::from(vec![arrow::datatypes::Field::new( + "selected", + DataType::Int64, + true, + )]); assert!(validate_field_names(&schema, Some(&selected), false).is_err()); - let selected = HashSet::from(["unrelated".to_string()]); + let selected = Fields::from(vec![arrow::datatypes::Field::new( + "unrelated", + DataType::Int64, + true, + )]); validate_field_names(&schema, Some(&selected), false).unwrap(); } + #[test] + fn projected_fields_skip_unselected_nested_duplicates() { + let children = Fields::from(vec![arrow::datatypes::Field::new( + "other", + DataType::Int64, + true, + )]); + let item = Arc::new(arrow::datatypes::Field::new( + "element", + DataType::Struct(children.clone()), + true, + )); + let entries = Arc::new(arrow::datatypes::Field::new( + "entries", + DataType::Struct(Fields::from(vec![ + arrow::datatypes::Field::new("key", DataType::Utf8, false), + arrow::datatypes::Field::new("value", DataType::Struct(children.clone()), true), + ])), + false, + )); + for (physical, projected) in [ + ("optional group s { optional int64 dup; optional int64 dup; optional int64 other; }", DataType::Struct(children)), + ("optional group s (LIST) { repeated group list { optional group element { optional int64 dup; optional int64 dup; optional int64 other; } } }", DataType::List(item)), + ("optional group s (MAP) { repeated group key_value { required binary key (UTF8); optional group value { optional int64 dup; optional int64 dup; optional int64 other; } } }", DataType::Map(entries, false)), + ] { + let schema = parquet::schema::parser::parse_message_type(&format!("message root {{ {physical} }}")).unwrap(); + for case_sensitive in [true, false] { + let selected = Fields::from(vec![arrow::datatypes::Field::new("s", projected.clone(), true)]); + validate_field_names(&schema, Some(&selected), case_sensitive).unwrap(); + assert!(validate_field_names(&schema, None, case_sensitive).is_err()); + } + } + } + #[test] fn duplicate_names_in_list_element_are_rejected() { let schema = parquet::schema::parser::parse_message_type( diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 7c13d9732cc..d108c90e07b 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -195,7 +195,7 @@ pub(crate) fn init_datasource_exec( parquet_source.metrics(), ) .with_spark_variant_schema(projects_variant) - .with_required_schema(&required_schema, case_sensitive, use_field_id), + .with_required_schema(&required_schema, &spark_parquet_options), ); parquet_source = parquet_source.with_parquet_file_reader_factory(reader_factory); diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 5a4b3e16305..0c9b865ba01 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -107,7 +107,7 @@ fn schema_has_field_ids(schema: &SchemaRef) -> bool { /// Arrow's, allow everything else) would fail open: a future addition to /// `parquet_convert_array` that this predicate does not know to also exclude would silently /// start producing wrong results instead of just missing an optimization. -fn is_pure_structural_narrowing( +pub(crate) fn is_pure_structural_narrowing( physical_type: &DataType, target_type: &DataType, parquet_options: &SparkParquetOptions,