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.

[sunchao]

@0lai0 sorry for the late reply - could you resolve the conflict? I'll merge this afterwards

[0lai0]

Resolved, Thanks @sunchao

[sunchao]

Oops I'm sorry @0lai0 ! could you resolve the conflict again?

[0lai0]

No problem, @sunchao. Merged main

[andygrove]

This is a light fully automated review since there are so many PRs open.

The physical side of the check at native/core/src/parquet/schema_adapter.rs:549 is the reader's Arrow field, and the doc on is_variant_marked (line 497) treats that marker as the file's VARIANT annotation. As far as I can tell that only holds when the file has no embedded ARROW:schema. When it has one, convert_field in parquet 59.3's arrow/schema/complex.rs copies each field's metadata from the hint and never derives the extension from the Parquet logical type. For a plain-struct request projects_variant is false, so parquet_exec.rs:198 keeps the hint. A concrete case is a file written by arrow-rs ArrowWriter without the non-default variant_experimental feature, from a column built with parquet-variant-compute's VariantArray::field. logical_type_for_struct returns None without that feature, so the group has no annotation but the hint carries arrow.parquet.variant. Spark 4.1 infers and reads that column as struct<metadata binary, value binary>, and main reads it the same way, but with this change the read fails with _LEGACY_ERROR_TEMP_3071. The reverse holds too: an annotated group whose hint lacks the marker still returns the storage bytes. Could the physical side come from the Parquet schema instead, for example by reconciling the hint's Variant markers with the file's annotations in EagerPageIndexReaderFactory, which already rewrites the hint for Variant projections? A test that writes an unannotated group with a hand-built ARROW:schema carrying the marker would pin this down.

[andygrove]

dev/diffs/4.2.0.diff landed on main in #4950. It has the same IgnoreComet("https://github.com/apache/datafusion-comet/issues/5741") on "variant logical type annotation - ignore variant annotation" that this PR removes from the 4.1 diff. Since this PR closes #5741, could you remove it from the 4.2 diff too? Otherwise the Spark 4.2 suite will keep skipping that test after the fix lands. The Spark SQL Tests guide covers regenerating a diff.

