Cached reads that feed Spark operators are slower than Spark's own cache format

[andygrove]

What is the problem the feature request solves?

With Comet's cache serializer installed (spark.comet.exec.inMemoryCache.enabled=true), a query that reads a cached relation through Spark operators rather than Comet ones is slower than the same query against Spark's DefaultCachedBatch, by roughly 1.7x to 2.5x depending on projection width.

Measured on 5M rows and 6 columns, min of 5 runs after a discarded warm-up, with Comet execution disabled so the cached data feeds Spark operators:

Read shape	Spark cache	Comet cache
count(*)	54 ms	190 ms
1 of 6 columns	103 ms	260 ms
3 of 6 columns	308 ms	542 ms
6 of 6 columns	358 ms	595 ms

This is what remains after #5484, which made decode proportional to the projection. With Comet execution enabled the same reads are now 1.7x to 1.8x faster than Spark's cache, so the gap is specific to Spark consuming the cached data.

It matters because the format is chosen when the cache is materialized and cannot be changed afterwards. A pipeline with many fallbacks to Spark operators pays this on every cached read, which is the shape reported in #5051 (comment).

Describe the potential solution

Diagnose before choosing a fix. The cause is not yet established. supportsColumnarOutput returns true for any Comet-writable schema, so Spark's InMemoryTableScanExec takes a columnar path over CometVectors and converts to rows above it, where DefaultCachedBatch feeds OnHeapColumnVectors to the same conversion. Whether the cost is that conversion, the Arrow decode itself, or the loss of Spark's per-column decompression is a guess at this point and should be measured first. supportsColumnarOutput returning false for this case, sending the read through convertCachedBatchToInternalRow instead, is worth measuring as part of that.

Detecting the losing case at write time is not sound, and is worth writing down so it is not re-proposed. The payload format is fixed when the relation materializes, and the readers do not exist yet: one cached relation serves arbitrarily many later queries with different plan shapes, so there is no single correct answer for it. The serializer's inputs are the schema and the input RDD, neither of which carries consumer information.

Three variants that are workable, in descending order of soundness:

1. Report it at read time. CometExecRule knows with certainty whether it replaced InMemoryTableScanExec. When it does not, the query is paying for Comet's cache format without getting Comet's scan. Surfacing that through the existing fallback-reason mechanism, so it appears in EXPLAIN, needs no heuristic and tells the user to disable the feature for that workload. This is the cheapest honest improvement and does not depend on the diagnosis above.
2. Check the caching session's config. If spark.comet.exec.enabled=false when the relation materializes, Comet operators cannot consume it and Spark's format is strictly better. Sound but narrow, since anyone enabling inMemoryCache.enabled most likely has execution enabled too.
3. Use the cached plan as a proxy. If the plan being cached is not itself a Comet plan, the read side is likely to fall back as well. This is the signal that would have helped the report above, but it is a correlation rather than a guarantee, and it is awkward to wire: the serializer receives an RDD[ColumnarBatch], not the plan, so the decision would have to reach it from CometExecRule. The delegation mechanism already exists, since supportsSchema returning false hands the whole relation to DefaultCachedBatchSerializer.

Option 1 is worth doing regardless. Options 2 and 3 only make sense if the diagnosis shows the gap cannot be closed directly.

Additional context

Follow-up from #5051 and #5484. The cache path is off by default and the limitation is documented on spark.comet.exec.inMemoryCache.enabled, so this is a performance gap in an experimental opt-in feature rather than a regression in a shipped path.

Related: #4781, #5245, #5484.