apache / apache/iceberg

Spark: implement SupportsPushDownVariantExtractions for shredded variant column pruning (plan change)

Open
#16,448 1 comment 0 reactions 0 assignees View on GitHub
improvement
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 11h
Merged PRs (30d)
132

Description

### Feature Request / Improvement

Iceberg can write shredded variant columns to Parquet (#14297). On the read path,
`SparkScanBuilder` does not implement Spark 4.1's `SupportsPushDownVariantExtractions`,
so Spark never rewrites `variant_get(...)` into struct field accesses and never prunes the
scan output schema to the requested shredded fields.

As a result, even when queries only need one or two paths (e.g.
`variant_get(payload, '$.size', 'long')`), the optimized plan still evaluates
`variant_get` in a Spark `Filter` above `BatchScan`, and the scan output still
treats `payload` as `VariantType`. The reader therefore loads the full shredded
Parquet layout (hundreds of `typed_value.*` columns in practice), materializes a
full `VARIANT` per row, and applies path extraction only after the scan. That reconstruction
is expensive: on the GitHub Activities 1-day shredded Iceberg table (GHA), a simple
filter/count query was ~14× slower than the same workload on a plain JSON string column
(~63s vs ~4.4s), despite only ~1.6× more storage.

(Predicate pushdown of `variant_get` into `IcebergScan` filters for manifest
planning is handled separately #15385. it does not rewrite the plan or avoid reading
all shredded columns per file.)

This issue implements the DSv2 contract (plan rewrite). #16726 will wire the
annotated readSchema() into the Parquet reader and avoid full-variant reconstruction (I/O reduction).

**Plan rewrite example**
For this query:
```sql
CREATE TABLE events (
id INT,
type STRING,
payload VARIANT
) USING iceberg
TBLPROPERTIES ('format-version' = '3');

SELECT count(*) AS large_events
FROM events
WHERE type = 'PushEvent'
AND variant_get(payload, '$.size', 'long') > 5;
```

**Before (today — no SupportsPushDownVariantExtractions)**
```

Aggregate [count(1)]
+- Filter (... type = PushEvent)
AND (variant_get(payload#22, $.size, LongType, ...) > 5)) ← still a function call
+- RelationV2[type#19, payload#22]

Filter (variant_get(payload#22, $.size, ...) > 5) ← runs per row AFTER scan
+- BatchScan [type#15, payload#18]
IcebergScan(..., filters=type IS NOT NULL, payload IS NOT NULL, type = 'PushEvent')
ReadSchema: full payload variant / all shredded columns
```

**After (DSv2 plan rewrite)**
```
Aggregate [count(1)]
+- Filter (... type = PushEvent)
AND (payload#25.0 > 5)) ← struct field access, not variant_get
+- RelationV2[type#24, payload#25]

Filter (isnotnull(payload#25) AND (payload#25.0 > 5))
+- BatchScan [type#24, payload#25]
readSchema(): struct>
```

**Note**

This issue changes the logical/physical plan shape and readSchema() contract. Parquet may still read all shredded columns until the follow-on wires readSchema() into the reader.

### Query engine

Spark

### Willingness to contribute

- [x] I can contribute this improvement/feature independently
- [x] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time

Contributor guide

Open the contributing guide

Research direction

Start with SparkScanBuilder and Spark's SupportsPushDownVariantExtractions contract, then inspect how IcebergScan exposes readSchema(). Done means the optimized plan replaces variant_get(...) with struct field accesses and readSchema() contains only the requested shredded fields; the Parquet reader follow-up is explicitly out of scope.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.