fix: validate Parquet VARIANT annotation against the requested read type

[0lai0]

Which issue does this PR close?

Closes #5741.

Rationale for this change

Reading a VARIANT-annotated Parquet field as an ordinary struct<value binary, metadata binary> silently returned the storage bytes. Spark rejects that read in ParquetToSparkSchemaConverter.convertGroupField with _LEGACY_ERROR_TEMP_3071 unless spark.sql.parquet.ignoreVariantAnnotation is set.

CometScanRule cannot catch it: it only sees the requested schema, and a hand-written struct carries none of the VariantMetadata that isVariantStruct looks for. The annotation lives in the file, so the check has to happen in the native reader.

What changes are included in this PR?

arrow-rs surfaces the annotation as the arrow.parquet.variant Arrow extension type, and Comet's serde marks a requested VariantType the same way. check_variant_annotation compares the two sides symmetrically in SparkPhysicalExprAdapterFactory::create, so a marked request stays a legitimate Variant read and only an unmarked request against an annotated file is rejected. Matching on value/metadata child names instead would misclassify ordinary structs, which #5741 rules out.

• spark.sql.parquet.ignoreVariantAnnotation plumbed through NativeScanCommon, read by key since the conf is 4.1-only while this file compiles against 3.4 through 4.1.
• Gated to Spark 4.1+. The converter branch and the conf both arrived in 4.1, so on 3.4, 3.5 and 4.0 the check is off rather than inventing a failure Spark does not have.
• Only roots in the Spark read schema are checked. Spark's schema pruning keeps unrequested top-level fields in the relation's data schema, which Comet passes with a separate projection, so init_datasource_exec hands the factory its required_schema through with_required_schema. Requested roots are walked with their requested type, matching Spark clipping the Parquet schema before converting it.
• Struct fields are resolved through match_struct_fields, the same field-id, case-folding and ambiguity rules the conversion uses. Map children are paired by position, since MapArray::keys/values and Spark's converter read children 0 and 1 regardless of their names. List elements are paired directly.
• Runs at file open, before any row group, matching Spark rejecting during schema conversion. An empty file fails too when the annotated field is requested.
• SparkError::ParquetVariantAnnotationMismatch converted by the 4.x shim into Spark's AnalysisException wrapped in FAILED_READ_FILE.
• dev/diffs/4.1.3.diff regenerated per spark-sql-tests.md to drop the IgnoreComet exclusion.

Known gap, documented on is_variant_marked: the extension type carries no spec version, so with ignoreVariantAnnotation=true on a non-v1 annotation Comet reads a file Spark refuses. Both engines reject it when the conf is off, with different error classes.

How are these changes tested?

ParquetVariantShreddingSuite / variant logical type annotation - ignore variant annotation is no longer excluded. I ran the suite locally against Spark 4.1.3 with ENABLE_COMET=true: 7 tests pass, none ignored. With the check disabled it fails with Expected exception org.apache.spark.SparkException to be thrown, but no exception was thrown, the failure #5741 reports, so the suite is exercising this change.

21 native tests:

• schema_adapter.rs covers rejection at each nesting shape, a map value whose Parquet name is not value, a real VARIANT(1) file written through the low-level writer, ignoreVariantAnnotation=true, a marked Variant request not being rejected, an empty file, nested field-id resolution, an annotated field pruned out of the read schema, and probes asserting the annotation still reaches create as an extension type.
• parquet_exec/variant_tests.rs drives init_datasource_exec with the full data schema, a read schema and a projection, as CometNativeScan does. An annotated root outside the read schema reads successfully and one inside it is rejected, in both a file with rows and a file with no row group.

[sunchao]

Correctness

This fixes the missing error in #5741: a manually declared ordinary struct can reach the native reader even when the Parquet field is annotated as VARIANT. Checking the physical extension marker against the requested marker is the right distinction. Matching only the storage child names would incorrectly classify ordinary structs. The change carries the ignore setting through the scan protocol, gates the Parquet datasource check to Spark 4.1+, and translates the new error through the Spark 4.x shim.

I found two [P2] correctness issues, detailed inline. The factory validates the full relation schema rather than the actual projection, so selecting an unrelated scalar column can fail because an omitted column has a VARIANT annotation. The map recursion also matches the entries children by name or field ID, while the reader consumes map keys and values by position. A differently named physical value field can therefore bypass the annotation check.

The existing field-ID remap preserves physical metadata, and absent requested fields do not acquire an annotation to reject. Marked Variant requests and the explicit ignore setting remain accepted by the guard. The PR documents a remaining version limitation: the Arrow marker loses the annotation's spec version. That limitation predates this guard's acceptance path and is not counted as a new finding here.

Validation

Reviewed 03f9da00857d469afb7dccdf6bcfdd1f3dd8dcb4 against 38a6ec362096c7c205b379c5a5d7a7fc81c6b9e9, including all 13 changed files. I checked maintained Spark 3.5/4.0 sources for schema clipping and map matching, plus the exact DataFusion 55.1.0 and parquet-rs 59.3.0 sources verified against the lockfile checksums. An isolated Rust probe compiled and exercised the exact guard and factory block with minimal type doubles. It confirmed the two control-flow cases and checked marked requests, the ignore setting, and renamed or missing field IDs. It did not execute Arrow, DataFusion, JNI, Spark, or a real Parquet scan. Maintained Spark 3.4/4.1 branches were unavailable, so no coverage of those sources is claimed.

