feat: support unicode case sensitive field names for reading parquet - #5602
feat: support unicode case sensitive field names for reading parquet#5602comphead wants to merge 4 commits into
Conversation
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Reviewed at commit 00f9985 (schema_adapter branch).
Thanks for tracking this down. The premise checks out: ParquetReadSupport.clipParquetGroupFields in Spark 4.1.1 builds groupBy(_.getName.toLowerCase(Locale.ROOT)) at line 428 and looks up f.name.toLowerCase(Locale.ROOT) at line 447, so schema_adapter.rs really was the odd one out against both Spark and our own nested and projection paths. Routing everything through one helper is the right shape.
I spent most of the review on whether Rust's to_lowercase and Java's toLowerCase(Locale.ROOT) actually agree, since the whole fix rests on that. Details below, but the short version is that the change is a clear net win and should go in. I have some things I would like addressed first.
The Unicode equivalence is close, but not exact
I generated the lowercase mapping for all 1,114,112 codepoints under JDK 17.0.17 and under Rust 1.97, then diffed.
The good news is that every contextual and expanding case agrees exactly. Final sigma matches including the case-ignorable context rule (ΑΣ to ας, ΑΣΒ to ασβ, ΑΣ' to ας', ΣΑΣ to σας), and so do İ to i̇ (U+0069 U+0307), ß, fi, titlecase digraphs (Dž to dž), iota subscript (ᾼ to ᾳ), and Roman numerals. Rust implements the same SpecialCasing and Final_Sigma rules Java does.
Where they differ is 95 codepoints, all in one direction, where Rust folds case and JDK 17 does not:
U+1C89 Cyrillic Ext-C (1) Unicode 15.0
U+2C2F Glagolitic (1) Unicode 14.0
U+A7C0..U+A7DC Latin Ext-D (11) Unicode 14/15
U+10570..U+10595 Vithkuqi (35) Unicode 14.0
U+10D50..U+10D65 Garay (22) Unicode 16.0
U+16EA0..U+16EB8 Kirat Rai (25) Unicode 16.0
The cause is Unicode table version skew rather than any algorithm difference. JDK 17 ships Unicode 13.0, JDK 21 ships 15.0, JDK 11 ships 10.0, and Rust 1.97 ships 16.0. So the divergent set is not fixed, it moves with the JDK, and our support matrix spans JDK 11 through 21+. On JDK 21 it would be roughly the 47 Unicode 16 additions. On JDK 11 it would be larger than 95, picking up things like Georgian Mtavruli at U+1C90..U+1CBA. For those codepoints Comet would resolve a field that Spark leaves unmatched, so we return data where Spark returns null, and we raise a duplicate field error where Spark succeeds.
None of that argues against the change. 1367 non-ASCII codepoints have a real JDK 17 case mapping that eq_ignore_ascii_case was getting wrong, so this trades 1367 divergences for 95. I only want the doc comment on names_match reworded. Right now it says the two fold the same way, and a future reader will take that literally. Could we say it matches Spark's algorithm, including the Final_Sigma and SpecialCasing rules, and note that any residual difference tracks the JDK's Unicode table version?
Also worth saying that the test data here is well chosen. MÜNCHEN, Ω/ω, GRÜN, CAFÉ/RÉSUMÉ all have case mappings that have been stable since Unicode 1.x, so these tests will not turn JDK-version dependent on us. Worth keeping that constraint in mind for any tests added later.
The description contradicts itself on case-sensitive mode
The description says there is no behavior change in case-sensitive mode, and also says it fixes the generateFakeColumnName block folding case in case-sensitive mode. I think that second one really is a behavior change.
Take a logical schema of [A(id=5), a(no id)] read against a file containing [a(id=9)], in case-sensitive mode with field IDs on. Spark sends A to matchIdField, id 5 is absent, so A gets a fake name and reads null. Logical a carries no id so it goes to matchCaseSensitiveField and resolves against physical a. Before this change Comet computed unmatched_id_logical_names = {"A"}, then "A".eq_ignore_ascii_case("a") was true, so it renamed physical a to __comet_unmatched_field_id_1 and logical a read back null. After this change it resolves, which matches Spark.
That looks like the right fix to me. Worth noting the structural reason it works: Spark renames the logical field with f.copy(name = generateFakeColumnName) and never compares names at all, whereas we rename the physical field that would collide, so our comparison has to track whichever mode the downstream name match uses. That is exactly what the change does, it just is not written down anywhere.
Could we drop that sentence from the description and add a test? It is the only change here that lands outside case-insensitive mode, so I would like it pinned down.
Per-file cost in remap_physical_schema
remap_physical_schema is O(physical x logical) and now allocates two Strings per comparison, with the loop-invariant side recomputed inside every .find, .any, and .position closure. It runs once per file through PhysicalExprAdapterFactory::create. Measuring that nested loop in release mode:
| columns | eq_ignore_ascii_case |
to_lowercase |
precomputed lowercase map |
|---|---|---|---|
| 1000 | 17.8 ms | 33.9 ms | 0.12 ms |
| 4000 | 56.9 ms | 268 ms | 0.21 ms |
The quadratic shape predates this PR, but the constant gets 2x to 5x worse on a per-file path. Would you be up for building a lowercase name to index map once up front? It also lines up more closely with Spark, which builds groupBy(_.getName.toLowerCase(Locale.ROOT)) once rather than re-folding per comparison, and it gives duplicate detection for free. If that feels like too much for this PR, just hoisting the invariant to_lowercase() out of the closures recovers a good chunk on its own.
Smaller things
schema_adapter.rs:341, names_match(pf.name(), col_name, false). As far as I can tell this is correct, since check_column_duplicate is only reached when original_physical_schema is Some and that is only populated when !case_sensitive. But the literal false reads like an oversight at the call site. Could we thread the flag through from the caller, or add a line saying this path is case-insensitive by construction?
We now have three copies of the folding policy: names_match here, normalize_name in parquet_support.rs, and the inline branch at parquet_exec.rs:114. Those drifting apart is what produced this bug in the first place. Could names_match live somewhere the other two can call it, so the next change to the policy has to touch one place?
Last one, and this is not yours to fix here. Related to the parquet_duplicate_fields_case_insensitive_unicode test you added, the nested path at parquet_support.rs:313 has assert_eq!(field_name_to_index_map.len(), from_fields.len()). A struct containing Ω and ω collapses that map in case-insensitive mode and trips the assert, so we panic instead of raising Spark's duplicate field error. Plain ASCII B and b do the same. It came in with #4216 so it predates this PR, but it is the nested counterpart to the case you are covering at the top level. Could you file an issue and link it here? I would rather it be tracked than rediscovered later.
CI
Nothing failing as of this commit. All lint jobs, the Spark 4.1 build, and both native library builds are green. Still running: ubuntu-latest/rust-test, the macOS [scans] job that covers the new CometNativeReaderSuite test, [exec], [expressions], [shuffle], and the Spark 3.5 and 4.1 SQL suites.
| .fields() | ||
| .iter() | ||
| .filter(|pf| pf.name().eq_ignore_ascii_case(col_name)) | ||
| .filter(|pf| names_match(pf.name(), col_name, false)) |
There was a problem hiding this comment.
[P2] Preserve field-ID precedence in the Unicode duplicate check
The Unicode-name coverage is useful. Could we retain ID-based resolution when applying this duplicate check? With spark.sql.parquet.fieldId.read.enabled=true and spark.sql.caseSensitive=false, an explicit one-field schema ω (ID 2) can select a file's ω (ID 2) even when the file also contains Ω (ID 1). A case-sensitive writer with field-ID writing enabled can produce these two nullable integer columns with varying values. The ID remap selects ID 2, but this check examines the original physical schema and now counts both names, returning DuplicateFieldCaseInsensitive before the valid projection. Spark selects the unique ID before considering names, and the previous ASCII check admits this Unicode example. Please retain ambiguity errors for name-based reads and add a regression case for the ID-based read. This example is source-derived, not an executed reproduction.
| if case_sensitive { | ||
| a == b | ||
| } else { | ||
| a.to_lowercase() == b.to_lowercase() |
There was a problem hiding this comment.
[P2] Match Java's contextual sigma lowercasing
There is a contextual difference in addition to the Unicode-table version differences already discussed. In the inspected Rust 1.88.0 and OpenJDK 17u 17.0.16+8 default-provider sources, A1Σ lowercases to a1σ in Rust but a1ς in Java's Locale.ROOT path. Java searches across the digit within a word, whereas Rust stops its cased-character search at that digit. With field-ID matching disabled, an explicit nullable A1σ Parquet projection therefore binds the stored values through this comparison where Spark and the previous ASCII comparison supply nulls. Could we preserve the JVM Parquet name-resolution contract for this context and cover it in a regression test? This is source-derived, not a runtime reproduction or a claim about every JDK/compiler version.
|
I wonder if it is worth using JNI so that Rust can call back into JVM to perform lowercase of names? |
hm. that actually makes sense IMO, to extend JNI calls to support this thing. Let me check if its performance reasonable, which should be as this operation is not on a hot path |
|
@andygrove @sunchao changed the PR to JNI call, according to Claude analysis rust and JVM unicode are doing the same in lots of cases, but still having differences in some exotic symbols. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 92ee91089ae1af5b3b02aa7a9bc49eb8e1402778 against the previously reviewed 00f9985 update.
The Java helper addresses the contextual-sigma mismatch on the successful JVM path. Its exact source passed 11 Unicode cases on OpenJDK 21, including A1Σ, expanding mappings, supplementary characters and distinct normalization forms. The new cached JNI class handle has a separate P1 lifetime issue described inline.
The existing P2 field-ID precedence concern remains: the top-level duplicate check still examines folded names in the original physical schema without exempting an ID-resolved column. The nested lookup now gives IDs precedence. I have not opened a duplicate thread for that issue.
Current GitHub checks show 64 successful and 9 skipped. Validation was limited to source tracing, the exact Java helper and an isolated JNI lifetime probe with a passing global-reference control. No full Comet build or Parquet query was run.
| pub const JVM_CLASS: &'static str = "org/apache/comet/CometSchemaUtils"; | ||
|
|
||
| pub fn new(env: &mut Env<'a>) -> JniResult<CometSchemaUtils<'a>> { | ||
| let class = env.find_class(JNIString::new(Self::JVM_CLASS))?; |
There was a problem hiding this comment.
[P1] Retain a global reference for the cached helper class
Could this class be retained as a JNI global reference before the binding is stored in JVMClasses? find_class returns a local reference, but JVMClasses::init retains this binding process-wide by extending the Env lifetime. Once the initializing native call returns, a later field-name cache miss can call toLowerCaseRoot with an expired jclass. The name cache only hides this for names already encountered. An isolated probe using this exact constructor, JNI 0.22.4 and OpenJDK 21 successfully folded A1Σ inside the creation frame, then aborted under -Xcheck:jni when folding Ω after that frame closed: Bad global or local ref passed to JNI. The same probe retaining a global reference passed. Please keep that global reference alive with the cached binding and cover a call after the initialization frame closes. This is a focused JNI reproduction, not a complete Comet scan reproduction.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Reviewed at 92ee9108. This is a substantially different change from the one I looked at before, so I re-ran the audit from scratch rather than diffing against my earlier comments.
Short version: the JNI approach is right and the fold itself is now exact. I have five things I would like addressed, and the largest is that nothing in the PR tests the change you just made.
The JNI approach is the right call
Saying this clearly so nobody spends time relitigating it. The fold has to be applied to both sides of the match. The read schema is known to the JVM, but the file schema is only known natively, after the file is opened. So there is no version of this where we fold on the Scala side and ship the result down in the plan protobuf. Given that, the choice is a JNI callback or reimplementing the JDK's case tables in Rust, and the callback is the one that stays correct as the JDK moves. The 95 divergent codepoints from my earlier comment are gone.
Nothing tests the JNI fold
The four parquet_case_insensitive_unicode_* tests are in 00f9985 and gone in 3a4b7a6. That makes sense, since cargo test has no JVM to call and fold_uncached takes the ASCII branch. But it leaves the delta between the two revisions unguarded. The three Scala tests use É, Ü and Ω, and Rust and every JDK we support fold those identically, so all three still pass if fold_uncached goes back to to_lowercase().
U+A7DC () is a good witness. Rust folds it to U+019B (ƛ). I checked JDK 11.0.22, 17.0.10, 21.0.2 and 22 and all four leave it alone, so unlike the U+A7C0 and U+10570 families it will not turn JDK-version dependent on us. I added this test locally and confirmed it passes on your branch, then patched fold_uncached to fold in Rust and got Spark Answer [null][null][null] against Comet Answer [0][1][2], which is #5495 in the other direction.
test("native reader non-ASCII fold follows the JVM Unicode table") {
withTempPath { path =>
spark.range(3).selectExpr("id as ``").write.parquet(path.toString)
val readSchema = new StructType().add("ƛ", LongType, nullable = true)
withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
val df = spark.read.schema(readSchema).parquet(path.toString)
val (_, cometPlan) = checkSparkAnswerAndOperator(df)
assert(
cometPlan.collect { case n: CometNativeScanExec => n }.nonEmpty,
"Expected a CometNativeScanExec")
}
}
}Asserting through checkSparkAnswer rather than against a literal null is what keeps it honest on a future JDK that does fold U+A7DC.
The case-sensitive field id change still has no test
This is the one I asked about last time. remap_field_id_missing_fake_renames_colliding_physical passes case_sensitive = false, which is the direction that already worked before this PR. The case-sensitive direction is the one the fix actually changes.
Here is the scenario from my earlier comment as a unit test. It passes on your branch, and it fails with left: "__comet_unmatched_field_id_1", right: "a" if the collision check goes back to eq_ignore_ascii_case, so it does pin the change down.
#[test]
fn remap_field_id_missing_does_not_fake_rename_case_sensitive() {
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
let id_meta = |id: &str| {
std::collections::HashMap::from([(
PARQUET_FIELD_ID_META_KEY.to_string(),
id.to_string(),
)])
};
let logical = Arc::new(Schema::new(vec![
Field::new("A", DataType::Int32, true).with_metadata(id_meta("5")),
Field::new("a", DataType::Int32, true),
]));
let physical = Arc::new(Schema::new(vec![
Field::new("a", DataType::Int32, true).with_metadata(id_meta("9")),
]));
let (remapped, _name_map) =
super::remap_physical_schema(&logical, &physical, true, true, false).unwrap();
assert_eq!(remapped.field(0).name(), "a");
}The fallback goes back to the bug
fold_uncached falls back to to_ascii_lowercase(). That is the behavior this PR is fixing, so a JNI failure puts us straight back on #5495 with only a warn! to mark it. Using to_lowercase() there instead would be wrong on 95 codepoints against JDK 17, where ASCII folding is wrong on 1367. Would you switch both the JVM-failure branch and the no-JVM branch to it?
The no-JVM branch matters for a second reason. It is the one cargo test takes, and it is why the four unit tests had to go. With to_lowercase() there, MÜNCHEN / münchen and Ω / ω resolve again under cargo test and most of that coverage can come back.
An ASCII fast path would keep the lock off the hot path
Every name goes through the process-wide RwLock cache, including pure ASCII ones, which will be all of them for almost every schema. For an all-ASCII string, toLowerCase(Locale.ROOT) and to_ascii_lowercase are provably the same thing. Locale.ROOT excludes the Turkish and Lithuanian rules, no ASCII codepoint lowercases to a non-ASCII one, and Final_Sigma needs a sigma to fire. I checked all 128 ASCII codepoints and all 16,384 two-character ASCII contexts on JDK 11, 17 and 21 and got zero differences on all three.
Could fold_names return early on name.is_ascii() before it touches the cache? That confines the lock, the JVM crossing and the whole fallback question to the names that actually need them, and it makes the FOLD_CACHE_MAX_ENTRIES bound a non-question.
The per-batch nested fold got more expensive
parquet_convert_struct_to_struct runs per batch through CometCastColumnExpr::evaluate, and it now folds both field lists and rebuilds folded_to_indices every time. The from and to Fields are fixed for the life of the expression, so that is the same answer on every batch.
A caveat on the numbers. My release build kept getting killed, so instead of benchmarking the real function I extracted both loops into a standalone release-mode program. Cache warm and single threaded, which is the most favorable case for the new code, the new path costs 1.68x the old at 8 struct fields, 1.61x at 32 and 1.62x at 128, over 10,000 batches. Under real concurrency it should be worse, since every scan task contends on the one lock.
Could the fold and the index map move to expression construction rather than running per batch? The ASCII fast path above recovers roughly half of it on its own if the hoist is too much for this PR.
The top-level match is still quadratic
Pre-folding both schemas once did fix the regression I flagged last time. In the same harness the match loop is at parity with base at 1000 columns and slightly better at 4000, so that concern is closed. Thank you for picking it up.
The quadratic shape is still there though, and check_column_duplicate adds an O(physical) scan per referenced column per rewrite on top of it. A folded-name to indices map built once in create() measured 16x faster at 1000 columns and 50x at 4000, and it gives the duplicate detection for free. You have already written that map in parquet_support.rs as folded_to_indices. Could the top level use the same shape? As it stands the PR solves one problem two different ways in two files, which is the drift that produced this bug in the first place.
Smaller things
Replacing the assert_eq! in parquet_support.rs with a real error is a good fix, and I am glad it is here rather than deferred to a follow-up issue. One question on the gate. The check fires in both modes, and in case-sensitive mode the fold is identity, so two byte-identical sibling names would raise _LEGACY_ERROR_TEMP_2093, whose message says "in case-insensitive mode", while caseSensitive=true. The top-level path does not do this, because check_column_duplicate only runs when original_physical_schema is Some, which only happens when !case_sensitive. Should the nested check be gated the same way, so the two agree?
Both new duplicate tests assert only that the message contains "duplicate field", and Spark's own reader emits _LEGACY_ERROR_TEMP_2093 with that same substring for the same input. So both pass unchanged if Comet falls back to Spark for the scan, which means they cannot fail for the reason we care about. Could they assert the error class, or check the plan before the intercept? The nested one especially, since it covers the path that used to panic.
Consolidating the fold into one function is exactly what I was hoping for. On its home, parquet_support now imports fold_names from schema_adapter while schema_adapter imports spark_parquet_convert from parquet_support. Would parquet_support.rs, or a small name_fold.rs, sit better, given that all three of schema_adapter, parquet_support and parquet_exec call it?
Every other Comet-owned class we look up in JVMClasses lives in spark/src/main/java, and CometSchemaUtils is in common/. It resolves fine because common is shaded in, so this is only about keeping the JNI surface in one place. Was there a reason for common?
The description
It still describes the first revision. names_match and all four parquet_case_insensitive_unicode_* tests are gone at this commit, so anyone reviewing from the description is reviewing a design that is not here any more. Could you refresh it for the JNI approach, drop the case-sensitive contradiction we discussed last time, and add the two changes it does not currently mention, which are the nested duplicate-field fix in parquet_support.rs and the new SparkError::duplicate_field_case_insensitive helper? The nested one is a behavior change, and a reader would want it flagged rather than discovering it in the diff.
Local runs and CI
cargo test -p datafusion-comet --lib parquet:: passes 89. CometNativeReaderSuite passes 59 with the one pre-existing #4199 skip, and ParquetReadV1Suite passes 62, both on Spark 4.1.3 with JDK 17. I skipped Comet Fuzz on purpose, since its generator emits ASCII column names and it has no power against this change.
Lint Java (Spark 4.0, JDK 17) is red while Lint Java (Spark 4.0, JDK 21) is green, and those two run the identical scalafix command against the identical profile, so I suspect a flake rather than a real violation. Spotless and scalastyle are clean when I run them locally. Worth a re-run to confirm before anyone chases it.
- Retain a JNI global reference for the cached CometSchemaUtils class so it survives past JVMClasses::init (was a dangling local ref). - Consolidate field-name folding into native/core/src/parquet/name_fold.rs, called by the schema adapter, the nested Struct->Struct convert, and the plan-time projection (previously duplicated across all three). - Add an ASCII fast path; use Rust Unicode to_lowercase (not ASCII folding) as the no-JVM / JNI-failure fallback. - Skip the case-insensitive duplicate check for field-id-resolved columns (mirrors Spark's matchIdField, which selects the id before comparing names). - Gate the nested duplicate-field error to case-insensitive mode; build the top-level duplicate map once (folded name -> field indices) for O(1) lookup. - Move CometSchemaUtils.java from common/ to spark/ to match the convention for JVMClasses-looked-up classes. - Add Rust and Scala regression tests.
3 participants
Closes #5495
Rationale for this change
The Parquet schema adapter matched field names case-insensitively with
eq_ignore_ascii_case, which folds only ASCIIA-Z/a-z. Spark'sParquetReadSupport.clipParquetGroupFieldsfolds withname.toLowerCase(Locale.ROOT)(full Unicode) for both matching and duplicate detection. Under Comet's default (spark.sql.caseSensitive=false), a column differing only by non-ASCII case (fileMÜNCHEN, querymünchen) silently read back asnullwhile Spark resolved it.The nested-struct path (
parquet_support.rs) and the projection path (parquet_exec.rs) already usedto_lowercase(), soschema_adapter.rswas the lone ASCII-only outlier, making top-level and nested resolution inconsistent.What changes are included in this PR?
In
native/core/src/parquet/schema_adapter.rs:names_match(a, b, case_sensitive)helper: exact compare when case-sensitive, otherwiseto_lowercase()compare.str::to_lowercaseis the Unicode-aware equivalent of Java'stoLowerCase(Locale.ROOT).eq_ignore_ascii_casesites through it: name match inremap_physical_schema,check_column_duplicate, thewrap_all_type_mismatchesfield/index lookups, and the missing-column check inreplace_missing_with_defaults.generateFakeColumnNameblock (unmatched field-id guard) folded case even in case-sensitive mode; it now respectscase_sensitive.if case_sensitive { … } else { … }branches.No behavior change in case-sensitive mode. Nested-struct name folding was already Unicode-aware, so this only aligns top-level matching with it and with Spark.
How are these changes tested?
Rust unit tests in
schema_adapter.rs:parquet_case_insensitive_unicode_name_match— fileMÜNCHENresolves tomünchen.parquet_duplicate_fields_case_insensitive_unicode—Ω/ωfold to a duplicate-field error.parquet_case_insensitive_unicode_nested_struct_field— nestedGRÜNresolves togrün(carries aNULL).parquet_case_insensitive_unicode_top_level_and_nested_struct—CAFÉ/RÉSUMÉresolve tocafé/résumé.Scala end-to-end test in
CometNativeReaderSuite(native reader case-insensitive resolution for top-level and nested struct fields): ASCII and non-ASCII names, at top level and inside a struct, asserting aCometNativeScanExecand matching Spark underspark.sql.caseSensitive=false(fields resolve) andtrue(read backnull).