From a7f4c6c189d9455a2b2688d5169b6707c0123a4c Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Wed, 9 Sep 2026 00:15:39 -0300 Subject: [PATCH 1/8] 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 36d245992eb..97ae559f093 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 5d70f2aaa76..947c19c1b55 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -177,7 +177,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 93567756d75..07ab73e335e 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 9a4414a8d9190c3d8426322d4adc0bffd45043fe Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Wed, 9 Sep 2026 00:28:49 -0300 Subject: [PATCH 2/8] 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 97ae559f093..c782673d6a4 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 acee660648dbd2276be1afacee8eac08e1407618 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Sat, 12 Sep 2026 14:55:18 -0300 Subject: [PATCH 3/8] 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 c782673d6a4..4eae6720d75 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 947c19c1b55..fda99fb43af 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -195,7 +195,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 07ab73e335e..9e2fe983ae8 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 6aafaf8e34fb550ee66a13e64a7d09c10aad3ba1 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Sun, 13 Sep 2026 15:29:20 -0300 Subject: [PATCH 4/8] 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 9e2fe983ae8..b4fdcfea075 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 fac9a77aafc9a5aaa13cf1a36a8c3cddabe49eae Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Sun, 13 Sep 2026 15:29:47 -0300 Subject: [PATCH 5/8] 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 4eae6720d75..353df49a18f 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 fda99fb43af..7c4042bbdbc 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -196,7 +196,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 e06408b44e1..d3c305ca136 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, From 4ac32381dc1c6bdf43615916ae69ab8e28254e7c Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Mon, 21 Sep 2026 21:42:00 -0300 Subject: [PATCH 6/8] fix: scope duplicate Parquet field checks to referenced columns --- .../user-guide/latest/compatibility/scans.md | 17 +- .../src/execution/operators/iceberg_scan.rs | 56 +++ .../eager_page_index_reader_factory.rs | 357 +-------------- native/core/src/parquet/parquet_exec.rs | 135 +++++- native/core/src/parquet/parquet_support.rs | 20 +- native/core/src/parquet/schema_adapter.rs | 430 ++++++++++++++++-- .../comet/exec/CometNativeReaderSuite.scala | 180 ++++++-- 7 files changed, 735 insertions(+), 460 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index 353df49a18f..695c6221b43 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -62,13 +62,16 @@ 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 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 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 +- Selecting a field by name when multiple physical siblings match, including inside structs, + arrays, and maps. Comet reports a duplicate-field error instead of resolving byte-identical + names, in either case-sensitivity mode. Root checks cover referenced columns, including + predicates. Unselected roots do not prevent reading a unique field by name or field ID. + Exact-name projections of unique children in structs and arrays of structs remain supported; + casts that cannot use this pruning reject duplicate siblings anywhere in the decoded + physical subtree, including map and case-differing nested projections. Field-ID resolution + retains precedence, but selecting a byte-identically duplicated physical root name still + raises a duplicate-field error, even when the requested field is renamed. + 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` diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index fdcd6c8ba5a..cb653dcfb17 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -619,6 +619,62 @@ mod tests { FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Fs)).build() } + #[test] + fn issue_5783_projection_rejects_selected_duplicate_root() { + use arrow::array::Int64Array; + use arrow::datatypes::{DataType, Field, Schema as ArrowSchema}; + use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; + + let physical = Arc::new(ArrowSchema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + ])); + let factory = super::SparkPhysicalExprAdapterFactory::new( + super::SparkParquetOptions::new(super::EvalMode::Legacy, "UTC", false), + None, + ); + for name in ["a", "b"] { + let target = Arc::new(ArrowSchema::new(vec![Field::new( + name, + DataType::Int64, + false, + )])); + let adapter = factory + .create(Arc::clone(&target), Arc::clone(&physical)) + .unwrap(); + let result = super::build_projection_expressions(&target, &adapter); + if name == "a" { + let error = result + .expect_err("selected root must be ambiguous") + .to_string(); + assert!(error.contains("duplicate"), "{error}"); + } else { + let batch = super::RecordBatch::try_new( + Arc::clone(&physical), + vec![ + Arc::new(Int64Array::from(vec![1])), + Arc::new(Int64Array::from(vec![2])), + Arc::new(Int64Array::from(vec![3])), + ], + ) + .unwrap(); + let output = + super::adapt_batch_with_expressions(batch, &target, &result.unwrap()).unwrap(); + assert_eq!(output.num_rows(), 1); + assert_eq!( + output + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 3 + ); + } + } + } + fn task_with_deletes(deletes: Vec) -> FileScanTask { FileScanTask::builder() .with_file_size_in_bytes(0) 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 c7de3581446..d89a1772835 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -44,13 +44,9 @@ //! 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. Preserve the -//! duplicate-field validation when replacing this factory. +//! page-index load back into `FileMetadataCache` instead of bypassing it. -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 arrow::datatypes::{DataType, FieldRef, Schema}; use async_trait::async_trait; use bytes::Bytes; use datafusion::common::Result as DFResult; @@ -81,8 +77,7 @@ use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaDataBuilder}; use parquet::file::metadata::{ FooterTail, PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader, }; -use parquet::schema::types::{ColumnDescPtr, SchemaDescriptor, Type}; -use std::collections::HashSet; +use parquet::schema::types::{ColumnDescPtr, SchemaDescriptor}; use std::fmt::{Debug, Display, Formatter}; use std::ops::Range; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -168,7 +163,6 @@ 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, - projection: Option<(SchemaRef, SparkParquetOptions)>, } impl EagerPageIndexReaderFactory { @@ -197,20 +191,9 @@ impl EagerPageIndexReaderFactory { metadata_cache, scan_io_metrics, spark_variant_schema: false, - projection: None, } } - pub(crate) fn with_required_schema( - mut self, - schema: &SchemaRef, - options: &SparkParquetOptions, - ) -> Self { - // Field-ID projections can rename columns, so names cannot safely restrict the walk. - self.projection = (!options.use_field_id).then(|| (Arc::clone(schema), options.clone())); - self - } - pub fn with_spark_variant_schema(mut self, enabled: bool) -> Self { self.spark_variant_schema = enabled; self @@ -242,7 +225,6 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, spark_variant_schema: self.spark_variant_schema, - projection: self.projection.clone(), })) } } @@ -258,7 +240,6 @@ struct EagerPageIndexReader { metadata_cache: Arc, metadata_size_hint: Option, spark_variant_schema: bool, - projection: Option<(SchemaRef, SparkParquetOptions)>, } // Arrow infers ENUM as Binary, losing the distinction from raw binary that Spark needs. @@ -391,80 +372,6 @@ fn with_spark_arrow_schema(metadata: Arc) -> ParquetResult, - case_sensitive: bool, -) -> parquet::errors::Result<()> { - if let Type::GroupType { fields, .. } = schema { - let mut names = HashSet::with_capacity(fields.len()); - for field in fields { - 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() - ))); - } - 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 @@ -532,7 +439,6 @@ 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 projection = self.projection.clone(); async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -592,71 +498,6 @@ 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(), - selected.as_ref(), - projection - .as_ref() - .is_none_or(|(_, options)| options.case_sensitive), - )?; if spark_variant_schema { with_spark_arrow_schema(metadata) } else { @@ -1000,198 +841,6 @@ 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 = Fields::from(vec![arrow::datatypes::Field::new( - "b", - DataType::Int64, - true, - )]); - validate_field_names(&schema, Some(&selected), true).unwrap(); - 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()); - } - - #[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 = Fields::from(vec![arrow::datatypes::Field::new( - "selected", - DataType::Int64, - true, - )]); - assert!(validate_field_names(&schema, Some(&selected), false).is_err()); - 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( - "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 7c4042bbdbc..0c78fb98fc3 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -177,8 +177,7 @@ 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; when replacing this factory, preserve - // its duplicate-field validation (#5783). + // applied. Filed upstream as apache/datafusion#23978. // // 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. @@ -195,8 +194,7 @@ pub(crate) fn init_datasource_exec( scan_io_source, parquet_source.metrics(), ) - .with_spark_variant_schema(projects_variant) - .with_required_schema(&required_schema, &spark_parquet_options), + .with_spark_variant_schema(projects_variant), ); parquet_source = parquet_source.with_parquet_file_reader_factory(reader_factory); @@ -470,6 +468,135 @@ mod tests { .as_usize() } + #[tokio::test] + #[ignore = "bounded baseline/candidate file-open timing"] + async fn issue_5783_file_open_cost() { + use std::time::Instant; + + let deadline = Instant::now() + Duration::from_secs(15 * 60); + for width in [100, 250, 500, 1000] { + for path in ["flat", "id", "nested"] { + for duplicate in [false, true] { + let fields: Vec = (0..width) + .map(|i| Field::new(format!("c{i}"), DataType::Int32, false)) + .collect(); + let mut logical = Arc::new(Schema::new(fields.clone())); + let mut physical = fields; + if duplicate { + physical.push(physical[0].clone()); + } + if path == "id" { + for (i, field) in physical.iter_mut().enumerate() { + *field = field.clone().with_metadata( + [("PARQUET:field_id".to_string(), i.to_string())].into(), + ); + } + logical = Arc::new(Schema::new(physical[..width].to_vec())); + } + let physical = Arc::new(Schema::new(physical)); + let mut batch = RecordBatch::try_new( + Arc::clone(&physical), + (0..physical.fields().len()) + .map(|_| Arc::new(Int32Array::from(vec![1])) as arrow::array::ArrayRef) + .collect(), + ) + .unwrap(); + if path == "nested" { + let values = arrow::array::StructArray::new( + physical.fields().clone(), + batch.columns().to_vec(), + None, + ); + let nested = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(physical.fields().clone()), + false, + )])); + batch = RecordBatch::try_new(nested, vec![Arc::new(values)]).unwrap(); + let mut requested = logical + .fields() + .iter() + .map(|f| f.as_ref().clone()) + .collect::>(); + requested.push(Field::new("missing", DataType::Int32, true)); + logical = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Struct(requested.into()), + false, + )])); + } + let file = tempfile::NamedTempFile::new().unwrap(); + let mut writer = + ArrowWriter::try_new(file.reopen().unwrap(), batch.schema(), None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let partition = + PartitionedFile::from_path(file.path().to_str().unwrap().to_string()) + .unwrap(); + for warm in [false, true] { + for repetition in 0..3 { + let shared = Arc::new(SessionContext::new()); + let mut elapsed = Duration::ZERO; + for iteration in 0..120 { + assert!(Instant::now() < deadline, "measurement budget exceeded"); + let context = if warm { + Arc::clone(&shared) + } else { + Arc::new(SessionContext::new()) + }; + let scan = if path == "id" { + init_datasource_exec( + Arc::clone(&logical), + Some(Arc::clone(&logical)), + None, + ObjectStoreUrl::local_filesystem(), + ObjectStoreBackend::Local, + vec![vec![partition.clone()]], + None, + None, + None, + "UTC", + true, + false, + false, + false, + &context, + false, + true, + false, + ) + .unwrap() + } else { + init_test_scan( + Arc::clone(&logical), + Arc::clone(&logical), + partition.clone(), + None, + None, + &context, + ) + }; + let start = Instant::now(); + let mut stream = scan.execute(0, context.task_ctx()).unwrap(); + let first = stream.next().await.unwrap(); + let duration = start.elapsed(); + if duplicate { + assert!(first.unwrap_err().to_string().contains("duplicate")); + } else { + assert_eq!(first.unwrap().num_rows(), 1); + } + if iteration >= 20 { + elapsed += duration; + } + } + println!("issue_5783_open path={path} width={width} duplicate={duplicate} warm={warm} rep={repetition} mean_us={:.3}", elapsed.as_secs_f64() * 1e4); + } + } + } + } + } + } + fn reader_metric(metrics: &ExecutionPlanMetricsSet, name: &str) -> usize { metrics .clone_inner() diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index d6ac9bed636..4fc2e99de34 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -420,8 +420,8 @@ fn field_id(field: &arrow::datatypes::Field) -> Option { /// ID-bearing requested fields match ONLY by ID (a missing ID is a missing column, never a name /// fallback); other fields match by name, folded with the same `toLowerCase(Locale.ROOT)` fold /// the top-level schema adapter uses when `case_sensitive` is false. A requested field whose -/// folded name matches more than one file field in case-insensitive mode raises Spark's -/// `foundDuplicateFieldInCaseInsensitiveModeError`. +/// folded name matches more than one file field is rejected. Case-insensitive matching retains +/// Spark's `foundDuplicateFieldInCaseInsensitiveModeError`. /// /// Shared by the runtime convert (`parquet_convert_struct_to_struct`) and the plan-time /// conversion check in `schema_adapter`, so both resolve nested fields identically. @@ -473,14 +473,14 @@ pub(crate) fn match_struct_fields( // falling back to name match. (true, Some(id)) => Ok(from_id_to_index.get(&id).copied()), _ => match folded_to_indices.get(to_folded[to_pos].as_str()) { - // Mirror Spark's `foundDuplicateFieldInCaseInsensitiveModeError`: a - // requested field matching more than one file field is ambiguous. Gated on - // case-insensitive mode to match the top-level check (which only runs when - // `!case_sensitive`): when case-sensitive the fold is identity, so a - // collision means byte-identical sibling names, and raising an error whose - // message says "in case-insensitive mode" would be wrong. Fall through to - // the first match in that case. - Some(indices) if indices.len() > 1 && !parquet_options.case_sensitive => { + // Reject selected ambiguity before a decoder can multiply rows. + Some(indices) if indices.len() > 1 => { + if parquet_options.case_sensitive { + return Err(DataFusionError::Execution(format!( + "Found duplicate Parquet field name '{}'", + to_field.name() + ))); + } let matched: Vec<&str> = indices .iter() .map(|&i| from_fields[i].name().as_str()) diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index d3c305ca136..9e447410bb1 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -97,6 +97,8 @@ fn schema_has_field_ids(schema: &SchemaRef) -> bool { /// (`build_projection_read_plan`'s cast-clipping, apache/datafusion#24090) can see the cast /// and read only the requested Parquet leaves, instead of falling back to a full-column read /// because it can't recognize `CometCastColumnExpr`. +/// This is also a decoder-safety obligation: returning true bypasses full-subtree +/// duplicate validation, so every omitted sibling must actually be clipped from the read. /// /// This is deliberately an allow list, not a deny list: it only recurses through the two /// container shapes `nested_struct::cast_column` actually implements (Struct, List / @@ -107,7 +109,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. -pub(crate) fn is_pure_structural_narrowing( +fn is_pure_structural_narrowing( physical_type: &DataType, target_type: &DataType, parquet_options: &SparkParquetOptions, @@ -810,6 +812,36 @@ fn check_conversion( } } +/// A Comet cast is opaque to Parquet leaf clipping and decodes its entire input subtree. +/// Check physical byte-identical names, not requested-name resolution, before that read. +fn check_decoded_field_names(data_type: &DataType) -> DataFusionResult<()> { + match data_type { + DataType::Struct(fields) => { + let mut names = HashSet::with_capacity(fields.len()); + for field in fields { + if !names.insert(field.name()) { + return Err(DataFusionError::Execution(format!( + "Found duplicate Parquet field name '{}'", + field.name() + ))); + } + check_decoded_field_names(field.data_type())?; + } + } + DataType::List(field) + | DataType::LargeList(field) + | DataType::FixedSizeList(field, _) + | DataType::ListView(field) + | DataType::LargeListView(field) + | DataType::Map(field, _) => { + check_decoded_field_names(field.data_type())?; + } + DataType::Dictionary(_, value) => check_decoded_field_names(value)?, + _ => {} + } + Ok(()) +} + /// Whether `col_name` (with folded form `col_folded`) is case-insensitively ambiguous in the /// file. `folded_to_indices` maps each folded physical name to the indices of the original /// physical fields that fold to it (built once in `create`), so more than one index under the @@ -856,42 +888,36 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { let should_match_by_id = self.parquet_options.use_field_id && schema_has_field_ids(&logical_file_schema); let needs_remap = !case_sensitive || should_match_by_id; - let (adapted_physical_schema, logical_to_physical_names, original_physical_dup_check) = - if needs_remap { - let (remapped, logical_to_physical) = remap_physical_schema( - &logical_file_schema, - &physical_file_schema, - case_sensitive, - self.parquet_options.use_field_id, - self.parquet_options.ignore_missing_field_id, - )?; - // Build the folded-name -> original-physical-field-indices map once for per-column - // duplicate detection, paired with the original schema so the rare error path can - // resolve the colliding names. Only meaningful in case-insensitive mode; it mirrors - // the `folded_to_indices` map the nested convert builds in `parquet_support`, so - // both paths detect ambiguity the same way instead of drifting. - let original_physical_dup_check = if !case_sensitive { - let folded = fold_schema_names(&physical_file_schema, false)?; - let mut map: HashMap> = HashMap::new(); - for (i, folded_name) in folded.into_iter().enumerate() { - map.entry(folded_name).or_default().push(i); - } - Some((Arc::clone(&physical_file_schema), map)) - } else { + let (adapted_physical_schema, logical_to_physical_names) = if needs_remap { + let (remapped, logical_to_physical) = remap_physical_schema( + &logical_file_schema, + &physical_file_schema, + case_sensitive, + self.parquet_options.use_field_id, + self.parquet_options.ignore_missing_field_id, + )?; + ( + remapped, + if logical_to_physical.is_empty() { None - }; - ( - remapped, - if logical_to_physical.is_empty() { - None - } else { - Some(logical_to_physical) - }, - original_physical_dup_check, - ) - } else { - (Arc::clone(&physical_file_schema), None, None) - }; + } else { + Some(logical_to_physical) + }, + ) + } else { + (Arc::clone(&physical_file_schema), None) + }; + + let mut duplicates: HashMap> = HashMap::new(); + for (i, name) in fold_schema_names(&physical_file_schema, case_sensitive)? + .into_iter() + .enumerate() + { + duplicates.entry(name).or_default().push(i); + } + duplicates.retain(|_, indices| indices.len() > 1); + let original_physical_dup_check = + (!duplicates.is_empty()).then(|| (Arc::clone(&physical_file_schema), duplicates)); // Fold both schemas once here so the per-column rewrite paths reuse them instead of // re-folding on every `rewrite` call. Case-sensitive mode folds to identity. @@ -899,11 +925,10 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { let physical_folded = fold_schema_names(&adapted_physical_schema, case_sensitive)?; // Folded names of logical fields that resolve by Parquet field id. Spark's `matchIdField` - // selects these by id before comparing names, so the case-insensitive duplicate check must + // selects these by id before comparing names, so the duplicate check must // skip them: an explicit `ω` (id 2) can select the file's `ω` (id 2) even when the file - // also holds `Ω` (id 1). Derived from `logical_folded`, which is the case-insensitive fold - // here since this only runs when `!case_sensitive`. - let id_resolved_logical_folded = if should_match_by_id && !case_sensitive { + // also holds `Ω` (id 1). Use the configured fold in both case modes. + let id_resolved_logical_folded = if should_match_by_id { Some( logical_file_schema .fields() @@ -917,6 +942,34 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { None }; + let id_duplicate_roots = if should_match_by_id { + let mut exact_names = HashSet::new(); + let mut duplicate_names = HashSet::new(); + for field in physical_file_schema.fields() { + if !exact_names.insert(field.name()) { + duplicate_names.insert(field.name()); + } + } + let duplicated_ids: HashMap = physical_file_schema + .fields() + .iter() + .filter(|f| duplicate_names.contains(f.name())) + .filter_map(|f| parse_field_id(f).map(|id| (id, f.name().clone()))) + .collect(); + logical_file_schema + .fields() + .iter() + .zip(&logical_folded) + .filter_map(|(field, folded)| { + parse_field_id(field) + .and_then(|id| duplicated_ids.get(&id)) + .map(|name| (folded.clone(), name.clone())) + }) + .collect() + } else { + HashMap::new() + }; + let default_factory = DefaultPhysicalExprAdapterFactory; let default_adapter = default_factory.create( Arc::clone(&logical_file_schema), @@ -932,6 +985,7 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { logical_to_physical_names, original_physical_dup_check, id_resolved_logical_folded, + id_duplicate_roots, logical_folded, physical_folded, })) @@ -963,16 +1017,16 @@ struct SparkPhysicalExprAdapter { /// physical names so that downstream reassign_expr_columns can find /// columns in the actual stream schema. logical_to_physical_names: Option>, - /// Case-insensitive duplicate detection, built once in `create`: the original (un-remapped) + /// Duplicate detection, built once in `create`: the original (un-remapped) /// physical schema paired with a `folded physical name -> field indices` map. A referenced - /// column whose folded name maps to more than one index is the `_LEGACY_ERROR_TEMP_2093` - /// ambiguity Spark raises; the schema resolves the colliding names on that error path. `None` - /// in case-sensitive mode (no folding, so nothing to detect). + /// column whose folded name maps to more than one index is ambiguous. The schema resolves + /// colliding names on the error path. `None` when no names collide. original_physical_dup_check: Option<(SchemaRef, HashMap>)>, /// Folded names of logical fields resolved by Parquet field id (see `create`). Spark selects /// these by id before comparing names, so the duplicate check above must not fire for them. /// `None` when not matching by id. id_resolved_logical_folded: Option>, + id_duplicate_roots: HashMap, /// `logical_file_schema` field names pre-folded once (see `fold_names`), parallel to /// `logical_file_schema.fields()`. Lets the per-column rewrite fallbacks match by folded name /// without re-folding the schema on every `rewrite` call. @@ -984,8 +1038,7 @@ struct SparkPhysicalExprAdapter { impl PhysicalExprAdapter for SparkPhysicalExprAdapter { fn rewrite(&self, expr: Arc) -> DataFusionResult> { - // In case-insensitive mode, check if any Column in this expression references - // a field with multiple case-insensitive matches in the physical schema. + // Check if any Column references multiple physical fields under the configured fold. // Only the columns actually referenced trigger the error (not the whole schema). if let Some((orig_physical, folded_to_indices)) = &self.original_physical_dup_check { // Collect referenced column names, then fold them in one JVM crossing rather than one @@ -998,8 +1051,13 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter { Ok(Transformed::no(e)) }); let col_refs: Vec<&str> = col_names.iter().map(|s| s.as_str()).collect(); - let col_folded = fold_names(&col_refs, false)?; + let col_folded = fold_names(&col_refs, self.parquet_options.case_sensitive)?; for (name, folded) in col_names.iter().zip(&col_folded) { + if let Some(physical_name) = self.id_duplicate_roots.get(folded) { + return Err(DataFusionError::Execution(format!( + "Found duplicate Parquet field name '{physical_name}'" + ))); + } // Fields resolved by Parquet field id are selected by id before names are // compared, so an id-resolved column must not trip the name-ambiguity check // (mirrors Spark's `matchIdField`, which never raises the duplicate-field error). @@ -1010,6 +1068,11 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter { { continue; } + if self.parquet_options.case_sensitive && folded_to_indices.contains_key(folded) { + return Err(DataFusionError::Execution(format!( + "Found duplicate Parquet field name '{name}'" + ))); + } if let Some(err) = check_column_duplicate(name, folded, folded_to_indices, orig_physical) { @@ -1168,6 +1231,7 @@ impl SparkPhysicalExprAdapter { physical_type: leaf_physical_type, target_type: leaf_target_type, } => { + check_decoded_field_names(physical_field.data_type())?; return Ok(Transformed::yes(reject_on_non_empty_expr( remapped, logical_field, @@ -1178,6 +1242,7 @@ impl SparkPhysicalExprAdapter { } } + check_decoded_field_names(physical_field.data_type())?; let cast_expr: Arc = Arc::new( CometCastColumnExpr::try_new( remapped, @@ -1269,6 +1334,7 @@ impl SparkPhysicalExprAdapter { physical_type: leaf_physical_type, target_type: leaf_target_type, } => { + check_decoded_field_names(physical_type)?; return Ok(Transformed::yes(reject_on_non_empty_expr( child, cast.target_field(), @@ -1312,6 +1378,8 @@ impl SparkPhysicalExprAdapter { return Ok(Transformed::no(expr)); } + check_decoded_field_names(physical_type)?; + // Complex casts (including changes in list representation), timestamp tz relabel // (e.g. Timestamp(us, None) -> Timestamp(us, Some("UTC")) for INT96 reads), and // Timestamp -> Int64 @@ -2993,6 +3061,274 @@ mod test { Ok(()) } + #[test] + fn issue_5783_referenced_root_duplicate() { + for nullable in [false, true] { + let physical = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("a", DataType::Int64, false), + Field::new("b", DataType::Int64, false), + ])); + let logical = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, nullable), + Field::new("b", DataType::Int64, false), + ])); + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.case_sensitive = true; + let adapter = SparkPhysicalExprAdapterFactory::new(options, None) + .create(logical, physical) + .unwrap(); + let selected = adapter.rewrite(Arc::new(Column::new("a", 0))); + let error = selected + .expect_err("selected duplicate root must fail") + .to_string(); + assert!(error.contains("duplicate"), "{error}"); + assert!(!error.contains("case-insensitive"), "{error}"); + let predicate = datafusion::physical_expr::expressions::BinaryExpr::new( + Arc::new(Column::new("a", 0)), + datafusion::logical_expr::Operator::Gt, + Arc::new(datafusion::physical_expr::expressions::Literal::new( + datafusion::common::ScalarValue::Int64(Some(0)), + )), + ); + assert!(adapter + .rewrite(Arc::new(predicate)) + .unwrap_err() + .to_string() + .contains("duplicate")); + let safe = adapter.rewrite(Arc::new(Column::new("b", 1))).unwrap(); + assert_eq!(safe.downcast_ref::().unwrap().index(), 2); + } + } + + #[test] + fn issue_5783_nested_name_duplicate() { + let fields = vec![ + Arc::new(Field::new("dup", DataType::Int64, true)), + Arc::new(Field::new("dup", DataType::Int64, true)), + ]; + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.case_sensitive = true; + let error = super::match_struct_fields(&fields, &fields[..1], &options) + .expect_err("selected duplicate child must fail") + .to_string(); + assert!(error.contains("duplicate"), "{error}"); + assert!(!error.contains("case-insensitive"), "{error}"); + } + + #[test] + fn issue_5783_fallback_deferred_rejection_checks_decoded_subtree() { + let logical = struct_schema(vec![ + Field::new("other", DataType::Int32, true), + Field::new("force_fallback", DataType::Int32, true), + ]); + for duplicate in [false, true] { + let mut fields = vec![ + Field::new("other", DataType::Int64, true), + Field::new( + "force_fallback", + DataType::List(Arc::new(Field::new("element", DataType::Int32, true))), + true, + ), + Field::new("dup", DataType::Int64, true), + ]; + if duplicate { + fields.push(Field::new("dup", DataType::Int64, true)); + } + let physical = struct_schema(fields); + let column: Arc = Arc::new(Column::new("s", 0)); + let default = super::DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical), Arc::clone(&physical)) + .unwrap(); + assert!( + default.rewrite(Arc::clone(&column)).is_err(), + "fixture must reach the default-adapter fallback" + ); + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.case_sensitive = true; + let adapter = SparkPhysicalExprAdapterFactory::new(options, None) + .create(Arc::clone(&logical), Arc::clone(&physical)) + .unwrap(); + let result = adapter.rewrite(column); + if duplicate { + let error = result + .expect_err("decoded duplicates must fail") + .to_string(); + assert!( + error.contains("duplicate Parquet field name 'dup'"), + "{error}" + ); + } else { + let expr = result.unwrap(); + assert!(expr.downcast_ref::().is_some()); + let empty = RecordBatch::new_empty(physical); + assert_eq!( + expr.evaluate(&empty).unwrap().into_array(0).unwrap().len(), + 0 + ); + } + } + } + + #[tokio::test] + async fn issue_5783_non_pruning_scan_rejects_before_output() { + for mode in ["deferred", "missing", "dictionary"] { + let values: ArrayRef = Arc::new(Int64Array::from(vec![900, 901, 902])); + let other = if mode == "dictionary" { + arrow::compute::cast( + &values, + &DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Int64)), + ) + .unwrap() + } else { + Arc::clone(&values) + }; + let fields = Fields::from(vec![ + Field::new("dup", DataType::Int64, true), + Field::new("dup", DataType::Int64, true), + Field::new("other", other.data_type().clone(), true), + ]); + let array = StructArray::new( + fields.clone(), + vec![Arc::clone(&values), values, other], + None, + ); + let batch = RecordBatch::try_new( + struct_schema(fields.iter().map(|f| f.as_ref().clone()).collect()), + vec![Arc::new(array)], + ) + .unwrap(); + let mut requested = vec![Field::new( + "other", + if mode == "deferred" { + DataType::Int32 + } else { + DataType::Int64 + }, + true, + )]; + if mode == "missing" { + requested.push(Field::new("missing", DataType::Int64, true)); + } + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.case_sensitive = true; + let mut stream = scan_parquet(&batch, struct_schema(requested), options).unwrap(); + let error = stream + .next() + .await + .unwrap() + .expect_err("must fail before first batch"); + assert!( + error + .to_string() + .contains("duplicate Parquet field name 'dup'"), + "{mode}: {error}" + ); + } + } + + #[tokio::test] + async fn issue_5783_physical_duplicates_survive_arrow_and_fail_before_output() { + use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + + for shape in 0..5 { + let mut children = vec![ + Field::new("dup", DataType::Int64, true), + Field::new("dup", DataType::Int64, true), + ]; + if shape == 1 { + children.push(Field::new("dup", DataType::Int64, true)); + } + if shape == 2 { + children.push(Field::new("other", DataType::Int64, true)); + } + let fields = Fields::from(children); + let values: ArrayRef = Arc::new(Int64Array::from(vec![0, 1, 2])); + let mut array: ArrayRef = Arc::new(StructArray::new( + fields.clone(), + (0..fields.len()).map(|_| Arc::clone(&values)).collect(), + None, + )); + let mut requested = + DataType::Struct(Fields::from(vec![Field::new("dup", DataType::Int64, true)])); + if shape == 3 { + array = Arc::new(ListArray::new( + Arc::new(Field::new("element", array.data_type().clone(), true)), + OffsetBuffer::new(vec![0, 1, 2, 3].into()), + array, + None, + )); + requested = DataType::List(Arc::new(Field::new("element", requested, true))); + } + if shape == 4 { + let entries = StructArray::new( + Fields::from(vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", array.data_type().clone(), true), + ]), + vec![ + Arc::new(arrow::array::StringArray::from(vec!["k", "k", "k"])), + array, + ], + None, + ); + array = Arc::new(arrow::array::MapArray::new( + Arc::new(Field::new("entries", entries.data_type().clone(), false)), + OffsetBuffer::new(vec![0, 1, 2, 3].into()), + entries, + None, + false, + )); + requested = DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct(Fields::from(vec![ + Field::new("key", DataType::Utf8, false), + Field::new("value", requested, true), + ])), + false, + )), + false, + ); + } + let physical = Arc::new(Schema::new(vec![Field::new( + "s", + array.data_type().clone(), + true, + )])); + let batch = RecordBatch::try_new(physical, vec![array]).unwrap(); + let required = Arc::new(Schema::new(vec![Field::new("s", requested, true)])); + let mut bytes = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut bytes, batch.schema(), None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let reader = + ParquetRecordBatchReaderBuilder::try_new(bytes::Bytes::from(bytes)).unwrap(); + assert_eq!( + reader.schema().field(0).data_type(), + batch.schema().field(0).data_type() + ); + assert_ne!( + reader.schema().field(0).data_type(), + required.field(0).data_type() + ); + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.case_sensitive = true; + let mut stream = scan_parquet(&batch, required, options).unwrap(); + let error = stream + .next() + .await + .unwrap() + .expect_err("duplicate must fail before first batch"); + assert!( + error + .to_string() + .contains("duplicate Parquet field name 'dup'"), + "{error}" + ); + } + } + #[tokio::test] async fn parquet_duplicate_fields_case_insensitive() { // Parquet file has columns "A", "B", "b" - reading "b" in case-insensitive mode 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 b4fdcfea075..dd3c2974ba1 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,24 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + test("duplicate Parquet field names - Spark logical schema legality") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(3).coalesce(1).write.parquet(path.toString) + Seq( + "s struct", + "s array>", + "s map>", + "s struct>").foreach { schema => + val error = intercept[org.apache.spark.sql.AnalysisException] { + spark.read.schema(schema).parquet(path.toString).collect() + } + assert(error.getMessage.contains("COLUMN_ALREADY_EXISTS"), error.getMessage) + } + } + } + } + Seq( ("two children", "named_struct('dup', id, 'dup', id + 100)", "struct"), ( @@ -72,29 +90,34 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper "map value", "map('key', named_struct('dup', id, 'dup', id + 100))", "map>")).foreach { case (shape, expression, readType) => - 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) + Seq(1, 4096).foreach { batchSize => + test(s"duplicate Parquet field names fail clearly - $shape batch $batchSize") { + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "true", + CometConf.COMET_BATCH_SIZE.key -> batchSize.toString) { + 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) } - 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) } } } @@ -112,7 +135,10 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper Seq(true, false).foreach { caseSensitive => withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) { val name = if (caseSensitive) "id" else "ID" - val df = spark.read.schema(s"$name bigint").parquet(path.toString) + val df = spark.read + .schema(s"$name bigint, s struct") + .parquet(path.toString) + .select(name) assert( find(df.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) (1 to 2).foreach { _ => @@ -167,13 +193,22 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper 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) + (1 to 2).foreach { _ => checkAnswer(byId, Seq(Row(3L))) } + for (id <- Seq(1L, 2L); name <- Seq("a", "renamed_a")) { + val duplicateSchema = new StructType().add( + name, + LongType, + nullable = true, + new MetadataBuilder().putLong("parquet.field.id", id).build()) + val duplicate = spark.read.schema(duplicateSchema).parquet(path.toString) + val error = intercept[Exception](duplicate.collect()) + val messages = Iterator + .iterate[Throwable](error)(_.getCause) + .takeWhile(_ != null) + .map(_.getMessage) + .mkString("\n") + assert(messages.contains("duplicate Parquet field name 'a'"), messages) + } } } } @@ -397,7 +432,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } - test("duplicate Parquet field names outside a nested projection remain readable") { + test("duplicate Parquet field names - exact-name projection works in both resolver modes") { withTempPath { path => withSQLConf(CometConf.COMET_ENABLED.key -> "false") { spark @@ -409,17 +444,36 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } 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) + val df = spark.read.schema("s struct").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()) + } + } + } + } + + Seq( + ( + "map", + "map('k', named_struct('dup', id, 'dup', id + 100, 'other', id + 900))", + "s map>"), + ( + "case", + "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)", + "S struct")).foreach { case (shape, expression, schema) => + test(s"duplicate Parquet field names - non-pruning $shape fails clearly") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(3).coalesce(1).selectExpr(s"$expression as s").write.parquet(path.toString) + assert(spark.read.schema(schema).parquet(path.toString).collect().length == 3) + } + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val df = spark.read.schema(schema).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) @@ -431,6 +485,56 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + Seq( + ( + "struct", + "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)", + "struct", + "s.other", + (n: Long) => Row(Row(n))), + ( + "deeper", + "named_struct('t', named_struct('dup', id, 'dup', id + 100, 'other', id + 900))", + "struct>", + "s.t.other", + (n: Long) => Row(Row(Row(n)))), + ( + "array", + "array(named_struct('dup', id, 'dup', id + 100, 'other', id + 900))", + "array>", + "s[0].other", + (n: Long) => Row(Seq(Row(n))))) + .foreach { case (shape, expression, readType, predicate, expected) => + test(s"duplicate Parquet field names - $shape projection and positive filter") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(3) + .coalesce(1) + .selectExpr(s"$expression as s") + .write + .parquet(path.toString) + } + Seq(1, 4096).foreach { batchSize => + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "true", + CometConf.COMET_BATCH_SIZE.key -> batchSize.toString, + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> "true") { + val df = spark.read.schema(s"s $readType").parquet(path.toString) + assert( + find(df.queryExecution.executedPlan)( + _.isInstanceOf[CometNativeScanExec]).isDefined) + checkAnswer(df, Seq(900L, 901L, 902L).map(expected)) + checkSparkAnswer(df) + val filtered = df.where(s"$predicate >= 901") + checkAnswer(filtered, Seq(901L, 902L).map(expected)) + checkSparkAnswer(filtered) + } + } + } + } + } + test("native reader - read simple STRUCT fields") { testSingleLineQuery( """ From f5e1dc5948276f9fceb8c833f1ff0af417b43d43 Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Tue, 22 Sep 2026 10:12:47 -0300 Subject: [PATCH 7/8] fix: validate decoded duplicate fields on every cast path Also make the Iceberg adapter test exercise case-sensitive mode, cover multi-file reads with and without mergeSchema, correct the compatibility note, and drop the ignored file-open timing harness. --- .../user-guide/latest/compatibility/scans.md | 21 +-- .../src/execution/operators/iceberg_scan.rs | 7 +- native/core/src/parquet/parquet_exec.rs | 129 ------------------ native/core/src/parquet/schema_adapter.rs | 3 + .../comet/exec/CometNativeReaderSuite.scala | 85 +++++++++--- 5 files changed, 81 insertions(+), 164 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index 695c6221b43..e58000fffa3 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -63,16 +63,17 @@ 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: - Selecting a field by name when multiple physical siblings match, including inside structs, - arrays, and maps. Comet reports a duplicate-field error instead of resolving byte-identical - names, in either case-sensitivity mode. Root checks cover referenced columns, including - predicates. Unselected roots do not prevent reading a unique field by name or field ID. - Exact-name projections of unique children in structs and arrays of structs remain supported; - casts that cannot use this pruning reject duplicate siblings anywhere in the decoded - physical subtree, including map and case-differing nested projections. Field-ID resolution - retains precedence, but selecting a byte-identically duplicated physical root name still - raises a duplicate-field error, even when the requested field is renamed. - 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 + arrays, and maps. Comet raises a duplicate-field error instead of resolving the collision. + Checks cover referenced columns, including predicates; unselected roots do not prevent + reading a unique field by name or field ID. Exact-name projections of unique children in + structs and arrays of structs remain supported. Casts that cannot use this pruning reject + byte-identical duplicate siblings anywhere in the decoded physical subtree, including maps. + Field-ID resolution retains precedence, but selecting a byte-identically duplicated physical + root name still raises a duplicate-field error, even when the requested field is renamed. + Names in separate groups do not collide. In case-sensitive mode Spark instead silently picks + one sibling, so disable Comet for the query with an explicit read schema to use that + resolution; a schema inferred from a file containing duplicate names is rejected by Spark + before Comet runs. 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 diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index cb653dcfb17..9f465329ca0 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -630,10 +630,9 @@ mod tests { Field::new("a", DataType::Int64, false), Field::new("b", DataType::Int64, false), ])); - let factory = super::SparkPhysicalExprAdapterFactory::new( - super::SparkParquetOptions::new(super::EvalMode::Legacy, "UTC", false), - None, - ); + let mut options = super::SparkParquetOptions::new(super::EvalMode::Legacy, "UTC", false); + options.case_sensitive = true; + let factory = super::SparkPhysicalExprAdapterFactory::new(options, None); for name in ["a", "b"] { let target = Arc::new(ArrowSchema::new(vec![Field::new( name, diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 0c78fb98fc3..93ac29e824b 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -468,135 +468,6 @@ mod tests { .as_usize() } - #[tokio::test] - #[ignore = "bounded baseline/candidate file-open timing"] - async fn issue_5783_file_open_cost() { - use std::time::Instant; - - let deadline = Instant::now() + Duration::from_secs(15 * 60); - for width in [100, 250, 500, 1000] { - for path in ["flat", "id", "nested"] { - for duplicate in [false, true] { - let fields: Vec = (0..width) - .map(|i| Field::new(format!("c{i}"), DataType::Int32, false)) - .collect(); - let mut logical = Arc::new(Schema::new(fields.clone())); - let mut physical = fields; - if duplicate { - physical.push(physical[0].clone()); - } - if path == "id" { - for (i, field) in physical.iter_mut().enumerate() { - *field = field.clone().with_metadata( - [("PARQUET:field_id".to_string(), i.to_string())].into(), - ); - } - logical = Arc::new(Schema::new(physical[..width].to_vec())); - } - let physical = Arc::new(Schema::new(physical)); - let mut batch = RecordBatch::try_new( - Arc::clone(&physical), - (0..physical.fields().len()) - .map(|_| Arc::new(Int32Array::from(vec![1])) as arrow::array::ArrayRef) - .collect(), - ) - .unwrap(); - if path == "nested" { - let values = arrow::array::StructArray::new( - physical.fields().clone(), - batch.columns().to_vec(), - None, - ); - let nested = Arc::new(Schema::new(vec![Field::new( - "s", - DataType::Struct(physical.fields().clone()), - false, - )])); - batch = RecordBatch::try_new(nested, vec![Arc::new(values)]).unwrap(); - let mut requested = logical - .fields() - .iter() - .map(|f| f.as_ref().clone()) - .collect::>(); - requested.push(Field::new("missing", DataType::Int32, true)); - logical = Arc::new(Schema::new(vec![Field::new( - "s", - DataType::Struct(requested.into()), - false, - )])); - } - let file = tempfile::NamedTempFile::new().unwrap(); - let mut writer = - ArrowWriter::try_new(file.reopen().unwrap(), batch.schema(), None).unwrap(); - writer.write(&batch).unwrap(); - writer.close().unwrap(); - let partition = - PartitionedFile::from_path(file.path().to_str().unwrap().to_string()) - .unwrap(); - for warm in [false, true] { - for repetition in 0..3 { - let shared = Arc::new(SessionContext::new()); - let mut elapsed = Duration::ZERO; - for iteration in 0..120 { - assert!(Instant::now() < deadline, "measurement budget exceeded"); - let context = if warm { - Arc::clone(&shared) - } else { - Arc::new(SessionContext::new()) - }; - let scan = if path == "id" { - init_datasource_exec( - Arc::clone(&logical), - Some(Arc::clone(&logical)), - None, - ObjectStoreUrl::local_filesystem(), - ObjectStoreBackend::Local, - vec![vec![partition.clone()]], - None, - None, - None, - "UTC", - true, - false, - false, - false, - &context, - false, - true, - false, - ) - .unwrap() - } else { - init_test_scan( - Arc::clone(&logical), - Arc::clone(&logical), - partition.clone(), - None, - None, - &context, - ) - }; - let start = Instant::now(); - let mut stream = scan.execute(0, context.task_ctx()).unwrap(); - let first = stream.next().await.unwrap(); - let duration = start.elapsed(); - if duplicate { - assert!(first.unwrap_err().to_string().contains("duplicate")); - } else { - assert_eq!(first.unwrap().num_rows(), 1); - } - if iteration >= 20 { - elapsed += duration; - } - } - println!("issue_5783_open path={path} width={width} duplicate={duplicate} warm={warm} rep={repetition} mean_us={:.3}", elapsed.as_secs_f64() * 1e4); - } - } - } - } - } - } - fn reader_metric(metrics: &ExecutionPlanMetricsSet, name: &str) -> usize { metrics .clone_inner() diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 9e447410bb1..de5e95df8a0 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -1154,6 +1154,8 @@ impl SparkPhysicalExprAdapter { return Ok(expr); }; + check_decoded_field_names(physical_field.data_type())?; + Ok(Arc::new( CometCastColumnExpr::try_new( expr, @@ -1290,6 +1292,7 @@ impl SparkPhysicalExprAdapter { .target_field() .has_valid_extension_type::() { + check_decoded_field_names(physical_type)?; let comet_cast: Arc = Arc::new( CometCastColumnExpr::try_new( child, 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 dd3c2974ba1..3d1fe07773a 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,9 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + private def causeMessages(error: Throwable): String = + causeChain(error).flatMap(e => Option(e.getMessage)).mkString("\n") + test("duplicate Parquet field names - Spark logical schema legality") { withTempPath { path => withSQLConf(CometConf.COMET_ENABLED.key -> "false") { @@ -111,13 +114,65 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper 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") + val messages = causeMessages(error) assert(messages.contains("duplicate Parquet field name 'dup'"), messages) + assert(!messages.toLowerCase.contains("case-insensitive"), messages) + } + } + } + } + } + + test("duplicate Parquet field names - multiple files and schema merge") { + withTempPath { cleanPath => + withTempPath { duplicatePath => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(3) + .coalesce(1) + .selectExpr("named_struct('dup', id, 'other', id + 900) as s") + .write + .parquet(cleanPath.toString) + spark + .range(3) + .coalesce(1) + .selectExpr("named_struct('dup', id, 'dup', id + 100, 'other', id + 900) as s") + .write + .parquet(duplicatePath.toString) + } + val paths = Seq(cleanPath.toString, duplicatePath.toString) + Seq("true", "false").foreach { mergeSchema => + Seq(true, false).foreach { caseSensitive => + withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) { + // mergeSchema only affects inference, and Spark already rejects inference for a + // duplicate-bearing file, so an explicit schema reads the unique sibling natively. + val unique = spark.read + .option("mergeSchema", mergeSchema) + .schema("s struct") + .parquet(paths: _*) + assert(find(unique.queryExecution.executedPlan)( + _.isInstanceOf[CometNativeScanExec]).isDefined) + checkSparkAnswerAndOperator(unique) + checkAnswer(unique, Seq(900L, 901L, 902L, 900L, 901L, 902L).map(n => Row(Row(n)))) + checkAnswer( + unique.where("s.other >= 901"), + Seq(901L, 902L, 901L, 902L).map(n => Row(Row(n)))) + val duplicate = spark.read + .option("mergeSchema", mergeSchema) + .schema("s struct") + .parquet(paths: _*) + assert(find(duplicate.queryExecution.executedPlan)( + _.isInstanceOf[CometNativeScanExec]).isDefined) + val error = intercept[Exception](duplicate.collect()) + assert(causeMessages(error).toLowerCase.contains("duplicate"), causeMessages(error)) + } + } + } + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val inferred = intercept[org.apache.spark.sql.AnalysisException] { + spark.read.option("mergeSchema", "true").parquet(paths: _*).collect() } + assert(inferred.getMessage.contains("COLUMN_ALREADY_EXISTS"), inferred.getMessage) } } } @@ -174,16 +229,12 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper 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") + val messages = causeMessages(error) 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))) + (1 to 2).foreach { _ => checkAnswer(valid, Seq(Row(3L))) } withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { val schema = new StructType().add( "renamed_b", @@ -202,11 +253,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper new MetadataBuilder().putLong("parquet.field.id", id).build()) val duplicate = spark.read.schema(duplicateSchema).parquet(path.toString) val error = intercept[Exception](duplicate.collect()) - val messages = Iterator - .iterate[Throwable](error)(_.getCause) - .takeWhile(_ != null) - .map(_.getMessage) - .mkString("\n") + val messages = causeMessages(error) assert(messages.contains("duplicate Parquet field name 'a'"), messages) } } @@ -474,11 +521,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper 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") + val messages = causeMessages(error) assert(messages.contains("duplicate Parquet field name 'dup'"), messages) } } From 2d082c054a5e7e59b818eeee7700e3d40664fffd Mon Sep 17 00:00:00 2001 From: Erik Bogado Date: Wed, 23 Sep 2026 00:54:07 -0300 Subject: [PATCH 8/8] fix: finish duplicate Parquet field review Make the Spark inference assertion deterministic and exercise both Variant adapter paths. Centralize decoded-field checks and diagnostics, and avoid per-column index vectors when names do not collide. Clarify Spark fallback behavior and link the related issues. Refs #5783 --- .../user-guide/latest/compatibility/scans.md | 11 +- native/core/src/parquet/parquet_support.rs | 9 +- native/core/src/parquet/schema_adapter.rs | 164 +++++++++++++----- .../comet/exec/CometNativeReaderSuite.scala | 5 +- 4 files changed, 135 insertions(+), 54 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index e58000fffa3..d479e75ffd8 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -70,11 +70,12 @@ The following limitations raise an error at scan time rather than falling back t byte-identical duplicate siblings anywhere in the decoded physical subtree, including maps. Field-ID resolution retains precedence, but selecting a byte-identically duplicated physical root name still raises a duplicate-field error, even when the requested field is renamed. - Names in separate groups do not collide. In case-sensitive mode Spark instead silently picks - one sibling, so disable Comet for the query with an explicit read schema to use that - resolution; a schema inferred from a file containing duplicate names is rejected by Spark - before Comet runs. Spark-compatible resolution is tracked in - [#5884](https://github.com/apache/datafusion-comet/issues/5884). + Names in separate groups do not collide. Spark may read a duplicate-bearing file with an + explicit schema in case-sensitive mode, but its choice of sibling depends on the field shape + and can produce unexpected values. Spark rejects schema inference from a single file with + duplicate names; inference across files can depend on merge order. + Resolution is tracked in [#5884](https://github.com/apache/datafusion-comet/issues/5884), + with mixed-type behavior in [#5964](https://github.com/apache/datafusion-comet/issues/5964). - 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/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 4fc2e99de34..521f95d4590 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -61,6 +61,10 @@ use super::objectstore::s3_blob_fs_support::{ normalize_object_store_url, NormalizedObjectStoreUrl, }; +pub(crate) fn duplicate_parquet_field_error(name: &str) -> DataFusionError { + DataFusionError::Execution(format!("Found duplicate Parquet field name '{name}'")) +} + // This file originates from cast.rs. While developing native scan support and implementing // SparkSchemaAdapter we observed that Spark's type conversion logic on Parquet reads does not // always align to the CAST expression's logic, so it was duplicated here to adapt its behavior. @@ -476,10 +480,7 @@ pub(crate) fn match_struct_fields( // Reject selected ambiguity before a decoder can multiply rows. Some(indices) if indices.len() > 1 => { if parquet_options.case_sensitive { - return Err(DataFusionError::Execution(format!( - "Found duplicate Parquet field name '{}'", - to_field.name() - ))); + return Err(duplicate_parquet_field_error(to_field.name())); } let matched: Vec<&str> = indices .iter() diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index de5e95df8a0..dc6a7799588 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -18,7 +18,7 @@ use crate::parquet::cast_column::CometCastColumnExpr; use crate::parquet::name_fold::{fold_name, fold_names, fold_schema_names}; use crate::parquet::parquet_support::{ - match_struct_fields, spark_parquet_convert, SparkParquetOptions, + duplicate_parquet_field_error, match_struct_fields, spark_parquet_convert, SparkParquetOptions, }; use arrow::array::new_empty_array; use arrow::compute::can_cast_types; @@ -97,8 +97,8 @@ fn schema_has_field_ids(schema: &SchemaRef) -> bool { /// (`build_projection_read_plan`'s cast-clipping, apache/datafusion#24090) can see the cast /// and read only the requested Parquet leaves, instead of falling back to a full-column read /// because it can't recognize `CometCastColumnExpr`. -/// This is also a decoder-safety obligation: returning true bypasses full-subtree -/// duplicate validation, so every omitted sibling must actually be clipped from the read. +/// This is also a decoder-safety obligation: returning true bypasses full-subtree duplicate +/// validation, so every omitted sibling must actually be clipped from the read. /// /// This is deliberately an allow list, not a deny list: it only recurses through the two /// container shapes `nested_struct::cast_column` actually implements (Struct, List / @@ -820,10 +820,7 @@ fn check_decoded_field_names(data_type: &DataType) -> DataFusionResult<()> { let mut names = HashSet::with_capacity(fields.len()); for field in fields { if !names.insert(field.name()) { - return Err(DataFusionError::Execution(format!( - "Found duplicate Parquet field name '{}'", - field.name() - ))); + return Err(duplicate_parquet_field_error(field.name())); } check_decoded_field_names(field.data_type())?; } @@ -842,6 +839,15 @@ fn check_decoded_field_names(data_type: &DataType) -> DataFusionResult<()> { Ok(()) } +/// Check an expression before it can decode its entire physical subtree. +fn checked_decoded_expr( + physical_type: &DataType, + expr: Arc, +) -> DataFusionResult> { + check_decoded_field_names(physical_type)?; + Ok(expr) +} + /// Whether `col_name` (with folded form `col_folded`) is case-insensitively ambiguous in the /// file. `folded_to_indices` maps each folded physical name to the indices of the original /// physical fields that fold to it (built once in `create`), so more than one index under the @@ -908,21 +914,36 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { (Arc::clone(&physical_file_schema), None) }; - let mut duplicates: HashMap> = HashMap::new(); - for (i, name) in fold_schema_names(&physical_file_schema, case_sensitive)? - .into_iter() - .enumerate() - { - duplicates.entry(name).or_default().push(i); - } - duplicates.retain(|_, indices| indices.len() > 1); - let original_physical_dup_check = - (!duplicates.is_empty()).then(|| (Arc::clone(&physical_file_schema), duplicates)); - // Fold both schemas once here so the per-column rewrite paths reuse them instead of // re-folding on every `rewrite` call. Case-sensitive mode folds to identity. let logical_folded = fold_schema_names(&logical_file_schema, case_sensitive)?; let physical_folded = fold_schema_names(&adapted_physical_schema, case_sensitive)?; + let original_folded = if Arc::ptr_eq(&adapted_physical_schema, &physical_file_schema) { + None + } else { + Some(fold_schema_names(&physical_file_schema, case_sensitive)?) + }; + let original_folded = original_folded.as_ref().unwrap_or(&physical_folded); + + // Only allocate per-column index vectors when a folded name actually collides. + let mut seen = HashSet::new(); + let mut collisions = HashSet::new(); + for name in original_folded { + if !seen.insert(name.as_str()) { + collisions.insert(name.as_str()); + } + } + let original_physical_dup_check = if collisions.is_empty() { + None + } else { + let mut duplicates: HashMap> = HashMap::new(); + for (i, name) in original_folded.iter().enumerate() { + if collisions.contains(name.as_str()) { + duplicates.entry(name.clone()).or_default().push(i); + } + } + Some((Arc::clone(&physical_file_schema), duplicates)) + }; // Folded names of logical fields that resolve by Parquet field id. Spark's `matchIdField` // selects these by id before comparing names, so the duplicate check must @@ -1026,6 +1047,9 @@ struct SparkPhysicalExprAdapter { /// these by id before comparing names, so the duplicate check above must not fire for them. /// `None` when not matching by id. id_resolved_logical_folded: Option>, + /// Folded logical name -> byte-identical duplicate physical name. Populated only when + /// matching by field ID, then checked before the ID-resolved name skip so decoded duplicate + /// roots still fail. id_duplicate_roots: HashMap, /// `logical_file_schema` field names pre-folded once (see `fold_names`), parallel to /// `logical_file_schema.fields()`. Lets the per-column rewrite fallbacks match by folded name @@ -1054,9 +1078,7 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter { let col_folded = fold_names(&col_refs, self.parquet_options.case_sensitive)?; for (name, folded) in col_names.iter().zip(&col_folded) { if let Some(physical_name) = self.id_duplicate_roots.get(folded) { - return Err(DataFusionError::Execution(format!( - "Found duplicate Parquet field name '{physical_name}'" - ))); + return Err(duplicate_parquet_field_error(physical_name)); } // Fields resolved by Parquet field id are selected by id before names are // compared, so an id-resolved column must not trip the name-ambiguity check @@ -1069,9 +1091,7 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter { continue; } if self.parquet_options.case_sensitive && folded_to_indices.contains_key(folded) { - return Err(DataFusionError::Execution(format!( - "Found duplicate Parquet field name '{name}'" - ))); + return Err(duplicate_parquet_field_error(name)); } if let Some(err) = check_column_duplicate(name, folded, folded_to_indices, orig_physical) @@ -1154,9 +1174,7 @@ impl SparkPhysicalExprAdapter { return Ok(expr); }; - check_decoded_field_names(physical_field.data_type())?; - - Ok(Arc::new( + let cast: Arc = Arc::new( CometCastColumnExpr::try_new( expr, Arc::clone(physical_field), @@ -1164,7 +1182,8 @@ impl SparkPhysicalExprAdapter { None, )? .with_parquet_options(self.parquet_options.clone()), - )) + ); + checked_decoded_expr(physical_field.data_type(), cast) } /// Wrap ALL Column expressions that have type mismatches with CometCastColumnExpr. @@ -1233,18 +1252,20 @@ impl SparkPhysicalExprAdapter { physical_type: leaf_physical_type, target_type: leaf_target_type, } => { - check_decoded_field_names(physical_field.data_type())?; - return Ok(Transformed::yes(reject_on_non_empty_expr( + let rejected = reject_on_non_empty_expr( remapped, logical_field, &column, &leaf_physical_type, &leaf_target_type, - ))); + ); + return Ok(Transformed::yes(checked_decoded_expr( + physical_field.data_type(), + rejected, + )?)); } } - check_decoded_field_names(physical_field.data_type())?; let cast_expr: Arc = Arc::new( CometCastColumnExpr::try_new( remapped, @@ -1254,7 +1275,10 @@ impl SparkPhysicalExprAdapter { )? .with_parquet_options(self.parquet_options.clone()), ); - return Ok(Transformed::yes(cast_expr)); + return Ok(Transformed::yes(checked_decoded_expr( + physical_field.data_type(), + cast_expr, + )?)); } else if column.index() != phys_idx { return Ok(Transformed::yes(remapped)); } @@ -1292,16 +1316,16 @@ impl SparkPhysicalExprAdapter { .target_field() .has_valid_extension_type::() { - check_decoded_field_names(physical_type)?; let comet_cast: Arc = Arc::new( CometCastColumnExpr::try_new( child, - input_field, + Arc::clone(&input_field), Arc::clone(cast.target_field()), None, )? .with_parquet_options(self.parquet_options.clone()), ); + let comet_cast = checked_decoded_expr(physical_type, comet_cast)?; return Ok(Transformed::yes(comet_cast)); } @@ -1337,14 +1361,17 @@ impl SparkPhysicalExprAdapter { physical_type: leaf_physical_type, target_type: leaf_target_type, } => { - check_decoded_field_names(physical_type)?; - return Ok(Transformed::yes(reject_on_non_empty_expr( + let rejected = reject_on_non_empty_expr( child, cast.target_field(), &column, &leaf_physical_type, &leaf_target_type, - ))); + ); + return Ok(Transformed::yes(checked_decoded_expr( + physical_type, + rejected, + )?)); } } @@ -1381,8 +1408,6 @@ impl SparkPhysicalExprAdapter { return Ok(Transformed::no(expr)); } - check_decoded_field_names(physical_type)?; - // Complex casts (including changes in list representation), timestamp tz relabel // (e.g. Timestamp(us, None) -> Timestamp(us, Some("UTC")) for INT96 reads), and // Timestamp -> Int64 @@ -1411,12 +1436,13 @@ impl SparkPhysicalExprAdapter { let comet_cast: Arc = Arc::new( CometCastColumnExpr::try_new( child, - input_field, + Arc::clone(&input_field), Arc::clone(cast.target_field()), None, )? .with_parquet_options(self.parquet_options.clone()), ); + let comet_cast = checked_decoded_expr(physical_type, comet_cast)?; return Ok(Transformed::yes(comet_cast)); } @@ -1437,7 +1463,10 @@ impl SparkPhysicalExprAdapter { None, )); - return Ok(Transformed::yes(spark_cast as Arc)); + return Ok(Transformed::yes(checked_decoded_expr( + physical_type, + spark_cast as Arc, + )?)); } Ok(Transformed::no(expr)) @@ -3568,6 +3597,57 @@ mod test { .has_valid_extension_type::()); } + #[test] + fn variant_with_duplicate_physical_children_is_rejected() { + let physical_type = DataType::Struct(Fields::from(vec![ + Field::new("value", DataType::Binary, false), + Field::new("value", DataType::Binary, false), + Field::new("metadata", DataType::Binary, false), + ])); + let canonical_type = DataType::Struct(Fields::from(vec![ + Field::new("value", DataType::Binary, false), + Field::new("metadata", DataType::Binary, false), + ])); + for logical_type in [physical_type.clone(), canonical_type] { + let identical = logical_type == physical_type; + let logical = Arc::new(Schema::new(vec![ + Field::new("v", logical_type, true).with_extension_type(VariantType) + ])); + let physical = Arc::new(Schema::new(vec![Field::new( + "v", + physical_type.clone(), + true, + ) + .with_extension_type(VariantType)])); + let default = super::DefaultPhysicalExprAdapterFactory + .create(Arc::clone(&logical), Arc::clone(&physical)) + .unwrap() + .rewrite(Arc::new(Column::new("v", 0))) + .unwrap(); + if identical { + assert!(default.downcast_ref::().is_some()); + } else { + assert!(default + .downcast_ref::() + .is_some()); + } + let adapter = SparkPhysicalExprAdapterFactory::new( + SparkParquetOptions::new(EvalMode::Legacy, "UTC", false), + None, + ) + .create(logical, physical) + .unwrap(); + let error = adapter + .rewrite(Arc::new(Column::new("v", 0))) + .expect_err("variant decoding must reject duplicate physical children") + .to_string(); + assert!( + error.contains("duplicate Parquet field name 'value'"), + "{error}" + ); + } + } + #[test] fn variant_field_id_wins_over_a_shadowing_name() { let storage = DataType::Struct(Fields::from(vec![ 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 3d1fe07773a..61fdd35bca5 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -144,8 +144,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper Seq("true", "false").foreach { mergeSchema => Seq(true, false).foreach { caseSensitive => withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) { - // mergeSchema only affects inference, and Spark already rejects inference for a - // duplicate-bearing file, so an explicit schema reads the unique sibling natively. + // mergeSchema only affects inference; an explicit schema reads the unique sibling. val unique = spark.read .option("mergeSchema", mergeSchema) .schema("s struct") @@ -170,7 +169,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } withSQLConf(CometConf.COMET_ENABLED.key -> "false") { val inferred = intercept[org.apache.spark.sql.AnalysisException] { - spark.read.option("mergeSchema", "true").parquet(paths: _*).collect() + spark.read.parquet(duplicatePath.toString).schema } assert(inferred.getMessage.contains("COLUMN_ALREADY_EXISTS"), inferred.getMessage) }