sunchao commented on code in PR #5681:
URL: https://github.com/apache/datafusion-comet/pull/5681#discussion_r3935083920
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -312,39 +335,343 @@ fn is_string_or_binary(dt: &DataType) -> bool {
}
/// Build a Spark-shaped `SchemaColumnConvertNotSupportedException` carrier
for a
-/// rejected Parquet -> Spark conversion. The bracketed column wrapping mirrors
+/// rejected Parquet -> Spark conversion. `column` is the Spark-style column
path (`a`, or
+/// `s, x` for a nested leaf); the bracketed wrapping mirrors
/// `Arrays.toString(descriptor.getPath())` in Spark's vectorized reader.
fn parquet_schema_convert_err(
- field_name: &str,
+ column: &str,
physical_type: &DataType,
target_type: &DataType,
) -> DataFusionError {
DataFusionError::External(Box::new(SparkError::ParquetSchemaConvert {
file_path: String::new(),
- column: format!("[{}]", field_name),
+ column: format!("[{}]", column),
physical_type: parquet_primitive_name(physical_type).to_string(),
spark_type: spark_catalog_name(target_type),
}))
}
/// Build a `RejectOnNonEmpty` expr wrapping `child`. The rejection fires only
/// when the input batch is non-empty (mirrors Spark's per-row-group check).
+/// `column` is the Spark-style column path, as for
[`parquet_schema_convert_err`].
fn reject_on_non_empty_expr(
child: Arc<dyn PhysicalExpr>,
target_field: &FieldRef,
- field_name: &str,
+ column: &str,
physical_type: &DataType,
target_type: &DataType,
) -> Arc<dyn PhysicalExpr> {
Arc::new(RejectOnNonEmpty {
child,
target_field: Arc::clone(target_field),
- column: format!("[{}]", field_name),
+ column: format!("[{}]", column),
physical_type: parquet_primitive_name(physical_type).to_string(),
spark_type: spark_catalog_name(target_type),
})
}
+/// Outcome of checking one Parquet (physical) -> Spark (logical) type pair
against the
+/// conversion rules of Spark's vectorized Parquet reader.
+enum ConversionCheck {
+ /// Spark has an updater for the pair (for a same-shape complex pair: for
every leaf).
+ Accept,
+ /// Spark rejects the pair; raised at plan time.
+ Reject(DataFusionError),
+ /// Spark rejects the pair, but only while decoding a row group, so the
rejection is
+ /// deferred to runtime via [`RejectOnNonEmpty`] (SPARK-26709). Carries
the offending
+ /// leaf's column path and physical / requested types for the error
message.
+ RejectOnNonEmpty {
+ column: String,
+ physical_type: DataType,
+ target_type: DataType,
+ },
+}
+
+/// Apply the rejection matrix of Spark's
`ParquetVectorUpdaterFactory.getUpdater` to a single
+/// physical/logical leaf pair. `column` is the Spark-style column path used
in the error (`a`
+/// for a top-level column, `s, x` for a nested leaf, mirroring
+/// `Arrays.toString(descriptor.getPath())`). The rules and their order are
exactly those the
+/// adapter applies to top-level columns; [`check_conversion`] applies them to
nested leaves.
+fn check_leaf_conversion(
+ physical_type: &DataType,
+ target_type: &DataType,
+ column: &str,
+ options: &SparkParquetOptions,
+) -> ConversionCheck {
+ // arrow-rs surfaces a column whose file carries an `ARROW:schema` with a
dictionary
+ // encoding as `Dictionary(_, value)`; Spark only ever sees the value's
Parquet type.
+ let physical_type = match physical_type {
+ DataType::Dictionary(_, value_type) => value_type.as_ref(),
+ other => other,
+ };
+ if physical_type == target_type {
+ return ConversionCheck::Accept;
+ }
+ let reject = || {
+ ConversionCheck::Reject(parquet_schema_convert_err(
+ column,
+ physical_type,
+ target_type,
+ ))
+ };
+ let reject_on_non_empty = || ConversionCheck::RejectOnNonEmpty {
+ column: column.to_string(),
+ physical_type: physical_type.clone(),
+ target_type: target_type.clone(),
+ };
+
+ // Reject reading a string/binary Parquet column as anything else. Spark's
+ // `ParquetVectorUpdaterFactory.getUpdater` BINARY case allows StringType /
+ // BinaryType, or DecimalType only when the column carries a
+ // `DecimalLogicalTypeAnnotation` (which arrow-rs surfaces as `Decimal128`,
+ // not `Binary`). Without this guard, runtime cast paths silently return
+ // nulls, parse strings, or surface as a generic Arrow type-mismatch error.
+ // See #4088 and #4351.
+ if is_string_or_binary(physical_type) && !is_string_or_binary(target_type)
{
+ return reject();
+ }
+
+ // Reject reading a primitive numeric Parquet column as StringType /
+ // BinaryType. Spark has no `int -> string` etc. updater. Defer to
+ // runtime via `RejectOnNonEmpty` so empty Parquet files (SPARK-26709)
+ // pass and the JVM shim translates to
+ // `SchemaColumnConvertNotSupportedException`.
+ let physical_is_primitive_numeric = matches!(
+ physical_type,
+ DataType::Boolean
+ | DataType::Int8
+ | DataType::Int16
+ | DataType::Int32
+ | DataType::Int64
+ | DataType::Float32
+ | DataType::Float64
+ );
+ if physical_is_primitive_numeric && is_string_or_binary(target_type) {
+ return reject_on_non_empty();
+ }
+
+ // Decimal-to-decimal narrowing. Spark's `isDecimalTypeMatched` (the
+ // `DecimalLogicalTypeAnnotation` branch) allows the read only when
+ // `dst_scale >= src_scale` AND
+ // `dst_precision - dst_scale >= src_precision - src_scale`.
+ // Either failure means silently dropping fractional digits or losing
+ // integer-side magnitude. See #4089 and #4343.
+ if let (DataType::Decimal128(src_p, src_s), DataType::Decimal128(dst_p,
dst_s)) =
+ (physical_type, target_type)
+ {
+ let src_int_precision = i32::from(*src_p) - i32::from(*src_s);
+ let dst_int_precision = i32::from(*dst_p) - i32::from(*dst_s);
+ if dst_s < src_s || dst_int_precision < src_int_precision {
+ return reject();
+ }
+ }
+
+ // Integer-to-decimal narrowing. Spark's `canReadAsDecimal` requires
+ // `precision - scale >= 10` for an INT32 source and `>= 20` for INT64.
+ // Unconditional in all Spark versions, so reject at plan time. See #4344.
+ let int_decimal_min_int_precision = match physical_type {
+ DataType::Int8 | DataType::Int16 | DataType::Int32 => Some(10i32),
+ DataType::Int64 => Some(20i32),
+ _ => None,
+ };
+ if let Some(min_int_precision) = int_decimal_min_int_precision {
+ let dst_precision_scale = match target_type {
+ DataType::Decimal128(p, s) | DataType::Decimal256(p, s) =>
Some((*p, *s)),
+ _ => None,
+ };
+ if let Some((dst_p, dst_s)) = dst_precision_scale {
+ let dst_int_precision = i32::from(dst_p) - i32::from(dst_s);
+ if dst_int_precision < min_int_precision {
+ return reject();
+ }
+ }
+ }
+
+ // Type promotion (widening). When `allow_type_promotion` is false,
+ // reject the three widenings (INT32→INT64, FLOAT→DOUBLE, INT32→DOUBLE)
+ // that Spark 3.x's vectorized reader rejects. The flag tracks Comet's
+ // per-Spark-version constant in ShimCometConf. Deferred to runtime so
+ // empty files (SPARK-26709) pass.
+ if !options.allow_type_promotion {
+ let is_disallowed_promotion = matches!(
+ (physical_type, target_type),
+ (DataType::Int32, DataType::Int64)
+ | (DataType::Float32, DataType::Float64)
+ | (DataType::Int32, DataType::Float64)
+ );
+ if is_disallowed_promotion {
+ return reject_on_non_empty();
+ }
+ }
+
+ // Reject primitive Parquet conversions Spark's vectorized reader rejects
+ // on every supported version (no matching branch in
+ // `ParquetVectorUpdaterFactory.getUpdater`):
+ //
+ // - `INT64 -> Int*` truncates lower bits.
+ // - `INT64 -> Float*` and `INT32 -> Float32` lose precision.
+ // - `Float* -> Int*` and `Float64 -> Float32` truncate / overflow.
+ // - `INT32 -> Timestamp` / `INT64 -> Date32` / `INT64 -> Timestamp`:
+ // date/timestamp-annotated columns surface as Date32 / Timestamp,
+ // so reaching this branch means the column was un-annotated.
+ // - `Date32 -> Timestamp(LTZ)`: Spark only allows Date -> TimestampNTZ.
+ // - `Timestamp -> Date32`: no Timestamp updater branches into Date.
+ //
+ // Deferred to runtime (SPARK-26709). See #4297.
+ let is_spark_rejected_conversion = matches!(
+ (physical_type, target_type),
+ // Long -> narrower int.
+ (
+ DataType::Int64,
+ DataType::Int8 | DataType::Int16 | DataType::Int32,
+ )
+ // Long -> floating point.
+ | (DataType::Int64, DataType::Float32 | DataType::Float64)
+ // Long -> date / timestamp (raw INT64; annotated columns surface as
Date32/Timestamp).
+ | (DataType::Int64, DataType::Date32)
+ | (DataType::Int64, DataType::Timestamp(_, _))
+ // Int -> float (DoubleType is allowed via IntegerToDoubleUpdater;
FloatType is not).
+ | (
+ DataType::Int8 | DataType::Int16 | DataType::Int32,
+ DataType::Float32,
+ )
+ // Int -> timestamp (raw INT32; DATE-annotated columns surface as
Date32).
+ | (
+ DataType::Int8 | DataType::Int16 | DataType::Int32,
+ DataType::Timestamp(_, _),
+ )
+ // Float -> int / Double -> int (no integer branches under
FLOAT/DOUBLE).
+ | (
+ DataType::Float32 | DataType::Float64,
+ DataType::Int8 | DataType::Int16 | DataType::Int32 |
DataType::Int64,
+ )
+ // Double -> float (narrowing).
+ | (DataType::Float64, DataType::Float32)
+ // Date -> Timestamp(LTZ). Spark allows Date -> TimestampNTZ only.
+ | (DataType::Date32, DataType::Timestamp(_, Some(_)))
+ // Timestamp -> Date.
+ | (DataType::Timestamp(_, _), DataType::Date32)
+ );
+ if is_spark_rejected_conversion {
+ return reject_on_non_empty();
+ }
+
+ // Spark 3.x refuses to read a Parquet TimestampLTZ column as
+ // TimestampNTZ (SPARK-36182); Spark 4.0 (SPARK-47447) lifted that.
+ // The flag tracks Comet's per-Spark-version constant in
+ // ShimCometConf. Deferred to runtime so empty files (SPARK-26709)
+ // still pass. See #4219.
+ //
+ // This catches all LTZ physical encodings: TIMESTAMP_MICROS /
+ // TIMESTAMP_MILLIS arrive as `Timestamp(_, Some(_))` directly, and
+ // INT96 arrives as `Timestamp(_, Some("UTC"))` because `coerce_int96_tz`
+ // attaches the UTC timezone (see `get_options`) instead of letting
+ // `coerce_int96` strip it to a timezone-free `Timestamp(_, None)`.
+ if !options.allow_timestamp_ltz_to_ntz
+ && matches!(
+ (physical_type, target_type),
+ (
+ DataType::Timestamp(_, Some(_)),
+ DataType::Timestamp(_, None)
+ )
+ )
+ {
+ return reject_on_non_empty();
+ }
+
+ // Scalar/complex mismatch (e.g. TIMESTAMP read as ARRAY<TIMESTAMP>):
+ // Spark's vectorized reader rejects with
+ // SchemaColumnConvertNotSupportedException (SPARK-45604). Same-shape
+ // complex pairs never reach this leaf check (`check_conversion` walks
their
+ // leaves instead), so two complex types here differ in shape (e.g. STRUCT
+ // read as ARRAY), which Spark rejects just the same.
+ let is_complex = |t: &DataType| {
+ matches!(
+ t,
+ DataType::Struct(_) | DataType::List(_) | DataType::Map(_, _)
+ )
+ };
+ if is_complex(physical_type) || is_complex(target_type) {
+ return reject();
+ }
+
+ ConversionCheck::Accept
+}
+
+/// Check a physical/logical type pair the way Spark's vectorized reader does.
Spark runs
+/// `getUpdater` on every *leaf* column regardless of nesting, so same-shape
complex pairs
+/// (struct / list / map, at any depth) are walked and
[`check_leaf_conversion`] is applied to
+/// each leaf, extending the column path the way `descriptor.getPath()` does
(struct field
+/// names, the map entries field plus its `key` / `value`, the list element
field). Requested
+/// struct fields resolve to file fields with the same field-id / case-fold
rules the runtime
+/// convert uses ([`match_struct_fields`]); requested fields missing from the
file are skipped
+/// (they read as null / default, as before). The first non-`Accept` verdict
in leaf order
+/// wins, like Spark, which raises for the first offending column it
initializes.
+fn check_conversion(
+ physical_type: &DataType,
+ target_type: &DataType,
+ column: &str,
+ options: &SparkParquetOptions,
+) -> DataFusionResult<ConversionCheck> {
+ match (physical_type, target_type) {
+ (DataType::Struct(physical_fields), DataType::Struct(target_fields))
=> {
+ let physical_indices = match_struct_fields(physical_fields,
target_fields, options)?;
+ for (target_field, physical_index) in
target_fields.iter().zip(physical_indices) {
+ let Some(physical_index) = physical_index else {
+ continue;
+ };
+ let physical_field = &physical_fields[physical_index];
+ let check = check_conversion(
+ physical_field.data_type(),
+ target_field.data_type(),
+ &format!("{column}, {}", physical_field.name()),
+ options,
+ )?;
+ if !matches!(check, ConversionCheck::Accept) {
+ return Ok(check);
+ }
+ }
+ Ok(ConversionCheck::Accept)
+ }
+ (DataType::List(physical_item), DataType::List(target_item)) =>
check_conversion(
+ physical_item.data_type(),
+ target_item.data_type(),
Review Comment:
[P2] Update the nested-list fixture for the stricter leaf check
The new recursion correctly rejects the narrowing in
`execution::planner::tests::test_nested_types_list_of_struct_by_index`, but the
fixture still creates nested `a` with DataFusion SQL's untyped `1` (`Int64`)
and requests `Int32`. The [Rust CI
job](https://github.com/apache/datafusion-comet/actions/runs/33876125486/job/101034777878)
now fails with `[c0, item, a]`, `INT64 -> int`.
Using the same SQL and a real Parquet scan, the base adapter succeeds and
the head adapter rejects it. An explicitly typed `CAST(1 AS INT)` succeeds with
the head adapter and preserves the intended `a`/`c` projection. Could you type
the fixture as `INT`, or change its requested and expected type to `Int64`?
This fixes the deterministic CI failure without weakening the new rejection
rule, which agrees with both maintained Spark branches.
--
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]
