[SPARK-58381][SQL] Gate the Arrow cache zero-copy write path on physical congruence with the cache schema - #57577
Open
viirya wants to merge 1 commit into
Open
Conversation
…cal congruence with the cache schema The zero-copy cache write path serializes the input vectors' buffers verbatim under the cache's own schema: serializeBatch writes only the record batch, and the read path reconstructs the schema from cacheSchema and loads the buffers positionally into it. Eligibility was decided by a blocklist (large var-width vectors and interchange-shaped nanosecond timestamp / CalendarInterval vectors), so any other physical divergence between an input vector and the corresponding cache schema field -- view or dictionary encodings, list offset widths, tagged structs carrying extra children, map entry children in the wrong order, timestamp units -- was forwarded verbatim and silently reinterpreted under the canonical layout on read. A map vector with entry children in [value, key] order, which Arrow tolerates and ArrowColumnVector reads correctly by name, comes back from the cache with every key and value swapped. Replace the blocklist with the per-column physical-congruence check ArrowUtils.isCompatibleWithDeclaredField (SPARK-58258), which compares the input vector's field tree against the cache schema field. Incongruent input falls back to the row-based conversion, which rewrites the values through ArrowWriter under the cache schema. Co-authored-by: Claude Code
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Replace the blocklist that decides zero-copy eligibility on the Arrow cache write path (
ColumnarBatchToArrowCachedBatchIterator) with a per-column physical-congruence check against the cache schema, usingArrowUtils.isCompatibleWithDeclaredField(introduced by SPARK-58258 for the columnar Python UDF pass-through). Input vectors whose field trees are not physically congruent with the corresponding cache schema field now take the row-based conversion path, which rewrites the values throughArrowWriterunder the cache schema. ThecontainsCacheSchemaMismatchblocklist is removed.Why are the changes needed?
The zero-copy path serializes the input vectors' buffers verbatim under the cache's own schema:
serializeBatchwrites only the record batch, and the read path reconstructs the schema fromcacheSchemaand loads the buffers positionally into it. The blocklist covered only large var-width vectors and interchange-shaped nanosecond timestamp /CalendarIntervalvectors, so every other physical divergence was forwarded verbatim and silently reinterpreted under the canonical layout when the cached batch is read back -- the same defect class SPARK-58258 eliminated on the UDF read side, one lifecycle stage earlier. Shapes the blocklist missed include view and dictionary encodings, list offset widths (LargeList,ListView), tagged structs carrying extra children, map entry children in the wrong order, and timestamp units.The corruption is concrete: an Arrow map vector whose entry-struct children sit in
[value, key]order is tolerated by Arrow and read correctly byArrowColumnVector(key/value are addressed by name), but its buffers are positional, so a cached{123: 7}comes back as{7: 123}-- silently, and for the lifetime of the cached relation. The new test demonstrates this: it fails on master withArray((7, 123)) did not equal Array((123, 7))and passes with the gate.In-tree columnar producers happened to be covered by the blocklist (the cache scan itself produces canonical vectors; Python UDF output's large var-width and interchange nanos/interval shapes were the blocked entries), so this is reachable today only from external Arrow-backed DSv2 columnar sources --
supportsColumnarInputaccepts any of them. The congruence check closes the whole class rather than the enumerated instances, and keeps the write gate symmetric with the read-side gate from SPARK-58258.A false reject is safe by construction: the row-based fallback re-encodes values under the cache schema (with its value-domain guards), so eligible batches lose only the zero-copy optimization, never correctness.
Does this PR introduce any user-facing change?
No. The Arrow cache serializer (SPARK-57268) is unreleased.
How was this patch tested?
New test in
ArrowCachedBatchSerializerSuitethat builds a spec-tolerated but non-canonical map vector (entry children[value, key]), verifies the source reads correctly, round-trips it through the serializer's columnar write and read paths, and asserts the values survive. Without the fix it fails with the keys and values swapped. FullArrowCachedBatchSerializerSuitepasses (72 tests), confirming canonical shapes remain zero-copy eligible.Was this patch authored or co-authored using generative AI tooling?
Yes, this pull request and its description were written by Claude Code.