feat: Support native scans with unprojected Spark 4 VARIANT columns - #5377
feat: Support native scans with unprojected Spark 4 VARIANT columns#5377sunchao wants to merge 3 commits into
Conversation
| val fullSchema = IcebergReflection.toSparkSchema(metadata.tableSchema) | ||
| typeChecker.isSchemaSupported(fullSchema, fallbackReasons) | ||
| val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol) | ||
| val unprojectedTypeChecker = new CometScanTypeChecker() { |
There was a problem hiding this comment.
This is being called for all the top level variant fields. What about a nested variant field?
There was a problem hiding this comment.
Nested Variant fields are handled recursively. DataTypeSupport.isTypeSupported dispatches back to this overridden checker for struct fields, array elements, and map keys/values, so an unprojected top-level Iceberg field may contain Variants at any depth. If that top-level field is projected, the strict checker still rejects it and the scan falls back to Spark. I clarified this in the comment and extended the Iceberg regression to cover nested structs, ARRAY<VARIANT>, and MAP<STRING, VARIANT>.
|
|
||
| statement | ||
| CREATE TABLE test_variant(id INT, v VARIANT) USING parquet | ||
| CREATE TABLE test_variant(id INT, v VARIANT, tail STRING) USING parquet |
There was a problem hiding this comment.
Should we also add a test with ARRAY<VARIANT> and/or MAP<STRING, VARIANT>
There was a problem hiding this comment.
Added explicit ARRAY<VARIANT> and MAP<STRING, VARIANT> coverage for both Parquet and Iceberg. The tests verify that native scans remain enabled when the collection columns are unprojected and fall back to Spark when either collection is projected. They also cover SQL-null elements and map values, Variant JSON null, and null collection roots.
Why are the changes needed?
Spark 4 introduces
VARIANTfor semi-structured data, so ordinary analytical tables can now contain a mixture of familiar typed columns and a JSON-likeVARIANTcolumn. Adding that one column should not prevent Comet from accelerating queries that never read it.For example, consider a table such as:
The query only needs
event_idandevent_type; it does not decode, filter, or otherwise inspectpayload. 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
payloadcauses the entire native scan to fall back, even though the query projects only supported scalar columns. For a regular Parquet table, Spark can correctly prunepayloadfrom the requested projection, but Comet still serializes the original full relation schema when building its native plan. BecauseVARIANThas no native Spark-type serialization, the supposedly supported scan can fail during plan construction.The practical result is that introducing a
VARIANTcolumn 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 needVARIANTdata should continue to run natively, while queries that actually need to readVARIANTmust 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
VARIANTcolumn is omitted. If a requested struct contains both a supported field and an unrequestedVARIANTsibling, Comet uses Spark's already-pruned version of that struct, so a query such asSELECT details.label FROM nested_eventscan still execute natively whendetailshas typeSTRUCT<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 representVARIANTin schema metadata but cannot materialize a projectedVARIANTfield 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 containVARIANT; 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:
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
VARIANTnatively. Direct projections,variant_getpredicates, shredded or otherwise unsupported nested forms, and delete paths that would require unsupportedVARIANTmaterialization continue to fail closed. Full nativeVARIANTexecution 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
VARIANTcolumns, columns appearing after the omitted field, supported nested siblings, nullable parent structs, partition columns, and file metadata. It also verifies that directVARIANTprojections and predicates still fall back, that SQLNULLremains distinct from aVARIANTcontaining JSONnull, and that an ordinarySTRUCT<value: BINARY, metadata: BINARY>is not mistaken forVARIANT.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
VARIANTcolumn is physically present but omitted from its projection, so the reference read intentionally includes theVARIANTcolumn 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 OFregression also passes, and the shared tests compile under both Spark 3.5 and Spark 4.0 despite their differentwithSQLConfreturn signatures.The focused runs passed three Iceberg
VARIANTcases, one Parquet SQL file, and one Spark 3.5 historical-snapshot case. Spotless, Scalastyle, andgit diff --checkalso passed.