feat: support direct Variant projection in native Parquet scans

[peterxcli]

Which issue does this PR close?

Closes #5551.
Closes #5546.

Rationale for this change

Allow Spark to project whole Variant values through the native Parquet scan with matching bytes and fallback behavior.

What changes are included in this PR?

Enable opt-in top-level Variant projection with allowReadingShredded=true and pushVariantIntoScan=false. Rebuild shredded values with Spark's byte encoding, preserve unshredded bytes, and fall back for non-null Variant existence defaults and unsupported consumers or reader settings.

Keep the empty-key and missing-metadata compatibility workarounds linked from #5477. Document the required configuration and the remaining whole-value pushdown work in #5519.

How are these changes tested?

The projection tests and Spark's unchanged VariantShreddingSuite pass locally on Spark 4.0.4 and 4.1.3. A wrapper checks native scan use and runs the upstream assertions in regular Linux/macOS CI. Native casting tests and clippy pass.

Added matched scan and normalization-allocation benchmarks. Local measurements show canonical reads are close, ordinary shredded reads take 1.3–1.6 times as long as Spark, and empty-key reads take about twice as long. No performance benefit is claimed.

Benchmark results (September 16, 2026)

Environment: Apple M4, 24 GiB RAM, macOS 26.6.2, JDK 21.0.6, Spark 4.0.4, Rust 1.96.0,
DataFusion 55.0.0, Arrow/Parquet 59.3.0. Native code used the optimized ci profile with jemalloc
(no LTO, debug assertions enabled). The JVM heap was 4 GiB. Results are specific to this local,
warm filesystem workload; CPU placement and thermal state were not controlled.

Matched scans

CometVariantReadBenchmark writes one Parquet file per fixture with 100,000 repeated objects,
a 1,024-byte string payload, and Parquet dictionary encoding enabled. Both readers use the same
file and hash every returned Variant's value and metadata bytes through a Dataset action.
Planning, scanning, row conversion, and consumption are included; file creation is excluded.
The benchmark checks byte equality and native scan engagement before timing. Ordinary shredded
fixtures include all schema keys in metadata; the empty-key fixture exercises metadata repair.

Two runs reverse the reader order. Each case has 7–17 measured iterations after warmup.
Cells show average ± standard deviation in milliseconds, with Spark-first / Comet-first runs.

Fixture	Spark (ms)	Comet (ms)
Canonical	132 ± 8 / 128 ± 4	125 ± 4 / 136 ± 5
Partially shredded	153 ± 4 / 152 ± 1	242 ± 2 / 242 ± 5
Fully shredded	161 ± 4 / 147 ± 1	207 ± 3 / 209 ± 4
Empty key	148 ± 3 / 150 ± 2	310 ± 2 / 311 ± 3

After building and installing the ci library and Spark 4.0 artifacts:

SPARK_LOCAL_IP=127.0.0.1 make -o release \
  benchmark-org.apache.spark.sql.benchmark.CometVariantReadBenchmark \
  PROFILES=-Pspark-4.0 BENCH_HEAP=4g -- 100000 1024
# Repeat with --reverse-cases appended.

Native normalization allocations

The ignored benchmark_variant_buffer_reuse test isolates normalization with 4,096 rows,
4,096-byte payloads and repeated metadata dictionaries. It constructs Arrow inputs before
timing, warms up three batches, then normalizes 30 batches. Jemalloc's thread counter measures
cumulative allocated bytes, including temporary and output buffers. This measures allocation
traffic, not retained memory or allocation counts, and excludes Parquet decoding and JVM work.
These larger Arrow fixtures are separate from the scan fixtures above.

Fixture	Allocated bytes per row	Mean ms per batch
Canonical	10,332	3.762
Partially shredded	57,402	15.925
Fully shredded	52,320	11.796
Empty key	83,399	22.268

cd native
cargo test -p datafusion-comet --profile ci --features jemalloc \
  benchmark_variant_buffer_reuse --lib -- --ignored --nocapture

Shredded reconstruction currently pays for Arrow unshredding plus Spark byte reconstruction.
These measurements leave reducing that allocation traffic as follow-up work.

