feat: support direct Variant projection in native Parquet scans

[peterxcli]

Which issue does this PR close?

Closes #5551.
Closes #5546.

Rationale for this change

Complete the ordinary-Parquet admission step in #5546 so Spark can project whole Variant values through a native scan.

What changes are included in this PR?

Enable direct top-level Variant projection with allowReadingShredded=true and pushVariantIntoScan=false. Carry missing-column defaults as constant [value, metadata] structs while preserving their schema indexes, and retain Spark fallbacks for unsupported consumers and reader settings.

Reuse the Variant normalizer to handle Spark's empty object keys until Comet upgrades to an Arrow release containing apache/arrow-rs#10352. Document the supported surface and add removal TODOs for the compatibility paths, including #5477.

How are these changes tested?

Focused Spark 4.0/4.1 tests cover value parity, defaults, Unicode field matching, vector layout, and fallback behavior. Native Variant tests and clippy pass, and Spark 3.5 compiles. The new projection suite runs in Linux and macOS CI.

[sunchao]

Reviewed 6e556c94 against base 8b818b53. No verified P1/P2 findings.

Correctness

Previously, the native Parquet reader had Variant storage adaptation, but the scan and execution rules still rejected direct Variant output. This change admits a Variant at a required top-level Parquet field and preserves Spark's logical Variant identity over binary children ordered [value, metadata]. Nested Variant projections and pushed Variant extraction remain on Spark. The default-value serializer now keeps each value paired with its original required-schema index; if any default cannot be serialized, the scan falls back instead of shifting later defaults. Native planning accepts only literals or a constant, correctly typed Variant storage struct and checks index bounds.

The reconstruction changes match the maintained Spark 4.0 source for missing object fields, required shredding states, scalar/array typed-value precedence, and Java UTF-16 object ordering. Adding shredded field names to the metadata dictionary also remaps residual field IDs. The empty-key retry checks the original metadata encoding before rebuilding it. Parent SQL nulls remain distinct from a Variant null. Spark handles strict-reader layout validation, nondefault timestamp inference, encryption, Variant consumers, and columnar-to-row conversion.

At 2026-09-12 19:35:47 UTC, the head has 60 successful and 10 skipped checks. The Spark 4.0 scan job passed 510 tests and the Spark 4.1 scan job passed 518; both executed all 11 new projection tests. Native CI passed the Variant reconstruction tests, and Spark 4.1 exec passed the Variant Arrow-representation and Python-fallback cases. These jobs checked out merge 020dfcb2 (6751af02 + 6e556c94), whose base and tree differ from the assigned pair. The authored Variant implementation and tests match, but DataFusion and schema-adaptation context differ, so this is qualified merge-CI evidence. No local product build or tests ran. Maintained Spark 3.4/4.1 branches were unavailable for source comparison; CI does not close those source gaps.

Performance

The implementation retains scan pruning: unread Variant roots are omitted from the native data schema, and pruning a Variant child can leave supported siblings eligible for native scanning. It reuses the existing scan and normalization path without adding an extra execution operator. Output buffers are rebuilt lazily when values actually need rewriting, preserving the unchanged path.

Shredded values still require recursive reconstruction, metadata lookups and, when keys are missing, per-row dictionary rebuilding and residual remapping. The empty-key compatibility retry adds work only after normalization fails. I found no verified performance regression, but this review has no benchmark establishing a speedup. Could you add a matched scan microbenchmark separating canonical, partially shredded, fully shredded and empty-key data, including normalization allocations? Please include repeated metadata dictionaries so the results show whether reconstruction does avoidable per-row work.

Design

The scan-only exception is appropriately bounded. Exempting CometScanExec from the execution-level Variant guard permits direct projection while retaining the guard for operators consuming or producing Variant. Spark's columnar-to-row path uses the two binary children through its Variant getter; the explicit Python guard covers both input and output. This keeps scan eligibility independent from expression, shuffle, write and Python support.

