feat: project Spark 4 VARIANT columns in native Parquet scans - #5407
feat: project Spark 4 VARIANT columns in native Parquet scans#5407peterxcli wants to merge 14 commits into
Conversation
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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.
| val schemaSupported = scanExec.requiredSchema.fields.forall { field => | ||
| isVariantType(field.dataType) || | ||
| typeChecker.isTypeSupported(field.dataType, field.name, fallbackReasons) |
There was a problem hiding this comment.
[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.
| } | ||
|
|
||
| let variant = VariantArray::try_new(array.as_ref())?; | ||
| let unshredded = unshred_variant(&variant)?; |
There was a problem hiding this comment.
[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.
| )); | ||
| } | ||
|
|
||
| let variant = VariantArray::try_new(array.as_ref())?; |
There was a problem hiding this comment.
[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.
| if (!op.isInstanceOf[CometScanExec] && | ||
| (op.output ++ op.children.flatMap(_.output)).exists(attr => | ||
| containsVariantType(attr.dataType))) { |
There was a problem hiding this comment.
[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.
| op, | ||
| "Native operators do not support schemas containing type VariantType") | ||
| return None |
There was a problem hiding this comment.
[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.
| unsafe { | ||
| std::ptr::write(array_ptr, FFI_ArrowArray::new(self)); | ||
| std::ptr::write(schema_ptr, FFI_ArrowSchema::try_from(self.data_type())?); | ||
| std::ptr::write(schema_ptr, FFI_ArrowSchema::try_from(field)?); |
There was a problem hiding this comment.
[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.
| query | ||
| SELECT v FROM test_variant |
There was a problem hiding this comment.
[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.
|
Thanks for the detailed review. I pushed
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 |
| ))); | ||
| } | ||
|
|
||
| let metadata = VariantMetadata::try_new(metadata.value(index))?; |
There was a problem hiding this comment.
[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.
| SET spark.sql.variant.writeShredding.enabled=true | ||
|
|
||
| statement | ||
| SET spark.sql.variant.forceShreddingSchemaForTest=k00 BIGINT |
There was a problem hiding this comment.
[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.
|
@peterxcli thanks for the PR! can you check the above CI failures? |
| } | ||
|
|
||
| let array = decode_variant_metadata_dictionary(array)?; | ||
| let variant = prepare_variant_for_unshredding(&VariantArray::try_new(array.as_ref())?)?; |
There was a problem hiding this comment.
[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.
|
Follow-up on |
| } | ||
|
|
||
| let array = decode_variant_metadata_dictionary(array)?; | ||
| let array = widen_unsigned_variant_typed_value(&array)?; |
There was a problem hiding this comment.
[P2] Normalize millisecond shredded timestamps before validation
Could we convert Spark-readable TIMESTAMP(MILLIS, ...) children before constructing VariantArray, or retain Spark fallback for them? An ordinary Parquet group read as v VARIANT, with typed_value INT64 (TIMESTAMP(MILLIS,true)) and value 1704067200123, returns 2024-01-01 00:00:00.123+00:00 on Spark 4.0.4 and 4.1.3. Arrow 58.4 restores Timestamp(Millisecond, Some("UTC")), however, and this normalizer throws Illegal shredded value type: Timestamp(ms, "UTC"). The false/NTZ annotation fails the same way; these files have no embedded ARROW:schema. Both Spark reader modes and a MICROS control pass. The direct Variant scan gate admits these files, but neither preprocessing step handles their timestamp units before the constructor rejects them. Recursive millisecond-to-microsecond normalization, preserving LTZ/NTZ, needs native-read coverage here.
|
|
||
| let array = decode_variant_metadata_dictionary(array)?; | ||
| let array = widen_unsigned_variant_typed_value(&array)?; | ||
| let variant = VariantArray::try_new(array.as_ref())?; |
There was a problem hiding this comment.
[P2] Normalize restored fixed-size lists before VariantArray construction
Could we convert restored FixedSizeList typed containers to supported List storage before this constructor, or keep these files on Spark? I reproduced an Arrow-written file read with schema v VARIANT, containing an ordinary physical Parquet LIST and FixedSizeList<Struct<typed_value:Int64>,2> retained in ARROW:schema: Spark 4.0.4 and 4.1.3 return [42,43] with both readers, but exact-head normalization throws Illegal shredded value type: FixedSizeList(...). The fixture uses signed Int64 children and no dictionary, so the existing unsigned/dictionary fixes do not cover it. Casting only that container to List makes the same normalizer return [42,43]. The new FixedSizeList reconstruction branches below are unreachable because Arrow 58.4 rejects the container first; please cover this retained-schema representation in a native-read regression.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed ba37a68, including the changes since ed39c4c. Two new [P2] findings are attached.
The existing Unicode and empty-key cases still reproduce for nested residuals. The dictionary family still fails for dictionary-encoded value/typed_value, and the unsigned family still fails for UInt64; the reported UInt8/16/32 cases now pass. These are continuations of those discussions, so I have not duplicated them inline.
| // errors itself: https://issues.apache.org/jira/browse/SPARK-47546 | ||
| if (scanExec.requiredSchema.fields.exists(field => isVariantType(field.dataType)) && | ||
| !SQLConf.get | ||
| .getConfString("spark.sql.variant.allowReadingShredded", "true") |
There was a problem hiding this comment.
[P2] Honor Spark 4.0's registered strict-reader default
The two-argument getConfString returns its supplied default when the setting is unset, but Spark 4.0 registers spark.sql.variant.allowReadingShredded as false. Consequently, the ordinary unset case bypasses this fallback even though Spark's strict reader is active. I reproduced this on Spark 4.0.4 with a Spark-written shredded Variant file: the effective setting is false and Spark rejects it with INVALID_VARIANT_FROM_PARQUET.WRONG_NUM_FIELDS, while this head selects CometNativeScanExec and returns the value. Explicit false correctly falls back, so the new test misses the default case. Please read the registered default through the version shim (or the no-default lookup for versions with Variant) and cover unset as well as explicit false/true.
| DataType::UInt8 => Some(DataType::Int16), | ||
| DataType::UInt16 => Some(DataType::Int32), | ||
| DataType::UInt32 => Some(DataType::Int64), | ||
| DataType::Timestamp(TimeUnit::Millisecond, timezone) => { |
There was a problem hiding this comment.
[P2] Preserve integer semantics for nanosAsLong Variant children
This normalizer leaves nanosecond timestamps unchanged, but with spark.sql.legacy.parquet.nanosAsLong=true, Spark's vectorized Parquet reader interprets those children as integers. On Spark 4.0.4, with spark.sql.variant.allowReadingShredded=true, a whole-Variant read of typed_value: INT64 TIMESTAMP(NANOS,true) containing 1704067200123000000 returns that integer without Comet; this head, with CometNativeScanExec asserted in the executed plan, instead returns a timestamp (2024-01-01 00:00:00.123 when cast to STRING; schema_of_variant reports TIMESTAMP instead of Spark's BIGINT). The fixture has no embedded ARROW:schema. A non-microsecond-aligned value, 1704067200123456789, instead fails with UNKNOWN_PRIMITIVE_TYPE_IN_VARIANT (type 18; the NTZ case produces 19). Please preserve the configured raw-Int64 interpretation before Variant construction, or fall back for these inputs. Merely converting nanos to micros would still change the Variant value type. The passing baseline here is Spark's vectorized reader; its row reader rejects the fixture.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed a784bea3911d3ee6a96253fc1b025739d01f1df2, including the full diff and increment from ba37a688.
The registered-default and nanosAsLong reports no longer reproduce; the earlier nested Unicode/empty-key, dictionary and UInt64 controls now agree with Spark. Three new P2 findings are attached.
Validation on chao-reviews-1: 36 focused native tests pass; 9/11 focused JVM tests pass, with both failing default tests passing under an explicit reader opt-in. Separate Spark/native probes reproduce both type mismatches with native scan assertions. This is focused validation, not a full-suite result.
| .collect::<Vec<_>>(); | ||
| changed.then(|| DataType::Struct(fields.into())) | ||
| } | ||
| _ => None, |
There was a problem hiding this comment.
[P2] Preserve binary semantics for unannotated fixed-length children
This leaves FixedSizeBinary unchanged, although Spark reads an unannotated Parquet FIXED_LEN_BYTE_ARRAY typed child as Binary. On this exact head with Spark 4.0.4, an ordinary Parquet v group containing binary metadata and typed_value: FIXED_LEN_BYTE_ARRAY(16) (bytes 00..0f, no UUID annotation or ARROW:schema) returns BINARY and "AAECAwQFBgcICQoLDA0ODw==" under both Spark readers, but an asserted CometNativeScanExec returns UUID and "00010203-0405-0607-0809-0a0b0c0d0e0f". Lengths 8 and 20 instead fail with Illegal shredded value type: FixedSizeBinary(...) while Spark succeeds; ordinary BINARY is a passing control. These scans are admitted with allowReadingShredded=true, pushdown disabled and nanosAsLong false. Please preserve the physical binary interpretation before unshredding, or fall back for these inputs.
| // Spark interprets TIMESTAMP(NANOS) leaves as raw longs in this legacy mode. The logical | ||
| // Variant schema does not expose whether a file contains such a shredded child, so preserve | ||
| // Spark semantics by falling back before reading any requested Variant value. | ||
| if (hasVariant && SQLConf.get.legacyParquetNanosAsLong) { |
There was a problem hiding this comment.
[P2] Honor disabled timestamp-NTZ inference inside Variant
nanosAsLong is not the only reader setting that changes a shredded leaf's logical type. With spark.sql.parquet.inferTimestampNTZ.enabled=false, Spark 4.0.4's vectorized reader treats TIMESTAMP(MICROS,false) as TIMESTAMP, while this admitted native Variant path preserves Arrow's absent timezone and returns TIMESTAMP_NTZ. I reproduced this with raw value 1704067200123456, allowReadingShredded=true, pushdown disabled and nanosAsLong false: schema_of_variant(v) differs, and in America/Los_Angeles the STRING results are 2023-12-31 16:00:00.123456 versus native 2024-01-01 00:00:00.123456. Every Comet query asserted one native scan. MILLIS reproduces too; default-inference and adjusted-to-UTC controls agree. The baseline here is explicitly vectorized Spark; its row reader ignores this setting for Variant. Please propagate this interpretation or gate Variant scans when it cannot be honored.
| sql("ALTER TABLE variant_defaults ADD COLUMNS(n INT DEFAULT 7)") | ||
| } | ||
|
|
||
| withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "false") { |
There was a problem hiding this comment.
[P2] Opt both native-default tests into the permissive reader
This scope and the native-read scope in native scan fills a Variant existence default for an old Parquet file only disable pushdown. Spark 4.0 registers allowReadingShredded=false, so the corrected production guard now falls back: the pairing test fails its native-scan count, and the missing-column test reaches Spark's unsupported vectorized Variant-default assignment. Both fail in the exact-head Spark 4.0.4 focused run; the current macOS scans job also records both failures. An external subclass reusing these unchanged tests and changing only sparkConf to set allowReadingShredded=true passes 2/2. Please explicitly set that option in both native-read scopes and retain the separate unset/false/true strict-reader regression.
2 participants
Which issue does this PR close?
Implements the ordinary-Parquet phase of #4295.
Related work:
This PR enables whole-value projection of direct, top-level Spark 4
VariantTypecolumns through Comet's ordinary native Parquet scan:Iceberg Variant projection and general native Variant expressions remain outside this phase.
Rationale for this change
Spark 4 exposes semi-structured values as the atomic
VariantType. At its Arrow boundary, Spark represents a Variant as a Struct with non-null Binary children in[value, metadata]order.ColumnVector.getVariantconsumes those child ordinals directly:Arrow identifies the logical value with the Field-level extension name
arrow.parquet.variant, backed by Struct storage. Parquet represents shredding as a group with requiredmetadata, optionalvalue, and optionaltyped_valuefields (storage contract, reconstruction semantics). Arrow-rs locates those children by name, andunshred_variantreconstructs a whole value.The physical Struct shape alone is therefore insufficient. Comet must preserve the logical Field marker, normalize the reader's physical representation, and present Spark with exactly
[value, metadata].The ordinary native scan path also needs an explicit Variant identity across its existing boundaries:
FFI_ArrowSchemafromArrayData.data_type(). Arrow'sTryFrom<&DataType>exports only the datatype shape, whileTryFrom<&Field>also exports the top-level name, nullability, and metadata; andStructType.Several reader-compatibility cases must be handled at the same boundary:
VariantArray::try_newreaches a binary-only validator. Upstream tracking is in arrow-rs #10802.typed_valuerepresentations that Spark reads after widening, including unsigned integers. The compatibility path and its upstream removal conditions are documented innormalize_variant_typed_value, with arrow-rs #10416 and PR #10417 tracking native support.masterandbranch-4.x: the landed code writes canonical unsigned UTF-8 order and performs canonical lookup with a legacy UTF-16 fallback, with backport789106b0. Remove Variant UTF-16 output rewriting #5474 tracks deletion of Comet's output rewrite when those semantics are available in every supported profile.spark.sql.variant.allowReadingShredded=false, Spark intentionally performs strict validation of the legacy two-child layout and reportsINVALID_VARIANT_FROM_PARQUETanalysis errors. Comet must retain that Spark error boundary for malformed files.What changes are included in this PR?
The end-to-end projection path is:
Preserve Variant identity and keep admission narrow
VARIANT = 21without renumbering existing protobuf values.VariantTypethrough the Spark-version shim while Spark 3.x remains a no-op (datatype serialization, Spark 4 shim).Struct<value: Binary, metadata: Binary>and attaches the canonical extension marker to the Arrow Field (native datatype and Field mapping, extension identity).VariantStructremain unsupported (scan type gate).Unread Variant-bearing roots continue to use the pruning behavior from #5377: they are removed from the native data schema instead of decoded. A requested root is replaced only by its already validated, pruned required field (required-schema construction).
Normalize the Parquet value once
The ordinary Parquet schema adapter installs
CometCastColumnExprwhenever the logical target is a marked Variant Field, including physical/logical identity casts (schema adaptation, identity-cast replacement).CometCastColumnExprroutes that array to the dedicated Variant module (cast boundary).At that single boundary,
normalize_variant_array:metadatabefore constructingVariantArray;unshred_variantto mergetyped_valueinto the whole value;[value, metadata]with the original parent null bitmap; the unchanged target Field remains the output schema Field used at FFI export.The compatibility code is isolated in
cast_column/variant.rswith focused tests incast_column/variant/tests.rs. JVM code does not duplicate unshredding, and the PR adds no Variant dependency beyond Arrow/Parquet 58.4.Apply Variant existence defaults without losing schema indexes
CometNativeScankeeps each serialized existence default paired with its required-schema index. Ordinary defaults remain literals. A SparkVariantValis transported only for this scan path as a constantCreateNamedStruct(value, metadata)using Variant's physical Arrow storage layout.The native planner evaluates either constant form into the
ScalarValueconsumed by the existing Parquet schema adapter (default evaluation, validated index mapping). The value is substituted only when the physical file lacks the column. If any present default cannot be serialized, the whole scan stays on Spark rather than emitting mismatched value/index lists; a schema with no defaults remains native.Preserve the Arrow Field through FFI and restore Spark VariantType
The shared FFI boundary exports each array with its corresponding RecordBatch Field instead of recreating the schema from the array datatype. Offset-normalized arrays use the same original output Field (batch export, Field-based schema export). If a top-level output name contains NUL, only the C-compatible exported name is substituted; datatype, nullability, and metadata remain unchanged.
On the JVM,
Utils.fromArrowFieldmaps only the explicitARROW:extension:name = arrow.parquet.variantmarker to the version-shim Variant type. Unmarked Structs keep their existingStructTypebehavior. The existingCometStructVectorpreserves the two child ordinals, so Spark's inheritedgetVariantworks without a new vector class.Keep unsupported consumers on Spark
The supported surface is direct, top-level, whole-value projection from ordinary Parquet. These paths remain explicit Spark fallbacks:
PushVariantIntoScan/ markedVariantStructoutput;variant_get, predicates, casts,parse_json,to_variant, and other Variant expressions;WriteFilesExec(write boundary);spark.sql.variant.allowReadingShredded=false.How are these changes tested?
Focused coverage verifies:
SELECT vandSELECT id, v, tailretain a Comet native Parquet scan;VariantType, exact[value, metadata]Binary children, and preserved Field metadata/nullability;INVALID_VARIANT_FROM_PARQUETAnalysisException;VariantStruct, and Iceberg remain fallbacks.The end-to-end vector assertions cover unshredded and shredded projection, strict malformed-input fallback, existence defaults, and dictionary metadata. SQL surface and fallback coverage is in
variant.sql, and the focused native normalization cases are invariant/tests.rs.Focused validation passed:
The Rust
parquet::cast_columnselection passed 27 tests. The Spark 4.0/4.1 SQL suites and malformed-input regressions passed, as did the Spark 4.1 native projection regression and the Spark 4.0 writer/Python fallback selections. Spark 3.5 compilation/package validation passed. Rust formatting, Maven Spotless/Scalastyle, andgit diff --checkpassed.