Comet cache decode ignores column projection, making narrow reads of wide cached relations slower than Spark

[andygrove]

What is the problem the feature request solves?

#5051 adds a Comet cache serializer that stores each cached batch as a single compressed Arrow IPC stream covering every cached column (CometCachedBatch). Column projection happens after the stream is decoded: convertCachedBatchToColumnarBatch calls Utils.decodeBatches(cb.bytes, "CometCache") and then projectBatch(batch, indices).

The consequence is that read cost is flat in the width of the projection. A scan that needs one column of a wide cached relation decompresses and Arrow-decodes all of them. Spark's DefaultCachedBatch stores each column separately, so its cost falls away as the projection narrows.

Measured on a 5M-row, 6-column relation (1 long key, 2 longs, 3 strings), cached and then read repeatedly by Spark operators, comparing spark.sql.cache.serializer set to Comet's serializer against Spark's default:

read shape	Spark cache	Comet cache	ratio
1 of 6 columns	353 ms	1175 ms	3.3x slower
3 of 6 columns	1078 ms	1859 ms	1.7x slower
6 of 6 columns	1227 ms	1728 ms	1.4x slower
materialize cache	5050 ms	2468 ms	2.0x faster

Materialization is faster, so a cache-once/read-many workload that projects a subset of a wide relation can be a net regression even though the write path improved. This is the shape reported against #5051 in #5051 (comment), where a pipeline with many fallbacks to Spark operators went from under 10 minutes to over 15.

Note that the gap does not close entirely at full projection, so decode and row conversion cost is a second, smaller contributor on top of the missing pruning.

Describe the potential solution

Make decode cost proportional to the projection. Options, roughly in increasing order of work:

1. Write one Arrow IPC stream per column (or per column group) into CometCachedBatch and decode only the streams the scan selects. This keeps the payload Arrow and leaves the stats and pruning logic untouched, at the cost of more, smaller buffers per batch.
2. Keep a single stream but record per-column byte ranges in CometCachedBatch, so a selected subset can be decoded without inflating the rest. Whether this is workable depends on the compression codec being applied per column rather than across the whole stream, which today it is not: Utils.serializeBatches wraps the entire ArrowStreamWriter in one codec.compressedOutputStream.
3. Push projection into the decode itself so unselected columns are skipped while reading the stream.

Option 1 is the most direct and reuses the existing serialization path per column.

Whichever route is taken, the benchmark needs a genuine baseline to measure against. CometInMemoryCacheBenchmark currently cannot provide one: spark.sql.cache.serializer is a static conf, so both of its cases read a Comet-written cache and "Comet cache disabled" means Spark execution over CometCachedBatch, not Spark's own cache format. A cross-serializer comparison needs a second SparkSession, and InMemoryRelation memoizes the resolved serializer in a JVM-static field, so InMemoryRelation.clearSerializer() has to be called between sessions or every phase after the first silently reuses the first phase's serializer and the comparison is vacuous.

Additional context

Follow-up from review and testing of #5051. The cache path is off by default (spark.comet.exec.inMemoryCache.enabled=false) and the limitation is documented on that config, so this is a performance gap in an experimental opt-in feature rather than a regression in a shipped path.

Related: #4781 (cache performance follow-ups), #5245 (AQE test coverage).

[andygrove]

Fixed in #5051 rather than as a follow-up, since a cache format that loses to Spark's on narrow reads is not worth shipping even behind a disabled-by-default flag.

CometCachedBatch now stores one compressed Arrow IPC stream per column and decodes only the selected ones, which is option 1 from the description. Two related cases went with it: an empty selection stays empty instead of expanding to every column, and CometInMemoryTableScanExec asks for one cheap column rather than the whole cache schema when a query needs only the row count. That widening was why count(*) was the worst case rather than the best.

Results on the same 5M row, 6 column relation, min of 5 runs after a discarded warm-up:

Read shape	Spark cache	Before	After
count(*)	62 ms	241 ms	77 ms
1 of 6 columns	105 ms	592 ms	57 ms
3 of 6 columns	339 ms	655 ms	204 ms
6 of 6 columns	375 ms	448 ms	448 ms

Per-column framing costs a schema block and compression framing per column per batch and gives up cross-column compression: footprint grows 2.5% at 6 columns and 32% at 60, where the cached relation is still 22x smaller than Spark's format.

Not fixed here, and worth a separate issue if anyone wants to pursue it: reads that feed Spark operators rather than Comet ones are still 1.7x to 2.5x behind Spark's cache, because they pay a row conversion that Spark's format avoids with a codegen'd path over its own layout. That is unrelated to projection.

Closing when #5051 merges.