The settings that change physical interpretation are checked before conversion to a native scan. Unsupported defaults also fail planning as a whole. The tests check both returned values and the expected execution/fallback nodes, including present nulls versus missing defaults, later default indexes, Unicode field matching, encrypted files and malformed layouts.

Abstraction & complexity

The additional complexity is concentrated in the existing Variant normalization module and one narrowly scoped default-value helper. It does not create a second general expression evaluator or broaden Variant support in unrelated type checkers. The Spark 3/4 shim keeps version-specific Variant objects out of shared code.

The metadata-extension and empty-key paths share residual rewriting instead of maintaining separate reconstruction engines. Their comments identify the Arrow follow-ups that can remove these compatibility paths. Retaining those removal conditions and the buffer-reuse/null-state tests will help keep this temporary machinery contained. No additional abstraction or blocking simplification is needed for this change.

[andygrove]

The reconstruction path does not look like it produces the same Variant binary that Spark produces. Spark's ShreddingUtils.rebuild goes through VariantBuilder.appendLong, which narrows to the smallest integer encoding, and it builds a fresh metadata dictionary holding only the keys it emits. unshred_variant in arrow-rs maps an Int32 typed_value to Variant::Int32 and passes the file's dictionary through unchanged. VariantVal.equals is byte-wise, so checkAnswer on a Variant column sees those as different values.

That looks like the reason checkVariantAnswer compares value.toString rather than the value, and the reason VariantShreddingSuite.checkExpr had to be relaxed in dev/diffs/4.1.3.diff with the note that native unshredding may use different integer widths and metadata dictionaries. Since spark.sql.variant.inferShreddingSchema and spark.sql.variant.writeShredding.enabled both default to true on Spark 4.1 and later, shredded files are the common case rather than an edge case, and the differing bytes get persisted when someone reads through Comet and writes back out. Could the normalizer re-encode to Spark's canonical form so the output is byte for byte identical? If that is not practical right now, would it make sense to record the difference in the Variant section of datatypes.md and open a tracking issue instead? Rewriting checkExpr for every Variant-typed expectation also takes away that suite's ability to catch a future regression in the reconstruction.

Related to that, dev/diffs/4.1.3.diff picks up the change but dev/diffs/4.0.4.diff does not, and Spark 4.0's copy of VariantShreddingSuite has the same shape. Its testWithTempPath sets spark.sql.variant.allowReadingShredded=true and runs every case with spark.sql.variant.pushVariantIntoScan both true and false, and there are three checkExpr(path, "v", ...) call sites comparing the whole Variant byte-wise. The missing-fields case shreds int fields and expects {"a":1}, {"b":2} and {"a":3,"b":4}, which is the same integer-width situation. spark_4_0 is behind the run-spark-4.0-tests label so that job was skipped here and a failure would only surface after merge. Could you add the label and get a green Spark 4.0 run before this goes in?

The existence-default case looks like the one place Comet now returns a value Spark cannot produce. WritableColumnVector.appendObjects has no VariantVal branch, so Spark's vectorized reader raises Cannot assign default column value to result column batch in vectorized Parquet reader for a missing Variant column carrying an EXISTS_DEFAULT. What makes it worth a second look is that PushVariantIntoScan.addVariantFields skips a Variant column whose existence default is not null, so on Spark 4.1 and 4.2 defaults this is reachable with no config change and users quietly get a different answer. Would it be safer to fall back when a required Variant field carries an existence default, until Spark itself supports it? If the intent is to keep the improved behavior, could the two cases where v is present in the file use sparkRows(...) so at least those are checked against Spark rather than hand-written rows?

Smaller point on the docs. The paragraph in datatypes.md lists spark.sql.variant.pushVariantIntoScan=false as a requirement but does not say it defaults to true on Spark 4.1 and later, where PushVariantIntoScan rewrites even a bare SELECT v into a marked one-field struct that CometScanTypeChecker declines. Combined with allowReadingShredded being false on 4.0.x, the required pair holds by default on no supported version, so direct projection is opt-in everywhere. Could that be stated plainly and linked to #5519?