I've added the run-spark-4.2-tests label so CI runs the Spark 4.2 SQL suite on this PR. The first run will fail in "Pre-compile Spark Test classes", because main's 4.2 diff still asks for Comet 1.1.0-SNAPSHOT (#6380). #6398 fixes that. Once it lands, merging main into this branch will rerun the suite and give a real result.

[0lai0]

Thanks @andygrove. You are right, and both directions were broken. With a hint present convert_field copies its metadata and never derives the extension from the logical type, so the marker the check read was the hint's, not the file's.

Fixed in EagerPageIndexReaderFactory, as you suggested: with_reconciled_variant_markers takes the hinted schema and the annotation-only schema (parquet_to_arrow_schema with and without the key-value metadata) and moves the markers to match the annotations, leaving everything else the hint decides alone. A file with no hint returns after one scan of the key-value metadata.

Two tests, one per direction, each writing the data with one schema and the hint arrow-rs writes for another. Both fail with the reconciliation disabled.

Also worth flagging: probe_variant_annotation built its own ParquetSource and never installed the reader factory, so no Variant test exercised the footer rewrites and none of them could have caught this. It installs the factory now.

cargo test --lib: 530 passed. dev/local-ci.sh spark sql_core-1 (4.1.3): 12835 passed, 0 failed.

On the 4.2 diff: will do. It is not on this branch yet — #4950 landed it after my last merge — so I will merge main and regenerate it without the IgnoreComet. Waiting for #6398 first, since until it lands the 4.2 run stops in "Pre-compile Spark Test classes" and tells us nothing about the fix.

[0lai0]

Status summary

Comet's scan never runs Spark's schema converter, so nothing compared a Parquet file's VARIANT annotation against the type the read asked for: a VARIANT column read as struct<value binary, metadata binary> silently returned the storage bytes where Spark raises _LEGACY_ERROR_TEMP_3071. This adds that check. Closes #5741.

The five commits, in reading order

Commit	What it does
5508366	The check itself: check_variant_annotation in schema_adapter.rs, the error carrier, the serde and proto plumbing for spark.sql.parquet.ignoreVariantAnnotation
03f9da0	Re-enables the Spark test the 4.1 diff was skipping for #5741
8c1cb78	Scopes the check to the requested read schema, and pairs map children by position the way MapArray reads them
96087ba	Review fix: takes the marker from the file's annotation rather than its ARROW:schema hint (below)
a29268d	Stops skipping the same Spark test on 4.2, whose diff landed on main in #4950

Both review points are addressed

@andygrove found that the physical side of the check was the reader's Arrow field, which parquet-rs builds from the ARROW:schema hint when a file carries one, never from the Parquet logical type.
That made the marker the hint's rather than the file's, so Comet rejected a column Spark reads (hint marked, group unannotated) and read one Spark rejects (the reverse). 96087ba reconciles the hint's Variant markers against the annotations in EagerPageIndexReaderFactory, before the adapter sees either, and adds a test per direction; both fail with the reconciliation disabled. It also installs the reader factory in probe_variant_annotation, which had been building its own ParquetSource without it — the reason no Variant test could have caught this.

a29268d regenerates dev/diffs/4.2.0.diff from a clean v4.2.0 clone per the Spark SQL Tests guide. The only change from the previous diff is that one hunk's removal.

[comphead]

The once-per-file check and the positional map pairing look right and sunchao has approved. Three comments, and a note that this needs another merge of main.

[comphead]

This runs on every open of a file that carries an ARROW:schema hint, whether or not the variant check is enabled. Each run does two parquet_to_arrow_schema conversions and decodes the hint, and the metadata cache keeps the original footer, so a cached file repeats the work on every open. The reconciled marker is only read by check_variant_annotation, and CometParquetUtils.ignoreVariantAnnotation is true below Spark 4.1. Comet's own writer, pyarrow and other arrow-rs writers emit the hint. Could this be skipped when ignore_variant_annotation is set? The options are already on the factory through with_conversion_check. I have not measured the cost.

[0lai0]

Ohh good catch, and it is worse than you put it: ignoreVariantAnnotation is !isSpark41Plus || conf, so every scan below Spark 4.1 paid for a rewrite whose only consumer never runs. Gated behind with_reconcile_variant_markers, set from !ignore_variant_annotation. Nothing else reads the physical side's marker, and the_footer_is_left_alone_when_the_annotation_check_is_off pins the skip.

[comphead]

The message says Comet rejects an unversioned annotation "that Spark would accept", but the doc on this test and the doc on is_variant_marked both say Spark rejects it with PARQUET_TYPE_NOT_RECOGNIZED. Could the message match them? The is_variant_marked doc around line 518 also ends with two similar "Closing it ..." sentences, so one can go.

[0lai0]

Both fixed. The message now reads "as Spark does", and the doc keeps only the sentence about reading the spec version.

[comphead]

variant_annotation_is_rejected_on_an_empty_file looks covered by requested_annotated_root_is_rejected_on_an_empty_file in variant_tests.rs, which goes through init_datasource_exec. The four variant_annotation_survives_* tests also each have a *_is_rejected twin for the same shape. A rejection can only happen when the marker reached create, so the twins may already prove what the survives tests check. Could those be merged or dropped? ProbeFactory, observed_physical and assert_annotation_survived would then be unused. variant_annotation_survives_with_embedded_arrow_schema looks like the one distinct shape.

[0lai0]

Dropped, 102 lines. You are right that the twins carry it: each asserts the full dotted path (v, ns.nv, av.element, mv.entries.value), which the error can only produce if the marker was found at that position. The empty-file reasoning moved onto requested_annotated_root_is_rejected_on_an_empty_file.

One correction: the helpers stay in use, since survives_with_embedded_arrow_schema still calls them. Clippy is clean.

[0lai0]

Merging main also extracted the serde body into buildNativeScanCommon, which contrib/delta-spark's CometDeltaNativeScan also calls, so setIgnoreVariantAnnotation now covers the Delta arm.
That is the behavior it should have had: before, Delta took the proto default and would raise _LEGACY_ERROR_TEMP_3071 on Spark versions whose converter has no variant branch at all.

893 tests pass, clippy and fmt clean, ./mvnw -DskipTests compile green.