Spark: Add selective shredded variant extraction Parquet readers

[qlong]

Changes

This PR is part of the work to support variant extraction pushdown, the core change is to introduce new parquet readers to read selected variant paths instead of the whole variant:

• Add selective Parquet readers (ParquetVariantExtractionReaders, VariantExtractionPathResolver) to read only shredded typed_value columns for requested extraction paths.
• Add Spark row reader adapter (SparkVariantExtractionReaders, SparkParquetReaders) to materialize extraction slots from the engine read schema instead of full variant blobs.
• Wire engine read schema from SparkBatch through SparkInputPartition to RowDataReader only (row Parquet path).
• Update PruneColumnsWithoutReordering so annotated extraction structs map back to Iceberg VARIANT columns in the scan projection.

Issue: #16726

Note for reviewers

• PathUtil.java is mostly copied from the existing PR Api: Support variant extract and fix manifest bounds byte order #15384, will rebase once that PR is merged.
• To reduce the scope, the new selective readers are only wired in for batch row scan. We can wire to other readers as a follow up.
• To reduce the scope, only supports extracting mostly used data types. Do not support extracting arrays, struct / nested struct.Request shredded columns for unsupported types will lead to read the whole variant (extraction pushdown rejected).
• Merge order:
  1. Spark: Add selective shredded variant extraction Parquet readers #16714
  2. Spark: Implement variant extraction pushdown for shredded VARIANT columns #16715

End to end testing

Requires #16715 for end-to-end testing. To try the full pushdown + selective read path without merging locally, use this branch:

https://github.com/qlong/iceberg/tree/variant-extraction-integration-test

Test results

Use 1-day Github activities data, ingested as json and shredded variants with 299 shreddred columns.

Baseline: gha-payload-iceberg-20260605 · variant + extraction pushdown ON + selective shredded variant extraction Parquet readers
Compare A: same run with payload stored as string_json
Compare B: gha-payload-iceberg-nopushed-20260605 · variant + pushdown OFF, read whole variant
Median of 3 timed runs per query (Spark Time taken:).

Query	Variant + pushdown (s)	string_json (s)	Δ vs baseline	Variant no-pushdown (s)	Δ vs baseline
c-q01	2.605	2.373	−8.9%	2.945	+13.0%
c-q04	3.875	6.412	+65.5%	72.082	+1760%
c-q05b	3.506	5.683	+62.1%	39.154	+1017%
c-q06	4.668	6.583	+41.0%	76.935	+1548%
c-q07	4.714	4.490	−4.8%	75.033	+1492%
c-q08	3.701	4.707	+27.2%	87.059	+2252%
c-q09	5.102	6.668	+30.7%	72.560	+1322%
c-q10	4.395	6.568	+49.4%	68.179	+1451%
c-q11	4.495	6.384	+42.0%	67.985	+1413%
c-q12	3.995	4.284	+7.2%	70.509	+1665%
c-q13	3.911	4.060	+3.8%	39.614	+913%
c-q14	4.769	5.450	+14.3%	63.331	+1228%
Total (Σ)	49.74	63.66	+28.0%	735.39	+1379%

Co-authored with Claude Sonnet 4.6

[qlong]

@rdblue @steveloughran @nssalian PTAL when you get a chance.

[tmater]

I may be missing something, but this looks like it overlaps with Spark's variant_get casting logic. toSparkValue handles the common cases, but it seems separate from Spark's existing behavior around failOnError, timeZoneId, and some cast edge cases.

Would it make sense to reuse Spark's cast path here if we can bridge from Iceberg's VariantValue to Spark's Variant / VariantVal?

[qlong]

Thanks for review. I assume you were referring VariantGet.cast in spark. toSparkValue in the connector is  required according to  DSV2 extraction pushdown contract. When Spark pushes extraction down, it delegates extraction and cast to the connector, the engine no longer calls VarianGett.cast one the values returned from connector.  The bridge from iceberg's VariantValue to spark's Variant already exists, it is triggered when extaction pushdown was rejected and connector returns the whole variant. It is expensive for shredded typed value, as they are put back into VariantValue then immediately extraced by Spark again.

The cast logic in Spark is more general than needed here, it handles cross-type coercisons that do not apply to typed_value, with the exception of type narrowing overflow. I added a fix for that.

[tmater]

I see, that makes sense, thanks for the explanation. Could you add a short method comment here explaining why this uses a narrow custom conversion instead of Spark’s VariantGet.cast?

[tmater]

One more thing about failOnError here, it looks like it is not taken into account.

Spark exposes two functions that map to VariantGet: variant_get (failOnError = true, throws on an invalid cast / overflow) and try_variant_get (failOnError = false, returns null). As written, toSparkValue returns null on overflow and type mismatch in both cases, so it seems to me this would behave like try_variant_get even when the user called variant_get.

[nssalian]

failOnError and timeZoneId are present in the extraction metadata (see the extractionField test helper), but extractionPath() reads only path, so they're dropped. On the gate side, #16715's pushVariantExtractions accepts an extraction based on path and target type alone, and isSupportedPushdownTargetType(DataType) only sees the type, so there's no point today where failOnError=true is rejected. That means variant_get (failOnError=true) gets routed through toSparkValue, which returns null on overflow and type mismatch, so it behaves like try_variant_get.

The fix is either to decline pushdown when failOnError=true, or to thread the flag through and throw on the overflow/mismatch branches. Declining is simpler since the readers stay try-only. A failOnError=true overflow test asserting a throw would mirror the existing ...ReturnsNull ones.

