feat: normalize marked Variant arrays at the native Parquet boundary - #5715
feat: normalize marked Variant arrays at the native Parquet boundary#5715peterxcli wants to merge 2 commits into
Conversation
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
This adds the native normalization step needed between Parquet's Variant storage and Spark's whole-value vector layout. It resolves metadata, value, and typed_value by name, prepares legacy residual objects, calls the existing unshred implementation once, and returns [value: Binary, metadata: Binary] with the requested Field and parent validity. Selecting this path before the adapter's identity shortcuts is necessary for a marked column whose physical and logical Fields already match.
The ordering conversion agrees with maintained branch-4.0-openai at 03f28fc4318024830a2ee8da7e83c42e0994d37a: Spark writes object fields with Java String.compareTo and uses that order for binary lookup from 32 fields onward. The tests include that boundary with supplementary Unicode keys. The output order and non-null binary child declarations also agree with Spark's Variant representation. Comet preserves its parent extension marker through return_field, which its Arrow importer recognizes. This is distinct from Spark's own Arrow child-metadata convention.
I found no new correctness P1/P2 in the declared native scope. Parent-null rows bypass value decoding. The residual walk propagates root metadata rows through nested objects and list offsets, while preserving parent validity and list buffers. Sliced-list offsets remain relative to the retained child array. SQL arrays, structs, and maps containing Variant still fail the existing recursive admission guards. This change therefore does not establish end-to-end Variant scan support. The PR explicitly assigns encoded children, coercions, empty-key compatibility, and reader policy to #5550. Maintained Spark 3.5 has no Variant type. Maintained 3.4 and 4.1 refs were unavailable, so those profiles are not independently source-qualified here.
Validation
Reviewed HEAD be5c547eab4335ae2ef54957a6258f21dd8750a4 against BASE 7190df631afe3795914839203c7afe57ea23903c. The authenticated discussion cutoff is 2026-09-05 21:06:20 UTC, with no prior comments or reviews. The Rust CI job explicitly passed all seven normalization tests and the marked-column name-remapping test. Its checkout was merge cc3806941504b16b6ec0fbf5a702d85fdbb3167d, whose parents are this exact base/head and whose tree equals the reviewed head. This is the recorded CI checkout. The current PR merge SHA was unavailable at the latest full REST refresh.
The snapshot has 69 successful checks, 9 skipped checks, and one failure. The macOS Spark 4.0 scans job reports a JVM SIGSEGV at address zero. I have not established a causal link to this patch. Skipped Spark SQL profiles and the skipped benchmark check are not passing evidence. Validation here combines source inspection with verified CI execution, not a local Spark/JNI run or a performance measurement.
Performance
There is one P2 in the unchanged-value path, detailed inline. reorder_variant_values still copies the entire Binary payload after determining that no ordering change is needed. rewrite_residual_values also fills a builder that it discards when changed stays false. These allocations are avoidable in this new native normalization path.
Please include a focused microbenchmark with already normalized scalar and ASCII-object values, partially shredded values, nested lists, and Unicode objects that require rewriting. Vary payload width and report allocations as well as throughput. Include an ordinary unmarked-column control. Source inspection shows that unmarked arrays only take the extension-name check here and do not enter the recursive normalization walk. The PR supplies correctness tests but no measurement of this cost.
Design
The boundary is well chosen: Parquet-specific storage reconstruction stays out of the generic Spark expression cast, and the existing upstream unshred implementation remains responsible for merging residual and typed values. Preserving the logical Field separately from its physical children keeps extension identity and field metadata available to consumers. The wrapper is installed before physical-name restoration, which preserves the existing case-insensitive mapping behavior.
The split from JVM admission is explicit and matches the current guards. Supporting a Variant nested inside an ordinary SQL container will need recursive reader integration before those guards are relaxed. The current residual recursion is for a single shredded Variant's internal objects and lists. It should not be treated as proof of that broader support.
Abstraction & complexity
The private module gives the compatibility work a clear boundary. SparkMetadataBuilder supplies sort keys while retaining original dictionary IDs, so it can reuse the upstream object builder without inventing another Variant encoding implementation. The row mapping is needed because flattened list children must use their originating root's metadata.
The main simplification is to create output buffers only after an actual rewrite is discovered, as requested inline. That preserves the compatibility checks while avoiding eager copies on the common unchanged path. The existing #5474 removal condition also keeps the temporary UTF-16 compatibility layer tied to supported Spark behavior.
| output.append_value(rebuilt.as_deref().unwrap_or_else(|| value.value(index))); | ||
| } | ||
|
|
||
| Ok(Arc::new(output.finish())) |
There was a problem hiding this comment.
Performance
[P2] Reuse unchanged Variant buffers instead of copying every value
Could we allocate the output only when a value actually needs rewriting? For a valid, non-null marked column already stored as [value: Binary, metadata: Binary], prepare_variant_for_unshredding and upstream unshred_variant both return without rebuilding, and an already ordered value makes rebuilt be None. This line nevertheless calls BinaryBuilder::append_value, which copies all its bytes, and the function always returns a fresh buffer. Even a column of scalar strings therefore pays an additional full-payload allocation and copy on every normalization. rewrite_residual_values has the same eager copy and then discards the builder when changed is false. Please retain the original array after validation when no rewrite or null-mask adjustment is needed, and initialize a builder lazily for changed batches. A focused allocation/throughput benchmark covering canonical and partially shredded inputs would verify this path. This concern is about the native API added here; JVM Variant scan admission remains closed.
There was a problem hiding this comment.
Addressed in b7338be. Both helpers now initialize a builder only at the first rewrite or required null-mask adjustment, copy the unchanged prefix once, and otherwise retain the original array after validation.
I compared the original eager-copy implementation against the lazy implementation on the same DataFusion 55.0.0 / Arrow-Parquet 59.3.0 dependencies. The baseline received only the accessor API adaptations needed to compile on 59.3.
Allocation results
Each batch contains 128 non-null rows with a 4 KiB string payload:
- Canonical: whole scalar strings already stored as
[value: Binary, metadata: Binary]; conversion to that layout happens before measurement. - Partially shredded: objects containing
known = 1and the string payload, withknownshredded to Int64 andpayloadretained in the residual value.
| Input | Eager-copy baseline | Lazy buffers | Reduction |
|---|---|---|---|
| Canonical | 1,052,007 B/batch | 768 B/batch | >99.9% |
| Partially shredded | 5,322,367 B/batch | 3,208,367 B/batch | 39.7% |
These are gross requested allocation bytes, counting alloc and realloc, not peak RSS or retained memory. A temporary System allocator wrapper measured one normalization after three warmups, excluding fixture construction. The normalization module was compiled with rustc -C opt-level=3 against the same debug-built dependencies for both implementations. The remaining canonical allocations are small wrappers; the payload buffers are reused. Partially shredded input still incurs upstream reconstruction and representation-conversion allocations.
The regressions check buffer identity, sliced inputs, late rewrites, parent-null adjustments, and validation errors. The buffer-reuse assertion fails against the original implementation. All 15 focused cast-column tests and 47 schema-adapter tests passed.
I also added an explicit throughput workload with 4,096 rows and 30 measured iterations. Local timings were unstable under memory pressure and concurrent compilation, so I am not claiming a throughput speedup. The temporary allocation instrumentation is not part of that checked-in test.
This measures the native normalization API only; JVM Variant scan admission remains closed.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Re-reviewed HEAD b7338bedb6f9d3dc376f892facaf539efffce9aa against BASE c348f775d5b20d91d1989756b72f62f366d67d32, including the update from be5c547eab4335ae2ef54957a6258f21dd8750a4. The prior P2 is addressed; I found no new P1/P2. The lazy builders preserve the unchanged prefix, sliced offsets and null rows. A parent-null row with a still-valid value starts a builder before appending null, while active rows continue to validate metadata and values before buffer reuse.
The output remains consistent with maintained Spark branch-4.0-openai at 03f28fc4318024830a2ee8da7e83c42e0994d37a: [value, metadata] Binary children and UTF-16 object-key ordering, including Spark's 32-field binary-search boundary. JSON null and SQL parent null remain distinct. Internal shredded lists retain their originating metadata-row mapping. JVM admission remains closed, including SQL containers containing Variant. Maintained Spark 3.5 has no Variant type; maintained 3.4 and 4.1 refs were unavailable for independent source qualification.
Validation
The current Rust CI job passed all 15 cast-column tests, including nine Variant tests, and all 47 schema-adapter tests. Its checkout was merge 8dc3b6eb632d23c076615e8ef0f2582825ae5227, with this exact base/head as parents and a tree identical to the reviewed head.
The final current-head CI refresh at 2026-09-05 22:49:34 UTC reported 57 successful, eight skipped, six running and one failed checks. The macOS scans failure reports a JVM SIGSEGV; I have not established a causal link to this change. This review uses source inspection and verified CI execution, with no local Rust/Spark/JNI execution or end-to-end Variant scan qualification. CI is not fully green.
Performance
The previous payload-copy finding is fixed in both helpers. Unchanged batches retain their original buffers; changed batches initialize a builder at the first required rewrite and copy the prefix once. The new identity assertions cover the direct helper, whole normalization, sliced input and idempotent output.
The author's allocation measurements support the expected reduction for canonical and partially shredded inputs. They measure gross requested allocation bytes using temporary instrumentation, not peak memory, and I have not independently reproduced them. The checked-in throughput workload is ignored by normal tests; it supplies neither an allocation counter nor a stable throughput result. No runtime speedup is established here.
Design
The merge preserves the adapter's Variant routing. The 73 other changed paths inherited from main match the new base exactly. The dependency manifests and lockfile likewise match the base; the relevant API adaptations agree with locked DataFusion 55.0.0 and Arrow/Parquet 59.3.0. In particular, value_column() is guaranteed to exist, including when a fully shredded input needs a synthesized null residual column.
Abstraction & complexity
The small binary_prefix_builder helper keeps prefix copying shared between the two lazy paths. It uses the array iterator, which preserves sliced offsets and nulls without separate indexing logic. The update adds no broader abstraction or reader-policy change, and I found no new complexity issue.
2 participants
Which issue does this PR close?
Closes #5549.
Third split of #5546
Rationale for this change
A marked Variant Field can reach the native Parquet schema adapter as a Struct containing
metadata, residualvalue, and optionaltyped_value. Spark's whole-value vector interface requires exactly two Binary children in[value, metadata]order. Reordering children alone cannot reconstruct shredded values.The adapter can also eliminate identity casts or leave identical physical and logical Fields as a bare column. A Variant with that layout still needs normalization, so the extension marker must select the conversion before those shortcuts.
What changes are included in this PR?
CometCastColumnExpr, including identical Fields and the adapter's complex-type fallback.cast_column/variant.rs. Reuse the existingparquet::variantimplementation to resolve storage children by name and callunshred_variantonce.[value: Binary, metadata: Binary], removetyped_value, preserve parent nulls, and retain the original marked output Field.münchenresolves to physicalMÜNCHENwhile keeping the normalization wrapper and output marker.This is native normalization infrastructure. JVM Variant scan admission remains closed, so this PR does not enable
SELECT von its own. Encoded children, physical-type coercions, empty-key metadata compatibility, reader-schema policy, and remaining field-ID handling belong to #5550. No dependency is added.How are these changes tested?
Focused Rust tests cover canonical, fully shredded, and partially shredded values; out-of-order storage children; objects, arrays, scalars, JSON null, SQL null, nullable parents; nested residual metadata-row mapping; Unicode object ordering; idempotent normalization; and Unicode schema-name remapping.