The PR adds 15 native cases and reports seven passing Spark 4.1.3 suite tests. I have not independently rerun them. At the September 15, 04:40 UTC refresh, only labeling passed. CI, CodeQL and the Delta gate were approval-required with zero jobs. The cached merge has the exact reviewed base/head parents and the same whole tree as the head, but there is no executed product CI to credit.

Performance

The guard walks in-memory schema metadata when the adapter is created. It adds no data-row loop or metadata fetch. It does allocate paths and lookup maps while visiting nested fields, including currently unrequested roots. Restricting validation to the actual read schema addresses that unnecessary work together with the first correctness issue. The existing ASCII fast path and cached non-ASCII name folding remain available. No benchmark or measured speedup is claimed for this validation change.

Design

The native reader is the appropriate boundary because the requested Spark schema alone cannot reveal a file annotation. An eager check also preserves rejection of an empty file when the incompatible field is requested. The check needs the same selection rules as the read itself: the requested schema for its scope, field-ID/name resolution for structs, and positional resolution for map children. Those changes address the findings without restricting legitimate projections or moving the error into per-row execution.

Abstraction & complexity

The option and structured error carrier fit the existing scan and error-conversion interfaces. The new recursive matcher duplicates struct-resolution logic already centralized in match_struct_fields. Reusing that helper for structs and handling map children explicitly would keep validation aligned with conversion, including ambiguity handling. The file-writing tests and probe factory have a useful purpose: they distinguish a real Parquet annotation reaching the adapter from an Arrow-schema hint that merely looks annotated. Their missing boundary cases are an omitted annotated root and a differently named map value, which the inline findings request.

[sunchao]

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.

[0lai0]

Thanks @sunchao. Sure, fixed. The validation now uses the requested read schema. init_datasource_exec passes required_schema to the factory via with_required_schema, so unrequested roots are skipped, and requested fields are still checked eagerly in create(), including empty files. Regressions in parquet_exec/variant_tests.rs cover an omitted annotated root (reads fine) and a requested one (rejected).

[sunchao]

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.

[0lai0]

Sure, fixed. The map branch now validates the key and value by position. variant_annotation_on_a_renamed_map_value_is_rejected covers an annotated map value named payload.

[sunchao]

Re-reviewed 8c1cb780428c22c731edb806ca2934a1d02cfd84. Both prior P2 findings are addressed: projection scope now uses the requested schema, and map validation pairs children by position. The shared struct matcher also removes the duplicated matching logic. Six new regression cases cover the fixes, including requested and unrequested annotated roots in empty files.

There is one new [P1] build integration issue, detailed inline: current main makes fold_schema_names fallible, while the new requested-schema loop treats its return value as a vector. This produces E0308 in the current merge result 3aa01e9b3cd7408e2d6cf6af3f604e6ef64ac2dc.

I reran the prior isolated reproductions and checked 19 current control-flow cases successfully. A separate compiler probe verifies that the exact new block compiles with the head's older API, fails with the merge's current API, and compiles when the error is propagated. These use minimal type doubles. I have not run the full native or Spark suites. Maintained Spark 3.4/4.1 sources remain unavailable. At the September 16, 18:46 UTC refresh, only labeling passed. CI, CodeQL and the title check still required workflow approval.

[sunchao]

Correctness

[P1] Propagate the fallible fold before building this map

Current main changed fold_schema_names to return DataFusionResult<Vec<String>> in #5845. This new call remains unchanged in the PR's merge commit. Iterating that Result yields a Vec<String>, so the closure produces Option<HashMap<Vec<String>, &FieldRef>> instead of the declared Option<HashMap<String, &FieldRef>>, and the current merge result cannot compile. A bounded rustc reproduction using the exact added block and folding function confirms E0308. the same block compiles with the head's older folding API. Please update this call to propagate the fold error before zipping, and make the optional-schema closure fallible (for example, return a Result and use transpose()), when updating to current main.

[0lai0]

Done. Merged current main, and the requested-schema fold now propagates its error through a fallible closure and transpose().

[sunchao]

Re-reviewed 1c714fac against 8c1cb780 and base 58ab5f61. The prior [P1] compilation issue is fixed: the requested-schema fold now propagates errors through the fallible closure and transpose(). I reran the original E0308 reproduction. The old merge block still fails, while the exact current head/merge block compiles and passes five controls, including injected fold failure.

Both earlier P2 fixes remain intact. The 19 isolated projection, nested-schema, missing-field, case-matching and positional-map cases pass. The full contribution adds no other authored behavior change since the last review. No new or remaining verified P1/P2 findings.

These are source checks and isolated compiler probes with type doubles, not a full native build or Spark/JNI execution. Maintained Spark 3.4/4.1 sources remain unavailable. At September 17, 08:33 UTC, CI and CodeQL require approval with zero jobs. The successful label job checked out the base and provides no product validation.