diff --git a/native/core/src/parquet/name_fold.rs b/native/core/src/parquet/name_fold.rs index 00626cb216..46d35243ae 100644 --- a/native/core/src/parquet/name_fold.rs +++ b/native/core/src/parquet/name_fold.rs @@ -39,25 +39,36 @@ use std::sync::{OnceLock, RwLock}; /// Pure-ASCII names are folded inline with `to_ascii_lowercase`: for an all-ASCII string this is /// provably identical to Java's `toLowerCase(Locale.ROOT)` (`Locale.ROOT` excludes the /// Turkish/Lithuanian rules, no ASCII codepoint lowercases to a non-ASCII one, and `Final_Sigma` -/// needs a sigma to fire), so it keeps the lock, the JVM crossing, and the fallback off the hot +/// needs a sigma to fire), so it keeps the lock and the JVM crossing off the hot /// path -- for almost every real schema every name is ASCII. /// /// Non-ASCII names are delegated to the JVM (`CometSchemaUtils.toLowerCaseRoot`) so Comet folds /// exactly as Spark does. Those folds are memoized process-wide (see [`fold_cache`]); the same /// field names recur across every batch and file, so this is one JVM crossing per distinct -/// non-ASCII name for the life of the process. Outside a Comet task there is no attached JVM (e.g. -/// Rust unit tests) or the JNI call fails, so fall back to Rust's own Unicode fold (see -/// [`fold_uncached`]). -pub(crate) fn fold_names(names: &[&str], case_sensitive: bool) -> Vec { +/// non-ASCII name until the cache fills. Returns owned names in input order. A missing JVM or JNI +/// failure is returned to the caller without caching any new folds: Rust's Unicode table can +/// differ from the JVM's and must never be used to continue a native scan after a failure. +pub(crate) fn fold_names(names: &[&str], case_sensitive: bool) -> DataFusionResult> { + fold_names_with(names, case_sensitive, jvm_fold_all) +} + +/// Apply the shared fast paths and cache using `fold` for non-ASCII cache misses. The callback +/// returns one owned fold per name in input order, or an error that aborts the entire batch. +/// Passing the JVM operation explicitly lets tests inject failures without process-wide hooks. +fn fold_names_with( + names: &[&str], + case_sensitive: bool, + fold: impl FnOnce(&[&str]) -> DataFusionResult>, +) -> DataFusionResult> { if case_sensitive { - return names.iter().map(|n| n.to_string()).collect(); + return Ok(names.iter().map(|n| n.to_string()).collect()); } // ASCII fast path: for an all-ASCII batch (the overwhelmingly common case) fold inline with no // `Option` buffer, no cache, and no JVM crossing. `to_ascii_lowercase` is provably identical to // Java's `toLowerCase(Locale.ROOT)` for ASCII (see the doc above). if names.iter().all(|n| n.is_ascii()) { - return names.iter().map(|n| n.to_ascii_lowercase()).collect(); + return Ok(names.iter().map(|n| n.to_ascii_lowercase()).collect()); } // Mixed batch: fold the ASCII names inline and route the non-ASCII ones through the cache/JVM. @@ -71,17 +82,24 @@ pub(crate) fn fold_names(names: &[&str], case_sensitive: bool) -> Vec { } } // `non_ascii_positions` is non-empty here (the all-ASCII case returned above). - fold_non_ascii(names, &non_ascii_positions, &mut result); + fold_non_ascii(names, &non_ascii_positions, &mut result, fold)?; - result + Ok(result .into_iter() .map(|folded| folded.expect("every name folded")) - .collect() + .collect()) } /// Fold the non-ASCII `names` at `positions`, writing each fold into `result`. Split out of /// [`fold_names`] so the common all-ASCII path stays a tight loop with no cache or JVM machinery. -fn fold_non_ascii(names: &[&str], positions: &[usize], result: &mut [Option]) { +/// Cached entries may be copied before `fold` fails, but misses are published only after the +/// whole callback succeeds. Callers must discard the partial output on error. +fn fold_non_ascii( + names: &[&str], + positions: &[usize], + result: &mut [Option], + fold: impl FnOnce(&[&str]) -> DataFusionResult>, +) -> DataFusionResult<()> { let cache = fold_cache(); let mut miss_positions: Vec = Vec::new(); { @@ -95,12 +113,12 @@ fn fold_non_ascii(names: &[&str], positions: &[usize], result: &mut [Option = miss_positions.iter().map(|&i| names[i]).collect(); - let (folded, cacheable) = fold_uncached(&miss_names); - if cacheable { + let folded = fold(&miss_names)?; + { let mut write = cache.write().unwrap(); for (pos, &i) in miss_positions.iter().enumerate() { // Bound the process-wide cache: an executor JVM is long-lived, so stop inserting @@ -115,27 +133,33 @@ fn fold_non_ascii(names: &[&str], positions: &[usize], result: &mut [Option String { +/// Returns an owned name, propagating JVM failures without substituting a different Unicode fold. +pub(crate) fn fold_name(name: &str, case_sensitive: bool) -> DataFusionResult { if case_sensitive { - return name.to_string(); + return Ok(name.to_string()); } if name.is_ascii() { - return name.to_ascii_lowercase(); + return Ok(name.to_ascii_lowercase()); } - let mut folded = fold_names(&[name], false); - folded + let mut folded = fold_names(&[name], false)?; + Ok(folded .pop() - .expect("fold_names returns one entry per input name") + .expect("fold_names returns one entry per input name")) } -/// Fold every field name in `schema`. See [`fold_names`]. -pub(crate) fn fold_schema_names(schema: &SchemaRef, case_sensitive: bool) -> Vec { +/// Return owned folds for every field name in `schema`, preserving order and propagating JVM +/// failures. Does not modify the schema. See [`fold_names`]. +pub(crate) fn fold_schema_names( + schema: &SchemaRef, + case_sensitive: bool, +) -> DataFusionResult> { let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect(); fold_names(&names, case_sensitive) } @@ -154,37 +178,21 @@ fn fold_cache() -> &'static RwLock> { CACHE.get_or_init(|| RwLock::new(HashMap::new())) } -/// Fold names that missed the cache. These are always non-ASCII (ASCII names never reach here). -/// Returns the folds and whether they may be cached: a transient JVM failure returns the fallback -/// but is NOT cached, so it cannot poison later lookups once the JVM recovers. -/// -/// The fallback is Rust's `str::to_lowercase` (full Unicode), not `to_ascii_lowercase`: ASCII -/// folding leaves every non-ASCII cased letter untouched (~1367 mismatches against JDK 17), -/// whereas Rust's Unicode fold only differs from the JVM on the handful of codepoints where the -/// JDK's Unicode table version differs from Rust's (~95 against JDK 17). It also lets the -/// no-JVM path resolve names whose case mapping is stable across Unicode versions (e.g. `Ω`/`ω`, -/// `MÜNCHEN`), which is what the Rust unit tests rely on. -fn fold_uncached(names: &[&str]) -> (Vec, bool) { - if crate::JAVA_VM.get().is_some() { - match jvm_fold_all(names) { - Ok(folded) => return (folded, true), - Err(e) => log::warn!( - "JVM case-fold failed; falling back to Rust Unicode lowercasing, which can differ \ - from Spark on codepoints where the JDK's Unicode table version differs: {e}" - ), - } - (names.iter().map(|n| n.to_lowercase()).collect(), false) - } else { - // No attached JVM (e.g. Rust unit tests): Rust's Unicode fold is the permanent mode and is - // cacheable. - (names.iter().map(|n| n.to_lowercase()).collect(), true) - } -} - /// Lower-case a batch of names via the JVM's `String.toLowerCase(Locale.ROOT)`, matching Spark's /// `ParquetReadSupport` byte-for-byte. Folds in chunks so at most `CHUNK * 2` JNI local refs are /// live in a single frame, keeping wide schemas within `DEFAULT_LOCAL_FRAME_CAPACITY` (32). +/// Returns owned strings only after every chunk succeeds; missing JVM and JNI errors propagate. fn jvm_fold_all(names: &[&str]) -> DataFusionResult> { + if crate::JAVA_VM.get().is_none() { + // Standalone Rust unit tests exercise stable Unicode mappings without starting Spark. + // This substitute is absent from production builds and cannot handle a JNI failure. + #[cfg(test)] + return Ok(names.iter().map(|name| name.to_lowercase()).collect()); + #[cfg(not(test))] + return Err(DataFusionError::Execution( + "JVM is not initialized for Parquet field-name folding".to_string(), + )); + } const CHUNK: usize = 16; let mut folded = Vec::with_capacity(names.len()); for chunk in names.chunks(CHUNK) { @@ -219,7 +227,7 @@ fn jvm_fold_all(names: &[&str]) -> DataFusionResult> { #[cfg(test)] mod test { /// The process-wide fold cache populates on first fold of a non-ASCII name and serves repeated - /// lookups. Runs under the Rust Unicode fallback (no JVM in `cargo test`), which exercises the + /// lookups. Runs under the test-only Rust Unicode substitute, which exercises the /// cacheable path. Uses a non-ASCII name because ASCII names take the inline fast path and are /// intentionally never cached. #[test] @@ -227,7 +235,7 @@ mod test { // Unique non-ASCII name so parallel tests don't share this cache entry. let name = "ΩFoldMemoUnique"; let folded = name.to_lowercase(); - let first = super::fold_names(&[name], false); + let first = super::fold_names(&[name], false).unwrap(); assert_eq!(first, vec![folded.clone()]); assert_eq!( super::fold_cache() @@ -238,7 +246,7 @@ mod test { Some(folded.as_str()) ); // Second call is served from the cache with the same result. - assert_eq!(super::fold_names(&[name], false), first); + assert_eq!(super::fold_names(&[name], false).unwrap(), first); } /// ASCII names take the inline fast path and must never touch the cache. @@ -246,7 +254,7 @@ mod test { fn fold_names_ascii_is_fast_path_and_uncached() { let name = "FoldAsciiUnique"; assert_eq!( - super::fold_names(&[name], false), + super::fold_names(&[name], false).unwrap(), vec!["foldasciiunique".to_string()] ); assert!(super::fold_cache().read().unwrap().get(name).is_none()); @@ -256,7 +264,83 @@ mod test { #[test] fn fold_names_case_sensitive_is_identity_and_uncached() { let name = "ΩFoldCaseSensitiveUnique"; - assert_eq!(super::fold_names(&[name], true), vec![name.to_string()]); + assert_eq!( + super::fold_names(&[name], true).unwrap(), + vec![name.to_string()] + ); assert!(super::fold_cache().read().unwrap().get(name).is_none()); } + + /// A failed JNI fold must abort the mixed batch without caching Rust's different Unicode + /// mapping; a later successful JVM fold must retain the distinct names and become cacheable. + #[test] + fn fold_names_propagates_jni_failure_without_caching_fallback() { + // Unique suffixes keep other parallel tests from populating these cache entries. JDK 17 + // leaves U+A7DC unchanged, while Rust lowercases it to U+019B and would falsely match. + let file_name = "Ƛ_jni_failure_unique"; + let requested_name = "ƛ_jni_failure_unique"; + let names = ["ASCII", file_name, requested_name]; + assert_eq!(file_name.to_lowercase(), requested_name); + { + let cache = super::fold_cache().read().unwrap(); + assert!(!cache.contains_key(file_name)); + assert!(!cache.contains_key(requested_name)); + } + + let error = super::fold_names_with(&names, false, |misses| { + assert_eq!(misses, &[file_name, requested_name]); + Err(super::DataFusionError::Execution( + "injected JNI fold failure".to_string(), + )) + }) + .unwrap_err(); + assert!(matches!( + error, + super::DataFusionError::Execution(message) if message == "injected JNI fold failure" + )); + { + let cache = super::fold_cache().read().unwrap(); + assert!(!cache.contains_key(file_name)); + assert!(!cache.contains_key(requested_name)); + } + + let expected = vec![ + "ascii".to_string(), + file_name.to_string(), + requested_name.to_string(), + ]; + let recovered = super::fold_names_with(&names, false, |misses| { + assert_eq!(misses, &[file_name, requested_name]); + // Simulate a recovered JDK 17 operation, whose Unicode table keeps this pair distinct. + Ok(misses.iter().map(|name| name.to_string()).collect()) + }) + .unwrap(); + assert_eq!(recovered, expected); + assert_eq!( + super::fold_names_with(&names, false, |_| panic!("cached folds must skip JNI")) + .unwrap(), + expected + ); + } + + /// ASCII folding and case-sensitive identity require no JVM, so even a failing JNI operation + /// must remain unused and the names must be returned with the appropriate local semantics. + #[test] + fn fold_names_fast_paths_skip_failing_jni_operation() { + for (names, case_sensitive, expected) in [ + (["ASCII", "MixedCase"], false, ["ascii", "mixedcase"]), + (["Ƛ", "ƛ"], true, ["Ƛ", "ƛ"]), + ] { + let mut called = false; + let folded = super::fold_names_with(&names, case_sensitive, |_| { + called = true; + Err(super::DataFusionError::Execution( + "unexpected JNI operation".to_string(), + )) + }) + .unwrap(); + assert!(!called); + assert_eq!(folded, expected); + } + } } diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 60a1027f97..1ae9d47dd5 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -120,8 +120,8 @@ pub(crate) fn init_datasource_exec( // Fold the data and required field names once (the same JVM `toLowerCase(Locale.ROOT)` // fold the schema adapter uses), then match on the folded names so this plan-time // projection stays consistent with the adapter's case-insensitive remap. - let data_folded = fold_schema_names(schema, case_sensitive); - let required_folded = fold_schema_names(&required_schema, case_sensitive); + let data_folded = fold_schema_names(schema, case_sensitive)?; + let required_folded = fold_schema_names(&required_schema, case_sensitive)?; let projection: Vec = required_folded .iter() .filter_map(|req| data_folded.iter().position(|d| d == req)) diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index c815944c46..1964f53174 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -445,7 +445,7 @@ pub(crate) fn match_struct_fields( let mut all_names: Vec<&str> = Vec::with_capacity(from_fields.len() + to_fields.len()); all_names.extend(from_fields.iter().map(|f| f.name().as_str())); all_names.extend(to_fields.iter().map(|f| f.name().as_str())); - let all_folded = fold_names(&all_names, parquet_options.case_sensitive); + let all_folded = fold_names(&all_names, parquet_options.case_sensitive)?; let (from_folded, to_folded) = all_folded.split_at(from_fields.len()); // Group file field indices by folded name so a case-insensitive collision is detected diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 5a4b3e1630..36a3af671f 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -111,8 +111,8 @@ fn is_pure_structural_narrowing( physical_type: &DataType, target_type: &DataType, parquet_options: &SparkParquetOptions, -) -> bool { - match (physical_type, target_type) { +) -> DataFusionResult { + Ok(match (physical_type, target_type) { (DataType::Struct(source_fields), DataType::Struct(target_fields)) => { // Comet matches by Parquet field id first when the target carries one; // DataFusion's generic cast has no field-id concept, so any field-id-bearing @@ -120,15 +120,15 @@ fn is_pure_structural_narrowing( if parquet_options.use_field_id && target_fields.iter().any(|f| parse_field_id(f).is_some()) { - return false; + return Ok(false); } // Fold the source field names once (O(sources), not O(targets x sources)), matching // this file's bulk-fold convention. let source_folded: Vec = source_fields .iter() .map(|f| fold_name(f.name(), parquet_options.case_sensitive)) - .collect(); - target_fields.iter().all(|target_field| { + .collect::>()?; + 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 // single source field. Two requirements: @@ -142,23 +142,29 @@ fn is_pure_structural_narrowing( // resolver. `struct` projecting `id` case-insensitively is ambiguous; // 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); + let folded_target = fold_name(target_field.name(), parquet_options.case_sensitive)?; let resolver_matches = source_folded .iter() .filter(|&f| f == &folded_target) .count(); - resolver_matches == 1 - && source_fields - .iter() - .find(|f| f.name() == target_field.name()) - .is_some_and(|source_field| { - is_pure_structural_narrowing( - source_field.data_type(), - target_field.data_type(), - parquet_options, - ) - }) - }) + if resolver_matches != 1 { + return Ok(false); + } + let Some(source_field) = source_fields + .iter() + .find(|f| f.name() == target_field.name()) + else { + return Ok(false); + }; + if !is_pure_structural_narrowing( + source_field.data_type(), + target_field.data_type(), + parquet_options, + )? { + return Ok(false); + } + } + true } (DataType::List(source_item), DataType::List(target_item)) | (DataType::LargeList(source_item), DataType::LargeList(target_item)) => { @@ -166,7 +172,7 @@ fn is_pure_structural_narrowing( source_item.data_type(), target_item.data_type(), parquet_options, - ) + )? } // Map is excluded structurally, not by the equality check below: `replace_with_spark_cast` // only reaches this predicate after its own top-level `physical_type == target_type` @@ -179,7 +185,7 @@ fn is_pure_structural_narrowing( // value or matching semantics that `nested_struct::cast_column` does not replicate, // and none of them arise from pruning alone. _ => physical_type == target_type, - } + }) } /// Remap physical schema field names to match logical schema field names. Mirrors Spark's @@ -253,8 +259,8 @@ fn remap_physical_schema( // JVM) so the O(physical x logical) name matching below compares pre-folded strings instead of // crossing into the JVM for every pair. Mirrors Spark's `caseInsensitiveParquetFieldMap`, which // groups file fields by their folded name a single time. - let logical_folded = fold_schema_names(logical_schema, case_sensitive); - let physical_folded = fold_schema_names(physical_schema, case_sensitive); + let logical_folded = fold_schema_names(logical_schema, case_sensitive)?; + let physical_folded = fold_schema_names(physical_schema, case_sensitive)?; // 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. @@ -865,7 +871,7 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { // the `folded_to_indices` map the nested convert builds in `parquet_support`, so // both paths detect ambiguity the same way instead of drifting. let original_physical_dup_check = if !case_sensitive { - let folded = fold_schema_names(&physical_file_schema, false); + let folded = fold_schema_names(&physical_file_schema, false)?; let mut map: HashMap> = HashMap::new(); for (i, folded_name) in folded.into_iter().enumerate() { map.entry(folded_name).or_default().push(i); @@ -889,8 +895,8 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { // Fold both schemas once here so the per-column rewrite paths reuse them instead of // re-folding on every `rewrite` call. Case-sensitive mode folds to identity. - let logical_folded = fold_schema_names(&logical_file_schema, case_sensitive); - let physical_folded = fold_schema_names(&adapted_physical_schema, case_sensitive); + let logical_folded = fold_schema_names(&logical_file_schema, case_sensitive)?; + let physical_folded = fold_schema_names(&adapted_physical_schema, case_sensitive)?; // Folded names of logical fields that resolve by Parquet field id. Spark's `matchIdField` // selects these by id before comparing names, so the case-insensitive duplicate check must @@ -992,7 +998,7 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter { Ok(Transformed::no(e)) }); let col_refs: Vec<&str> = col_names.iter().map(|s| s.as_str()).collect(); - let col_folded = fold_names(&col_refs, false); + let col_folded = fold_names(&col_refs, false)?; for (name, folded) in col_names.iter().zip(&col_folded) { // Fields resolved by Parquet field id are selected by id before names are // compared, so an id-resolved column must not trip the name-ambiguity check @@ -1111,7 +1117,7 @@ impl SparkPhysicalExprAdapter { expr.transform(|e| { if let Some(column) = e.downcast_ref::() { let col_name = column.name(); - let col_folded = fold_name(col_name, case_sensitive); + let col_folded = fold_name(col_name, case_sensitive)?; // Resolve fields by name because this is the fallback path // that runs on the original expression when the default @@ -1289,7 +1295,7 @@ impl SparkPhysicalExprAdapter { // alongside `test::nested_struct_narrowing_cast_matches_datafusion_generic_cast` // for the Struct-pruning shape that always does route through // `nested_struct::cast_column`. - if is_pure_structural_narrowing(physical_type, target_type, &self.parquet_options) { + if is_pure_structural_narrowing(physical_type, target_type, &self.parquet_options)? { // Not an assertion that `nested_struct::cast_column` specifically runs (it may // not, see above), only that *some* DataFusion cast path can actually perform // this pair, so a bug in the predicate surfaces here at plan time instead of @@ -1366,7 +1372,8 @@ impl SparkPhysicalExprAdapter { Ok(Transformed::no(expr)) } - /// Replace references to missing columns with default values. + /// Replace missing columns with defaults cast to the logical schema, propagating conversion + /// errors rather than retaining an unconverted value whose fields could not be matched. fn replace_missing_with_defaults( &self, expr: Arc, @@ -1382,45 +1389,33 @@ impl SparkPhysicalExprAdapter { // Build owned (column_name, default_value) pairs for columns missing from the physical file. // For each default: filter to only columns absent from physical schema, then type-cast // the value to match the logical schema's field type if they differ (using Spark cast semantics). - let case_sensitive = self.parquet_options.case_sensitive; - // Physical schema names were folded once in `create()`; reuse them here. - let physical_folded = &self.physical_folded; - let missing_column_defaults: Vec<(String, ScalarValue)> = defaults - .iter() - .filter_map(|(col, val)| { - let col_name = col.name(); - let col_folded = fold_name(col_name, case_sensitive); - - // Only include defaults for columns missing from the physical file schema - let is_missing = !physical_folded.iter().any(|f| f == &col_folded); + let mut missing_column_defaults = Vec::new(); + for (col, val) in defaults { + let col_name = col.name(); + let col_folded = fold_name(col_name, self.parquet_options.case_sensitive)?; + + // Only include defaults for columns missing from the physical file schema. + if self.physical_folded.iter().any(|name| name == &col_folded) { + continue; + } - if !is_missing { - return None; + let mut value = val.clone(); + if let Some(field) = self + .logical_file_schema + .field_with_name(col_name) + .ok() + .filter(|field| val.data_type() != *field.data_type()) + { + if let ColumnarValue::Scalar(converted) = spark_parquet_convert( + ColumnarValue::Scalar(value.clone()), + field.data_type(), + &self.parquet_options, + )? { + value = converted; } - - // Cast value to logical schema type if needed (only if types differ) - let value = self - .logical_file_schema - .field_with_name(col_name) - .ok() - .filter(|field| val.data_type() != *field.data_type()) - .and_then(|field| { - spark_parquet_convert( - ColumnarValue::Scalar(val.clone()), - field.data_type(), - &self.parquet_options, - ) - .ok() - .and_then(|cv| match cv { - ColumnarValue::Scalar(s) => Some(s), - _ => None, - }) - }) - .unwrap_or_else(|| val.clone()); - - Some((col_name.to_string(), value)) - }) - .collect(); + } + missing_column_defaults.push((col_name.to_string(), value)); + } let name_based: HashMap<&str, &ScalarValue> = missing_column_defaults .iter() @@ -1556,7 +1551,7 @@ mod test { DataType, Field, Fields, Int64Type, Schema, TimeUnit, TimestampMicrosecondType, }; use arrow::record_batch::RecordBatch; - use datafusion::common::DataFusionError; + use datafusion::common::{DataFusionError, ScalarValue}; use datafusion::datasource::listing::PartitionedFile; use datafusion::datasource::physical_plan::{FileGroup, FileScanConfigBuilder, ParquetSource}; use datafusion::datasource::source::DataSourceExec; @@ -3019,8 +3014,53 @@ mod test { ); } + /// Errors while matching nested default fields must abort rewriting instead of inserting the + /// original, unconverted struct literal for a missing column. + #[test] + fn missing_struct_default_propagates_nested_conversion_error() -> Result<(), DataFusionError> { + let logical = Arc::new(Schema::new(vec![Field::new( + "missing", + DataType::Struct(vec![Field::new("résumé", DataType::Int32, true)].into()), + true, + )])); + let physical = Arc::new(Schema::new(vec![Field::new( + "present", + DataType::Int32, + true, + )])); + let default = ScalarValue::Struct(Arc::new(StructArray::try_new( + vec![ + Field::new("RÉSUMÉ", DataType::Int32, true), + Field::new("résumé", DataType::Int32, true), + ] + .into(), + vec![ + Arc::new(Int32Array::from(vec![10])) as ArrayRef, + Arc::new(Int32Array::from(vec![20])) as ArrayRef, + ], + None, + )?)); + let defaults = HashMap::from([(Column::new("missing", 0), default)]); + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + options.case_sensitive = false; + let adapter = SparkPhysicalExprAdapterFactory::new(options, Some(defaults)) + .create(logical, physical)?; + + let error = adapter + .rewrite(Arc::new(Column::new("missing", 0))) + .expect_err("ambiguous nested defaults must not become literals"); + let message = error.to_string(); + assert!( + message.contains("Found duplicate field") + && message.contains("RÉSUMÉ") + && message.contains("résumé"), + "expected nested duplicate-field error, got: {message}" + ); + Ok(()) + } + /// Crate-level check of the case-insensitive remap. Under `cargo test` there is no attached - /// JVM, so `fold_names` uses the ASCII fallback; ASCII casing still distinguishes match from + /// JVM, but ASCII names use the inline fast path; ASCII casing still distinguishes match from /// no-match, so this documents that `remap_physical_schema` renames the physical field to the /// logical name and records the reverse mapping. #[test] @@ -3363,7 +3403,7 @@ mod test { let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); let target = struct_type(vec![("id", DataType::Int64)]); - assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); let physical = list_type(struct_type(vec![ ("id", DataType::Int64), @@ -3376,7 +3416,7 @@ mod test { "inner", struct_type(vec![("a", DataType::Int64)]), )])); - assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// A target field with no exact-name match in the source (here, only a case-insensitive @@ -3388,7 +3428,7 @@ mod test { let opts = default_options(); let physical = struct_type(vec![("ID", DataType::Int64)]); let target = struct_type(vec![("id", DataType::Int64)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } #[test] @@ -3412,7 +3452,7 @@ mod test { let mut opts = default_options(); opts.case_sensitive = case_sensitive; assert_eq!( - is_pure_structural_narrowing(&physical, &target, &opts), + is_pure_structural_narrowing(&physical, &target, &opts).unwrap(), case_sensitive, "{physical:?} -> {target:?}, case_sensitive={case_sensitive}" ); @@ -3448,7 +3488,7 @@ mod test { let opts = default_options(); let physical = struct_type(vec![("id", DataType::Int64)]); let target = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// When `use_field_id` is set and the target struct carries Parquet field ids, Comet @@ -3460,7 +3500,7 @@ mod test { opts.use_field_id = true; let physical = struct_type_with_field_id(vec![("id", DataType::Int64, 1)]); let target = struct_type_with_field_id(vec![("id", DataType::Int64, 1)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// Map value narrowing has no equivalent in `nested_struct::cast_column` (it has no Map @@ -3479,7 +3519,7 @@ mod test { let target = physical.clone(); // Even a no-op Map "narrowing" (target == physical) must not be routed through this // predicate; Map is excluded structurally, not by an equality shortcut. - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// A leaf-level type change (timestamp tz relabeling, `nanosAsLong`, decimal promotion, @@ -3491,7 +3531,7 @@ mod test { let opts = default_options(); let physical = struct_type(vec![("id", DataType::Int32)]); let target = struct_type(vec![("id", DataType::Int64)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// NTZ -> LTZ timestamp relabeling (INT96 reads) is a metadata-only reinterpretation @@ -3511,7 +3551,7 @@ mod test { "ts", DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), )]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// Dictionary-encoded columns get Comet's own dictionary-preserving or @@ -3529,7 +3569,7 @@ mod test { "d", DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::LargeUtf8)), )]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// Build a `SparkPhysicalExprAdapter` over a single "events" column and run the real @@ -3613,7 +3653,7 @@ mod test { opts.use_field_id = true; let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); let target = struct_type(vec![("id", DataType::Int64)]); - assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// The field-id bail must apply at every nesting level it is reached, not just the @@ -3631,7 +3671,7 @@ mod test { "outer", struct_type_with_field_id(vec![("id", DataType::Int64, 1)]), )]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// A target struct with zero field-name overlap against the source is denied, distinct @@ -3644,7 +3684,7 @@ mod test { let opts = default_options(); let physical = struct_type(vec![("left", DataType::Int64)]); let target = struct_type(vec![("right", DataType::Int64)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// `case_sensitive = true` with an exact-case match must still be allowed: the predicate's @@ -3656,7 +3696,7 @@ mod test { opts.case_sensitive = true; let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); let target = struct_type(vec![("id", DataType::Int64)]); - assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } /// `case_sensitive = true` with a case-differing name is denied, same as the @@ -3671,6 +3711,6 @@ mod test { opts.case_sensitive = true; let physical = struct_type(vec![("ID", DataType::Int64)]); let target = struct_type(vec![("id", DataType::Int64)]); - assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts).unwrap()); } }