Skip to content

Commit 53c58be

Browse files
peterxcliclaude
peterxcli
and
claude
committed
fix: restrict the Unicode fold guard to referenced scan columns
The fail-closed guard for Unicode-folded physical names ran over the full logical file schema inside remap_physical_schema, aborting scans over unused ordinary columns that Spark never reads, and was disabled globally whenever field-ID matching was enabled even though ID-less fields still fall through to ASCII name matching. Move it into the per-referenced- column walk in SparkPhysicalExprAdapter::rewrite, alongside the existing duplicate detection, and exempt only columns whose logical field itself carries a PARQUET:field_id. Co-Authored-By: Claude Fable 5 <[email protected]>
1 parent e78da57 commit 53c58be

2 files changed

Lines changed: 235 additions & 85 deletions

File tree

‎native/core/src/parquet/schema_adapter.rs‎

Lines changed: 194 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -102,43 +102,6 @@ fn remap_physical_schema(
102102
)));
103103
}
104104

105-
// Spark resolves case-insensitive names by Unicode lowercasing (`toLowerCase(Locale.ROOT)`
106-
// in `ParquetReadSupport`), while the matcher below folds ASCII only. A physical name such
107-
// as `K` (U+212A KELVIN SIGN) resolves to a requested ASCII `k` in Spark but stays
108-
// unmatched here, and the scan-planning gate cannot see physical file names, so the column
109-
// would be silently null-filled as missing. Until #5495 brings Spark-parity Unicode
110-
// matching, fail Variant-bearing scans whose resolution would differ under Unicode folding
111-
// rather than return values that disagree with Spark.
112-
if !case_sensitive
113-
&& !should_match_by_id
114-
&& logical_schema.fields().iter().any(|f| is_variant_field(f))
115-
{
116-
for logical_field in logical_schema.fields() {
117-
let name = logical_field.name();
118-
if physical_schema
119-
.fields()
120-
.iter()
121-
.any(|pf| pf.name().eq_ignore_ascii_case(name))
122-
{
123-
continue;
124-
}
125-
if let Some(pf) = physical_schema
126-
.fields()
127-
.iter()
128-
.find(|pf| !pf.name().is_ascii() && pf.name().to_lowercase() == name.to_lowercase())
129-
{
130-
return Err(DataFusionError::Execution(format!(
131-
"Case-insensitive scan of a Variant-bearing schema requires Unicode column \
132-
matching: requested column '{name}' resolves to physical Parquet column \
133-
'{phys}' only under Unicode case folding, which Comet's native scan does \
134-
not support yet (https://github.com/apache/datafusion-comet/issues/5495). \
135-
Retry with spark.comet.enabled=false, or align the column name case.",
136-
phys = pf.name(),
137-
)));
138-
}
139-
}
140-
}
141-
142105
// Build id -> all matching physical field names. We need the full list so we can mirror
143106
// Spark's `_LEGACY_ERROR_TEMP_2094` "Found duplicate field(s)" error when an ID-bearing
144107
// logical field would resolve to more than one physical field.
@@ -387,6 +350,31 @@ fn check_column_duplicate(col_name: &str, physical_schema: &SchemaRef) -> Option
387350
}
388351
}
389352

