andygrove opened a new issue, #5484: URL: https://github.com/apache/datafusion-comet/issues/5484
### 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 https://github.com/apache/datafusion-comet/pull/5051#issuecomment-5183261544, 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). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
