Skip to content

feat: Support native scans with unprojected Spark 4 VARIANT columns - #5377

Open
sunchao wants to merge 2 commits into
apache:mainfrom
sunchao:dev/chao/codex/comet-4295-unprojected-variant-scans
Open

feat: Support native scans with unprojected Spark 4 VARIANT columns#5377
sunchao wants to merge 2 commits into
apache:mainfrom
sunchao:dev/chao/codex/comet-4295-unprojected-variant-scans

Conversation

@sunchao

@sunchao sunchao commented Aug 15, 2026

Copy link
Copy Markdown
Member

Why are the changes needed?

Spark 4 introduces VARIANT for semi-structured data, so ordinary analytical tables can now contain a mixture of familiar typed columns and a JSON-like VARIANT column. Adding that one column should not prevent Comet from accelerating queries that never read it.

For example, consider a table such as:

CREATE TABLE events (
  event_id BIGINT,
  payload VARIANT,
  event_type STRING
) USING parquet;

SELECT event_id, event_type
FROM events
WHERE event_type = 'purchase';

The query only needs event_id and event_type; it does not decode, filter, or otherwise inspect payload. Nevertheless, the existing scan paths make decisions using the complete table schema instead of the columns the native reader actually needs.

For an Iceberg table, that means the presence of payload causes the entire native scan to fall back, even though the query projects only supported scalar columns. For a regular Parquet table, Spark can correctly prune payload from the requested projection, but Comet still serializes the original full relation schema when building its native plan. Because VARIANT has no native Spark-type serialization, the supposedly supported scan can fail during plan construction.

The practical result is that introducing a VARIANT column can either disable native acceleration for unrelated queries or make those queries fail outright. This PR addresses that narrower, immediately useful part of #4295: queries that do not need VARIANT data should continue to run natively, while queries that actually need to read VARIANT must still fall back safely to Spark.

What changes were proposed in this PR?

The change makes native-scan eligibility follow the actual data projection, while respecting the different ways Spark Parquet scans and Iceberg scans describe their input schemas.

For a regular Parquet scan, Spark has already determined which columns and nested fields are required. Comet now derives the schema sent to native execution from that information instead of blindly serializing every field in the relation schema. A completely unrequested VARIANT column is omitted. If a requested struct contains both a supported field and an unrequested VARIANT sibling, Comet uses Spark's already-pruned version of that struct, so a query such as SELECT details.label FROM nested_events can still execute natively when details has type STRUCT<label: STRING, payload: VARIANT>. Because removing a field changes subsequent positions, the native projection is rebuilt against the pruned schema so that later data columns, partition values, and file metadata continue to refer to the correct fields.

Iceberg requires a different boundary. Its complete table schema is still passed to iceberg-rust, which can represent VARIANT in schema metadata but cannot materialize a projected VARIANT field or a projected parent that contains one. The scan rule therefore distinguishes between roots that the reader will actually project and roots that remain entirely untouched. An omitted root may contain VARIANT; a projected root is still checked strictly, as are every other unsupported type and any scan whose projected field IDs cannot be determined safely. This also preserves fallback for empty and metadata-only projections, because the current Iceberg reader interprets an empty physical projection as a request for all columns.

That distinction must use Iceberg field IDs rather than column names. Column names can change while their IDs remain stable, which matters for time travel:

-- Snapshot 123 was created while the column was named "details".
ALTER TABLE iceberg_events RENAME COLUMN details TO renamed_details;

SELECT details.label
FROM iceberg_events VERSION AS OF 123;

The historical query exposes the old name, while the current table schema exposes the new one. Matching names would incorrectly treat the VARIANT-bearing root as unprojected and let the native reader fail later. Matching the stable field ID identifies the same logical root across both schemas and correctly keeps this unsupported nested projection on Spark.

This PR deliberately does not claim to decode VARIANT natively. Direct projections, variant_get predicates, shredded or otherwise unsupported nested forms, and delete paths that would require unsupported VARIANT materialization continue to fail closed. Full native VARIANT execution remains follow-up work once Comet's Spark-facing representation and upstream native-reader support can carry the type end to end.

How was this PR tested?

The Spark 4.0 Parquet SQL regression checks both native-plan selection and Spark-answer parity for omitted top-level VARIANT columns, columns appearing after the omitted field, supported nested siblings, nullable parent structs, partition columns, and file metadata. It also verifies that direct VARIANT projections and predicates still fall back, that SQL NULL remains distinct from a VARIANT containing JSON null, and that an ordinary STRUCT<value: BINARY, metadata: BINARY> is not mistaken for VARIANT.

The Spark 4.0 Iceberg coverage exercises supported native scalar projections and filters, projected nested-root fallback, empty and metadata-only projections, and the historical rename shown above. The historical-rename regression was run against the previous implementation first and failed because a native scan was incorrectly selected; it passes with field-ID-based validation. Iceberg 1.10 has a separate Spark-reader bug when an annotated VARIANT column is physically present but omitted from its projection, so the reference read intentionally includes the VARIANT column and removes it from the collected expected rows. This preserves a real Spark-versus-Comet comparison over the same files without relying on the unrelated upstream bug.

A Spark 3.5 Iceberg VERSION AS OF regression also passes, and the shared tests compile under both Spark 3.5 and Spark 4.0 despite their different withSQLConf return signatures.

mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.CometIcebergNativeSuite variant' test

mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.CometSqlFileTestSuite variant' test

mvn -o -ntp -Pspark-3.5 -Dtest=none \
  '-Dsuites=org.apache.comet.CometIcebergNativeSuite schema evolution - read old snapshot after column drop' test

The focused runs passed three Iceberg VARIANT cases, one Parquet SQL file, and one Spark 3.5 historical-snapshot case. Spotless, Scalastyle, and git diff --check also passed.

@sunchao sunchao changed the title Support native scans with unprojected Spark 4 VARIANT columns feat: Support native scans with unprojected Spark 4 VARIANT columns Aug 15, 2026
@sunchao
sunchao marked this pull request as ready for review August 16, 2026 03:16
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant