sunchao commented on code in PR #5407:
URL: https://github.com/apache/datafusion-comet/pull/5407#discussion_r3880260048
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -450,23 +476,63 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
// a field with multiple case-insensitive matches in the physical
schema.
// Only the columns actually referenced trigger the error (not the
whole schema).
if let Some(orig_physical) = &self.original_physical_schema {
+ // Until #5495 brings Spark-parity Unicode name matching,
Variant-bearing scans
+ // fail closed for referenced columns whose resolution would
differ under Unicode
+ // folding, rather than silently null-filling values Spark reads.
Restricting the
+ // check to referenced columns keeps unused ordinary columns in
the native data
+ // schema from aborting a scan that never reads them, and the
per-field ID
+ // exemption mirrors the name-match eligibility in
`remap_physical_schema`:
+ // ID-bearing logical fields resolve by ID, everything else falls
through to the
+ // ASCII name matcher and needs the guard.
+ let variant_scan = self
+ .logical_file_schema
+ .fields()
+ .iter()
+ .any(|f| is_variant_field(f));
+ let should_match_by_id = self.parquet_options.use_field_id
+ && schema_has_field_ids(&self.logical_file_schema);
// Walk the expression tree to find Column references
- let mut duplicate_err: Option<DataFusionError> = None;
+ let mut column_err: Option<DataFusionError> = None;
let _ = Arc::<dyn PhysicalExpr>::clone(&expr).transform(|e| {
if let Some(col) = e.downcast_ref::<Column>() {
if let Some((req, matched)) =
check_column_duplicate(col.name(), orig_physical)
{
- duplicate_err =
Some(DataFusionError::External(Box::new(
+ column_err = Some(DataFusionError::External(Box::new(
SparkError::DuplicateFieldCaseInsensitive {
required_field_name: req,
matched_fields: matched,
},
)));
}
+ if column_err.is_none() && variant_scan {
+ let id_matched = should_match_by_id
+ && self
+ .logical_file_schema
+ .field_with_name(col.name())
+ .ok()
Review Comment:
[P2] Apply the ID exemption before duplicate-name validation
A case-sensitive Spark Parquet write can produce valid, unshredded `v
VARIANT` with field ID 1 and `V INT` with field ID 2. Read it through an
explicit schema containing only `v VARIANT`/ID 1 with case-insensitive field-ID
reading, `allowReadingShredded=true`, `pushVariantIntoScan=false`, and
`parquet.filterPushdown=false` (no predicates). [Spark resolves the unique ID
before name
matching](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala#L445-L473).
Here `check_column_duplicate` has already recorded the `v`/`V` collision, so
`column_err.is_none()` prevents this per-field ID exemption from applying and
the query fails with `_LEGACY_ERROR_TEMP_2093`. BASE left this required Variant
on Spark; the PR newly admits it to this failing native path. Please exempt
ID-resolved columns from name-duplicate validation too, while retaining
duplicate-ID checks. This is
source-verified, not a locally executed reproduction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
