sunchao commented on code in PR #5932:
URL: https://github.com/apache/datafusion-comet/pull/5932#discussion_r4012091659
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -911,6 +1088,29 @@ impl PhysicalExprAdapterFactory for
SparkPhysicalExprAdapterFactory {
None
};
+ // Compare the file's VARIANT annotations against the requested types
before handing the
+ // schemas to the default adapter. `adapted_physical_schema` is used
so that field-id and
+ // case-insensitive resolution has already aligned the two sides'
top-level names.
+ if !self.parquet_options.ignore_variant_annotation {
+ let mut physical_by_folded: HashMap<&str, usize> = HashMap::new();
+ for (i, name) in physical_folded.iter().enumerate() {
+ physical_by_folded.entry(name.as_str()).or_insert(i);
+ }
+ let mut path = Vec::new();
+ for (logical_field, folded) in
logical_file_schema.fields().iter().zip(&logical_folded)
Review Comment:
### Correctness
[P2] Could this validation use the actual requested read schema rather than
every field of `logical_file_schema`? `CometNativeScan` keeps an ordinary
struct in `nativeDataSchema` even when it is unprojected, and
`init_datasource_exec` passes that full schema to `ParquetSource` with a
separate projection. DataFusion 55.1.0 consequently calls this factory with the
full schema before rewriting the projection. For a file with `id INT, v
VARIANT(1)`, a read declared as `id INT, v STRUCT<value BINARY, metadata
BINARY>` followed by `.select("id")` now fails on `v`, although no read of `v`
was requested. Spark clips its Parquet schema to the requested columns before
conversion. Please keep the eager check for requested fields, including empty
files, and add a regression covering an omitted annotated root.
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -434,6 +434,183 @@ fn is_string_or_binary(dt: &DataType) -> bool {
)
}
+/// Approximate Spark's `DataType.sql` for the variant rejection message.
`spark_catalog_name`
+/// bottoms out at "unknown" for nested types, which is exactly the shape a
hand-written Variant
+/// read schema has, so the nested cases are spelled out here.
+///
+/// This is close to `DataType.sql` but not identical. Spark's
`StructField.sql` wraps names in
+/// `QuotingUtils.quoteIfNeeded` and appends the nullability and comment DDL,
none of which is
+/// reproduced here, so a field named `has space` renders bare. The whole
message already differs
+/// from Spark's by design (see `variant_annotation_err`), and the rendering
exists to tell a user
+/// which type they asked for, so it is kept simple rather than made
byte-identical.
+fn spark_read_type_name(dt: &DataType) -> String {
+ match dt {
+ DataType::Struct(fields) => {
+ let rendered = fields
+ .iter()
+ .map(|field| {
+ format!(
+ "{}: {}",
+ field.name(),
+ spark_read_type_name(field.data_type())
+ )
+ })
+ .collect::<Vec<_>>()
+ .join(", ");
+ format!("STRUCT<{rendered}>")
+ }
+ DataType::List(item) | DataType::LargeList(item) => {
+ format!("ARRAY<{}>", spark_read_type_name(item.data_type()))
+ }
+ DataType::Map(entries, _) => match entries.data_type() {
+ DataType::Struct(fields) if fields.len() == 2 => format!(
+ "MAP<{}, {}>",
+ spark_read_type_name(fields[0].data_type()),
+ spark_read_type_name(fields[1].data_type())
+ ),
+ other => format!("MAP<{}>", spark_read_type_name(other)),
+ },
+ other => spark_catalog_name(other).to_uppercase(),
+ }
+}
+
+/// Build the carrier for a Parquet field whose VARIANT logical type
annotation is incompatible
+/// with the requested Spark read type. The JVM shim turns these into Spark's
+/// `_LEGACY_ERROR_TEMP_3071` `AnalysisException`.
+///
+/// `column` is the dotted path to the offending field. Spark's own message
interpolates the
+/// Parquet `Type` itself, roughly `optional group v (VARIANT(1)) { ... }`, so
the two messages
+/// deliberately differ. `ParquetVariantShreddingSuite` matches on `Invalid
Spark read type` and
+/// accepts either, and a path is the more useful of the two once the field is
nested.
+fn variant_annotation_err(column: &str, target_type: &DataType) ->
DataFusionError {
+
DataFusionError::External(Box::new(SparkError::ParquetVariantAnnotationMismatch
{
+ file_path: String::new(),
+ column: column.to_string(),
+ spark_type: spark_read_type_name(target_type),
+ }))
+}
+
+/// Whether `field` is marked as a Variant. On a physical field the marker is
the Parquet VARIANT
+/// logical type annotation, which arrow-rs surfaces as the
`arrow.parquet.variant` Arrow
+/// extension type for struct fields, list elements and map values alike. On a
requested field it
+/// is the same marker Comet's serde applies for `VariantType`.
`check_variant_annotation` relies
+/// on both sides using it, so it compares them through this one predicate.
+///
+/// The marker does not carry the annotation's spec version, and Spark keys
its handling off that
+/// version. `ParquetToSparkSchemaConverter.convertGroupField` matches
+/// `case v: VariantLogicalTypeAnnotation if v.getSpecVersion == 1`, and an
annotation failing
+/// that guard falls through to `case _ => throw
unrecognizedParquetTypeError(...)`, which is
+/// `PARQUET_TYPE_NOT_RECOGNIZED`. arrow-rs matches `LogicalType::Variant(_)`
for any version, so
+/// this predicate cannot tell the two apart. Spark's writer emits version 1.
An Arrow-based
+/// writer emits no version at all, which parquet-java reads back as 0.
+///
+/// Both engines therefore reject a non-v1 annotation, with different error
classes: Spark raises
+/// `PARQUET_TYPE_NOT_RECOGNIZED` and Comet raises `_LEGACY_ERROR_TEMP_3071`.
+/// `variant_annotation_without_a_spec_version_is_also_rejected` pins that.
+///
+/// The one behavioral gap is `ignore_variant_annotation` on a non-v1
annotation. Spark's guard
+/// fails before it reaches its own ignore branch, so Spark still raises
+/// `PARQUET_TYPE_NOT_RECOGNIZED`, while Comet skips the check entirely and
reads the plain
+/// struct. Comet is the more permissive of the two there. Closing it would
mean pairing the
+/// requested schema against the Parquet `SchemaDescriptor` rather than the
Arrow schema, which
+/// duplicates the field-id and case-folding rules `check_variant_annotation`
reuses. Closing it
+/// means either reading the version from the Parquet schema or arrow-rs
carrying it on the
+/// extension type.
+fn is_variant_marked(field: &Field) -> bool {
+ field.has_valid_extension_type::<VariantType>()
+}
+
+/// Reject reading a Parquet field that carries the VARIANT logical type
annotation as anything
+/// other than Spark's `VariantType`, mirroring the
`checkConversionRequirement` in Spark's
+/// `ParquetToSparkSchemaConverter.convertGroupField`.
+///
+/// Comet's scan never runs Spark's schema converter, so this is the only
place the file's
+/// annotation is ever compared against the requested type. The two sides are
compared
+/// symmetrically through `is_variant_marked`: a marked request is a
legitimate Variant read (see
+/// `parquet_exec::init_datasource_exec`'s `projects_variant`) and must not be
rejected.
+/// Identifying a Variant by its `value`/`metadata` child names instead would
misclassify ordinary
+/// structs, which is what apache/datafusion-comet#5741 rules out.
+///
+/// Runs once per file when the reader opens it, before any row group is
inspected, so a file with
+/// no row groups is rejected too. That matches Spark, which rejects while
converting the schema,
+/// and is the opposite of the type-promotion checks in this module, which
`RejectOnNonEmpty`
+/// defers to execution to match Spark's per-row-group behavior.
+fn check_variant_annotation(
+ logical: &FieldRef,
+ physical: &FieldRef,
+ parquet_options: &SparkParquetOptions,
+ path: &mut Vec<String>,
+) -> DataFusionResult<()> {
+ path.push(logical.name().clone());
+ if is_variant_marked(physical) && !is_variant_marked(logical) {
+ return Err(variant_annotation_err(&path.join("."),
logical.data_type()));
+ }
+ match (logical.data_type(), physical.data_type()) {
+ (DataType::Struct(logical_fields), DataType::Struct(physical_fields))
=> {
+ // Pair nested fields the same way `spark_parquet_convert` does,
so the field this
+ // check inspects is the one the read will actually pull from the
file: when the
+ // logical struct carries Parquet field IDs anywhere, ID-bearing
logical fields match
+ // ONLY by ID and the rest fall back to the folded name. A logical
field with no
+ // counterpart in the file is null-filled and carries no
annotation to check.
+ let should_match_by_id = parquet_options.use_field_id
+ && logical_fields.iter().any(|f| parse_field_id(f).is_some());
+ let physical_id_to_index: HashMap<i32, usize> = if
should_match_by_id {
+ let mut map = HashMap::new();
+ for (i, field) in physical_fields.iter().enumerate() {
+ if let Some(id) = parse_field_id(field) {
+ map.entry(id).or_insert(i);
+ }
+ }
+ map
+ } else {
+ HashMap::new()
+ };
+ let physical_names = physical_fields
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>();
+ let physical_folded = fold_names(&physical_names,
parquet_options.case_sensitive);
+ // First match wins on a folded-name collision, as
`spark_parquet_convert` does when it
+ // falls through its ambiguity check; `collect` would arbitrarily
keep the last.
+ let mut folded_to_index: HashMap<&str, usize> = HashMap::new();
+ for (i, folded) in physical_folded.iter().enumerate() {
+ folded_to_index.entry(folded.as_str()).or_insert(i);
+ }
+
+ for logical_child in logical_fields {
+ let physical_index = match (should_match_by_id,
parse_field_id(logical_child)) {
+ (true, Some(id)) => physical_id_to_index.get(&id).copied(),
+ _ => {
+ let folded =
+ fold_name(logical_child.name(),
parquet_options.case_sensitive);
+ folded_to_index.get(folded.as_str()).copied()
+ }
+ };
+ if let Some(i) = physical_index {
+ check_variant_annotation(
+ logical_child,
+ &physical_fields[i],
+ parquet_options,
+ path,
+ )?;
+ }
+ }
+ }
+ (DataType::List(logical_item), DataType::List(physical_item))
+ | (DataType::LargeList(logical_item),
DataType::LargeList(physical_item))
+ | (DataType::List(logical_item), DataType::LargeList(physical_item))
+ | (DataType::LargeList(logical_item), DataType::List(physical_item))
=> {
+ check_variant_annotation(logical_item, physical_item,
parquet_options, path)?;
+ }
+ (DataType::Map(logical_entries, _), DataType::Map(physical_entries,
_)) => {
Review Comment:
### Correctness
[P2] Could the map branch validate the key and value by position instead of
recursing through the entries struct's name/field-ID matcher? The requested
Arrow map uses `key`/`value`, but parquet-rs retains the file's child names. If
the second child is named `payload` and carries `VARIANT(1)`, the struct
matcher finds no `value` field and skips its annotation. Both
`check_conversion` and `parquet_convert_map_to_map` still read that second
child positionally, so the incompatible plain-struct map value gets past this
new guard. Spark's map reader also selects children 0 and 1 rather than
requiring those names. Please use the same positional pairing here and cover an
annotated map value whose Parquet field name differs from `value`.
--
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]
