[SPARK-58390][SQL] Emit row counts without deserializing Arrow payloads for empty-projection cache reads - #57583
Closed
viirya wants to merge 1 commit into
Closed
[SPARK-58390][SQL] Emit row counts without deserializing Arrow payloads for empty-projection cache reads#57583viirya wants to merge 1 commit into
viirya wants to merge 1 commit into
Conversation
…ds for empty-projection cache reads An empty projection over an Arrow-cached relation (e.g. a count aggregate through the row-based reader) selects no columns, yet convertCachedBatchToInternalRow still deserialized and decompressed every cached batch's Arrow payload just to iterate its rows. The row count is already recorded on the cached batch, so short-circuit the empty-projection case to emit that many reused 0-field UnsafeRows without touching the payload. Also fix the ArrowCachedBatch scaladoc, which listed the per-column statistics as (upperBound, lowerBound, ...) while both write paths produce the ColumnStats.collectedStatistics order (lowerBound, upperBound, nullCount, count, sizeInBytes). Co-authored-by: Claude Code
uros-b
approved these changes
Jul 28, 2026
uros-b
pushed a commit
that referenced
this pull request
Jul 28, 2026
…ds for empty-projection cache reads ### What changes were proposed in this pull request? Short-circuit `ArrowCachedBatchSerializer.convertCachedBatchToInternalRow` when the projection is empty: emit `numRows` reused 0-field `UnsafeRow`s per cached batch, without deserializing or decompressing the batch's Arrow payload. The row count is already recorded on `ArrowCachedBatch`. Also fixes the `ArrowCachedBatch` scaladoc, which listed the per-column statistics as `(upperBound, lowerBound, ...)` while both write paths produce the `ColumnStats.collectedStatistics` order `(lowerBound, upperBound, nullCount, count, sizeInBytes)`; the code was consistently lower-first everywhere, only the doc was wrong. ### Why are the changes needed? An empty projection (e.g. a count aggregate through the row-based reader, `spark.sql.inMemoryColumnarStorage.enableVectorizedReader=false`) selects no columns, yet the reader still paid full IPC deserialization and decompression for every cached batch just to iterate its rows. That cost is pure waste: the answer is a stored integer. ### Does this PR introduce _any_ user-facing change? No. The Arrow cache serializer (SPARK-57268) is unreleased, and the change is performance-only; results are identical. ### How was this patch tested? Two new tests in `ArrowCachedBatchSerializerSuite`: - `empty projection emits row counts without deserializing the Arrow payload`: hands the reader a cached batch whose Arrow payload is garbage bytes. The empty projection returns the correct number of empty rows purely from `numRows` (fails before this change with `IllegalArgumentException: capacity < 0` from the IPC reader, proving the payload used to be deserialized), while a projection that actually needs the payload still fails on the same batch, pinning that only the empty-projection case skips the read. - `count aggregate over the cached relation with the row-based reader`: end-to-end `count(*)` over a cached relation spanning many small Arrow batches with the vectorized reader disabled, plus a `sum` over the same cached data verifying projecting reads still decode the payload correctly. Full `ArrowCachedBatchSerializerSuite` (73 tests) and `ArrowCachedBatchKryoRegistrationSuite` pass. ### Was this patch authored or co-authored using generative AI tooling? Yes, this pull request and its description were written by Claude Code. Closes #57583 from viirya/arrow-cache-empty-projection. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com> (cherry picked from commit 073c8d8) Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
Member
Member
Author
|
Thanks @uros-b |
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
2 participants
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?
Short-circuit
ArrowCachedBatchSerializer.convertCachedBatchToInternalRowwhen the projection is empty: emitnumRowsreused 0-fieldUnsafeRows per cached batch, without deserializing or decompressing the batch's Arrow payload. The row count is already recorded onArrowCachedBatch.Also fixes the
ArrowCachedBatchscaladoc, which listed the per-column statistics as(upperBound, lowerBound, ...)while both write paths produce theColumnStats.collectedStatisticsorder(lowerBound, upperBound, nullCount, count, sizeInBytes); the code was consistently lower-first everywhere, only the doc was wrong.Why are the changes needed?
An empty projection (e.g. a count aggregate through the row-based reader,
spark.sql.inMemoryColumnarStorage.enableVectorizedReader=false) selects no columns, yet the reader still paid full IPC deserialization and decompression for every cached batch just to iterate its rows. That cost is pure waste: the answer is a stored integer.Does this PR introduce any user-facing change?
No. The Arrow cache serializer (SPARK-57268) is unreleased, and the change is performance-only; results are identical.
How was this patch tested?
Two new tests in
ArrowCachedBatchSerializerSuite:empty projection emits row counts without deserializing the Arrow payload: hands the reader a cached batch whose Arrow payload is garbage bytes. The empty projection returns the correct number of empty rows purely fromnumRows(fails before this change withIllegalArgumentException: capacity < 0from the IPC reader, proving the payload used to be deserialized), while a projection that actually needs the payload still fails on the same batch, pinning that only the empty-projection case skips the read.count aggregate over the cached relation with the row-based reader: end-to-endcount(*)over a cached relation spanning many small Arrow batches with the vectorized reader disabled, plus asumover the same cached data verifying projecting reads still decode the payload correctly.Full
ArrowCachedBatchSerializerSuite(73 tests) andArrowCachedBatchKryoRegistrationSuitepass.Was this patch authored or co-authored using generative AI tooling?
Yes, this pull request and its description were written by Claude Code.