353+
/// Detect a physical Parquet column that Spark's case-insensitive resolver would match to
354+
/// `col_name` but the ASCII-only matcher cannot. Spark groups physical names by Unicode
355+
/// lowercasing (`toLowerCase(Locale.ROOT)` in `ParquetReadSupport`), so a physical name such
356+
/// as `K` (U+212A KELVIN SIGN) resolves to a requested ASCII `k` there while staying
357+
/// unmatched here, and the scan-planning gate cannot see physical file names. Returns the
358+
/// folded physical name when no ASCII-insensitive match exists but a non-ASCII physical name
359+
/// Unicode-lowercases to the same string as `col_name`.
360+
fn check_column_unicode_fold_mismatch(
361+
col_name: &str,
362+
physical_schema: &SchemaRef,
363+
) -> Option<String> {
364+
if physical_schema
365+
.fields()
366+
.iter()
367+
.any(|pf| pf.name().eq_ignore_ascii_case(col_name))
368+
{
369+
return None;
370+
}
371+
physical_schema
372+
.fields()
373+
.iter()
374+
.find(|pf| !pf.name().is_ascii() && pf.name().to_lowercase() == col_name.to_lowercase())
375+
.map(|pf| pf.name().clone())
376+
}
377+
390378
impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory {
391379
fn create(
392380
&self,
@@ -488,23 +476,63 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
488476
// a field with multiple case-insensitive matches in the physical schema.
489477
// Only the columns actually referenced trigger the error (not the whole schema).
490478
if let Some(orig_physical) = &self.original_physical_schema {
479+
// Until #5495 brings Spark-parity Unicode name matching, Variant-bearing scans
480+
// fail closed for referenced columns whose resolution would differ under Unicode
481+
// folding, rather than silently null-filling values Spark reads. Restricting the
482+
// check to referenced columns keeps unused ordinary columns in the native data
483+
// schema from aborting a scan that never reads them, and the per-field ID
484+
// exemption mirrors the name-match eligibility in `remap_physical_schema`:
485+
// ID-bearing logical fields resolve by ID, everything else falls through to the
486+
// ASCII name matcher and needs the guard.
487+
let variant_scan = self
488+
.logical_file_schema
489+
.fields()
490+
.iter()
491+
.any(|f| is_variant_field(f));
492+
let should_match_by_id = self.parquet_options.use_field_id
493+
&& schema_has_field_ids(&self.logical_file_schema);
491494
// Walk the expression tree to find Column references
492-
let mut duplicate_err: Option<DataFusionError> = None;
495+
let mut column_err: Option<DataFusionError> = None;
493496
let _ = Arc::<dyn PhysicalExpr>::clone(&expr).transform(|e| {
494497
if let Some(col) = e.downcast_ref::<Column>() {
495498
if let Some((req, matched)) = check_column_duplicate(col.name(), orig_physical)
496499
{
497-
duplicate_err = Some(DataFusionError::External(Box::new(
500+
column_err = Some(DataFusionError::External(Box::new(
498501
SparkError::DuplicateFieldCaseInsensitive {
499502
required_field_name: req,
500503
matched_fields: matched,
501504
},
502505
)));
503506
}
507+
if column_err.is_none() && variant_scan {
508+
let id_matched = should_match_by_id
509+
&& self
510+
.logical_file_schema
511+
.field_with_name(col.name())
512+
.ok()
513+
.and_then(parse_field_id)
514+
.is_some();
515+
if !id_matched {
516+
if let Some(phys) =
517+
check_column_unicode_fold_mismatch(col.name(), orig_physical)
518+
{
519+
column_err = Some(DataFusionError::Execution(format!(
520+
"Case-insensitive scan of a Variant-bearing schema requires \
521+
Unicode column matching: requested column '{name}' resolves \
522+
to physical Parquet column '{phys}' only under Unicode case \
523+
folding, which Comet's native scan does not support yet \
524+
(https://github.com/apache/datafusion-comet/issues/5495). \
525+
Retry with spark.comet.enabled=false, or align the column \
526+
name case.",
527+
name = col.name(),
528+
)));
529+
}
530+
}
531+
}
504532
}
505533
Ok(Transformed::no(e))
506534
});
507-
if let Some(err) = duplicate_err {
535+
if let Some(err) = column_err {
508536
return Err(err);
509537
}
510538
}
@@ -2033,67 +2061,148 @@ mod test {
20332061
}
20342062

20352063
#[test]
2036-
fn test_case_insensitive_variant_scan_rejects_unicode_folded_physical_name() {
2037-
use super::remap_physical_schema;
2038-
2039-
let variant_storage = DataType::Struct(Fields::from(vec![
2040-
Field::new("value", DataType::Binary, true),
2041-
Field::new("metadata", DataType::Binary, true),
2042-
]));
2043-
let variant_logical: SchemaRef = Arc::new(Schema::new(vec![Field::new(
2044-
"k",
2045-
variant_storage.clone(),
2046-
true,
2047-
)
2048-
.with_extension_type(VariantType)]));
2064+
fn test_check_column_unicode_fold_mismatch() {
2065+
use super::check_column_unicode_fold_mismatch;
20492066

20502067
// Physical `K` (U+212A KELVIN SIGN) folds to `k` in Spark's Unicode resolver but not
2051-
// in the ASCII matcher: the scan must fail closed instead of null-filling `k`.
2068+
// in the ASCII matcher: the guard must report it.
20522069
let kelvin: SchemaRef = Arc::new(Schema::new(vec![Field::new(
20532070
"\u{212A}",
2054-
variant_storage.clone(),
2071+
DataType::Int32,
20552072
true,
20562073
)]));
2057-
let err =
2058-
remap_physical_schema(&variant_logical, &kelvin, false, false, false).unwrap_err();
2059-
assert!(err.to_string().contains("Unicode case folding"), "{err}");
2060-
assert!(err.to_string().contains("5495"), "{err}");
2074+
assert_eq!(
2075+
check_column_unicode_fold_mismatch("k", &kelvin),
2076+
Some("\u{212A}".to_string())
2077+
);
20612078

2062-
// ASCII case differences still match natively.
2063-
let ascii: SchemaRef = Arc::new(Schema::new(vec![Field::new(
2064-
"K",
2065-
variant_storage.clone(),
2066-
true,
2067-
)]));
2068-
let (remapped, name_map) =
2069-
remap_physical_schema(&variant_logical, &ascii, false, false, false).unwrap();
2070-
assert_eq!(remapped.field(0).name(), "k");
2071-
assert_eq!(name_map.get("k").map(String::as_str), Some("K"));
2079+
// ASCII case differences are handled by the existing matcher.
2080+
let ascii: SchemaRef = Arc::new(Schema::new(vec![Field::new("K", DataType::Int32, true)]));
2081+
assert_eq!(check_column_unicode_fold_mismatch("k", &ascii), None);
20722082

20732083
// `ſ` (U+017F) does not lowercase to `s`, so Spark also treats the requested column
2074-
// as missing; the existing missing-column path is kept.
2084+
// as missing; both engines agree on the missing-column path.
20752085
let long_s: SchemaRef = Arc::new(Schema::new(vec![Field::new(
20762086
"\u{17F}",
2077-
variant_storage.clone(),
2087+
DataType::Int32,
20782088
true,
20792089
)]));
2080-
let variant_s: SchemaRef = Arc::new(Schema::new(vec![Field::new(
2081-
"s",
2082-
variant_storage.clone(),
2083-
true,
2084-
)
2085-
.with_extension_type(VariantType)]));
2086-
assert!(remap_physical_schema(&variant_s, &long_s, false, false, false).is_ok());
2090+
assert_eq!(check_column_unicode_fold_mismatch("s", &long_s), None);
20872091

2088-
// Without a marked Variant field in the required schema the pre-#5495 behavior is
2089-
// unchanged.
2090-
let plain_logical: SchemaRef =
2091-
Arc::new(Schema::new(vec![Field::new("k", DataType::Int32, true)]));
2092-
let plain_kelvin: SchemaRef = Arc::new(Schema::new(vec![Field::new(
2093-
"\u{212A}",
2092+
// A genuinely missing column reports nothing.
2093+
let other: SchemaRef = Arc::new(Schema::new(vec![Field::new(
2094+
"other",
20942095
DataType::Int32,
20952096
true,
20962097
)]));
2097-
assert!(remap_physical_schema(&plain_logical, &plain_kelvin, false, false, false).is_ok());
2098+
assert_eq!(check_column_unicode_fold_mismatch("k", &other), None);
2099+
}
2100+
2101+
fn variant_field(name: &str, id: Option<i32>) -> Field {
2102+
let field = Field::new(
2103+
name,
2104+
DataType::Struct(Fields::from(vec![
2105+
Field::new("value", DataType::Binary, true),
2106+
Field::new("metadata", DataType::Binary, true),
2107+
])),
2108+
true,
2109+
)
2110+
.with_extension_type(VariantType);
2111+
match id {
2112+
Some(id) => {
2113+
let mut metadata = field.metadata().clone();
2114+
metadata.insert("PARQUET:field_id".to_string(), id.to_string());
2115+
field.with_metadata(metadata)
2116+
}
2117+
None => field,
2118+
}
2119+
}
2120+
2121+
fn int_field(name: &str, id: Option<i32>) -> Field {
2122+
let field = Field::new(name, DataType::Int32, true);
2123+
match id {
2124+
Some(id) => {
2125+
let mut metadata = std::collections::HashMap::new();
2126+
metadata.insert("PARQUET:field_id".to_string(), id.to_string());
2127+
field.with_metadata(metadata)
2128+
}
2129+
None => field,
2130+
}
2131+
}
2132+
2133+
fn rewrite_column(
2134+
logical: SchemaRef,
2135+
physical: SchemaRef,
2136+
use_field_id: bool,
2137+
column: &str,
2138+
) -> Result<(), DataFusionError> {
2139+
use datafusion::physical_expr::expressions::Column;
2140+
use datafusion::physical_expr::PhysicalExpr;
2141+
2142+
let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
2143+
options.case_sensitive = false;
2144+
options.use_field_id = use_field_id;
2145+
let factory = SparkPhysicalExprAdapterFactory::new(options, None);
2146+
let adapter = factory.create(Arc::clone(&logical), physical)?;
2147+
let index = logical.index_of(column).unwrap();
2148+
let expr: Arc<dyn PhysicalExpr> = Arc::new(Column::new(column, index));
2149+
adapter.rewrite(expr).map(|_| ())
2150+
}
2151+
2152+
#[test]
2153+
fn test_case_insensitive_variant_scan_fails_closed_on_unicode_folded_name() {
2154+
// Referencing `k` whose physical name is Kelvin `K` must fail closed rather than
2155+
// null-fill values Spark reads (#5495).
2156+
let logical: SchemaRef = Arc::new(Schema::new(vec![variant_field("k", None)]));
2157+
let physical: SchemaRef = Arc::new(Schema::new(vec![variant_field("\u{212A}", None)]));
2158+
let err = rewrite_column(logical, physical, false, "k").unwrap_err();
2159+
assert!(err.to_string().contains("Unicode case folding"), "{err}");
2160+
assert!(err.to_string().contains("5495"), "{err}");
2161+
}
2162+
2163+
#[test]
2164+
fn test_unicode_guard_skips_unreferenced_columns() {
2165+
// An unused ordinary column with a Unicode-folded physical name stays in the native
2166+
// data schema; only referenced columns may abort the scan.
2167+
let logical: SchemaRef = Arc::new(Schema::new(vec![
2168+
variant_field("v", None),
2169+
int_field("k", None),
2170+
]));
2171+
let physical: SchemaRef = Arc::new(Schema::new(vec![
2172+
variant_field("v", None),
2173+
int_field("\u{212A}", None),
2174+
]));
2175+
rewrite_column(Arc::clone(&logical), Arc::clone(&physical), false, "v").unwrap();
2176+
let err = rewrite_column(logical, physical, false, "k").unwrap_err();
2177+
assert!(err.to_string().contains("Unicode case folding"), "{err}");
2178+
}
2179+
2180+
#[test]
2181+
fn test_unicode_guard_applies_to_id_less_fields_in_field_id_scans() {
2182+
// With field-ID reads enabled, ID-bearing fields resolve by ID and are exempt, but an
2183+
// ID-less Variant field still falls through to ASCII name matching and needs the
2184+
// guard.
2185+
let logical: SchemaRef = Arc::new(Schema::new(vec![
2186+
int_field("sibling", Some(1)),
2187+
variant_field("k", None),
2188+
]));
2189+
let physical: SchemaRef = Arc::new(Schema::new(vec![
2190+
int_field("data", Some(1)),
2191+
variant_field("\u{212A}", None),
2192+
]));
2193+
rewrite_column(Arc::clone(&logical), Arc::clone(&physical), true, "sibling").unwrap();
2194+
let err = rewrite_column(logical, physical, true, "k").unwrap_err();
2195+
assert!(err.to_string().contains("Unicode case folding"), "{err}");
2196+
2197+
// The same Variant field carrying an ID resolves by ID and stays exempt.
2198+
let logical: SchemaRef = Arc::new(Schema::new(vec![
2199+
int_field("sibling", Some(1)),
2200+
variant_field("k", Some(2)),
2201+
]));
2202+
let physical: SchemaRef = Arc::new(Schema::new(vec![
2203+
int_field("data", Some(1)),
2204+
variant_field("\u{212A}", Some(2)),
2205+
]));
2206+
rewrite_column(logical, physical, true, "k").unwrap();
20982207
}
20992208
}

‎spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,47 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper
170170
}
171171
}
172172

173+
test("case-insensitive Variant scan ignores unused Unicode physical columns") {
174+
assume(isSpark40Plus, "VariantType requires Spark 4.0+")
175+
176+
withTempPath { path =>
177+
withSQLConf(
178+
CometConf.COMET_ENABLED.key -> "false",
179+
"spark.sql.variant.writeShredding.enabled" -> "false") {
180+
// Physical `K` (U+212A KELVIN SIGN) beside an ordinary Variant column.
181+
sql("SELECT parse_json('42') AS v, 7 AS `K`").write.parquet(path.toString)
182+
}
183+
184+
val table = s"variant_unicode_unused_${System.currentTimeMillis()}"
185+
withTable(table) {
186+
sql(s"CREATE TABLE $table (v VARIANT, `k` INT) USING parquet OPTIONS (path '$path')")
187+
withSQLConf(
188+
SQLConf.CASE_SENSITIVE.key -> "false",
189+
"spark.sql.variant.allowReadingShredded" -> "true",
190+
"spark.sql.variant.pushVariantIntoScan" -> "false") {
191+
// `k` stays in the native data schema even though `SELECT v` never reads it: the
192+
// Unicode fail-closed guard must only inspect referenced columns, so this scan
193+
// succeeds natively.
194+
val (_, cometPlan) = checkSparkAnswer(sql(s"SELECT v FROM $table"))
195+
assert(collect(cometPlan) { case s: CometNativeScanExec => s }.size == 1)
196+
197+
// Referencing `k` alongside the Variant column still fails closed (#5495).
198+
val df = sql(s"SELECT v, `k` FROM $table")
199+
assert(collect(df.queryExecution.executedPlan) { case s: CometNativeScanExec =>
200+
s
201+
}.size == 1)
202+
val error = intercept[Exception](df.collect())
203+
val messages = Iterator
204+
.iterate(error: Throwable)(_.getCause)
205+
.takeWhile(_ != null)
206+
.map(_.getMessage)
207+
.mkString("\n")
208+
assert(messages.contains("Unicode case folding"), messages)
209+
}
210+
}
211+
}
212+
}
213+
173214
test("Variant scan falls back when Parquet encryption is configured") {
174215
assume(isSpark40Plus, "VariantType requires Spark 4.0+")
175216

0 commit comments

Comments
 (0)