[qlong]

@tmater @nssalian Thanks for review. I have been on vacation, will address the comments in a few days.

[qlong]

Good call out on missing support for failOnError. Changed it so the cast behavior matches Spark:

• pass failOnError flag around
• For trivial conversion, use fast inline conversion for performance
• Delegate to Spark's own cast for other conversions. This path has a around 1KB overhead in VariantCastArgs, which is high for primitives.

Pass timezoneId around timezone sensitive cast (in delegation path).

[tmater]

If I am not mistaken, this utility is only used for variant extraction/shredding paths. Should we rename it to something like VariantPathUtil or VariantPath so it is clear this is not a general Iceberg path utility?

[qlong]

You are correct, agree that VariantPathUtil is a better name. I am going to defer the change for now, since this file is copied from #15384 to avoid stacked PRs. There are other in-fligh PRs that also copy this file. I will put up a follow up PR to rename once they are merged.

[tmater]

I think this loses some path semantics. PathUtil.parse distinguishes object names from array indexes, but parseObjectPath flattens both into string parts. For example, $[0] means array element 0, while $['[0]'] means an object field whose name is literally [0]; after flattening, both are represented as the same string segment "[0]".

[qlong]

Good call out. I removed parseObjectPath, and pass List through the Parquet extraction readers. Added tests

[tmater]

Slightly broader question: would it make sense to use an existing JSONPath parser here and then validate that the parsed path is within Iceberg's supported subset? This is probably the fourth project where I have seen a new VariantPath-style implementation over the past year, so I am a bit worried about adding another one unless we keep the scope very explicit.

For example, we could allow simple/singular paths like field access and array indexes, while rejecting wildcards, recursive descent, slices, and filter expressions for now.

[qlong]

My understanding is iceberg has strict dependency hygiene, adding new lib would require review and could add transtive dependences. The parser from #15384 is intentionlly minimal.. We probably should look into dedicated lib if we need to support wildcard, filter expressions.

[tmater]

I like the direction of separating Spark variant_get semantics from the optimized selective Parquet read path. One thing that still makes this PR difficult to review is that it introduces both the pushdown/delegation plumbing and the custom primitive fast path at the same time.

Would it make sense to split this into two PRs?

1. First PR: wire the selective reader path but delegate all Spark type conversion to Spark’s VariantGet.cast, so the new path is easier to validate against existing Spark semantics.
2. Follow-up PR: add the primitive inline fast path as a focused optimization, with benchmarks and targeted equivalence tests for the pairs it owns.

[tmater]

I think this mishandles root extraction ("$"). Eaerlier PathUtil.parse("$") returns an empty path, and for a shredded object resolveShreddedFieldGroup(variantGroup, emptyPath) returns the root typed_value group. That group contains object fields, not a variant value group with its own value/typed_value, so the leaf-reader check below fails and no reader is built unless the root serialized value fallback exists.

[qlong]

@tmater I like the simplicity of delegation, but I ran micro performance and found the overhead (memory allocation) was quite high for delegating to spark cast. Let me run some end to end perf tests first, will share the numbers here before we decide?

[qlong]

@tmater Here is 1 e2e test runs using 1-days Github activity data. partial delegation is what the current state of PR, full means all cast is done by Spark. 1 extaction per query.

c-q05, c-q05b, c-q08 all extract to long, they show biggest performance gain. c-q09 extract boolan, all other queries extract a string value, they do not show much difference.

I think in the long term fast inline cast is worth it at least for numbers, more so if more then 1 extraction per queries. But I am open to remove the inline cast and related cross check tests for now if you think the PR is already too big. If we go that way, we just need to keep inline cast for nanoseconds until spark supports it.

Compare: gha-payload-full-delegation-20260709 vs gha-payload-partial-delegation-20260709  (metric: query_median)
  Run A: /Users/qlong/opensources/variant-conformance-benchmark/results/gha-payload-full-delegation-20260709/gha-payload/timings-variant.csv
  Run B: /Users/qlong/opensources/variant-conformance-benchmark/results/gha-payload-partial-delegation-20260709/gha-payload/timings-variant.csv

query    median_A   median_B   delta_s   delta_%
------   --------   --------   -------   -------
c-q01          2.73       2.66    +0.08    +2.9%
c-q04          3.97       3.93    +0.03    +0.9%
c-q05         73.18      65.64    +7.54   +11.5%
c-q05b         3.76       3.33    +0.43   +12.9%
c-q06          4.42       4.34    +0.09    +2.0%
c-q07          4.18       4.17    +0.01    +0.2%
c-q08          3.54       3.29    +0.25    +7.7%
c-q09          4.26       4.30    -0.04    -1.0%
c-q10          4.40       4.26    +0.14    +3.2%
c-q11          4.25       4.26    -0.02    -0.4%
c-q12          3.80       3.96    -0.16    -3.9%
c-q13          3.66       3.65    +0.01    +0.3%
c-q14          4.15       4.10    +0.05    +1.3%

Summary: 13 queries, 13 comparable
  Geo-mean delta: +2.8%  (Run B faster)
  Total (query_median):   120.3s vs 111.9s

[steveloughran]

FWIW I'd like my jmh benchmark #15629 in alongside/before this, they are independent but the final predicate pushdown really needs it to show speedup. even this should show before/after speedup.