perf: speed up Spark consumers of Comet cache - #5859
Draft
peterxcli wants to merge 4 commits into
Draft
Conversation
peterxcli
force-pushed
the
codex/cache-spark-consumer-benchmark
branch
2 times, most recently
from
September 11, 2026 19:41
2e65a02 to
091eb00
Compare
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?
Related to #5485.
Numeric cache encoding and decoding is split into #5869, stacked on this PR.
Rationale for this change
Spark can read Comet's cached Arrow vectors through a row iterator even when the consumer supports code generation. This materializes an intermediate
UnsafeRowfor every row before the consumer reads its fields.What changes are included in this PR?
Feed eligible cache scans through Spark's
ColumnarToRowExec, which fuses vector reads into the generated consumer. Preserve AQE cache stages and existing columnar boundaries. For remaining row readers, generate an indexed iterator using Spark's reusableUnsafeRowwriter, removing adapters and the extra copy while preserving owned variable-width values and interpreted fallback.How are these changes tested?
Cache and iterator suites pass on Spark 3.4.3, 3.5.9, and 4.1.3, covering cold/warm AQE caches, codegen and interpreted paths, row ownership, nulls, nested values, sorting, joins, and batch boundaries.
Benchmark
The patch reduces cached-read time by 42–78% versus Comet main. Reading all six mixed columns takes 28% less time than vanilla Spark; reading six numeric columns still takes 15% more time.
Spark 4.1.3, JDK 21, Apple M4, 6 GiB heap, one local worker; 5M rows and six columns. Mixed uses three longs and three strings; numeric uses six longs. All queries use Spark operators with Comet native execution disabled. Vectorized cache reading is enabled for all three cases: vanilla Spark and Comet main choose row readers; the patch uses the fused columnar path.
Medians of 30 actions per cell across two fresh JVMs, with five warm-ups per query and reversed run order. Cache creation and planning are outside timing. Main:
8320ae481; measured patch:cd80194cd(reader code unchanged at091eb0020). Patch / Spark is the elapsed-time ratio; lower is better.A separate forced-row control takes 541 ms for six mixed columns and 226 ms for six numeric columns, versus 363 and 180 ms with the columnar path. All 1,080 measured actions, including this control, matched uncached answers. Spark six-column medians varied from 488–505 ms for mixed and 151–168 ms for numeric between JVMs. These are cached aggregate reads on one machine, not whole-application speedups.