Repository navigation
Conversation
A slice of a native string, binary, list or map array reaches the JVM with offsets that start past zero, with the rows before the slice still at the front of its data. Arrow Java writes those bytes into the IPC stream, and the VectorSchemaRootAppender that coalesces broadcast batches assumes offsets start at zero, so it prepended them to the first value of each slice. serializeBatches now writes such a batch through a slice of its root, which rebases the offsets to zero and leaves the earlier bytes out.
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.
Which issue does this PR close?
Closes #6779.
Rationale for this change
A Comet broadcast can return wrong values, or fail with
OversizedAllocationException, when itsbuild side emits sliced batches with string, binary, array or map columns, at any nesting level.
arrow-rs slices these arrays by moving the start of their offsets, so the JVM imports a slice
whose offsets start past zero and whose data buffer still begins with the rows before the slice.
zero_offsets(#6339) only handlesArrayData::offset, which arrow-rs keeps for booleans.DataFusion produces such slices whenever an operator emits one batch in
batch_sizepieces.Grouped aggregates do this for every
GROUP BYandDISTINCT, and so doLIMIT ... OFFSET,TopK and window aggregates.
Utils.serializeBatcheswrites these vectors with Arrow Java'sVectorUnloader, which keeps theoffsets as they are and writes the data buffer from byte 0.
coalesceBroadcastBatches(#3703)then appends the decoded batches with
VectorSchemaRootAppender. In Arrow Java 18.3 that assumesoffsets start at zero, so the first value of every slice absorbs all the bytes before it. The
issue's query returns 199976 rows instead of 200000, and the appended data grows with every
slice, which is where the
OversizedAllocationExceptioncomes from. Every release since 0.15.0has this.
Arrow Java fixed
VectorAppenderin apache/arrow-java#1180, but no release includes it yet.19.0.0 is the latest.
What changes are included in this PR?
serializeBatcheschecks each column, children included, for an offset buffer that starts pastzero. When one does, it writes
root.slice(0, rowCount)instead of the root. The slice rebasesthose offsets to zero and drops the bytes before them, and shares every other buffer. Batches
whose offsets already start at zero are written as before.
buffer from the start up to its own end, so the bytes collected to the driver grew with the
square of the number of slices per partition.
VectorSchemaRootAppender.appendcall notes that coalescing relies on this.The in-memory cache serializer (
CachedBatchIpc.serialize) also writes the bytes before the firstoffset. There they only inflate the payload and its size statistics, and the values read back
correctly, so this PR leaves it unchanged.
How are these changes tested?
CometJoinSuite"Broadcast coalescing keeps the values of sliced batches" broadcastsSELECT DISTINCT *over string, binary,array<string>and struct columns with nulls at batchsize 100. It compares every value with Spark and checks that the broadcast still coalesced.
UtilsSuite"serializeBatches rebases offsets that start past zero" coalesces string vectorsshaped like imported slices and checks the values. It also checks that each slice takes as many
bytes as the same values in a vector of their own.
Both tests fail without the fix. Locally, on Spark 4.1 against a build of recent
main, the newtests, the issue's reproducer, the rest of
UtilsSuiteand 25 existing broadcast tests inCometJoinSuite,CometExecSuiteandCometInMemoryCacheSuitepass. The Spark 3.x profiles werenot run locally.