You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/14 20:30:13 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-5845-683c7219e39928d267dc45c664d8656a4157da78 created (now dc6a201d1e)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5845-683c7219e39928d267dc45c664d8656a4157da78
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at dc6a201d1e fix: propagate Parquet field-name folding failures (#5845)
This branch includes the following new commits:
new dc6a201d1e fix: propagate Parquet field-name folding failures (#5845)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: propagate Parquet field-name folding failures (#5845)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5845-683c7219e39928d267dc45c664d8656a4157da78
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit dc6a201d1e17c91c329b5161a1c62a7b6c5cd120
Author: Chao Sun <su...@apache.org>
AuthorDate: Mon Sep 14 16:17:53 2026 +0000
fix: propagate Parquet field-name folding failures (#5845)
---
native/core/src/parquet/name_fold.rs | 192 +++++++++++++++++++--------
native/core/src/parquet/parquet_exec.rs | 4 +-
native/core/src/parquet/parquet_support.rs | 2 +-
native/core/src/parquet/schema_adapter.rs | 206 +++++++++++++++++------------
4 files changed, 264 insertions(+), 140 deletions(-)
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<String> {
+/// 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<Vec<String>> {
+ 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<Vec<String>>,
+) -> DataFusionResult<Vec<String>> {
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<String> {
}
}
// `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<String>]) {
+/// 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<String>],
+ fold: impl FnOnce(&[&str]) -> DataFusionResult<Vec<String>>,
+) -> DataFusionResult<()> {
let cache = fold_cache();
let mut miss_positions: Vec<usize> = Vec::new();
{
@@ -95,12 +113,12 @@ fn fold_non_ascii(names: &[&str], positions: &[usize], result: &mut [Option<Stri
}
if miss_positions.is_empty() {
- return;
+ return Ok(());
}
let miss_names: Vec<&str> = 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<Stri
for (pos, &i) in miss_positions.iter().enumerate() {
result[i] = Some(folded[pos].clone());
}
+ Ok(())
}
/// Fold a single field name. Convenience wrapper over [`fold_names`] for the per-column-reference
/// lookups; the schema side is always folded in bulk via [`fold_schema_names`]. Short-circuits the
/// common single-name cases (identity when case-sensitive, inline ASCII fold otherwise) so a lone
/// name never allocates a throwaway `Vec` or touches the cache; non-ASCII names use the bulk path.
-pub(crate) fn fold_name(name: &str, case_sensitive: bool) -> 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<String> {
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<String> {
+/// 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<Vec<String>> {
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<HashMap<String, String>> {
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<String>, 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<Vec<String>> {
+ 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<Vec<String>> {
#[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<usize> = 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<bool> {
+ 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<String> = source_fields
.iter()
.map(|f| fold_name(f.name(), parquet_options.case_sensitive))
- .collect();
- target_fields.iter().all(|target_field| {
+ .collect::<DataFusionResult<_>>()?;
+ 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<id,ID>` 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<String, Vec<usize>> = 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::<Column>() {
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<dyn PhysicalExpr>,
@@ -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());
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org