diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index d5ab79b4a9..0dc88124af 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -62,6 +62,15 @@ 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 + 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 d89a177283..c7de358144 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -44,9 +44,13 @@ //! 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 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; @@ -77,7 +81,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}; @@ -163,6 +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, + projection: Option<(SchemaRef, SparkParquetOptions)>, } impl EagerPageIndexReaderFactory { @@ -191,9 +197,20 @@ 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 @@ -225,6 +242,7 @@ 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(), })) } } @@ -240,6 +258,7 @@ 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. @@ -372,6 +391,80 @@ 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 @@ -439,6 +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 projection = self.projection.clone(); async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -498,6 +592,71 @@ 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 { @@ -841,6 +1000,198 @@ 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 60a1027f97..d108c90e07 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -176,7 +176,8 @@ pub(crate) fn init_datasource_exec( // `store_sales`), the page index is re-fetched, uncached, on every open (comet#3978). // `EagerPageIndexReaderFactory` forces the page index to load on the first fetch and be // cached with the footer, at the cost of losing the skip's benefit when it would have - // applied. Filed upstream as apache/datafusion#23978; revert this once that's fixed. + // applied. Filed upstream as apache/datafusion#23978; when replacing this factory, preserve + // its duplicate-field validation (#5783). // // Preserve bytes_scanned's existing requested data/Bloom-filter range accounting. Footer // and page-index reads through get_metadata bypass it, and coalescing may fetch extra bytes. @@ -193,7 +194,8 @@ pub(crate) fn init_datasource_exec( scan_io_source, parquet_source.metrics(), ) - .with_spark_variant_schema(projects_variant), + .with_spark_variant_schema(projects_variant) + .with_required_schema(&required_schema, &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 5a4b3e1630..0c9b865ba0 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, 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 25c0e93002..8babf51432 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,154 @@ 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) => + 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) + } + } + } + } + + 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 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 { _ => + 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") { + 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) @@ -249,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( """