feat: project Spark 4 VARIANT columns in native Parquet scans

[peterxcli]

Which issue does this PR close?

This is Phase A of #4295.
Relate to: #5438

It enables whole-value projection of direct, top-level Spark 4 VariantType columns through Comet's ordinary native Parquet scan. It uses Arrow/Parquet's whole-value unshredding support without taking on #3983's broader shredded writer, subfield-pruning, or predicate-pushdown scope. Iceberg Variant projection remains a later phase of #4295.

Rationale for this change

Spark 4 represents semi-structured values with the atomic VariantType. At the Arrow boundary, Spark uses a Struct containing non-null Binary children in [[value, metadata] order] (https://github.com/apache/spark/blob/v4.1.3/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala#L183-L197). ColumnVector.getVariant consumes that physical layout directly:

return new VariantVal(getChild(0).getBinary(rowId), getChild(1).getBinary(rowId));

Arrow identifies the logical value with the Field-level extension name arrow.parquet.variant, backed by Struct storage. Parquet readers may expose metadata, value, and optional shredded typed_value children in any order. Arrow-rs resolves those children by name and unshred_variant reconstructs a whole value without typed_value.

The base scan path has three independent gaps:

• the protobuf datatype enum has no Variant identity;
• the native exporter constructs FFI_ArrowSchema from ArrayData.data_type(), which cannot carry the parent Field marker; and
• the JVM Arrow bridge maps every Struct to StructType.

Comet therefore needs to preserve more than the physical Struct shape. The parent Field marker distinguishes Variant from an ordinary user Struct, while Spark requires exactly [value, metadata] at the JVM vector boundary. The scan also needs to handle these interoperability cases:

• VariantArray::try_new permits dictionary-encoded metadata in its contract, but arrow-rs 58.4's binary validation rejects the dictionary before construction.
• Spark writes and searches object fields using Java String.compareTo order (builder, 32-field binary search), which differs from Arrow's UTF-8 ordering for supplementary Unicode keys. [SPARK-58949][CORE] Use unsigned UTF-8 byte ordering for Variant object fields spark#58239
• Spark existence defaults are constant-folded into table-schema metadata to represent pre-existing rows without rewriting old files (creation, scan-time evaluation). A missing Variant column must receive that value without shifting later defaults to the wrong schema indexes.
• Arrow extension metadata belongs to a Field. FFI_ArrowSchema::try_from(&DataType) has no parent Field metadata, whereas TryFrom<&Field> exports the name, nullability, and metadata. Spark models top-level field names as String and passes them through Parquet conversion, while Arrow's C schema stores the name as a CString; a top-level name containing NUL therefore needs a safe export representation.

Without one consistent normalization and export path, Variant can lose its logical identity, expose children in the wrong order, produce incorrect Unicode object lookup, or fail on valid Parquet reader representations.

What changes are included in this PR?

The native projection path is:

Spark VariantType
  -> Comet protobuf VARIANT
  -> Arrow Field<Struct[value, metadata], arrow.parquet.variant>
  -> ordinary Parquet schema adapter
  -> dictionary decode (when needed)
  -> VariantArray::try_new + unshred_variant
  -> Spark-compatible object ordering
  -> Struct[value: Binary, metadata: Binary]
  -> Arrow C Data Interface Field
  -> CometStructVector with logical Spark VariantType
  -> ColumnVector.getVariant

Preserve Variant identity from Spark to Arrow

• Appends VARIANT = 21 without renumbering existing protobuf values.
• Serializes Spark 4 VariantType through version-specific shims while Spark 3.x remains inert (Spark serialization, Spark 4 shim).
• Maps the protobuf type to physical Struct<value: Binary, metadata: Binary> and attaches the canonical extension marker to the parent Arrow Field (native type mapping, Field construction).
• Keeps datatype transport separate from native expression support: general supportedDataType and expression/operator gates continue to reject Variant outside the direct scan path.

Normalize the Parquet value once

The ordinary Parquet schema adapter installs CometCastColumnExpr whenever the logical target is a marked Variant Field, including physical/logical identity casts (schema adaptation). Its normalization helper:

1. decodes a dictionary-encoded metadata child before constructing VariantArray;
2. converts a partially shredded residual value to Arrow UTF-8 object order while it passes through upstream validation;
3. calls unshred_variant to merge typed_value into the whole value;
4. converts BinaryView/LargeBinary children to ordinary Binary;
5. recursively rebuilds incompatible object values in Spark's UTF-16 order while preserving already-compatible bytes; and
6. returns exactly [value, metadata] with the original parent null bitmap.

The implementation and compatibility helpers are in cast_column.rs. This covers already-unshredded, fully shredded, and partially shredded reader output without adding another Variant dependency or duplicating unshredding in JVM code.

Apply Spark existence defaults without losing schema indexes

CometNativeScan keeps every serialized existence default paired with its original required-schema index. Ordinary values remain literals; a Spark VariantVal is transported only for this scan path as a constant CreateNamedStruct(value, metadata) using the physical Arrow storage layout.

The native planner evaluates that constant into the ScalarValue consumed by the existing Parquet schema adapter (default evaluation, index mapping). The value is substituted only when the physical file lacks that column. Any present default that cannot be serialized keeps the scan on Spark instead of emitting mismatched value/index lists; a schema with no defaults remains native.

Preserve the output Field through FFI and restore Spark VariantType

The shared FFI boundary exports each array with its corresponding RecordBatch Field, including offset-normalized arrays (batch export). When a top-level output field name contains NUL, only its exported C name is substituted; datatype, nullability, and metadata remain unchanged (FFI helper).

On the JVM, Utils.fromArrowField maps only the explicit ARROW:extension:name = arrow.parquet.variant marker to Spark's version-shim Variant type. Unmarked Arrow Structs retain their existing StructType behavior. The existing CometStructVector preserves the two child ordinals, so Spark's inherited getVariant can consume the result without a new vector class.

Keep unsupported consumers on Spark

The supported surface is direct, top-level, whole-value projection from ordinary Parquet. The following remain explicit fallbacks:

• PushVariantIntoScan / annotated VariantStruct output;
• variant_get, predicates, casts, parse_json, to_variant, and other native Variant expressions;
• native columnar-to-row, sort/limit, shuffle, and spill;
• native Parquet writes—the write guard inspects the actual input below WriteFilesExec (write boundary);
• Comet's accelerated MapInArrow/MapInPandas rewrite—Variant-bearing batches stay on Spark's ordinary Python path (Python boundary);
• nested Variant inside ARRAY/MAP/STRUCT; and
• Iceberg Variant projection and equality deletes.

Unread Variant roots continue to use #5377's pruning behavior and are removed rather than decoded. Whole-value SQL tests set spark.sql.variant.pushVariantIntoScan=false; a separate negative case verifies that Spark's pushed VariantStruct remains a safe fallback.

How are these changes tested?

The focused coverage verifies:

• SELECT v and SELECT id, v, tail retain a Comet native Parquet scan;
• unshredded, fully shredded, and partially shredded values reconstruct correctly;
• objects, arrays, scalars, JSON null, SQL null, and nullable parents round-trip;
• the exported vector has logical Spark VariantType, exact [value, metadata] Binary children, and a preserved parent marker/nullability;
• existence defaults fill physically missing Variant columns without shifting later defaults;
• dictionary-encoded Variant metadata is decoded before unshredding;
• 32-field objects containing supplementary and high-BMP Unicode keys remain compatible with Spark's binary lookup;
• a top-level Parquet field name containing NUL exports without losing Field metadata;
• native writes, the accelerated MapInArrow rewrite, expressions/casts, native C2R, shuffle/spill, nested Variant, pushed VariantStruct, and Iceberg remain fallbacks; and
• the Spark 3.5 profile retains its existing shim boundary and test compilation.

Focused local validation passed:

make core

cd native
cargo fmt --all -- --check
env DYLD_LIBRARY_PATH="$JAVA_HOME/lib/server" \
  cargo test -p datafusion-comet test_normalize_ --lib
env DYLD_LIBRARY_PATH="$JAVA_HOME/lib/server" \
  cargo test -p datafusion-comet test_create_default_value_from_literal_and_variant_struct --lib
env DYLD_LIBRARY_PATH="$JAVA_HOME/lib/server" \
  cargo test -p datafusion-comet test_ffi_schema_sanitizes_nul_name_and_preserves_metadata --lib
cargo clippy -p datafusion-comet --lib --tests -- -D warnings
cd ..

mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.CometSqlFileTestSuite variant' test
mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.parquet.ParquetReadV1Suite native scan' test
mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.comet.parquet.CometParquetWriterSuite parquet write with Variant input falls back to Spark' test
mvn -o -ntp -Pspark-4.0 -Dtest=none \
  '-Dsuites=org.apache.spark.sql.comet.CometMapInBatchSuite Variant-bearing input or output' test
mvn -o -ntp -Pspark-4.1 -Dtest=none \
  '-Dsuites=org.apache.comet.CometSqlFileTestSuite variant' test
mvn -o -ntp -Pspark-3.5 -DskipTests test-compile

git diff --check upstream/main...HEAD

The Rust normalization selection passed 6/6 tests. The Spark 4.0 native-scan selection passed 13/13 tests, including projection, existence defaults, a top-level NUL-containing name, and dictionary metadata. The focused writer and MapInBatch fallback tests passed 1/1 each; the Spark 4.0 and 4.1 Variant SQL suites passed; and Spark 3.5 test compilation passed. Maven Spotless and Scalastyle checks, Rust formatting and Clippy, and git diff --check also passed.

[sunchao]

Summary

I reviewed 45a0ed44ed9c58ede31410de077a0882e72fd4f8 against 92954d7884091d2c6fa3e109d11fc1b8cb4a7325, including the full 18-file diff and five independent review scopes. The scan-only design is a reasonable Phase A boundary: preserve Variant identity, reconstruct whole values at the Parquet boundary, and let Spark handle unsupported consumers. Seven P2 findings survived verification: six runtime compatibility regressions and one cross-version SQL-test failure confirmed in CI. Two runtime cases produce incorrect values, and four make previously valid operations fail.

Prior state and problem

Previously, a requested Spark Variant could not be serialized through Comet's type protobuf, and a native Arrow Struct could not be recovered as Spark's logical Variant type. Exporting an Arrow DataType rather than its Field also discarded the parent extension metadata needed to distinguish Variant storage from an ordinary Struct.

The existing pruning path already allowed scans to avoid unread Variant-bearing roots and retain supported siblings. This change extends that path to requested top-level Variant values, so it also makes previously unreachable default-value and downstream-consumer paths relevant.

Design approach

The PR appends VARIANT = 21, uses version shims to keep Spark 3 behavior inert, and represents Variant as a marked Arrow Field with Binary value and metadata children. The Parquet schema adapter installs a normalization expression that delegates reconstruction to Arrow's Variant implementation, converts its output to Binary, and restores Spark's child order.

Both native export call sites now pass Fields through the C Data Interface. The JVM recognizes the parent extension marker and reuses CometStructVector, allowing Spark's inherited getVariant to consume the two children.

Correctness / compatibility analysis

The ordinary projection path, name-based child lookup, and parent-null handling are supported by the added tests and focused review. The surviving problems are outside those examples: Variant existence defaults shift later defaults, reconstructed Unicode object keys are incompatible with Spark's lookup order, dictionary metadata is rejected before conversion, write wrappers evade the operator guard, the later Python rewrite bypasses it entirely, and Field export panics on a valid top-level NUL-containing name.

I built the JVM code at the requested head and used the macOS CI native artifact after verifying that its synthetic-merge tree is identical to the head tree. Five focused Scala cases and a PySpark comparison reproduced the six inline issues on Spark 4.0.4. Separate byte/serializer probes also checked the Unicode and Pandas cases against Spark 4.1.3. The PR's existing focused Variant projection test also passed locally, 1/1. These are targeted checks, not a full local Spark/native test run.

At the final refresh, CI had 48 successful, 2 failed, 13 running, and 7 skipped checks. The Spark 4.1 expressions job and Spark 4.2 expressions job both fail the new variant.sql:50 native-plan assertion because default Variant pushdown produces the deliberately unsupported VariantStruct representation. I inspected both job logs. Native builds, Rust tests, and the scan matrix are green, but the full CI run is not complete.

Key design decisions

Using an explicit extension marker is preferable to recognizing Variant from Struct shape, because ordinary user Structs must retain their existing meaning. Keeping datatype serialization separate from expression support is also appropriate for the proposed scope.

However, Arrow-compatible storage is not sufficient for every Spark consumer. Spark's object lookup order and Pandas Variant marker need compatibility handling or fallback. Similarly, a guard in tryConvertToComet cannot cover wrappers with empty outputs or operators introduced by a later transition rule.

Implementation sketch

On the Scala side, CometScanRule admits direct Variant roots, QueryPlanSerde transports the type, and CometNativeScan retains the requested logical field. Native schema construction attaches the extension marker, while the ordinary schema adapter wraps the physical reader column in CometCastColumnExpr.

Normalization resolves the reader's children by name, calls unshred_variant, and produces the two Binary children. The output batch's Field then accompanies its array through JNI, and Utils.fromArrowField restores Variant identity before Spark reads the vector. This is a compact path, but its new admission needs to be checked against existing default reconstruction and all downstream transitions.

Behavioral changes worth calling out

Whole-value top-level Variant scans can now remain native, including shredded layouts supported by the normalizer. Nested Variant, pushed VariantStruct, Iceberg projection, Variant expressions, shuffle, and native row conversion are intended to keep their existing fallback boundaries.

The change also affects more than direct projection: default expressions are now processed for admitted Variant schemas, and opt-in native-write and Python paths can receive Variant scans. Exporting the original Field name changes the common FFI path for non-Variant columns as well, which is why the NUL-name regression is included here.

Suggested improvements

Please address the six runtime cases with focused regressions and align the new native-projection SQL assertions with the supported scan configuration. Keep default values paired with their indexes, decode accepted reader representations before constructing VariantArray, and preserve Spark lookup behavior for reconstructed objects. The Unicode regression should include at least 32 keys with both supplementary and high-BMP characters, because a small ASCII-only object does not exercise Spark's binary-search path.

For the scan-only scope, apply fallback to the actual write input beneath WriteFilesExec and to the post-columnar Python rewrite instead of implicitly enabling those consumers. Preserve extension metadata without passing unsupported raw names to the C-string exporter. The existing successful projection tests should remain alongside these negative and compatibility cases. The SQL file also needs the same pushdown setting as the focused vector test so its native-plan assertions run on the intended path in Spark 4.1 and later.

[sunchao]

[P2] Reject unsupported Variant defaults before admitting the scan

Could we keep this scan on Spark when a required Variant field has a non-null existence default? CometNativeScan drops failed default serializations with flatMap, but retains every default index, and CometLiteral still rejects Variant. I reproduced this on Spark 4.0.4:

CREATE TABLE t(v VARIANT DEFAULT parse_json('1')) USING parquet;
INSERT INTO t VALUES (parse_json('42'));
ALTER TABLE t ADD COLUMNS(n INT DEFAULT 7);
SELECT v, n FROM t;

Spark returns (42, 7), while this head's native scan returns (42, NULL). The remaining default 7 is zipped to index 0 (v), where it is ignored because that column exists physically, leaving n without its default. Please reject an unserializable default or preserve and validate each value/index pair before enabling the scan.

[sunchao]

[P2] Preserve Spark lookup compatibility when rebuilding Unicode objects

Could we account for Spark's object-key ordering before returning these reconstructed bytes? Arrow sorts object keys in UTF-8 order, but the supported Spark versions use Java String.compareTo and switch to binary search at 32 fields. I wrote a shredded Parquet object with Spark containing k00 through k29, U+E000, and 😀. With pushVariantIntoScan=false and allowReadingShredded=true, variant_get(v, '$.😀', 'int') returns 531 on Spark but NULL with this native scan. The expression itself correctly falls back to Spark, but it consumes the incompatible reconstructed ordering. The byte-level mismatch also reproduces on Spark 4.1.3. Please normalize for the Spark consumer or retain fallback for affected values, with a 32-key Unicode regression.

[sunchao]

[P2] Decode dictionary metadata before constructing VariantArray

Could we decode dictionary-encoded metadata before this call? The canonical Arrow Variant representation permits it, and an Arrow-written Parquet file can retain metadata: Dictionary(Int32, Binary) in its embedded ARROW:schema while storing ordinary required BINARY children physically. I reproduced a file containing 42, 43, 44: Spark 4.0.4 reads it successfully, but this head's native scan throws Illegal shredded value type: Dictionary(Int32, Binary). Arrow/Parquet 58.4.0 restores the nested dictionary, which VariantArray::try_new rejects, so the Binary cast below is never reached. Decoding the metadata child first makes the same values readable.

[sunchao]

[P2] Check the write input beneath WriteFilesExec

Could this guard inspect the same unwrapped data-producing children used by requiresNativeChildren below? For DataWritingCommandExec(WriteFilesExec(CometNativeScan[Variant])), both the command output and WriteFilesExec.output are empty, so the Variant check misses the input. With spark.comet.parquet.write.enabled=true and spark.comet.operator.DataWritingCommandExec.allowIncompatible=true, copying a Spark-written Variant Parquet column now selects CometNativeWriteExec and fails in CometArrowStream.inputObjects -> Utils.toArrowSchema with Unsupported data type: ... VariantType ... variant. The intended Spark write fallback succeeds. Please apply the Variant check after unwrapping WriteFilesExec so this scan-only change does not enable the unsupported writer.

[sunchao]

[P2] Apply the Variant fallback to the later Python rewrite too

Could we apply this boundary to EliminateRedundantTransitions.EligibleMapInBatch as well? That later rule creates CometMapInBatchExec without passing through this guard. I reproduced a native Variant scan followed by df.mapInPandas(lambda batches: batches, df.schema): it succeeds with spark.comet.exec.pyarrowUDF.enabled=false, but fails with the flag enabled. The accelerated runner forwards the new Arrow schema, which lacks Spark's variant=true metadata on the metadata child. Spark's Pandas serializer therefore supplies a dict rather than VariantVal, and the identity result fails assert isinstance(variant, VariantVal) during output conversion. Keeping Variant-bearing inputs on the ordinary Spark Python path would preserve the intended fallback.

[sunchao]

[P2] Handle NUL-containing field names before C schema export

Could we preserve the Field metadata without passing an embedded-NUL name to Arrow's C-string exporter? Spark accepts a top-level Parquet column named v\u0000suffix. I wrote and read that ordinary BIGINT column successfully with Spark 4.0.4, but the native scan at this head fails with NulError. Arrow 58.4.0's FFI_ArrowSchema::try_from(field) calls CString::new(field.name()).unwrap(), whereas the previous datatype-only export did not serialize the parent name. This affects non-Variant columns too, and the unaligned branch has the same issue. Please use a safe exported name or an explicit pre-execution fallback while retaining the logical metadata.

[sunchao]

[P2] Pin Variant pushdown for the native projection SQL assertions

Could we set spark.sql.variant.pushVariantIntoScan=false for these native-projection cases, as the new Scala vector test does? Spark 4.1 and 4.2 enable that optimizer rule by default, so even SELECT v becomes the annotated VariantStruct representation that this PR deliberately keeps on Spark. The plain query assertion then requires a native plan that cannot be produced. This is the actual failure in both Spark 4.1 CI and Spark 4.2 CI: variant.sql:50 fails with Expected only Comet native operators, but found Project. Please configure the whole-value test path explicitly and keep a separate fallback assertion for pushed VariantStruct.

[peterxcli]

Thanks for the detailed review. I pushed 784c316cf with focused coverage for the six runtime cases and the cross-version SQL assertion:

• Existence defaults: I used the preserve-and-validate alternative rather than rejecting every Variant default. Spark's schema-held VariantVal is transported as a scan-only [value, metadata] constant, every value stays paired with its required-schema index, and the native schema adapter supplies it only when the Parquet field is absent. A present default that cannot serialize falls back safely.
• Unicode object ordering: reconstructed objects are normalized to Spark's Java UTF-16 order. For partially shredded input, the residual value is temporarily put in Arrow UTF-8 order for upstream unshredding and converted to Spark order afterward. The regression contains 32 keys, including U+E000 and 😀.
• Dictionary metadata: dictionary-encoded metadata is decoded before VariantArray::try_new. This has a Rust unit test and an end-to-end native SELECT v test using a generated Parquet fixture, since SQL cannot request nested Arrow dictionary encoding.
• Native writes: the Variant guard inspects the actual data-producing child beneath WriteFilesExec, keeping Variant input on Spark's writer. A focused write-plan test verifies that CometNativeWriteExec is not selected.
• Python rewrite: EliminateRedundantTransitions does not create CometMapInBatchExec when its input or output contains Variant. The Python operation remains on Spark's ordinary path, covered by a focused physical-rule test.
• Top-level NUL-containing FFI names: only the exported C field name substitutes NUL with U+FFFD; datatype, nullability, and Field metadata are preserved. Rust and native-scan tests cover this boundary, and the source evidence is pinned to Spark v4.1.3 and arrow-rs 58.4.0.
• PushVariantIntoScan: whole-value native SQL cases pin it to false, with a separate assertion that pushed VariantStruct remains an explicit fallback.

Focused Rust tests, Spark 4.0/4.1 Variant SQL tests, the Spark 4.0 native-scan selection, writer and MapInBatch fallback tests, Spark 3.5 test compilation, formatting, Clippy, and git diff --check pass. I also refreshed the PR description so it documents the complete data path, compatibility handling, tests, and deliberate fallback boundaries in one place.

[sunchao]

[P2] Allow empty object keys in Variant metadata

Could we preserve valid empty dictionary entries here? On Spark 4.0.4, writing parse_json('{"":1}') to ordinary unshredded Parquet and then reading v with spark.sql.variant.pushVariantIntoScan=false selects CometNativeScan but now fails with offsets not monotonically increasing. Spark reads the same file, and the previous normalizer accepts the identical bytes. Spark encodes this empty key with metadata 01 01 00 00: one dictionary entry with two equal offsets. Arrow/Parquet 58.4.0's full metadata validator requires strictly increasing offsets when the sorted bit is unset, so this new unconditional call rejects the value even though no key reordering is needed. A nested empty key fails the same way; ordinary keys and empty string values pass. Please allow these valid empty keys and add a native-read regression.

[sunchao]

[P2] Scope the forced shredding schema to this SQL fixture

Could we scope or restore spark.sql.variant.forceShreddingSchemaForTest? Because this key is absent from the fixture's header configs, the runner does not restore this SET when variant.sql finishes. On Spark 4.1/4.2, writeShredding.enabled is then restored to true, and later ordinary Parquet writes enter Spark's test-only forced-schema path. Both the 4.1 expression job and 4.2 expression job show variant.sql passing followed by 13 other fixture failures, starting with lag_lead.sql: ParquetWriteSupport.writeFields throws Index 3 out of bounds for length 3. Running the unchanged fixture through its actual runner reproduces a passing ordinary write before it, the same failing write afterward, and recovery after unsetting only this key. Please include this setting in the fixture's scoped configs or restore it so subsequent tests retain their original configuration.

[sunchao]

@peterxcli thanks for the PR! can you check the above CI failures?

[sunchao]

[P2] Widen unsigned shredded Variant children before normalization

Please widen unsigned typed_value children before constructing VariantArray, or keep these reads on Spark. With spark.sql.variant.pushVariantIntoScan=false and spark.sql.variant.allowReadingShredded=true, Spark 4.0.4/4.1.3's vectorized reader accepts ordinary Parquet files whose Variant typed_value is INT32 (INTEGER(8,false)), (16,false), or (32,false), including values 255, 65535, and 4294967295. I reproduced all three failures through the exact-head CometNativeScan: Arrow/Parquet 58.4 restores UInt8/UInt16/UInt32, which this constructor rejects with Illegal shredded value type: UInt8 (or 16/32). These files have no embedded ARROW:schema, and the new top-level Variant gate admits them. Widening only that child to Int16/Int32/Int64 makes the current normalizer accept the same rows. An unsigned upper-bound native-read regression would cover this boundary.