This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5786-244bb6ce73efaf28f72c8e2c71a07215554ee836 in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 50c42dda85943b5b2d5d63676aa30f4ff22727f4 Author: Erik Bogado <[email protected]> AuthorDate: Thu Sep 24 01:20:01 2026 +0000 fix: reject duplicate Parquet field names before decoding (#5786) * fix: reject duplicate Parquet field names before decoding * docs: remove issue reference from duplicate-field limitation * fix: scope duplicate checks to projected roots * test: cover pruned duplicate Parquet fields * 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. * fix: scope duplicate Parquet field checks to referenced columns * 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. * 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 --------- Co-authored-by: Andy Grove <[email protected]> --- .../user-guide/latest/compatibility/scans.md | 14 + .../core/src/execution/operators/iceberg_scan.rs | 55 +++ native/core/src/parquet/parquet_exec.rs | 2 +- native/core/src/parquet/parquet_support.rs | 21 +- native/core/src/parquet/schema_adapter.rs | 533 ++++++++++++++++++--- .../apache/comet/exec/CometNativeReaderSuite.scala | 328 +++++++++++++ 6 files changed, 885 insertions(+), 68 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index 36d245992e..d479e75ffd 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -62,6 +62,20 @@ 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 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. 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/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index fdcd6c8ba5..9f465329ca 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -619,6 +619,61 @@ 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 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, + 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::<Int64Array>() + .unwrap() + .value(0), + 3 + ); + } + } + } + fn task_with_deletes(deletes: Vec<FileScanTaskDeleteFile>) -> FileScanTask { FileScanTask::builder() .with_file_size_in_bytes(0) diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 5d70f2aaa7..93ac29e824 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -177,7 +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; revert this once that's fixed. + // 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. diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index d6ac9bed63..521f95d459 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. @@ -420,8 +424,8 @@ fn field_id(field: &arrow::datatypes::Field) -> Option<i32> { /// 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 +477,11 @@ 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(duplicate_parquet_field_error(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 e06408b44e..dc6a779958 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,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 / @@ -810,6 +812,42 @@ 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(duplicate_parquet_field_error(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(()) +} + +/// Check an expression before it can decode its entire physical subtree. +fn checked_decoded_expr( + physical_type: &DataType, + expr: Arc<dyn PhysicalExpr>, +) -> DataFusionResult<Arc<dyn PhysicalExpr>> { + 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 @@ -856,54 +894,62 @@ 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<String, Vec<usize>> = 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) + }; // 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<String, Vec<usize>> = 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 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 +963,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<i32, String> = 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 +1006,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 +1038,19 @@ struct SparkPhysicalExprAdapter { /// physical names so that downstream reassign_expr_columns can find /// columns in the actual stream schema. logical_to_physical_names: Option<HashMap<String, String>>, - /// 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<String, Vec<usize>>)>, /// 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<HashSet<String>>, + /// 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<String, String>, /// `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 +1062,7 @@ struct SparkPhysicalExprAdapter { impl PhysicalExprAdapter for SparkPhysicalExprAdapter { fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> DataFusionResult<Arc<dyn PhysicalExpr>> { - // 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 +1075,11 @@ 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(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 // (mirrors Spark's `matchIdField`, which never raises the duplicate-field error). @@ -1010,6 +1090,9 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter { { continue; } + if self.parquet_options.case_sensitive && folded_to_indices.contains_key(folded) { + return Err(duplicate_parquet_field_error(name)); + } if let Some(err) = check_column_duplicate(name, folded, folded_to_indices, orig_physical) { @@ -1091,7 +1174,7 @@ impl SparkPhysicalExprAdapter { return Ok(expr); }; - Ok(Arc::new( + let cast: Arc<dyn PhysicalExpr> = Arc::new( CometCastColumnExpr::try_new( expr, Arc::clone(physical_field), @@ -1099,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. @@ -1168,13 +1252,17 @@ impl SparkPhysicalExprAdapter { physical_type: leaf_physical_type, target_type: leaf_target_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, + )?)); } } @@ -1187,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)); } @@ -1228,12 +1319,13 @@ impl SparkPhysicalExprAdapter { let comet_cast: Arc<dyn PhysicalExpr> = 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)); } @@ -1269,13 +1361,17 @@ impl SparkPhysicalExprAdapter { physical_type: leaf_physical_type, target_type: leaf_target_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, + )?)); } } @@ -1340,12 +1436,13 @@ impl SparkPhysicalExprAdapter { let comet_cast: Arc<dyn PhysicalExpr> = 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)); } @@ -1366,7 +1463,10 @@ impl SparkPhysicalExprAdapter { None, )); - return Ok(Transformed::yes(spark_cast as Arc<dyn PhysicalExpr>)); + return Ok(Transformed::yes(checked_decoded_expr( + physical_type, + spark_cast as Arc<dyn PhysicalExpr>, + )?)); } Ok(Transformed::no(expr)) @@ -2993,6 +3093,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::<Column>().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<dyn PhysicalExpr> = 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::<super::RejectOnNonEmpty>().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 @@ -3229,6 +3597,57 @@ mod test { .has_valid_extension_type::<VariantType>()); } + #[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::<Column>().is_some()); + } else { + assert!(default + .downcast_ref::<datafusion::physical_expr::expressions::CastExpr>() + .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 93567756d7..61fdd35bca 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,235 @@ 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") { + spark.range(3).coalesce(1).write.parquet(path.toString) + Seq( + "s struct<dup: bigint, dup: bigint>", + "s array<struct<dup: bigint, dup: bigint>>", + "s map<string, struct<dup: bigint, dup: bigint>>", + "s struct<t: struct<dup: bigint, dup: bigint>>").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<dup: bigint>"), + ( + "three children", + "named_struct('dup', id, 'dup', id + 100, 'dup', id + 200)", + "struct<dup: bigint>"), + ( + "distinct sibling", + "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)", + "struct<dup: bigint, other: bigint>"), + ( + "array element", + "array(named_struct('dup', id, 'dup', id + 100))", + "array<struct<dup: bigint>>"), + ( + "map value", + "map('key', named_struct('dup', id, 'dup', id + 100))", + "map<string, struct<dup: bigint>>")).foreach { case (shape, expression, readType) => + Seq(1, 4096).foreach { batchSize => + test(s"duplicate Parquet field names fail 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 = 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; an explicit schema reads the unique sibling. + val unique = spark.read + .option("mergeSchema", mergeSchema) + .schema("s struct<other: bigint>") + .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<dup: bigint>") + .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.parquet(duplicatePath.toString).schema + } + assert(inferred.getMessage.contains("COLUMN_ALREADY_EXISTS"), inferred.getMessage) + } + } + } + } + + 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, s struct<dup: bigint>") + .parquet(path.toString) + .select(name) + 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 = 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) + (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", + 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) + (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 = causeMessages(error) + assert(messages.contains("duplicate Parquet field name 'a'"), messages) + } + } + } + } + } + + test( + "duplicate Parquet field names - distinct siblings and repeated names in separate groups") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(3) + .selectExpr( + "named_struct('dup', id, 'Dup', id + 100) as s", + "named_struct('dup', id + 200) as t") + .write + .parquet(path.toString) + } + def read = spark.read + .schema("s struct<dup: bigint, Dup: bigint>, t struct<dup: bigint>") + .parquet(path.toString) + assert( + find(read.queryExecution.executedPlan)(_.isInstanceOf[CometNativeScanExec]).isDefined) + checkSparkAnswer(read) + } + } + } + test("native reader case sensitivity") { withTempPath { path => spark.range(10).toDF("a").write.parquet(path.toString) @@ -249,6 +478,105 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + test("duplicate Parquet field names - exact-name projection works in both resolver modes") { + 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 df = spark.read.schema("s struct<other: 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)))) + } + } + } + } + + Seq( + ( + "map", + "map('k', named_struct('dup', id, 'dup', id + 100, 'other', id + 900))", + "s map<string, struct<other: bigint>>"), + ( + "case", + "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)", + "S struct<OTHER: bigint>")).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 = causeMessages(error) + assert(messages.contains("duplicate Parquet field name 'dup'"), messages) + } + } + } + } + + Seq( + ( + "struct", + "named_struct('dup', id, 'dup', id + 100, 'other', id + 900)", + "struct<other: bigint>", + "s.other", + (n: Long) => Row(Row(n))), + ( + "deeper", + "named_struct('t', named_struct('dup', id, 'dup', id + 100, 'other', id + 900))", + "struct<t: struct<other: bigint>>", + "s.t.other", + (n: Long) => Row(Row(Row(n)))), + ( + "array", + "array(named_struct('dup', id, 'dup', id + 100, 'other', id + 900))", + "array<struct<other: bigint>>", + "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( """ --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