[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?

[sunchao]

Re-reviewed 6555410ac22d against 8b818b53bd6a. Since my previous approval at 6e556c94, only the metadata-workaround comment changed; its new tracking links are accurate. I need to qualify that approval: these two concerns from andygrove's review remain unresolved.

• [P2] Whole-Variant byte compatibility and coverage. Maintained Spark 4.0 rebuilds shredded integer 1 as 0c01; Arrow 59.3.0 preserves Int32 as 1401000000 and retains the file dictionary. A focused probe using the maintained Spark Java classes confirms equal JSON but unequal VariantVal bytes, independently for width and metadata. This establishes representation/test parity divergence, not wrong JSON values. Please resolve the compatibility policy and retain targeted byte-level assertions. The unchanged Spark 4.0 upstream suite still compares whole Variants; its current CI job is skipped.
• [P2] Non-null Variant existence defaults. Maintained Spark 4.0's vectorized reader rejects a missing Variant column's default, while this serializer admits it and the native scan returns it. The new test acknowledges that difference and uses literal expectations. Please resolve this behavior difference and compare present-file values/nulls with Spark separately.

Current CI ran all 11 projection tests on Linux Spark 4.0/4.1 successfully. Producer/consumer artifact IDs and digests match. It executed merge 81f38d2a with main 0690d38d, not the assigned base; DataFusion is 55.1 there versus 55.0 in this head. The three red checks trace to an Iceberg Gradle download failure and downstream missing-shard checks. macOS was skipped.

No duplicate inline findings. My matched scan/allocation benchmark request remains open; no performance benefit is established. I did not run Comet/JNI locally. Maintained Spark 3.4/4.1/4.2 source gaps remain.

[sunchao]

Re-reviewed 497062ad against the previous review. The two earlier code issues are addressed: typed Variant values are rebuilt with Spark-compatible widths and metadata, byte-level assertions are restored, and non-null Variant existence defaults now force scan fallback.

The final CI refresh exposed one additional P2 in the existing scan-test expectations. The native test job fails two decimal tests because they still expect Decimal16(123, 2), while reconstruction now returns Decimal4(123, 2). The normalizer and test file are identical between this head and the executed merge. Please update those assertions to the Spark-normalized output without removing the wide physical input fixtures. Details are in the inline comment. This run stopped after 358 passes and two failures, leaving 1,096 tests unrun.

The isolated source-matched scalar checks covered 26 integer, decimal and NaN cases against maintained Spark 4.0. They are not full Arrow/JNI tests. The added matched and reverse-order benchmarks report shredded reads about 1.3–1.6 times as slow as Spark, and about twice as slow for the empty-key fixture, with substantial allocation traffic. The opt-in documentation now discloses that cost. I did not independently reproduce those timings.

Other CI jobs are still running. The successful JVM build and failed native job used merge b8a4fdb6, whose main parent differs from the reviewed base. Spark 4.0 upstream tests and macOS are skipped. Maintained Spark 3.4/4.1 refs were unavailable locally. No approval while the new P2 remains.

[sunchao]

Correctness

[P2] Could you update the two existing scan-test expectations alongside this reconstruction change, keeping their wide physical decimal fixtures? variant_scan_uses_parquet_physical_types_instead_of_arrow_schema_hints and variant_scan_reads_wide_physical_decimal_as_decimal128 still compare against Variant::Decimal16 for the unscaled value 123 at scale 2. The Spark-compatible narrowing here correctly returns Decimal4, so both assertions fail in the current native test job. I verified that both the normalizer and test file are identical between this head and the executed merge. This is a test-expectation update, not a request to restore the wide encoding. Please rerun the native suite after updating it. The current fail-fast run leaves 1,096 tests unexecuted.

[sunchao]

Re-reviewed 9a4cd83b59807522f4310778be78f9190a5dde59 against my last published review. The decimal-test P2 is addressed: both assertions now expect Decimal4, while retaining the Decimal256(38,2) input and the 17- and 32-byte physical fixtures with positive and negative values.

The current native CI job passed 1,456 tests, with 5 skipped, including both previously failing fixtures. Its log confirms merge checkout 27c93d96. The fixture and normalizer files match this head, although the merge's main parent and broader context differ from the reviewed base.

The benchmark report moved into the PR description with its methods, numbers and caveats intact. The guide links there and still warns about slower shredded reads. These remain author-reported measurements. I did not independently reproduce them.

No new or remaining P1/P2 found. At September 15, 19:50 UTC, 3 CI checks are still running, alongside 15 successful and 11 skipped. Spark 4.0 upstream tests and macOS are skipped. No local Parquet/Arrow/JNI rerun was performed. Maintained Spark 3.4/4.1 source gaps remain.

[sunchao]

cc @andygrove @viirya @comphead let me know if you want to take another pass before I merge this

[sunchao]

Merging, thanks @peterxcli !

[dwsmith1983]

Hi @andygrove @sunchao @peterxcli

The last three merge-queue runs for this PR fail the same test on Linux and macOS, native scan declines top-level fields that repeat a Parquet field id, with the plan still holding a CometNativeScan (for example https://github.com/apache/datafusion-comet/actions/runs/35418573600). It also failed that way in the groups that included #5874 behind this PR.

The cause looks like the isSchemaSupported hunk in CometScanRule.scala: the per-field forall over requiredSchema.fields replaces the typeChecker.isSchemaSupported(scanExec.requiredSchema, ...) call, and #6004 put the root-level duplicate-field-id check in exactly that override, so it no longer runs. Keeping the schema-level call and applying the Variant admission on top of it, or checking duplicateFieldIds(requiredSchema.fields) before the loop, should bring the test back.

[sunchao]

Re-reviewed unchanged head 9a4cd83b after the merge-queue regression report.

[P2] The root duplicate-field-ID fallback still needs to be preserved. The new requiredSchema.fields.forall loop calls isTypeSupported for each field, bypassing the schema-level guard added by #6004. This affects ordinary BIGINT columns with repeated Parquet IDs when field-ID reading is enabled, without any Variant column or Variant opt-in. The nested-struct guard still runs. Please retain validation across all required root fields while allowing the intended top-level Variant case.

I verified queue commit 23f756cd and its source contribution against this PR. The existing regression test fails with a retained CometNativeScan in both Linux Spark 4.1 and macOS Spark 4.0. Those jobs respectively report 541 passed / 1 failed and 531 passed / 1 failed, and both pass the nested-ID test. The failure is at the fallback-plan assertion before row collection, so these logs establish lost fallback, not an observed wrong-result execution.

The earlier decimal-fixture fix remains intact, and both fixtures pass in the queue's Rust job. This remaining P2 qualifies my earlier approval. I am acknowledging the existing report without duplicating its finding inline. No local Spark/JNI rerun or independent benchmark was performed. Maintained Spark 3.4/4.1 source gaps remain.

[sunchao]

Re-reviewed 2fba70d1 against 5442c937. The earlier [P2] top-level duplicate-field-ID fallback issue is fixed. Required schemas now go through CometScanTypeChecker.isSchemaSupported, restoring the root ID check. The Variant exception applies only to immediate required fields, while recursive and partition checks still use the strict checker.

Current scan CI passes both root and nested ID regressions. Those tests verify Spark fallback and its duplicate-ID error with ID matching enabled, and native execution with matching disabled. The 13 Variant projection tests and eight Spark shredding tests also pass. This job ran merge 0d12aff5, whose tree is identical to the reviewed head. It reports 558 passed, five canceled and one ignored. Rust CI reports 1,584 passed and five skipped, including both corrected decimal fixtures. Execution CI reports 945 passed and five ignored, including Variant FFI and Python fallback coverage.

No remaining or new P1/P2 findings in this follow-up. The revision adds a planning-time checker and keeps the previously reviewed reconstruction code unchanged. I did not rerun benchmarks or local full suites. Current checks show 23 successes and 14 skips. macOS and dedicated Spark SQL jobs were skipped. Maintained Spark 3.5/4.0 sources confirm the duplicate-ID error behavior, while maintained 3.4/4.1 source branches remain unavailable.