From d29794c71503a4e5a745cb62a7d8f4ad47af39d4 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sat, 24 Oct 2026 16:08:02 +0800 Subject: [PATCH 1/7] fix: resolve conflicting Parquet field claims consistently --- native/core/src/parquet/field_resolver.rs | 271 ++++++++++++- native/core/src/parquet/parquet_support.rs | 51 ++- native/core/src/parquet/schema_adapter.rs | 233 +++++++---- .../comet/exec/CometNativeReaderSuite.scala | 368 ++++++++++++++++++ 4 files changed, 836 insertions(+), 87 deletions(-) diff --git a/native/core/src/parquet/field_resolver.rs b/native/core/src/parquet/field_resolver.rs index 060fe5c8286..c80e5501ca0 100644 --- a/native/core/src/parquet/field_resolver.rs +++ b/native/core/src/parquet/field_resolver.rs @@ -47,6 +47,43 @@ pub(crate) enum Selection<'a> { ByName(&'a [usize]), } +/// The strongest claim on one file field. Root remapping and nested conversion give a file +/// field to this claimant only. Repeated id claims keep the last requested field, matching +/// Spark parquet-mr's `ParquetRowConverter` id-to-index `toMap`. Equal name claims keep the +/// earliest requested field, retaining the root remap's name-matching behavior. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum FieldClaim { + /// The requested field selects this file field by Parquet field id. + IdMatch(usize), + /// An id-bearing requested field has this file field's folded name. It must not reach this + /// field by name when its id is missing or selects a different field. Only an id match can + /// override this claim. The root hides it with a placeholder; nested fields read NULL. + Shadowed, + /// The requested field selects this file field by folded name. + NameMatch(usize), + /// No requested field claims this file field. The root keeps its original name. + Unchanged, +} + +impl FieldClaim { + fn strength(self) -> u8 { + match self { + Self::IdMatch(_) => 3, + Self::Shadowed => 2, + Self::NameMatch(_) => 1, + Self::Unchanged => 0, + } + } + + fn claim(&mut self, claim: Self) { + if claim.strength() > self.strength() + || matches!((*self, claim), (Self::IdMatch(_), Self::IdMatch(_))) + { + *self = claim; + } + } +} + /// Resolves the requested fields of one group against the file fields of the same group, as /// Spark's `clipParquetGroupFields` does. The caller folds both sides' names once (see /// `name_fold`) and passes them in, parallel to the fields. @@ -204,6 +241,55 @@ impl<'a> FieldResolver<'a> { )), } } + + /// Resolve claims in file order, with id matches ahead of shadows and name matches. + /// Repeated requested ids keep the last claimant, as Spark parquet-mr does. Duplicate file ids are + /// checked across the whole requested group, including unprojected root fields (#7006). + /// Name ambiguity is left to the caller: the root checks referenced columns only, whereas + /// nested conversion validates every requested field in [`Self::resolve_all`]. + pub(crate) fn file_claims(&self) -> DataFusionResult> { + let mut claims = vec![FieldClaim::Unchanged; self.file.len()]; + for requested in 0..self.requested.len() { + match self.select(requested)? { + Selection::ById(matched) => { + if let Some(file) = matched { + claims[file].claim(FieldClaim::IdMatch(requested)); + } + for &file in self.name_matches(requested) { + claims[file].claim(FieldClaim::Shadowed); + } + } + Selection::ByName(matches) => { + for &file in matches { + claims[file].claim(FieldClaim::NameMatch(requested)); + } + } + } + } + Ok(claims) + } + + /// Resolve a nested group in requested order, giving each file field to its strongest + /// claimant only. Missing and losing fields read NULL. Validate independent selections + /// first, in requested order, so arbitration cannot hide ambiguity or change which of + /// several invalid selections reports its error first. + pub(crate) fn resolve_all(&self) -> DataFusionResult>> { + let mut resolved = (0..self.requested.len()) + .map(|requested| self.resolve(requested)) + .collect::>>()?; + let claims = self.file_claims()?; + for (requested, file) in resolved.iter_mut().enumerate() { + if let Some(index) = *file { + if !matches!( + claims[index], + FieldClaim::IdMatch(owner) | FieldClaim::NameMatch(owner) if owner == requested + ) { + *file = None; + } + } + } + Ok(resolved) + } } /// The error for requested field `name` when the file fields at `matches` all share its folded @@ -260,7 +346,7 @@ impl Indices { #[cfg(test)] mod test { - use super::{FieldResolver, Selection}; + use super::{FieldClaim, FieldResolver, Selection}; use crate::parquet::name_fold::fold_names; use arrow::datatypes::{DataType, Field, FieldRef}; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; @@ -349,4 +435,187 @@ mod test { assert_eq!(resolver.select(0).unwrap(), Selection::ByName(&[1])); }); } + + #[test] + fn conflicting_claims_prefer_id_in_either_requested_order() { + for file_name in ["A", "a"] { + for case_sensitive in [false, true] { + for use_field_id in [false, true] { + for reverse in [false, true] { + let mut requested = [field("x", Some(1)), field("a", None)]; + if reverse { + requested.reverse(); + } + let file = [field(file_name, Some(1))]; + with_resolver( + &requested, + &file, + case_sensitive, + use_field_id, + |resolver| { + let mut expected = if use_field_id { + vec![Some(0), None] + } else { + vec![None, (!case_sensitive || file_name == "a").then_some(0)] + }; + if reverse { + expected.reverse(); + } + assert_eq!(resolver.resolve_all().unwrap(), expected); + }, + ); + } + } + } + } + } + + #[test] + fn group_claims_include_shadowing_and_unclaimed_fields() { + let requested = [ + field("x", Some(1)), + field("a", None), + field("b", Some(2)), + field("B", None), + field("C", None), + ]; + let file = [ + field("A", Some(1)), + field("B", None), + field("c", None), + field("d", None), + ]; + with_resolver(&requested, &file, false, true, |resolver| { + assert_eq!( + resolver.file_claims().unwrap(), + vec![ + FieldClaim::IdMatch(0), + FieldClaim::Shadowed, + FieldClaim::NameMatch(4), + FieldClaim::Unchanged, + ] + ); + assert_eq!( + resolver.resolve_all().unwrap(), + vec![Some(0), None, None, None, Some(2)] + ); + }); + } + + #[test] + fn shadowing_respects_case_sensitivity_and_a_stronger_id_claim() { + // A's id resolves elsewhere, but its name still shadows a name-only claimant of `a`. + // An independent id claimant of `a` can override that shadow in either order. + for reverse in [false, true] { + for case_sensitive in [false, true] { + for claim_by_id in [false, true] { + let mut requested = [ + field("A", Some(1)), + field("a", None), + field("z", claim_by_id.then_some(2)), + ]; + let mut expected = vec![ + Some(1), + (case_sensitive && !claim_by_id).then_some(0), + claim_by_id.then_some(0), + ]; + if reverse { + requested.reverse(); + expected.reverse(); + } + let file = [field("a", Some(2)), field("b", Some(1))]; + with_resolver(&requested, &file, case_sensitive, true, |resolver| { + assert_eq!(resolver.resolve_all().unwrap(), expected); + }); + } + } + } + } + + #[test] + fn id_ties_keep_last_and_name_ties_keep_first() { + for use_field_id in [false, true] { + for reverse in [false, true] { + let mut requested = [ + field("x", Some(1)), + field("y", Some(1)), + field("Ab", None), + field("aB", None), + ]; + if reverse { + requested.reverse(); + } + let file = [field("p", Some(1)), field("AB", None)]; + with_resolver(&requested, &file, false, use_field_id, |resolver| { + let expected = match (reverse, use_field_id) { + (false, true) => vec![None, Some(0), Some(1), None], + (false, false) => vec![None, None, Some(1), None], + (true, true) => vec![Some(1), None, None, Some(0)], + (true, false) => vec![Some(1), None, None, None], + }; + assert_eq!(resolver.resolve_all().unwrap(), expected); + }); + } + } + } + + #[test] + fn three_or_more_id_claims_keep_last_over_name_and_shadow() { + for count in [3, 4] { + for reverse in [false, true] { + let mut requested: Vec<_> = ["x", "y", "A", "z"][..count] + .iter() + .map(|name| field(name, Some(1))) + .collect(); + requested.insert(1, field("a", None)); + requested.push(field("b", None)); + let mut expected = vec![None; requested.len()]; + expected[count] = Some(0); + expected[count + 1] = Some(1); + if reverse { + requested.reverse(); + // Reversal changes the winning id claimant to x, now at the last position. + expected.fill(None); + expected[0] = Some(1); + expected[count + 1] = Some(0); + } + let file = [field("a", Some(1)), field("b", None)]; + with_resolver(&requested, &file, false, true, |resolver| { + assert_eq!(resolver.resolve_all().unwrap(), expected); + }); + } + } + } + + #[test] + fn arbitration_does_not_hide_ambiguous_name_or_reorder_errors() { + let file = [ + field("a", Some(1)), + field("A", Some(2)), + field("b", Some(1)), + ]; + for reverse in [false, true] { + let mut requested = [field("a", None), field("x", Some(1))]; + if reverse { + requested.reverse(); + } + with_resolver(&requested, &file, false, true, |resolver| { + let first_error = resolver.resolve(0).unwrap_err().to_string(); + assert_eq!(resolver.resolve_all().unwrap_err().to_string(), first_error); + }); + } + // Both physical fields are owned by id claimants, but the independent name selection + // is still ambiguous. Root claims allow it so the root can defer its name check. + let requested = [field("x", Some(1)), field("y", Some(2)), field("a", None)]; + with_resolver(&requested, &file[..2], false, true, |resolver| { + assert_eq!( + resolver.file_claims().unwrap(), + vec![FieldClaim::IdMatch(0), FieldClaim::IdMatch(1)] + ); + assert_eq!( + resolver.resolve_all().unwrap_err().to_string(), + resolver.resolve(2).unwrap_err().to_string() + ); + }); + } } diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index d38f0ef701d..26f2e7508b0 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -432,10 +432,11 @@ pub(crate) fn field_id(field: &arrow::datatypes::Field) -> Option { /// Resolve each requested (`to`) struct field to the index of the file (`from`) field it reads /// from, or `None` when the file holds no such field, with Spark's `clipParquetGroupFields` rules -/// as [`FieldResolver::resolve`] applies them: an id-bearing requested field matches only by id +/// as [`FieldResolver::resolve_all`] applies them: an id-bearing requested field matches only by id /// when `use_field_id` is set, and every other field matches by name, folded with the same /// `toLowerCase(Locale.ROOT)` fold the top-level schema adapter uses when `case_sensitive` is -/// false. A requested field that selects more than one file field is rejected. +/// false. A requested field that selects more than one file field is rejected. Each file field +/// belongs to its strongest claimant only; a losing requested field reads NULL. /// /// 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. @@ -461,9 +462,7 @@ pub(crate) fn match_struct_fields( parquet_options.case_sensitive, parquet_options.use_field_id, ); - (0..to_fields.len()) - .map(|to_pos| resolver.resolve(to_pos)) - .collect() + resolver.resolve_all() } /// Cast between struct types based on logic in @@ -2533,4 +2532,46 @@ mod tests { .unwrap(); assert_eq!(col.value(0), 44); } + + #[test] + fn shadowed_struct_fields_follow_all_missing_null_policy() { + use super::{parquet_convert_array, SparkParquetOptions}; + use arrow::array::{Array, ArrayRef, Int32Array, StructArray}; + use arrow::buffer::NullBuffer; + use arrow::datatypes::{DataType, Field, Fields}; + use datafusion_comet_spark_expr::EvalMode; + use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use std::collections::HashMap; + use std::sync::Arc; + + let parent_nulls = NullBuffer::from(vec![true, false, true]); + let from: ArrayRef = Arc::new(StructArray::new( + Fields::from(vec![Field::new("b", DataType::Int32, true)]), + vec![Arc::new(Int32Array::from(vec![1, 2, 3]))], + Some(parent_nulls.clone()), + )); + let to_type = DataType::Struct(Fields::from(vec![ + Field::new("b", DataType::Int64, true).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "99".to_string(), + )])), + Field::new("B", DataType::Int32, true), + ])); + for null_struct in [false, true] { + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); + options.use_field_id = true; + options.return_null_struct_if_all_fields_missing = null_struct; + let converted = parquet_convert_array(Arc::clone(&from), &to_type, &options).unwrap(); + let converted = converted.as_any().downcast_ref::().unwrap(); + assert!(converted + .columns() + .iter() + .all(|column| column.null_count() == 3)); + if null_struct { + assert_eq!(converted.null_count(), 3); + } else { + assert_eq!(converted.nulls(), Some(&parent_nulls)); + } + } + } } diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 9a256f33a63..34f274ed5f9 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -20,7 +20,7 @@ use crate::parquet::datetime_rebase::{ resolve_file_rebase_policies, wrap_datetime_rebase, FileRebasePolicies, RebaseReadMode, SessionRebaseModes, }; -use crate::parquet::field_resolver::{duplicate_name_error, FieldResolver, Selection}; +use crate::parquet::field_resolver::{duplicate_name_error, FieldClaim, FieldResolver}; use crate::parquet::name_fold::{fold_name, fold_names, fold_schema_names}; use crate::parquet::parquet_support::{ duplicate_parquet_field_error, field_id, match_struct_fields, spark_parquet_convert, @@ -123,6 +123,7 @@ fn is_pure_structural_narrowing( .iter() .map(|f| fold_name(f.name(), parquet_options.case_sensitive)) .collect::>()?; + let mut claimed_names = HashSet::with_capacity(target_fields.len()); for target_field in target_fields { // DataFusion's retained `CastExpr` resolves struct fields by *exact* name, so // keeping it is only sound when Spark's configured resolver would pick the same @@ -138,6 +139,11 @@ fn is_pure_structural_narrowing( // Spark and Comet's converter reject it, but DataFusion's cast would silently // return the exact-case field, so the cast must not be retained. let folded_target = fold_name(target_field.name(), parquet_options.case_sensitive)?; + // The generic cast can give one source to several targets. Comet keeps only + // the earliest equal-strength name claim, so that shape needs its converter. + if !claimed_names.insert(folded_target.clone()) { + return Ok(false); + } let resolver_matches = source_folded .iter() .filter(|&f| f == &folded_target) @@ -183,77 +189,6 @@ fn is_pure_structural_narrowing( }) } -/// What the root remap does with one file column, strongest claim first. Spark's -/// `clipParquetGroupFields` resolves every requested field on its own, but a rename gives a file -/// column one name. A column that more than one requested field claims therefore takes the -/// strongest claim, and between equal claims the earliest requested field. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum RootResolution { - /// Requested field `.0` selects the column by Parquet field id (Spark's `matchIdField`). - /// The column takes that field's name. - IdMatch(usize), - /// The column has the folded name of an id-bearing requested field that did not select it. - /// `matchIdField` never compares names, so the column takes a placeholder name (Spark's - /// `generateFakeColumnName`), which keeps DataFusion's name lookup from reaching it. - Shadowed, - /// Requested field `.0`, which carries no id or reads with id matching off, selects the - /// column by folded name. The column takes that field's name, which differs from the file - /// name at most in case. - NameMatch(usize), - /// No requested field claims the column. It keeps its file name. - Unchanged, -} - -impl RootResolution { - fn strength(self) -> u8 { - match self { - RootResolution::IdMatch(_) => 3, - RootResolution::Shadowed => 2, - RootResolution::NameMatch(_) => 1, - RootResolution::Unchanged => 0, - } - } - - /// Keep `claim` if it is stronger than the current resolution. Requested fields claim in - /// order, so the earliest one wins a tie. - fn claim(&mut self, claim: RootResolution) { - if claim.strength() > self.strength() { - *self = claim; - } - } -} - -/// Resolve every root file column against the requested root fields. Raises Spark's -/// `_LEGACY_ERROR_TEMP_2094` "Found duplicate field(s)" when an id-bearing logical field would -/// resolve to more than one physical field. That check covers every logical field, which is the -/// whole data schema rather than the projection, so it also fails a read that never references -/// the field, where Spark reads the file (#7006). An ambiguous name fails only for a referenced -/// column (see `check_duplicate_columns`). -fn resolve_root_fields(resolver: &FieldResolver) -> DataFusionResult> { - let mut resolutions = vec![RootResolution::Unchanged; resolver.file_fields().len()]; - for requested in 0..resolver.requested_fields().len() { - match resolver.select(requested)? { - Selection::ById(matched) => { - if let Some(file) = matched { - resolutions[file].claim(RootResolution::IdMatch(requested)); - } - // All ID-bearing targets resolve by ID, even when a different file column has - // the requested name. Hide that shadowing column after giving its own ID match - // precedence, whether the target's ID was missing or resolved elsewhere. - for &file in resolver.name_matches(requested) { - resolutions[file].claim(RootResolution::Shadowed); - } - } - Selection::ByName(matches) => { - for &file in matches { - resolutions[file].claim(RootResolution::NameMatch(requested)); - } - } - } - } - Ok(resolutions) -} - /// `field` under `name`, keeping everything else the file says about it. fn renamed_field(field: &FieldRef, name: impl Into) -> FieldRef { Arc::new(field.as_ref().clone().with_name(name)) @@ -285,14 +220,14 @@ fn rename_to_logical( fn remap_physical_schema( resolver: &FieldResolver, ) -> DataFusionResult<(SchemaRef, HashMap)> { - let resolutions = resolve_root_fields(resolver)?; + let resolutions = resolver.file_claims()?; // Placeholder names for shadowed file columns, Spark's `generateFakeColumnName`. A // placeholder skips every folded logical and physical name, the names downstream lookups // compare, so no requested column can reach a shadowed column through its placeholder, even // one whose name differs from the placeholder only in case. Most reads shadow nothing, so // the reserved names are gathered only when one does. - let reserved: HashSet<&str> = if resolutions.contains(&RootResolution::Shadowed) { + let reserved: HashSet<&str> = if resolutions.contains(&FieldClaim::Shadowed) { resolver .requested_folded() .iter() @@ -317,12 +252,12 @@ fn remap_physical_schema( .iter() .zip(resolutions) .map(|(field, resolution)| match resolution { - RootResolution::IdMatch(logical) | RootResolution::NameMatch(logical) => { + FieldClaim::IdMatch(logical) | FieldClaim::NameMatch(logical) => { let logical_name = resolver.requested_fields()[logical].name(); rename_to_logical(field, logical_name, &mut name_map) } - RootResolution::Shadowed => renamed_field(field, placeholder()), - RootResolution::Unchanged => Arc::clone(field), + FieldClaim::Shadowed => renamed_field(field, placeholder()), + FieldClaim::Unchanged => Arc::clone(field), }) .collect(); @@ -3100,6 +3035,134 @@ pub(crate) mod test { Ok(()) } + /// The losing name claimant reads NULL even when converting that physical column to its + /// requested type would fail. This exercises both the file-open check and runtime adapter. + #[tokio::test] + async fn conflicting_field_claims_read_null_at_every_depth() -> Result<(), DataFusionError> { + use arrow::buffer::NullBuffer; + + for depth in 0..=2 { + for reverse in [false, true] { + for losing_type in [DataType::Int64, DataType::Int32] { + let values: ArrayRef = + Arc::new(Int64Array::from(vec![Some(11), Some(22), None, Some(44)])); + let mut batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new("A", DataType::Int64, true).with_metadata(id_meta("1")) + ])), + vec![Arc::clone(&values)], + )?; + let mut fields = vec![ + Field::new("x", DataType::Int64, true).with_metadata(id_meta("1")), + Field::new("a", losing_type.clone(), true), + ]; + let mut columns = vec![values, new_null_array(&losing_type, 4)]; + if reverse { + fields.reverse(); + columns.reverse(); + } + let mut expected = + RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?; + for level in 0..depth { + let wrap = |batch: RecordBatch| -> Result { + let array = StructArray::try_new( + batch.schema().fields().clone(), + batch.columns().to_vec(), + Some(NullBuffer::from(vec![true, false, true, true])), + )?; + Ok(RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + format!("s{level}"), + array.data_type().clone(), + true, + )])), + vec![Arc::new(array)], + )?) + }; + batch = wrap(batch)?; + expected = wrap(expected)?; + } + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); + options.use_field_id = true; + let mut stream = scan_parquet(&batch, expected.schema(), options)?; + let read = stream.next().await.unwrap()?; + assert_eq!( + read, expected, + "depth={depth}, reverse={reverse}, losing_type={losing_type}" + ); + assert!(stream.next().await.is_none()); + } + } + } + Ok(()) + } + + /// Spark parquet-mr assigns repeated requested ids to the last field. The first field's + /// incompatible type must not reject a read after that field loses its claim. + #[tokio::test] + async fn repeated_requested_ids_read_last_field_at_root_and_nested( + ) -> Result<(), DataFusionError> { + for count in [2, 3, 4] { + for nested in [false, true] { + for reverse in [false, true] { + let values = Arc::new(Int64Array::from(vec![Some(11), Some(22), None])); + let physical = + Field::new("A", DataType::Int64, true).with_metadata(id_meta("1")); + let mut names = vec!["x", "y", "z", "w"]; + names.truncate(count); + if reverse { + names.reverse(); + } + let requested: Vec<_> = names + .iter() + .enumerate() + .map(|(index, name)| { + let data_type = if index + 1 == count { + DataType::Int64 + } else { + DataType::Int32 + }; + Field::new(*name, data_type, true).with_metadata(id_meta("1")) + }) + .collect(); + let (batch, required) = if nested { + ( + struct_batch(physical, values.clone())?, + struct_schema(requested), + ) + } else { + ( + RecordBatch::try_new( + Arc::new(Schema::new(vec![physical])), + vec![values.clone()], + )?, + Arc::new(Schema::new(requested)), + ) + }; + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); + options.use_field_id = true; + let mut stream = scan_parquet(&batch, required, options)?; + let read = stream.next().await.unwrap()?; + let columns = if nested { + read.column(0).as_struct().columns() + } else { + read.columns() + }; + for column in &columns[..count - 1] { + assert_eq!(column.data_type(), &DataType::Int32); + assert_eq!(column.null_count(), values.len()); + } + assert_eq!( + columns[count - 1].as_primitive::(), + values.as_ref() + ); + assert!(stream.next().await.is_none()); + } + } + } + Ok(()) + } + /// Disallowed widening (`INT32 -> bigint` with `allow_type_promotion` off) inside a /// struct defers to `RejectOnNonEmpty`, like the top level: a non-empty file fails ... #[tokio::test] @@ -4084,10 +4147,10 @@ pub(crate) mod test { ); } - /// Between equal claims the earliest requested field wins, for an id match and for a name - /// match alike. + /// Repeated requested ids keep the last claimant, matching Spark parquet-mr's + /// ParquetRowConverter. Equal name claims retain the earliest claimant. #[test] - fn root_resolution_earliest_requested_field_wins_a_tie() { + fn root_resolution_id_ties_keep_last_and_name_ties_keep_first() { let logical = Arc::new(Schema::new(vec![ Field::new("x", DataType::Int32, true).with_metadata(id_meta("1")), Field::new("y", DataType::Int32, true).with_metadata(id_meta("1")), @@ -4099,7 +4162,7 @@ pub(crate) mod test { Field::new("AB", DataType::Int32, true), ])); let (remapped, _name_map) = remap(&logical, &physical, false, true); - assert_eq!(field_names(&remapped), ["x", "Ab"]); + assert_eq!(field_names(&remapped), ["y", "Ab"]); } /// Field-id precedence in the case-insensitive duplicate check: an explicit `ω` (id 2) @@ -4570,6 +4633,14 @@ pub(crate) mod test { assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } + #[test] + fn is_pure_structural_narrowing_denies_conflicting_name_claims() { + let opts = default_options(); + let physical = struct_type(vec![("a", DataType::Int32), ("b", DataType::Int32)]); + let target = struct_type(vec![("a", DataType::Int32), ("a", DataType::Int32)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); + } + /// NTZ -> LTZ timestamp relabeling (INT96 reads) is a metadata-only reinterpretation /// Comet applies via a raw `reinterpret_cast`; Arrow's generic timestamp cast performs an /// actual timezone conversion instead, changing the value. Locked in as its own test 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 d5dff80cb2f..8a373dff900 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -31,6 +31,7 @@ import org.apache.parquet.io.api.RecordConsumer import org.apache.parquet.schema.MessageTypeParser import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.comet.CometNativeScanExec +import org.apache.spark.sql.execution.FileSourceScanExec import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions.{array, col} import org.apache.spark.sql.internal.SQLConf @@ -1269,6 +1270,373 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper nullable = true, new MetadataBuilder().putLong("parquet.field.id", id.toLong).build()) + private def withFieldClaimFile(nested: Boolean)(f: (String, StructType) => Unit): Unit = { + val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() + def wrapSchema(fields: StructType): StructType = + if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) + else fields + val writeSchema = wrapSchema(StructType(Seq(withFieldId("A", 1)))) + val readSchema = wrapSchema( + StructType(Seq(withFieldId("x", 1), StructField("a", LongType, nullable = true)))) + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "false", + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = (1L to 3L).map(value => if (nested) Row(Row(value)) else Row(value)) + spark + .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) + .write + .parquet(path.getCanonicalPath) + } + f(path.getCanonicalPath, readSchema) + } + } + } + + private def fieldClaimReference(query: => DataFrame): Seq[Row] = { + var rows = Seq.empty[Row] + withSQLConf( + CometConf.COMET_ENABLED.key -> "false", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false", + SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "false") { + val df = query + val scans = collect(df.queryExecution.executedPlan) { case scan: FileSourceScanExec => + scan + } + assert(scans.nonEmpty && scans.forall(!_.supportsColumnar)) + rows = df.collect().toSeq + } + rows + } + + private def checkNativeFieldClaimAnswer(df: DataFrame, expected: Seq[Row]): Unit = { + assert( + collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec => scan }.nonEmpty, + s"expected native field claim resolution:\n${df.queryExecution.executedPlan}") + checkAnswer(df, expected) + } + + for { + nested <- Seq(false, true) + pruning <- (if (nested) Seq(false, true) else Seq(false)) + selected <- Seq("both", "id", "name") + } { + test( + s"Parquet field claims projection - nested=$nested pruning=$pruning selected=$selected") { + withFieldClaimFile(nested) { (path, schema) => + withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> pruning.toString) { + def query: DataFrame = { + val df = spark.read.schema(schema).parquet(path) + val prefix = if (nested) "s." else "" + val names = selected match { + case "both" => Seq("x", "a") + case "id" => Seq("x") + case "name" => Seq("a") + } + df.select(names.map(name => col(prefix + name)): _*) + } + val expected = fieldClaimReference(query) + checkNativeFieldClaimAnswer(query, expected) + } + } + } + } + + for { + nested <- Seq(false, true) + pushdown <- Seq(false, true) + predicate <- Seq("missing is null", "missing is not null", "id greater than one") + } { + test( + s"Parquet field claims filters - nested=$nested pushdown=$pushdown predicate=$predicate") { + withFieldClaimFile(nested) { (path, schema) => + withSQLConf( + SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true", + CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key -> pushdown.toString) { + def query: DataFrame = { + val df = spark.read.schema(schema).parquet(path) + val prefix = if (nested) "s." else "" + val x = col(prefix + "x") + val a = col(prefix + "a") + val condition = predicate match { + case "missing is null" => a.isNull + case "missing is not null" => a.isNotNull + case "id greater than one" => x > 1L + } + df.where(condition).select(x, a) + } + val expected = fieldClaimReference(query) + val values = predicate match { + case "missing is null" => Seq(1L, 2L, 3L) + case "missing is not null" => Seq.empty[Long] + case "id greater than one" => Seq(2L, 3L) + } + assert(expected.sortBy(_.toString) == values.map(value => Row(value, null))) + val df = query + checkNativeFieldClaimAnswer(df, expected) + if (!nested && pushdown && predicate == "id greater than one") { + val scans = collect(df.queryExecution.executedPlan) { + case scan: CometNativeScanExec => + scan + } + assert( + scans.exists(scan => + scan.metrics("pushdown_rows_pruned").value + + scan.metrics("pushdown_rows_matched").value > 0)) + } + } + } + } + } + + for { + nested <- Seq(false, true) + reverse <- Seq(false, true) + kind <- Seq("struct", "array", "map") + } { + test(s"Parquet field claims complex source - nested=$nested reverse=$reverse kind=$kind") { + val (dataType, values): (DataType, Seq[Any]) = kind match { + case "struct" => + (new StructType().add("v", LongType), Seq(Row(11L), Row(null), null)) + case "array" => + ( + ArrayType(LongType, containsNull = true), + Seq(Seq(11L, null, 33L), Seq.empty[Long], null)) + case "map" => + ( + MapType(StringType, LongType, valueContainsNull = true), + Seq(Map("k" -> 11L, "nil" -> null), Map.empty[String, Long], null)) + } + val id1 = new MetadataBuilder().putLong("parquet.field.id", 1L).build() + val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() + def wrapSchema(fields: StructType): StructType = + if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) else fields + def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields + val writeSchema = + wrapSchema(StructType(Seq(StructField("A", dataType, nullable = true, id1)))) + val fields = Seq( + StructField("x", dataType, nullable = true, id1), + StructField("a", dataType, nullable = true)) + val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else fields)) + val expected = + values.map(value => wrapRow(if (reverse) Row(null, value) else Row(value, null))) + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "false", + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .createDataFrame( + spark.sparkContext.parallelize(values.map(value => wrapRow(Row(value))), 1), + writeSchema) + .write + .parquet(path.getCanonicalPath) + } + def query: DataFrame = spark.read.schema(readSchema).parquet(path.getCanonicalPath) + withSQLConf( + CometConf.COMET_ENABLED.key -> "false", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") { + val df = query + val scans = collect(df.queryExecution.executedPlan) { case scan: FileSourceScanExec => + scan + } + assert(scans.nonEmpty && scans.forall(!_.supportsColumnar)) + checkAnswer(df, expected) + } + checkNativeFieldClaimAnswer(query, expected) + } + } + } + } + + for { + nested <- Seq(false, true) + reverse <- Seq(false, true) + count <- Seq(2, 3, 4) + } { + test( + s"Parquet field claims - repeated requested Parquet ids match parquet-mr - nested=$nested reverse=$reverse count=$count") { + val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() + def wrapSchema(fields: StructType): StructType = + if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) + else fields + def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields + + val fields = Seq("x", "y", "z", "w").take(count).map(name => withFieldId(name, 1)) + val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else fields)) + val writeSchema = wrapSchema(StructType(Seq(withFieldId("A", 1)))) + val expected = Seq(11L, 22L, 33L).map(value => + wrapRow(Row.fromSeq(Seq.fill[Any](count - 1)(null) :+ value))) + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "false", + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = Seq(11L, 22L, 33L).map(value => wrapRow(Row(value))) + spark + .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) + .write + .parquet(path.getCanonicalPath) + } + // ParquetRowConverter builds its requested-id-to-index map with toMap, so the last + // field with the repeated id owns the value. Check the Spark result in both orders. + val sparkRows = fieldClaimReference { + spark.read.schema(readSchema).parquet(path.getCanonicalPath) + } + assert(sparkRows.sortBy(_.toString) == expected.sortBy(_.toString)) + withSQLConf( + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.key -> "true") { + val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) + // Keep the existing JVM guard: native matching is tested separately by the Rust + // Parquet roundtrip, which bypasses this guard and exercises the actual converter. + val plan = df.queryExecution.executedPlan + assert(collect(plan) { case scan: CometNativeScanExec => + scan + }.isEmpty) + val reasons = new ExtendedExplainInfo().getFallbackReasons(plan) + assert(reasons.exists(_.contains("duplicate Parquet field ids")), s"$reasons\n$plan") + assert(df.collect().toSeq.sortBy(_.toString) == sparkRows.sortBy(_.toString)) + } + } + } + } + } + + for { + nested <- Seq(false, true) + offheap <- Seq(false, true) + } { + test( + s"Parquet field claims - issue 6994 original reproduction - nested=$nested offheap=$offheap") { + val id1 = new MetadataBuilder().putLong("parquet.field.id", 1L).build() + val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() + def wrapSchema(fields: StructType): StructType = + if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) + else fields + def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields + + val writeSchema = + wrapSchema(StructType(Seq(StructField("A", IntegerType, nullable = true, id1)))) + val readSchema = wrapSchema( + StructType( + Seq( + StructField("x", IntegerType, nullable = true, id1), + StructField("a", IntegerType, nullable = true)))) + val expected = (1 to 3).map(value => wrapRow(Row(value, null))) + + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "false", + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true", + "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true", + "spark.sql.columnVector.offheap.enabled" -> offheap.toString) { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = (1 to 3).map(value => wrapRow(Row(value))) + spark + .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) + .write + .parquet(path.getCanonicalPath) + } + for ((reader, comet, vectorized) <- Seq( + ("Spark vectorized", false, true), + ("Spark parquet-mr", false, false), + ("Comet native", true, true))) { + withSQLConf( + CometConf.COMET_ENABLED.key -> comet.toString, + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) { + val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) + val plan = df.queryExecution.executedPlan + val native = collect(plan) { case scan: CometNativeScanExec => scan }.nonEmpty + assert(native == comet, s"unexpected scan for $reader:\n$plan") + if (!comet) { + val scans = collect(plan) { case scan: FileSourceScanExec => scan } + assert(scans.nonEmpty && scans.forall(_.supportsColumnar == vectorized)) + } + val rows = df.collect().toSeq + info( + s"Spark ${spark.version}, nested=$nested, offheap=$offheap, $reader: ${rows.mkString(", ")}") + // Observe Spark's vectorized result without pinning its known bug as expected + // behavior. Comet and parquet-mr must return NULL for the losing name claimant. + if (comet || !vectorized) { + assert(rows.sortBy(_.toString) == expected.sortBy(_.toString)) + } + } + } + } + } + } + } + + for { + shape <- Seq("root", "struct", "nested struct", "array", "map") + reverse <- Seq(false, true) + fileName <- Seq("A", "a") + } { + test(s"Parquet field claims prefer id over name - $shape reverse=$reverse file=$fileName") { + def wrapSchema(fields: StructType): StructType = shape match { + case "root" => fields + case "struct" => new StructType().add("s", fields) + case "nested struct" => new StructType().add("s", new StructType().add("t", fields)) + case "array" => new StructType().add("s", ArrayType(fields, containsNull = true)) + case "map" => + new StructType().add("s", MapType(StringType, fields, valueContainsNull = true)) + } + + def wrapRow(fields: Row): Row = shape match { + case "root" => fields + case "struct" => Row(fields) + case "nested struct" => Row(Row(fields)) + case "array" => Row(Seq(fields, null)) + case "map" => Row(Map("value" -> fields, "null" -> null)) + } + + val values = Seq[java.lang.Long](11L, null, 33L, 44L, 55L) + val parentNulls = if (shape == "root") Seq.empty else Seq(Row(null)) + val writeSchema = wrapSchema(StructType(Seq(withFieldId(fileName, 1)))) + val fields = Seq(withFieldId("x", 1), StructField("a", LongType, nullable = true)) + val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else fields)) + withSQLConf( + // Cover folded matching for A and exact, case-sensitive matching for a. + SQLConf.CASE_SENSITIVE.key -> (fileName == "a").toString, + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + CometConf.COMET_BATCH_SIZE.key -> "2") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = values.map(value => wrapRow(Row(value))) ++ parentNulls + spark + .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) + .write + .parquet(path.getCanonicalPath) + } + for (useFieldId <- Seq(true, false)) { + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> useFieldId.toString) { + val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) + assert( + collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec => + scan + }.nonEmpty, + "expected a native scan to exercise Comet's field claim arbitration") + val expected = values.map { value => + val fields = if (useFieldId) Seq(value, null) else Seq(null, value) + wrapRow(Row.fromSeq(if (reverse) fields.reverse else fields)) + } ++ parentNulls + // Spark's vectorized reader returns zeros or stale values for this conflict + // (#6994). Pin the intended rows directly rather than using it as the oracle. + checkAnswer(df, expected) + } + } + } + } + } + } + /** * Write a one-row file and assert Comet hands the read back to Spark under field id matching, * and keeps it native without. From 8d0a12d7c7613f85520cc76d73eeacf805966ec8 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sat, 24 Oct 2026 16:31:42 +0800 Subject: [PATCH 2/7] test: use explicit Arc cloning in field claim regression tests --- native/core/src/parquet/schema_adapter.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 34f274ed5f9..d3a0e465462 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -3127,14 +3127,14 @@ pub(crate) mod test { .collect(); let (batch, required) = if nested { ( - struct_batch(physical, values.clone())?, + struct_batch(physical, Arc::clone(&values) as ArrayRef)?, struct_schema(requested), ) } else { ( RecordBatch::try_new( Arc::new(Schema::new(vec![physical])), - vec![values.clone()], + vec![Arc::clone(&values) as ArrayRef], )?, Arc::new(Schema::new(requested)), ) From 5369e01f0010cc7ee8eb5c32397e4ad9adad8062 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sun, 25 Oct 2026 10:14:42 +0800 Subject: [PATCH 3/7] fix: preserve Parquet field validation and projected reads --- native/core/Cargo.toml | 4 + .../core/benches/parquet_field_resolution.rs | 137 ++++++++++++ native/core/src/parquet/field_resolver.rs | 139 ++++++++---- native/core/src/parquet/parquet_support.rs | 14 +- native/core/src/parquet/schema_adapter.rs | 203 ++++++++++++++++-- .../org/apache/comet/DataTypeSupport.scala | 15 ++ .../serde/operator/CometNativeScan.scala | 19 +- .../comet/exec/CometNativeReaderSuite.scala | 154 ++++++++++++- 8 files changed, 618 insertions(+), 67 deletions(-) create mode 100644 native/core/benches/parquet_field_resolution.rs diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index a01bbb7f829..009e36c101d 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -174,6 +174,10 @@ harness = false name = "parquet_timestamp_conversion" harness = false +[[bench]] +name = "parquet_field_resolution" +harness = false + [[bench]] name = "sort_row_partitions" harness = false diff --git a/native/core/benches/parquet_field_resolution.rs b/native/core/benches/parquet_field_resolution.rs new file mode 100644 index 00000000000..edbbb712732 --- /dev/null +++ b/native/core/benches/parquet_field_resolution.rs @@ -0,0 +1,137 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Conflict-free field-ID matching must not repeatedly select each child per batch. +//! Compare optimized builds with Criterion's --save-baseline / --baseline options. + +use std::{collections::HashMap, hint::black_box, sync::Arc}; + +use arrow::array::{Array, ArrayRef, Int32Array, StructArray}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::record_batch::RecordBatch; +use bytes::Bytes; +use comet::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions}; +use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput}; +use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_spark_expr::EvalMode; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; + +const ROWS: usize = 32_768; +const CHILDREN: usize = 64; + +fn field(name: String, id: usize) -> Arc { + Arc::new( + Field::new(name, DataType::Int32, true).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])), + ) +} + +fn benchmark(c: &mut Criterion) { + let fields = (0..CHILDREN) + .map(|id| field(format!("file_{id}"), id)) + .collect::>(); + let columns: Vec = (0..CHILDREN) + .map(|id| Arc::new(Int32Array::from(vec![id as i32; ROWS])) as ArrayRef) + .collect(); + let input: ArrayRef = Arc::new(StructArray::new(fields.into(), columns.clone(), None)); + let target_fields = (0..CHILDREN) + .map(|id| field(format!("requested_{id}"), id)) + .collect::>(); + let target = DataType::Struct(target_fields.clone().into()); + let expected: ArrayRef = Arc::new(StructArray::new(target_fields.into(), columns, None)); + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "s", + input.data_type().clone(), + true, + )])), + vec![Arc::clone(&input)], + ) + .unwrap(); + let mut writer = ArrowWriter::try_new(Vec::new(), batch.schema(), None).unwrap(); + writer.write(&batch).unwrap(); + let bytes = Bytes::from(writer.into_inner().unwrap()); + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); + options.use_field_id = true; + + let mut group = c.benchmark_group("parquet_field_resolution"); + for batch_size in [64, 8192] { + let decoded = ParquetRecordBatchReaderBuilder::try_new(bytes.clone()) + .unwrap() + .with_batch_size(batch_size) + .build() + .unwrap() + .next() + .unwrap() + .unwrap(); + let input = decoded.column(0); + let converted = + spark_parquet_convert(ColumnarValue::Array(Arc::clone(input)), &target, &options) + .unwrap(); + let ColumnarValue::Array(converted) = converted else { + unreachable!() + }; + assert_eq!(converted.as_ref(), expected.slice(0, batch_size).as_ref()); + + group.throughput(Throughput::Elements(batch_size as u64)); + group.bench_with_input( + BenchmarkId::new("convert", batch_size), + input, + |b, input| { + b.iter(|| { + spark_parquet_convert( + ColumnarValue::Array(Arc::clone(black_box(input))), + black_box(&target), + black_box(&options), + ) + .unwrap() + }) + }, + ); + group.throughput(Throughput::Elements(ROWS as u64)); + group.bench_function(BenchmarkId::new("decode_and_convert", batch_size), |b| { + b.iter(|| { + let reader = ParquetRecordBatchReaderBuilder::try_new(bytes.clone()) + .unwrap() + .with_batch_size(batch_size) + .build() + .unwrap(); + let mut rows = 0; + for batch in reader { + let batch = batch.unwrap(); + rows += batch.num_rows(); + black_box( + spark_parquet_convert( + ColumnarValue::Array(Arc::clone(batch.column(0))), + &target, + &options, + ) + .unwrap(), + ); + } + assert_eq!(rows, ROWS); + }) + }); + } + group.finish(); +} + +criterion_group!(benches, benchmark); +criterion_main!(benches); diff --git a/native/core/src/parquet/field_resolver.rs b/native/core/src/parquet/field_resolver.rs index c80e5501ca0..24bb636748b 100644 --- a/native/core/src/parquet/field_resolver.rs +++ b/native/core/src/parquet/field_resolver.rs @@ -66,6 +66,10 @@ pub(crate) enum FieldClaim { } impl FieldClaim { + fn owned_by(self, requested: usize) -> bool { + matches!(self, Self::IdMatch(owner) | Self::NameMatch(owner) if owner == requested) + } + fn strength(self) -> u8 { match self { Self::IdMatch(_) => 3, @@ -84,6 +88,38 @@ impl FieldClaim { } } +/// Independent selections and ownership for one nested group. Keeping a losing selection +/// lets schema validation inspect its shape before the runtime replaces its value with NULL. +#[derive(Debug)] +pub(crate) struct FieldMatches { + selected: Vec>, + claims: Vec, +} + +impl FieldMatches { + /// Each independent file index paired with whether the requester owns that file field. + pub(crate) fn iter(&self) -> impl Iterator, bool)> + '_ { + self.selected.iter().enumerate().map(|(requested, file)| { + ( + *file, + file.is_some_and(|index| self.claims[index].owned_by(requested)), + ) + }) + } + + /// Reuse the selection vector for runtime conversion, masking only the losing fields. + pub(crate) fn into_owned(mut self) -> Vec> { + for (requested, file) in self.selected.iter_mut().enumerate() { + if let Some(index) = *file { + if !self.claims[index].owned_by(requested) { + *file = None; + } + } + } + self.selected + } +} + /// Resolves the requested fields of one group against the file fields of the same group, as /// Spark's `clipParquetGroupFields` does. The caller folds both sides' names once (see /// `name_fold`) and passes them in, parallel to the fields. @@ -91,7 +127,7 @@ impl FieldClaim { /// [`FieldResolver::select`] raises a requested id that more than one file field carries, which /// both the root and nested fields report at once, as Spark does. It does not raise an /// ambiguous name: the root reports one only for a column the read references, and -/// [`FieldResolver::resolve`] raises it for a nested field. +/// [`FieldResolver::resolve_all`] raises it for a nested field. pub(crate) struct FieldResolver<'a> { requested: &'a [FieldRef], requested_folded: &'a [String], @@ -228,8 +264,17 @@ impl<'a> FieldResolver<'a> { /// no such field. An ambiguous selection is an error, raised before a decoder can multiply /// rows: see [`FieldResolver::select`] for a repeated id and [`duplicate_name_error`] for a /// name. - pub(crate) fn resolve(&self, requested: usize) -> DataFusionResult> { - match self.select(requested)? { + #[cfg(test)] + fn resolve(&self, requested: usize) -> DataFusionResult> { + self.resolve_selection(requested, self.select(requested)?) + } + + fn resolve_selection( + &self, + requested: usize, + selection: Selection<'_>, + ) -> DataFusionResult> { + match selection { Selection::ById(index) => Ok(index), Selection::ByName([]) => Ok(None), Selection::ByName([index]) => Ok(Some(*index)), @@ -250,45 +295,41 @@ impl<'a> FieldResolver<'a> { pub(crate) fn file_claims(&self) -> DataFusionResult> { let mut claims = vec![FieldClaim::Unchanged; self.file.len()]; for requested in 0..self.requested.len() { - match self.select(requested)? { - Selection::ById(matched) => { - if let Some(file) = matched { - claims[file].claim(FieldClaim::IdMatch(requested)); - } - for &file in self.name_matches(requested) { - claims[file].claim(FieldClaim::Shadowed); - } - } - Selection::ByName(matches) => { - for &file in matches { - claims[file].claim(FieldClaim::NameMatch(requested)); - } - } - } + self.record_claims(&mut claims, requested, self.select(requested)?); } Ok(claims) } - /// Resolve a nested group in requested order, giving each file field to its strongest - /// claimant only. Missing and losing fields read NULL. Validate independent selections - /// first, in requested order, so arbitration cannot hide ambiguity or change which of - /// several invalid selections reports its error first. - pub(crate) fn resolve_all(&self) -> DataFusionResult>> { - let mut resolved = (0..self.requested.len()) - .map(|requested| self.resolve(requested)) - .collect::>>()?; - let claims = self.file_claims()?; - for (requested, file) in resolved.iter_mut().enumerate() { - if let Some(index) = *file { - if !matches!( - claims[index], - FieldClaim::IdMatch(owner) | FieldClaim::NameMatch(owner) if owner == requested - ) { - *file = None; + fn record_claims(&self, claims: &mut [FieldClaim], requested: usize, selection: Selection<'_>) { + match selection { + Selection::ById(matched) => { + if let Some(file) = matched { + claims[file].claim(FieldClaim::IdMatch(requested)); + } + for &file in self.name_matches(requested) { + claims[file].claim(FieldClaim::Shadowed); + } + } + Selection::ByName(matches) => { + for &file in matches { + claims[file].claim(FieldClaim::NameMatch(requested)); } } } - Ok(resolved) + } + + /// Select each nested field once and record claims from that selection. Validate selections + /// in requested order so arbitration cannot hide ambiguity or reorder selection errors. + /// The caller can validate losing selections before masking them with [`FieldMatches::into_owned`]. + pub(crate) fn resolve_all(&self) -> DataFusionResult { + let mut selected = Vec::with_capacity(self.requested.len()); + let mut claims = vec![FieldClaim::Unchanged; self.file.len()]; + for requested in 0..self.requested.len() { + let selection = self.select(requested)?; + selected.push(self.resolve_selection(requested, selection)?); + self.record_claims(&mut claims, requested, selection); + } + Ok(FieldMatches { selected, claims }) } } @@ -461,7 +502,7 @@ mod test { if reverse { expected.reverse(); } - assert_eq!(resolver.resolve_all().unwrap(), expected); + assert_eq!(resolver.resolve_all().unwrap().into_owned(), expected); }, ); } @@ -496,12 +537,30 @@ mod test { ] ); assert_eq!( - resolver.resolve_all().unwrap(), + resolver.resolve_all().unwrap().into_owned(), vec![Some(0), None, None, None, Some(2)] ); }); } + #[test] + fn independent_selections_distinguish_losing_and_missing_fields() { + let requested = [ + field("x", Some(1)), + field("a", None), + field("missing", Some(2)), + ]; + let file = [field("A", Some(1))]; + with_resolver(&requested, &file, false, true, |resolver| { + let matches = resolver.resolve_all().unwrap(); + assert_eq!( + matches.iter().collect::>(), + vec![(Some(0), true), (Some(0), false), (None, false)] + ); + assert_eq!(matches.into_owned(), vec![Some(0), None, None]); + }); + } + #[test] fn shadowing_respects_case_sensitivity_and_a_stronger_id_claim() { // A's id resolves elsewhere, but its name still shadows a name-only claimant of `a`. @@ -525,7 +584,7 @@ mod test { } let file = [field("a", Some(2)), field("b", Some(1))]; with_resolver(&requested, &file, case_sensitive, true, |resolver| { - assert_eq!(resolver.resolve_all().unwrap(), expected); + assert_eq!(resolver.resolve_all().unwrap().into_owned(), expected); }); } } @@ -553,7 +612,7 @@ mod test { (true, true) => vec![Some(1), None, None, Some(0)], (true, false) => vec![Some(1), None, None, None], }; - assert_eq!(resolver.resolve_all().unwrap(), expected); + assert_eq!(resolver.resolve_all().unwrap().into_owned(), expected); }); } } @@ -581,7 +640,7 @@ mod test { } let file = [field("a", Some(1)), field("b", None)]; with_resolver(&requested, &file, false, true, |resolver| { - assert_eq!(resolver.resolve_all().unwrap(), expected); + assert_eq!(resolver.resolve_all().unwrap().into_owned(), expected); }); } } diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 26f2e7508b0..ae0a504c5cb 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -16,7 +16,7 @@ // under the License. use crate::execution::operators::ExecutionError; -use crate::parquet::field_resolver::FieldResolver; +use crate::parquet::field_resolver::{FieldMatches, FieldResolver}; use crate::parquet::name_fold::fold_names; use arrow::array::{ make_array, FixedSizeBinaryArray, GenericListViewArray, MapArray, OffsetSizeTrait, StringArray, @@ -430,13 +430,14 @@ pub(crate) fn field_id(field: &arrow::datatypes::Field) -> Option { .and_then(|v| v.parse::().ok()) } -/// Resolve each requested (`to`) struct field to the index of the file (`from`) field it reads -/// from, or `None` when the file holds no such field, with Spark's `clipParquetGroupFields` rules +/// Resolve each requested (`to`) struct field independently and record value ownership, +/// with Spark's `clipParquetGroupFields` rules /// as [`FieldResolver::resolve_all`] applies them: an id-bearing requested field matches only by id /// when `use_field_id` is set, and every other field matches by name, folded with the same /// `toLowerCase(Locale.ROOT)` fold the top-level schema adapter uses when `case_sensitive` is /// false. A requested field that selects more than one file field is rejected. Each file field -/// belongs to its strongest claimant only; a losing requested field reads NULL. +/// belongs to its strongest claimant only. Keep losing selections for schema validation; +/// runtime conversion masks them with [`FieldMatches::into_owned`] so they read NULL. /// /// 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. @@ -444,7 +445,7 @@ pub(crate) fn match_struct_fields( from_fields: &[FieldRef], to_fields: &[FieldRef], parquet_options: &SparkParquetOptions, -) -> DataFusionResult>> { +) -> DataFusionResult { // Fold the file (`from`) and requested (`to`) field names once via the JVM's // `toLowerCase(Locale.ROOT)` (the same fold the top-level schema adapter uses), so // nested case-insensitive matching is byte-for-byte consistent with the top level. @@ -476,7 +477,8 @@ fn parquet_convert_struct_to_struct( ) -> DataFusionResult { match (from_type, to_type) { (DataType::Struct(from_fields), DataType::Struct(to_fields)) => { - let from_indices = match_struct_fields(from_fields, to_fields, parquet_options)?; + let from_indices = + match_struct_fields(from_fields, to_fields, parquet_options)?.into_owned(); let mut field_overlap = false; let mut cast_fields: Vec = Vec::with_capacity(to_fields.len()); diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index d3a0e465462..637059d6e87 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -659,8 +659,9 @@ fn combine_checks( /// encoding used by Spark. Legacy or custom group names cannot be recovered from this schema. /// Requested struct fields resolve to file fields with the same field-id / case-fold rules the runtime /// convert uses ([`match_struct_fields`]); requested fields missing from the file are skipped -/// (they read as null / default, as before). A pair Spark rejects when it opens the file is an -/// error, and sibling verdicts combine as in [`combine_checks`]. +/// (they read as null / default, as before). Independently selected losing fields still undergo +/// open-time shape validation, but do not contribute deferred decoder rejections because their +/// values are not read. Sibling verdicts combine as in [`combine_checks`]. fn check_conversion( physical_type: &DataType, target_type: &DataType, @@ -674,17 +675,29 @@ fn check_conversion( match (physical_type, target_type) { (DataType::Struct(physical_fields), DataType::Struct(target_fields)) => { let physical_indices = match_struct_fields(physical_fields, target_fields, options)?; - combine_checks(target_fields.iter().zip(physical_indices).filter_map( - |(target_field, physical_index)| { - let physical_field = &physical_fields[physical_index?]; - Some(check_conversion( - physical_field.data_type(), - target_field.data_type(), - &format!("{column}, {}", physical_field.name()), - options, - )) - }, - )) + combine_checks( + target_fields + .iter() + .zip(physical_indices.iter()) + .filter_map(|(target_field, (physical_index, owns_value))| { + let physical_field = &physical_fields[physical_index?]; + Some( + check_conversion( + physical_field.data_type(), + target_field.data_type(), + &format!("{column}, {}", physical_field.name()), + options, + ) + .map(|check| { + if owns_value { + check + } else { + ConversionCheck::Accept + } + }), + ) + }), + ) } ( DataType::List(physical_item) @@ -3574,6 +3587,166 @@ pub(crate) mod test { } } + #[test] + fn absent_container_selections_do_not_add_conversion_errors() -> Result<(), DataFusionError> { + let physical = DataType::Struct( + vec![Arc::new( + Field::new("A", DataType::Int64, true).with_metadata(id_meta("1")), + )] + .into(), + ); + for (case_sensitive, missing_id) in [(true, false), (false, true)] { + let mut options = default_options(); + options.use_field_id = true; + options.case_sensitive = case_sensitive; + let mut absent = Field::new("a", struct_type(vec![("v", DataType::Int64)]), true); + if missing_id { + absent = absent.with_metadata(id_meta("2")); + } + let target = DataType::Struct( + vec![ + Arc::new(Field::new("x", DataType::Int64, true).with_metadata(id_meta("1"))), + Arc::new(absent), + ] + .into(), + ); + assert!(matches!( + check_conversion(&physical, &target, "s", &options)?, + ConversionCheck::Accept + )); + } + Ok(()) + } + + #[tokio::test] + async fn wide_field_id_struct_preserves_values_and_parent_nulls_across_batches( + ) -> Result<(), DataFusionError> { + use arrow::buffer::NullBuffer; + + let rows = 16_401; + let fields = (1..=64) + .map(|id| { + Arc::new( + Field::new(format!("file_{id}"), DataType::Int32, true) + .with_metadata(id_meta(&id.to_string())), + ) + }) + .collect::>(); + let columns = (1..=64) + .map(|id| { + Arc::new(Int32Array::from( + (0..rows) + .map(|row| (row % 11 != 0).then_some(id * 100_000 + row as i32)) + .collect::>(), + )) as ArrayRef + }) + .collect::>(); + let nulls = Some(NullBuffer::from( + (0..rows).map(|row| row % 7 != 0).collect::>(), + )); + let input = StructArray::new(fields.into(), columns.clone(), nulls.clone()); + let requested = (1..=64) + .rev() + .map(|id| { + Arc::new( + Field::new(format!("requested_{id}"), DataType::Int32, true) + .with_metadata(id_meta(&id.to_string())), + ) + }) + .collect::>(); + let expected = + StructArray::new(requested.into(), columns.into_iter().rev().collect(), nulls); + let wrap = |array: StructArray| { + RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "s", + array.data_type().clone(), + true, + )])), + vec![Arc::new(array)], + ) + }; + let input = wrap(input)?; + let expected = wrap(expected)?; + let mut options = default_options(); + options.use_field_id = true; + let mut stream = scan_parquet(&input, expected.schema(), options)?; + let mut offset = 0; + let mut batches = 0; + while let Some(batch) = stream.next().await { + let batch = batch?; + assert_eq!(batch, expected.slice(offset, batch.num_rows())); + offset += batch.num_rows(); + batches += 1; + } + assert_eq!(offset, rows); + assert!(batches > 1); + Ok(()) + } + + /// Clipping validates a selected container even when another claimant owns its value. + /// Losing leaf conversions do not install a decoder and must not add deferred failures. + #[test] + fn losing_field_claims_preserve_open_errors_but_not_deferred_rejections( + ) -> Result<(), DataFusionError> { + let mut options = default_options(); + options.use_field_id = true; + for reverse in [false, true] { + for wrap in 0..4 { + for losing_type in [ + DataType::Int32, + struct_type(vec![("v", DataType::Int64)]), + list_type(struct_type(vec![("v", DataType::Int64)])), + map_type(struct_type(vec![("v", DataType::Int64)])), + ] { + let physical = DataType::Struct( + vec![Arc::new( + Field::new("A", DataType::Int64, true).with_metadata(id_meta("1")), + )] + .into(), + ); + let mut requested = vec![ + Arc::new( + Field::new("x", DataType::Int64, true).with_metadata(id_meta("1")), + ), + Arc::new(Field::new("a", losing_type.clone(), true)), + ]; + if reverse { + requested.reverse(); + } + let target = DataType::Struct(requested.into()); + let (physical, target) = match wrap { + 0 => (physical, target), + 1 => (list_type(physical), list_type(target)), + 2 => (map_type(physical), map_type(target)), + _ => ( + DataType::Dictionary(Box::new(DataType::Int32), Box::new(physical)), + target, + ), + }; + let result = check_conversion(&physical, &target, "s", &options); + if matches!(losing_type, DataType::Int32) { + assert!(matches!(result?, ConversionCheck::Accept)); + } else { + let error = result + .err() + .expect("losing container must still be clipped"); + let message = error.to_string(); + assert!(message.contains("INT64"), "{message}"); + assert!(message.contains("A"), "{message}"); + let required = Arc::new(Schema::new(vec![Field::new("s", target, true)])); + let physical = Arc::new(Schema::new(vec![Field::new("s", physical, true)])); + assert!( + super::check_file_conversions(required, physical, options.clone()) + .is_err() + ); + } + } + } + } + Ok(()) + } + /// Positive: `INT32 -> bigint` inside a struct still converts when type promotion is /// allowed (Spark 4.x behaviour). #[tokio::test] @@ -3720,7 +3893,9 @@ pub(crate) mod test { assert!(!error.contains("case-insensitive"), "{error}"); // A case-sensitive read compares exact names, so `Dup` resolves to itself alone. assert_eq!( - super::match_struct_fields(&fields, &fields[2..], &options).unwrap(), + super::match_struct_fields(&fields, &fields[2..], &options) + .unwrap() + .into_owned(), vec![Some(2)] ); } diff --git a/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala b/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala index 4d482c156ea..a2cb75bdda1 100644 --- a/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala +++ b/spark/src/main/scala/org/apache/comet/DataTypeSupport.scala @@ -132,6 +132,21 @@ object DataTypeSupport { ids.distinct.length != ids.length } + /** + * True when a field named in `requiredFieldNames` shares its id with another field in `fields`. + * Names are matched exactly, as in the native scan's projection into its serialized schema. + * Like the sibling-only check above, this is only meaningful when field-id matching is enabled. + */ + def hasDuplicateFieldIds( + fields: Array[StructField], + requiredFieldNames: Set[String]): Boolean = { + val fieldsById = fields.flatMap(field => fieldId(field).map(_ -> field.name)).groupBy(_._1) + fieldsById.values.exists(matches => + matches.length > 1 && matches.exists { case (_, name) => + requiredFieldNames.contains(name) + }) + } + private def fieldId(field: StructField): Option[Int] = { if (!ParquetUtils.hasFieldId(field)) None else { diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index caab7a72bee..a19a95963c5 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala @@ -227,11 +227,26 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS // GEOGRAPHY) would otherwise crash schema serialization; fall back instead. Validate the same // pruned data schema `convert` serializes, and the full partition schema, which is never // pruned. - val serializedFields = + val serializedDataSchema = nativeDataSchema( scanExec.relation.dataSchema, scanExec.requiredSchema, - scanExec.conf.resolver).fields ++ scanExec.relation.partitionSchema.fields + scanExec.conf.resolver) + + // Spark clips only required roots, but native root remapping sees every serialized root. + // An unread alias with the same id can take a required root's value. The type checker only + // sees requiredSchema, so also check the required roots against the actual serialized schema. + if (CometParquetUtils.readFieldId(scanExec.conf) && + DataTypeSupport.hasDuplicateFieldIds( + serializedDataSchema.fields, + scanExec.requiredSchema.fieldNames.toSet)) { + withFallbackReason( + scanExec, + "Native Parquet scan does not support duplicate Parquet field ids shared by " + + "a required root and another serialized data root") + } + + val serializedFields = serializedDataSchema.fields ++ scanExec.relation.partitionSchema.fields serializedFields.foreach { field => if (serializeDataType(field.dataType).isEmpty) { withFallbackReason( 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 8a373dff900..3d5c62bf2e6 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -1508,6 +1508,149 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + test("Parquet field claims - pruned losing container does not add a schema error") { + withFieldClaimFile(nested = true) { (path, _) => + val fields = StructType( + Seq( + withFieldId("x", 1), + StructField("a", new StructType().add("v", LongType), nullable = true))) + val schema = new StructType().add("s", fields) + withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") { + def query: DataFrame = spark.read.schema(schema).parquet(path).select("s.x") + val expected = fieldClaimReference(query) + assert(expected == Seq(Row(1L), Row(2L), Row(3L))) + checkNativeFieldClaimAnswer(query, expected) + } + } + } + + for { + shape <- Seq("struct", "array", "map") + reverse <- Seq(false, true) + empty <- Seq(false, true) + } { + test( + s"Parquet field claims - losing container still fails schema validation - shape=$shape reverse=$reverse empty=$empty") { + def wrapType(fields: StructType): DataType = shape match { + case "struct" => fields + case "array" => ArrayType(fields, containsNull = true) + case "map" => MapType(StringType, fields, valueContainsNull = true) + } + val writeType = wrapType(StructType(Seq(withFieldId("A", 1)))) + val fields = Seq( + withFieldId("x", 1), + StructField("a", new StructType().add("v", LongType), nullable = true)) + val readType = wrapType(StructType(if (reverse) fields.reverse else fields)) + val value: Any = shape match { + case "struct" => Row(7L) + case "array" => Seq(Row(7L)) + case "map" => Map("k" -> Row(7L)) + } + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "false", + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val rows = if (empty) Seq.empty[Row] else Seq(Row(value)) + spark + .createDataFrame( + spark.sparkContext.parallelize(rows, 1), + new StructType().add("s", writeType)) + .write + .parquet(path.getCanonicalPath) + } + def query: DataFrame = + spark.read.schema(new StructType().add("s", readType)).parquet(path.getCanonicalPath) + + for (vectorized <- Seq(false, true)) { + withSQLConf( + CometConf.COMET_ENABLED.key -> "false", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) { + val error = intercept[Exception](query.collect()) + assert(causeMessages(error).contains("is not a group"), causeMessages(error)) + } + } + val df = query + assert( + collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec => + scan + }.nonEmpty, + s"expected native schema validation:\n${df.queryExecution.executedPlan}") + val error = intercept[Exception](df.collect()) + val messages = causeMessages(error) + assert(messages.contains("INT64") && messages.contains("A"), messages) + } + } + } + } + + for { + reverse <- Seq(false, true) + readIds <- Seq(false, true) + selected <- Seq("x", "y", "z") + } { + test( + s"Parquet field claims - repeated root ids single-column projection - reverse=$reverse readIds=$readIds selected=$selected") { + val fields = Seq(withFieldId("x", 1), withFieldId("y", 1), withFieldId("z", 2)) + val readSchema = StructType(if (reverse) fields.reverse else fields) + val writeSchema = StructType(Seq(withFieldId("A", 1), withFieldId("z", 2))) + val expected = Seq(Row(if (selected == "z") 9L else if (readIds) 7L else null)) + withSQLConf( + SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", + SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readIds.toString) { + withTempPath { path => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .createDataFrame(spark.sparkContext.parallelize(Seq(Row(7L, 9L)), 1), writeSchema) + .write + .parquet(path.getCanonicalPath) + } + def query: DataFrame = + spark.read.schema(readSchema).parquet(path.getCanonicalPath).select(selected) + + // Either Spark reader can read one alias alone. Keep both aliases in dataSchema to + // exercise the difference between Spark's projection and native root remapping. + for (vectorized <- Seq(false, true)) { + withSQLConf( + CometConf.COMET_ENABLED.key -> "false", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) { + val df = query + val scans = collect(df.queryExecution.executedPlan) { + case scan: FileSourceScanExec => + scan + } + assert(scans.nonEmpty && scans.forall(_.supportsColumnar == vectorized)) + scans.foreach { scan => + assert(scan.requiredSchema.fieldNames.toSeq == Seq(selected)) + assert(scan.relation.dataSchema.fieldNames.toSeq == readSchema.fieldNames.toSeq) + } + checkAnswer(df, expected) + } + } + withSQLConf( + CometConf.COMET_ENABLED.key -> "true", + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") { + val df = query + checkAnswer(df, expected) + val plan = df.queryExecution.executedPlan + val native = collect(plan) { case scan: CometNativeScanExec => scan }.nonEmpty + if (readIds && selected != "z") { + assert(!native, s"expected a fallback for a projected repeated id:\n$plan") + val reasons = new ExtendedExplainInfo().getFallbackReasons(plan) + assert( + reasons.exists(_.contains("duplicate Parquet field ids")), + s"$reasons\n$plan") + } else { + // Unread duplicate IDs must not disable an unrelated root or name-based reads. + assert(native, s"unexpected fallback for an unrelated root or disabled IDs:\n$plan") + } + } + } + } + } + } + for { nested <- Seq(false, true) offheap <- Seq(false, true) @@ -1559,12 +1702,13 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper val scans = collect(plan) { case scan: FileSourceScanExec => scan } assert(scans.nonEmpty && scans.forall(_.supportsColumnar == vectorized)) } - val rows = df.collect().toSeq - info( - s"Spark ${spark.version}, nested=$nested, offheap=$offheap, $reader: ${rows.mkString(", ")}") - // Observe Spark's vectorized result without pinning its known bug as expected - // behavior. Comet and parquet-mr must return NULL for the losing name claimant. + // Only inspect the plan for Spark's vectorized reader: the losing field can + // contain stale memory with off-heap vectors. Comet and parquet-mr must return + // NULL for the losing name claimant. if (comet || !vectorized) { + val rows = df.collect().toSeq + info( + s"Spark ${spark.version}, nested=$nested, offheap=$offheap, $reader: ${rows.mkString(", ")}") assert(rows.sortBy(_.toString) == expected.sortBy(_.toString)) } } From 29e3f9fb6e9eaf69796d4a07546dfd0582cf29d3 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sat, 31 Oct 2026 22:35:02 +0800 Subject: [PATCH 4/7] fix: preserve deferred type rejection for losing Parquet fields --- native/core/src/parquet/schema_adapter.rs | 149 +++++++++++++----- .../comet/exec/CometNativeReaderSuite.scala | 60 ++++++- 2 files changed, 169 insertions(+), 40 deletions(-) diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index e6e19371e28..cc50af694fd 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -693,8 +693,8 @@ fn check_conversion( /// Requested struct fields resolve to file fields with the same field-id / case-fold rules the runtime /// convert uses ([`match_struct_fields`]); requested fields missing from the file are skipped /// (they read as null / default, as before). Independently selected losing fields still undergo -/// open-time shape validation, but do not contribute deferred decoder rejections because their -/// values are not read. Sibling verdicts combine as in [`combine_checks`]. +/// open-time shape validation and deferred leaf checks, like Spark's `getUpdater`. Only runtime +/// conversion masks their values with NULL. Sibling verdicts combine as in [`combine_checks`]. /// /// `physical_field` may carry [`NANOS_TIMESTAMP_KEY`]. fn check_field_conversion( @@ -737,24 +737,15 @@ fn check_nested_conversion( target_fields .iter() .zip(physical_indices.iter()) - .filter_map(|(target_field, (physical_index, owns_value))| { + .filter_map(|(target_field, (physical_index, _))| { let physical_field = &physical_fields[physical_index?]; - Some( - check_nested_conversion( - physical_field.data_type(), - target_field.data_type(), - &format!("{column}, {}", physical_field.name()), - options, - is_nanos_timestamp(physical_field), - ) - .map(|check| { - if owns_value { - check - } else { - ConversionCheck::Accept - } - }), - ) + Some(check_nested_conversion( + physical_field.data_type(), + target_field.data_type(), + &format!("{column}, {}", physical_field.name()), + options, + is_nanos_timestamp(physical_field), + )) }), ) } @@ -3744,10 +3735,11 @@ pub(crate) mod test { Ok(()) } - /// The losing name claimant reads NULL even when converting that physical column to its - /// requested type would fail. This exercises both the file-open check and runtime adapter. + /// Losing name claims read NULL, but a nested leaf must still pass Spark's conversion + /// check. Root losing-field validation is tracked separately in #7207. #[tokio::test] - async fn conflicting_field_claims_read_null_at_every_depth() -> Result<(), DataFusionError> { + async fn conflicting_field_claims_validate_losing_fields_at_every_depth( + ) -> Result<(), DataFusionError> { use arrow::buffer::NullBuffer; for depth in 0..=2 { @@ -3794,7 +3786,25 @@ pub(crate) mod test { let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); options.use_field_id = true; let mut stream = scan_parquet(&batch, expected.schema(), options)?; - let read = stream.next().await.unwrap()?; + let read = stream.next().await.unwrap(); + if depth > 0 && losing_type == DataType::Int32 { + let error = read.expect_err("losing nested leaf must still be validated"); + let message = error.to_string(); + let column = (0..depth) + .rev() + .map(|level| format!("s{level}")) + .chain(std::iter::once("A".to_string())) + .collect::>() + .join(", "); + assert!( + message.contains(&format!("Column: [[{column}]]")), + "{message}" + ); + assert!(message.contains("Expected: int"), "{message}"); + assert!(message.contains("Found: INT64"), "{message}"); + continue; + } + let read = read?; assert_eq!( read, expected, "depth={depth}, reverse={reverse}, losing_type={losing_type}" @@ -3806,11 +3816,11 @@ pub(crate) mod test { Ok(()) } - /// Spark parquet-mr assigns repeated requested ids to the last field. The first field's - /// incompatible type must not reject a read after that field loses its claim. + /// Repeated root ids keep their existing last-wins behavior. Inside a struct, every + /// selected leaf still contributes its type check, including the losing id claims. #[tokio::test] - async fn repeated_requested_ids_read_last_field_at_root_and_nested( - ) -> Result<(), DataFusionError> { + async fn repeated_requested_ids_preserve_losing_field_validation() -> Result<(), DataFusionError> + { for count in [2, 3, 4] { for nested in [false, true] { for reverse in [false, true] { @@ -3851,12 +3861,14 @@ pub(crate) mod test { let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); options.use_field_id = true; let mut stream = scan_parquet(&batch, required, options)?; - let read = stream.next().await.unwrap()?; - let columns = if nested { - read.column(0).as_struct().columns() - } else { - read.columns() - }; + let read = stream.next().await.unwrap(); + if nested { + let error = read.expect_err("losing id claims must still be validated"); + assert!(error.to_string().contains("Column: [[s, A]]"), "{error}"); + continue; + } + let read = read?; + let columns = read.columns(); for column in &columns[..count - 1] { assert_eq!(column.data_type(), &DataType::Int32); assert_eq!(column.null_count(), values.len()); @@ -3964,6 +3976,62 @@ pub(crate) mod test { Ok(()) } + /// A losing leaf's rejection is deferred too: neither an empty file nor a pruned row + /// group may raise it before any rows are decoded. + #[tokio::test] + async fn losing_field_claim_rejection_passes_for_empty_and_pruned_reads( + ) -> Result<(), DataFusionError> { + let nested = struct_batch( + Field::new("A", DataType::Int64, true).with_metadata(id_meta("1")), + Arc::new(Int64Array::from(vec![7])), + )?; + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![ + nested.schema().field(0).clone(), + Field::new("id", DataType::Int32, false), + ])), + vec![ + Arc::clone(nested.column(0)), + Arc::new(Int32Array::from(vec![1])), + ], + )?; + for reverse in [false, true] { + let mut fields = vec![ + Field::new("x", DataType::Int64, true).with_metadata(id_meta("1")), + Field::new("a", DataType::Int32, true), + ]; + if reverse { + fields.reverse(); + } + let required = Arc::new(Schema::new(vec![ + struct_schema(fields).field(0).clone(), + Field::new("id", DataType::Int32, false), + ])); + let mut options = default_options(); + options.use_field_id = true; + let id_is_100: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("id", 1)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int32(Some(100)))), + )); + for (input, predicate) in [ + (RecordBatch::new_empty(batch.schema()), None), + (batch.clone(), Some(id_is_100)), + ] { + let mut stream = scan_parquet_with_predicate( + &input, + Arc::clone(&required), + options.clone(), + predicate, + )?; + while let Some(read) = stream.next().await { + assert_eq!(read?.num_rows(), 0, "reverse={reverse}"); + } + } + } + Ok(()) + } + /// The same holds for a file whose only row group a filter prunes (#6506). #[tokio::test] async fn rejected_conversion_passes_for_pruned_row_group() -> Result<(), DataFusionError> { @@ -4380,11 +4448,9 @@ pub(crate) mod test { Ok(()) } - /// Clipping validates a selected container even when another claimant owns its value. - /// Losing leaf conversions do not install a decoder and must not add deferred failures. + /// Losing selections preserve both open-time shape errors and deferred leaf rejections. #[test] - fn losing_field_claims_preserve_open_errors_but_not_deferred_rejections( - ) -> Result<(), DataFusionError> { + fn losing_field_claims_preserve_open_and_deferred_errors() -> Result<(), DataFusionError> { let mut options = default_options(); options.use_field_id = true; for reverse in [false, true] { @@ -4422,7 +4488,14 @@ pub(crate) mod test { }; let result = check_conversion(&physical, &target, "s", &options); if matches!(losing_type, DataType::Int32) { - assert!(matches!(result?, ConversionCheck::Accept)); + assert!(matches!( + result?, + ConversionCheck::RejectOnNonEmpty { + column, + physical_type: DataType::Int64, + target_type: DataType::Int32, + } if column.ends_with(", A") + )); } else { let error = result .err() 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 28da7ad551d..57d0db3b9d6 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -33,6 +33,7 @@ import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.comet.CometNativeScanExec import org.apache.spark.sql.execution.FileSourceScanExec import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException import org.apache.spark.sql.functions.{array, col} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ @@ -1338,7 +1339,8 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper nullable = true, new MetadataBuilder().putLong("parquet.field.id", id.toLong).build()) - private def withFieldClaimFile(nested: Boolean)(f: (String, StructType) => Unit): Unit = { + private def withFieldClaimFile(nested: Boolean, empty: Boolean = false)( + f: (String, StructType) => Unit): Unit = { val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() def wrapSchema(fields: StructType): StructType = if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) @@ -1352,7 +1354,9 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { withTempPath { path => withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - val rows = (1L to 3L).map(value => if (nested) Row(Row(value)) else Row(value)) + val rows = + if (empty) Seq.empty[Row] + else (1L to 3L).map(value => if (nested) Row(Row(value)) else Row(value)) spark .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) .write @@ -1592,6 +1596,58 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } + for { + reverse <- Seq(false, true) + empty <- Seq(false, true) + } { + test( + s"Parquet field claims - losing leaf retains deferred rejection - reverse=$reverse empty=$empty") { + withFieldClaimFile(nested = true, empty = empty) { (path, schema) => + val fields = Seq(withFieldId("x", 1), StructField("a", IntegerType, nullable = true)) + val readType = StructType(if (reverse) fields.reverse else fields) + val readSchema = StructType(Seq(schema("s").copy(dataType = readType))) + def query: DataFrame = spark.read.schema(readSchema).parquet(path) + + def checkDeferredRejection(df: DataFrame): Unit = { + if (empty) { + checkAnswer(df, Seq.empty[Row]) + } else { + val error = intercept[Exception](df.collect()) + val messages = causeMessages(error) + assert( + causeChain(error).exists(_.isInstanceOf[SchemaColumnConvertNotSupportedException]), + messages) + assert(messages.contains("[s, A]") && messages.contains("INT64"), messages) + } + } + + withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") { + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val df = query + val scans = collect(df.queryExecution.executedPlan) { case scan: FileSourceScanExec => + scan + } + assert(scans.nonEmpty && scans.forall(_.supportsColumnar)) + checkDeferredRejection(df) + } + val df = query + assert( + collect(df.queryExecution.executedPlan) { case scan: CometNativeScanExec => + scan + }.nonEmpty, + s"expected native losing-leaf validation:\n${df.queryExecution.executedPlan}") + checkDeferredRejection(df) + } + + // parquet-mr has no deferred updater check on the losing leaf and returns NULL. + val expected = + if (empty) Seq.empty[Row] + else (1L to 3L).map(value => Row(if (reverse) Row(null, value) else Row(value, null))) + assert(fieldClaimReference(query).sortBy(_.toString) == expected.sortBy(_.toString)) + } + } + } + for { shape <- Seq("struct", "array", "map") reverse <- Seq(false, true) From 13f6234539149bb58845a371e2d93a7aa7373ec0 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sat, 31 Oct 2026 22:43:24 +0800 Subject: [PATCH 5/7] docs: clarify Parquet field claim differences between readers --- docs/source/user-guide/latest/parquet.md | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/docs/source/user-guide/latest/parquet.md b/docs/source/user-guide/latest/parquet.md index 5c9d4f7c1f7..74e94a57ae9 100644 --- a/docs/source/user-guide/latest/parquet.md +++ b/docs/source/user-guide/latest/parquet.md @@ -297,10 +297,6 @@ for production, so do not turn it off to avoid an entry on this page. differ only in such bytes distinct: they match Spark in a release build, and a debug build panics on top-level values ([#7032](https://github.com/apache/datafusion-comet/issues/7032), part of [#4764](https://github.com/apache/datafusion-comet/issues/4764)). -- **Field-id reads where one file column matches a field by id and a sibling by name.** With - `spark.sql.parquet.fieldId.read.enabled=true`, Comet returns NULL for the sibling at the top - level. Inside a struct, it returns the values of the field matched by id. Spark's own readers - disagree here ([#6994](https://github.com/apache/datafusion-comet/issues/6994)). - **`TIMESTAMP_MILLIS` values that overflow when converted to microseconds.** In nested fields, and in every column of a scan that has data filters, Comet returns NULL where Spark fails with `ArithmeticException: long overflow` @@ -386,6 +382,13 @@ With `spark.comet.write.parquet.enabled=true`: ### Differences that are not Comet bugs +- **Field-id reads where one file column matches a field by id and a sibling by name.** With + `spark.sql.parquet.fieldId.read.enabled=true`, when both siblings are requested and their types + are compatible with the file column, Comet's native reader gives the value to the id match and + returns NULL for the name match, both at the top level and inside structs. Spark's parquet-mr + reader does the same. Spark's vectorized reader can return zeros for the losing sibling with + on-heap column vectors, or uninitialized memory values with off-heap column vectors + ([#6994](https://github.com/apache/datafusion-comet/issues/6994)). - **Field-id reads with a read schema that swaps two nested field names.** With `spark.sql.parquet.fieldId.read.enabled=true`, Spark's vectorized reader pairs the file's columns with the read fields by the file's names, so it fails with a type mismatch or returns no rows. From 4503caf9d049d33ddc46537b57205531314358b7 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sat, 31 Oct 2026 23:06:10 +0800 Subject: [PATCH 6/7] refactor: simplify Parquet field claims and regression tests --- native/core/src/parquet/field_resolver.rs | 68 +------ native/core/src/parquet/parquet_support.rs | 42 ----- native/core/src/parquet/schema_adapter.rs | 101 +---------- .../comet/exec/CometNativeReaderSuite.scala | 167 +++++------------- 4 files changed, 49 insertions(+), 329 deletions(-) diff --git a/native/core/src/parquet/field_resolver.rs b/native/core/src/parquet/field_resolver.rs index 24bb636748b..b6e7f21b4ba 100644 --- a/native/core/src/parquet/field_resolver.rs +++ b/native/core/src/parquet/field_resolver.rs @@ -48,9 +48,8 @@ pub(crate) enum Selection<'a> { } /// The strongest claim on one file field. Root remapping and nested conversion give a file -/// field to this claimant only. Repeated id claims keep the last requested field, matching -/// Spark parquet-mr's `ParquetRowConverter` id-to-index `toMap`. Equal name claims keep the -/// earliest requested field, retaining the root remap's name-matching behavior. +/// field to this claimant only. Equal-strength claims keep the earliest requested field, +/// retaining the root remap's tie order. Repeated required ids are declined during scan planning. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum FieldClaim { /// The requested field selects this file field by Parquet field id. @@ -80,9 +79,7 @@ impl FieldClaim { } fn claim(&mut self, claim: Self) { - if claim.strength() > self.strength() - || matches!((*self, claim), (Self::IdMatch(_), Self::IdMatch(_))) - { + if claim.strength() > self.strength() { *self = claim; } } @@ -288,8 +285,8 @@ impl<'a> FieldResolver<'a> { } /// Resolve claims in file order, with id matches ahead of shadows and name matches. - /// Repeated requested ids keep the last claimant, as Spark parquet-mr does. Duplicate file ids are - /// checked across the whole requested group, including unprojected root fields (#7006). + /// Duplicate file ids are checked across the whole requested group, including unprojected + /// root fields (#7006). /// Name ambiguity is left to the caller: the root checks referenced columns only, whereas /// nested conversion validates every requested field in [`Self::resolve_all`]. pub(crate) fn file_claims(&self) -> DataFusionResult> { @@ -591,61 +588,6 @@ mod test { } } - #[test] - fn id_ties_keep_last_and_name_ties_keep_first() { - for use_field_id in [false, true] { - for reverse in [false, true] { - let mut requested = [ - field("x", Some(1)), - field("y", Some(1)), - field("Ab", None), - field("aB", None), - ]; - if reverse { - requested.reverse(); - } - let file = [field("p", Some(1)), field("AB", None)]; - with_resolver(&requested, &file, false, use_field_id, |resolver| { - let expected = match (reverse, use_field_id) { - (false, true) => vec![None, Some(0), Some(1), None], - (false, false) => vec![None, None, Some(1), None], - (true, true) => vec![Some(1), None, None, Some(0)], - (true, false) => vec![Some(1), None, None, None], - }; - assert_eq!(resolver.resolve_all().unwrap().into_owned(), expected); - }); - } - } - } - - #[test] - fn three_or_more_id_claims_keep_last_over_name_and_shadow() { - for count in [3, 4] { - for reverse in [false, true] { - let mut requested: Vec<_> = ["x", "y", "A", "z"][..count] - .iter() - .map(|name| field(name, Some(1))) - .collect(); - requested.insert(1, field("a", None)); - requested.push(field("b", None)); - let mut expected = vec![None; requested.len()]; - expected[count] = Some(0); - expected[count + 1] = Some(1); - if reverse { - requested.reverse(); - // Reversal changes the winning id claimant to x, now at the last position. - expected.fill(None); - expected[0] = Some(1); - expected[count + 1] = Some(0); - } - let file = [field("a", Some(1)), field("b", None)]; - with_resolver(&requested, &file, false, true, |resolver| { - assert_eq!(resolver.resolve_all().unwrap().into_owned(), expected); - }); - } - } - } - #[test] fn arbitration_does_not_hide_ambiguous_name_or_reorder_errors() { let file = [ diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 2314352c1e1..42d4630dde8 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -3040,46 +3040,4 @@ mod tests { .unwrap(); assert_eq!(col.value(0), 44); } - - #[test] - fn shadowed_struct_fields_follow_all_missing_null_policy() { - use super::{parquet_convert_array, SparkParquetOptions}; - use arrow::array::{Array, ArrayRef, Int32Array, StructArray}; - use arrow::buffer::NullBuffer; - use arrow::datatypes::{DataType, Field, Fields}; - use datafusion_comet_spark_expr::EvalMode; - use parquet::arrow::PARQUET_FIELD_ID_META_KEY; - use std::collections::HashMap; - use std::sync::Arc; - - let parent_nulls = NullBuffer::from(vec![true, false, true]); - let from: ArrayRef = Arc::new(StructArray::new( - Fields::from(vec![Field::new("b", DataType::Int32, true)]), - vec![Arc::new(Int32Array::from(vec![1, 2, 3]))], - Some(parent_nulls.clone()), - )); - let to_type = DataType::Struct(Fields::from(vec![ - Field::new("b", DataType::Int64, true).with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - "99".to_string(), - )])), - Field::new("B", DataType::Int32, true), - ])); - for null_struct in [false, true] { - let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); - options.use_field_id = true; - options.return_null_struct_if_all_fields_missing = null_struct; - let converted = parquet_convert_array(Arc::clone(&from), &to_type, &options).unwrap(); - let converted = converted.as_any().downcast_ref::().unwrap(); - assert!(converted - .columns() - .iter() - .all(|column| column.null_count() == 3)); - if null_struct { - assert_eq!(converted.null_count(), 3); - } else { - assert_eq!(converted.nulls(), Some(&parent_nulls)); - } - } - } } diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index cc50af694fd..ba0914cc535 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -123,7 +123,7 @@ fn is_pure_structural_narrowing( .iter() .map(|f| fold_name(f.name(), parquet_options.case_sensitive)) .collect::>()?; - let mut claimed_names = HashSet::with_capacity(target_fields.len()); + // Read schemas with duplicate sibling names are declined by the JVM scan checks. for target_field in target_fields { // DataFusion's retained `CastExpr` resolves struct fields by *exact* name, so // keeping it is only sound when Spark's configured resolver would pick the same @@ -139,11 +139,6 @@ fn is_pure_structural_narrowing( // Spark and Comet's converter reject it, but DataFusion's cast would silently // return the exact-case field, so the cast must not be retained. let folded_target = fold_name(target_field.name(), parquet_options.case_sensitive)?; - // The generic cast can give one source to several targets. Comet keeps only - // the earliest equal-strength name claim, so that shape needs its converter. - if !claimed_names.insert(folded_target.clone()) { - return Ok(false); - } let resolver_matches = source_folded .iter() .filter(|&f| f == &folded_target) @@ -3816,74 +3811,6 @@ pub(crate) mod test { Ok(()) } - /// Repeated root ids keep their existing last-wins behavior. Inside a struct, every - /// selected leaf still contributes its type check, including the losing id claims. - #[tokio::test] - async fn repeated_requested_ids_preserve_losing_field_validation() -> Result<(), DataFusionError> - { - for count in [2, 3, 4] { - for nested in [false, true] { - for reverse in [false, true] { - let values = Arc::new(Int64Array::from(vec![Some(11), Some(22), None])); - let physical = - Field::new("A", DataType::Int64, true).with_metadata(id_meta("1")); - let mut names = vec!["x", "y", "z", "w"]; - names.truncate(count); - if reverse { - names.reverse(); - } - let requested: Vec<_> = names - .iter() - .enumerate() - .map(|(index, name)| { - let data_type = if index + 1 == count { - DataType::Int64 - } else { - DataType::Int32 - }; - Field::new(*name, data_type, true).with_metadata(id_meta("1")) - }) - .collect(); - let (batch, required) = if nested { - ( - struct_batch(physical, Arc::clone(&values) as ArrayRef)?, - struct_schema(requested), - ) - } else { - ( - RecordBatch::try_new( - Arc::new(Schema::new(vec![physical])), - vec![Arc::clone(&values) as ArrayRef], - )?, - Arc::new(Schema::new(requested)), - ) - }; - let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); - options.use_field_id = true; - let mut stream = scan_parquet(&batch, required, options)?; - let read = stream.next().await.unwrap(); - if nested { - let error = read.expect_err("losing id claims must still be validated"); - assert!(error.to_string().contains("Column: [[s, A]]"), "{error}"); - continue; - } - let read = read?; - let columns = read.columns(); - for column in &columns[..count - 1] { - assert_eq!(column.data_type(), &DataType::Int32); - assert_eq!(column.null_count(), values.len()); - } - assert_eq!( - columns[count - 1].as_primitive::(), - values.as_ref() - ); - assert!(stream.next().await.is_none()); - } - } - } - Ok(()) - } - /// Disallowed widening (`INT32 -> bigint` with `allow_type_promotion` off) inside a /// struct defers to `RejectOnNonEmpty`, like the top level: a non-empty file fails ... #[tokio::test] @@ -5091,24 +5018,6 @@ pub(crate) mod test { ); } - /// Repeated requested ids keep the last claimant, matching Spark parquet-mr's - /// ParquetRowConverter. Equal name claims retain the earliest claimant. - #[test] - fn root_resolution_id_ties_keep_last_and_name_ties_keep_first() { - let logical = Arc::new(Schema::new(vec![ - Field::new("x", DataType::Int32, true).with_metadata(id_meta("1")), - Field::new("y", DataType::Int32, true).with_metadata(id_meta("1")), - Field::new("Ab", DataType::Int32, true), - Field::new("aB", DataType::Int32, true), - ])); - let physical = Arc::new(Schema::new(vec![ - Field::new("p", DataType::Int32, true).with_metadata(id_meta("1")), - Field::new("AB", DataType::Int32, true), - ])); - let (remapped, _name_map) = remap(&logical, &physical, false, true); - assert_eq!(field_names(&remapped), ["y", "Ab"]); - } - /// Field-id precedence in the case-insensitive duplicate check: an explicit `ω` (id 2) /// reading a file that holds both `ω` (id 2) and `Ω` (id 1) must resolve by id (Spark's /// `matchIdField` selects the id before ever comparing names) rather than raising a @@ -5577,14 +5486,6 @@ pub(crate) mod test { assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } - #[test] - fn is_pure_structural_narrowing_denies_conflicting_name_claims() { - let opts = default_options(); - let physical = struct_type(vec![("a", DataType::Int32), ("b", DataType::Int32)]); - let target = struct_type(vec![("a", DataType::Int32), ("a", DataType::Int32)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); - } - /// NTZ -> LTZ timestamp relabeling (INT96 reads) is a metadata-only reinterpretation /// Comet applies via a raw `reinterpret_cast`; Arrow's generic timestamp cast performs an /// actual timezone conversion instead, changing the value. Locked in as its own test 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 57d0db3b9d6..469af1f4db8 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -1524,57 +1524,25 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } - for { - nested <- Seq(false, true) - reverse <- Seq(false, true) - count <- Seq(2, 3, 4) - } { + for (nested <- Seq(false, true)) { test( - s"Parquet field claims - repeated requested Parquet ids match parquet-mr - nested=$nested reverse=$reverse count=$count") { - val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() - def wrapSchema(fields: StructType): StructType = - if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) - else fields - def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields - - val fields = Seq("x", "y", "z", "w").take(count).map(name => withFieldId(name, 1)) - val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else fields)) - val writeSchema = wrapSchema(StructType(Seq(withFieldId("A", 1)))) - val expected = Seq(11L, 22L, 33L).map(value => - wrapRow(Row.fromSeq(Seq.fill[Any](count - 1)(null) :+ value))) - withSQLConf( - SQLConf.CASE_SENSITIVE.key -> "false", - SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", - SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true", - SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") { - withTempPath { path => - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - val rows = Seq(11L, 22L, 33L).map(value => wrapRow(Row(value))) - spark - .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) - .write - .parquet(path.getCanonicalPath) - } - // ParquetRowConverter builds its requested-id-to-index map with toMap, so the last - // field with the repeated id owns the value. Check the Spark result in both orders. - val sparkRows = fieldClaimReference { - spark.read.schema(readSchema).parquet(path.getCanonicalPath) - } - assert(sparkRows.sortBy(_.toString) == expected.sortBy(_.toString)) - withSQLConf( - CometConf.COMET_ENABLED.key -> "true", - CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.key -> "true") { - val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) - // Keep the existing JVM guard: native matching is tested separately by the Rust - // Parquet roundtrip, which bypasses this guard and exercises the actual converter. - val plan = df.queryExecution.executedPlan - assert(collect(plan) { case scan: CometNativeScanExec => - scan - }.isEmpty) - val reasons = new ExtendedExplainInfo().getFallbackReasons(plan) - assert(reasons.exists(_.contains("duplicate Parquet field ids")), s"$reasons\n$plan") - assert(df.collect().toSeq.sortBy(_.toString) == sparkRows.sortBy(_.toString)) - } + s"Parquet field claims - repeated requested ids fall back to parquet-mr - nested=$nested") { + withFieldClaimFile(nested) { (path, _) => + val aliases = StructType(Seq(withFieldId("x", 1), withFieldId("y", 1))) + val schema = if (nested) new StructType().add("s", aliases) else aliases + def query: DataFrame = spark.read.schema(schema).parquet(path) + val expected = + (1L to 3L).map(value => if (nested) Row(Row(null, value)) else Row(null, value)) + assert(fieldClaimReference(query).sortBy(_.toString) == expected.sortBy(_.toString)) + withSQLConf( + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false", + CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.key -> "true") { + val df = query + val plan = df.queryExecution.executedPlan + assert(collect(plan) { case scan: CometNativeScanExec => scan }.isEmpty) + val reasons = new ExtendedExplainInfo().getFallbackReasons(plan) + assert(reasons.exists(_.contains("duplicate Parquet field ids")), s"$reasons\n$plan") + checkAnswer(df, expected) } } } @@ -1776,78 +1744,19 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } for { - nested <- Seq(false, true) - offheap <- Seq(false, true) - } { - test( - s"Parquet field claims - issue 6994 original reproduction - nested=$nested offheap=$offheap") { - val id1 = new MetadataBuilder().putLong("parquet.field.id", 1L).build() - val id10 = new MetadataBuilder().putLong("parquet.field.id", 10L).build() - def wrapSchema(fields: StructType): StructType = - if (nested) StructType(Seq(StructField("s", fields, nullable = true, id10))) - else fields - def wrapRow(fields: Row): Row = if (nested) Row(fields) else fields - - val writeSchema = - wrapSchema(StructType(Seq(StructField("A", IntegerType, nullable = true, id1)))) - val readSchema = wrapSchema( - StructType( - Seq( - StructField("x", IntegerType, nullable = true, id1), - StructField("a", IntegerType, nullable = true)))) - val expected = (1 to 3).map(value => wrapRow(Row(value, null))) - - withSQLConf( - SQLConf.CASE_SENSITIVE.key -> "false", - SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> "true", - SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true", - "spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true", - "spark.sql.columnVector.offheap.enabled" -> offheap.toString) { - withTempPath { path => - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - val rows = (1 to 3).map(value => wrapRow(Row(value))) - spark - .createDataFrame(spark.sparkContext.parallelize(rows, 1), writeSchema) - .write - .parquet(path.getCanonicalPath) - } - for ((reader, comet, vectorized) <- Seq( - ("Spark vectorized", false, true), - ("Spark parquet-mr", false, false), - ("Comet native", true, true))) { - withSQLConf( - CometConf.COMET_ENABLED.key -> comet.toString, - SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) { - val df = spark.read.schema(readSchema).parquet(path.getCanonicalPath) - val plan = df.queryExecution.executedPlan - val native = collect(plan) { case scan: CometNativeScanExec => scan }.nonEmpty - assert(native == comet, s"unexpected scan for $reader:\n$plan") - if (!comet) { - val scans = collect(plan) { case scan: FileSourceScanExec => scan } - assert(scans.nonEmpty && scans.forall(_.supportsColumnar == vectorized)) - } - // Only inspect the plan for Spark's vectorized reader: the losing field can - // contain stale memory with off-heap vectors. Comet and parquet-mr must return - // NULL for the losing name claimant. - if (comet || !vectorized) { - val rows = df.collect().toSeq - info( - s"Spark ${spark.version}, nested=$nested, offheap=$offheap, $reader: ${rows.mkString(", ")}") - assert(rows.sortBy(_.toString) == expected.sortBy(_.toString)) - } - } - } - } - } - } - } - - for { - shape <- Seq("root", "struct", "nested struct", "array", "map") + (shape, dataType) <- Seq( + ("root", LongType), + ("struct", LongType), + ("nested struct", LongType), + ("array", LongType), + ("map", LongType), + ("root", IntegerType), + ("struct", IntegerType)) reverse <- Seq(false, true) fileName <- Seq("A", "a") } { - test(s"Parquet field claims prefer id over name - $shape reverse=$reverse file=$fileName") { + test( + s"Parquet field claims prefer id over name - $shape $dataType reverse=$reverse file=$fileName") { def wrapSchema(fields: StructType): StructType = shape match { case "root" => fields case "struct" => new StructType().add("s", fields) @@ -1865,10 +1774,16 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper case "map" => Row(Map("value" -> fields, "null" -> null)) } - val values = Seq[java.lang.Long](11L, null, 33L, 44L, 55L) + val values: Seq[Any] = dataType match { + case IntegerType => Seq[java.lang.Integer](11, null, 33, 44, 55) + case LongType => Seq[java.lang.Long](11L, null, 33L, 44L, 55L) + } val parentNulls = if (shape == "root") Seq.empty else Seq(Row(null)) - val writeSchema = wrapSchema(StructType(Seq(withFieldId(fileName, 1)))) - val fields = Seq(withFieldId("x", 1), StructField("a", LongType, nullable = true)) + val writeSchema = + wrapSchema(StructType(Seq(withFieldId(fileName, 1).copy(dataType = dataType)))) + val fields = Seq( + withFieldId("x", 1).copy(dataType = dataType), + StructField("a", dataType, nullable = true)) val readSchema = wrapSchema(StructType(if (reverse) fields.reverse else fields)) withSQLConf( // Cover folded matching for A and exact, case-sensitive matching for a. @@ -1895,8 +1810,12 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper val fields = if (useFieldId) Seq(value, null) else Seq(null, value) wrapRow(Row.fromSeq(if (reverse) fields.reverse else fields)) } ++ parentNulls - // Spark's vectorized reader returns zeros or stale values for this conflict - // (#6994). Pin the intended rows directly rather than using it as the oracle. + // Spark's vectorized reader can return zeros or stale values for this conflict + // (#6994). Compare with parquet-mr, whose losing field is defined to be NULL. + val reference = fieldClaimReference { + spark.read.schema(readSchema).parquet(path.getCanonicalPath) + } + assert(reference.sortBy(_.toString) == expected.sortBy(_.toString)) checkAnswer(df, expected) } } From 23cfe0a17969236c423801c096697b42789896c4 Mon Sep 17 00:00:00 2001 From: sam-1112 Date: Sat, 31 Oct 2026 23:40:48 +0800 Subject: [PATCH 7/7] perf: skip Parquet claim arbitration for ID-only groups Clarify field ID resolution comments to match the existing behavior. --- native/core/src/parquet/field_resolver.rs | 96 +++++++++++++++++++--- native/core/src/parquet/parquet_support.rs | 6 +- native/core/src/parquet/schema_adapter.rs | 5 +- 3 files changed, 92 insertions(+), 15 deletions(-) diff --git a/native/core/src/parquet/field_resolver.rs b/native/core/src/parquet/field_resolver.rs index b6e7f21b4ba..539661d354f 100644 --- a/native/core/src/parquet/field_resolver.rs +++ b/native/core/src/parquet/field_resolver.rs @@ -90,7 +90,9 @@ impl FieldClaim { #[derive(Debug)] pub(crate) struct FieldMatches { selected: Vec>, - claims: Vec, + /// All-id groups need no arbitration: repeated required ids are declined by scan planning, + /// and only a name selection can lose to an id match or a shadow. + claims: Option>, } impl FieldMatches { @@ -99,17 +101,23 @@ impl FieldMatches { self.selected.iter().enumerate().map(|(requested, file)| { ( *file, - file.is_some_and(|index| self.claims[index].owned_by(requested)), + file.is_some_and(|index| { + self.claims + .as_ref() + .is_none_or(|claims| claims[index].owned_by(requested)) + }), ) }) } /// Reuse the selection vector for runtime conversion, masking only the losing fields. pub(crate) fn into_owned(mut self) -> Vec> { - for (requested, file) in self.selected.iter_mut().enumerate() { - if let Some(index) = *file { - if !self.claims[index].owned_by(requested) { - *file = None; + if let Some(claims) = self.claims { + for (requested, file) in self.selected.iter_mut().enumerate() { + if let Some(index) = *file { + if !claims[index].owned_by(requested) { + *file = None; + } } } } @@ -318,13 +326,28 @@ impl<'a> FieldResolver<'a> { /// Select each nested field once and record claims from that selection. Validate selections /// in requested order so arbitration cannot hide ambiguity or reorder selection errors. /// The caller can validate losing selections before masking them with [`FieldMatches::into_owned`]. + /// Defer claims until the first name selection: an all-id group has no competing claims + /// after scan planning has excluded repeated requested ids. File-side ambiguity is still + /// checked by every [`Self::select`] call. pub(crate) fn resolve_all(&self) -> DataFusionResult { - let mut selected = Vec::with_capacity(self.requested.len()); - let mut claims = vec![FieldClaim::Unchanged; self.file.len()]; + let mut selected: Vec> = Vec::with_capacity(self.requested.len()); + let mut claims = None; for requested in 0..self.requested.len() { let selection = self.select(requested)?; - selected.push(self.resolve_selection(requested, selection)?); - self.record_claims(&mut claims, requested, selection); + let file = self.resolve_selection(requested, selection)?; + if matches!(selection, Selection::ByName(_)) && claims.is_none() { + let mut group_claims = vec![FieldClaim::Unchanged; self.file.len()]; + // Earlier selections were all by id. Reuse their validated indices instead + // of selecting them again, including the shadows a name claimant may meet. + for (previous, &file) in selected.iter().enumerate() { + self.record_claims(&mut group_claims, previous, Selection::ById(file)); + } + claims = Some(group_claims); + } + if let Some(claims) = &mut claims { + self.record_claims(claims, requested, selection); + } + selected.push(file); } Ok(FieldMatches { selected, claims }) } @@ -558,6 +581,59 @@ mod test { }); } + #[test] + fn id_only_group_preserves_missing_fields_and_file_ambiguity() { + let file = [ + field("y", Some(1)), + field("x", Some(2)), + field("m", Some(4)), + ]; + let requested = [ + field("x", Some(1)), + field("m", Some(3)), + field("y", Some(2)), + ]; + with_resolver(&requested, &file, false, true, |resolver| { + let matches = resolver.resolve_all().unwrap(); + assert_eq!( + matches.iter().collect::>(), + vec![(Some(0), true), (None, false), (Some(1), true)] + ); + assert_eq!(matches.into_owned(), vec![Some(0), None, Some(1)]); + }); + let ambiguous = [field("a", Some(1)), field("b", Some(1))]; + with_resolver(&requested, &ambiguous, false, true, |resolver| { + assert_eq!( + resolver.resolve_all().unwrap_err().to_string(), + resolver.select(0).unwrap_err().to_string() + ); + }); + } + + #[test] + fn name_selection_after_id_prefix_preserves_independent_validation() { + let file = [field("A", Some(1)), field("B", Some(2))]; + for reverse in [false, true] { + let mut requested = [field("x", Some(1)), field("y", Some(2)), field("a", None)]; + let mut expected = vec![(Some(0), true), (Some(1), true), (Some(0), false)]; + if reverse { + requested.reverse(); + expected.reverse(); + } + with_resolver(&requested, &file, false, true, |resolver| { + let matches = resolver.resolve_all().unwrap(); + assert_eq!(matches.iter().collect::>(), expected); + assert_eq!( + matches.into_owned(), + expected + .iter() + .map(|&(file, owns)| if owns { file } else { None }) + .collect::>() + ); + }); + } + } + #[test] fn shadowing_respects_case_sensitivity_and_a_stronger_id_claim() { // A's id resolves elsewhere, but its name still shadows a name-only claimant of `a`. diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 42d4630dde8..aee6f725da8 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -101,9 +101,9 @@ pub struct SparkParquetOptions { /// non-null parents keep their structure with all-null leaves. Comet sets it to true /// on Spark 3.4 through 4.0, which always use the legacy behavior. pub return_null_struct_if_all_fields_missing: bool, - /// When true, resolve fields by parquet.field.id metadata instead of name - /// (mirrors Spark's `spark.sql.parquet.fieldId.read.enabled`). Only takes effect - /// when both physical and logical fields actually carry IDs. + /// When true, requested fields carrying parquet.field.id metadata resolve only by id; + /// an absent file id never falls back to a name match. Requested fields without ids + /// still resolve by name (mirrors Spark's `spark.sql.parquet.fieldId.read.enabled`). pub use_field_id: bool, /// Whether type promotion (schema evolution) is allowed, e.g. INT32 -> INT64, /// FLOAT -> DOUBLE. Mirrors spark.comet.schemaEvolution.enabled. diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index ba0914cc535..7a131101665 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -204,8 +204,9 @@ fn rename_to_logical( } /// Remap physical schema field names to match logical schema field names. Mirrors Spark's -/// `clipParquetGroupFields`: prefer ID match for any logical field that carries a -/// `PARQUET:field_id`, fall back to case-insensitive name match otherwise. `resolver` resolves +/// `clipParquetGroupFields`: when id matching is enabled, a logical field carrying +/// `PARQUET:field_id` resolves only by id, never falling back to name if the id is missing. +/// Other fields resolve by name using the configured case sensitivity. `resolver` resolves /// the logical root fields against the physical ones. /// /// The remap only changes top-level field NAMES so that `DefaultPhysicalExprAdapter`'s