andygrove commented on PR #5051: URL: https://github.com/apache/datafusion-comet/pull/5051#issuecomment-5432138279
Fixed #5484 here rather than leaving it as a follow-up. A cache format that loses to Spark's on the reads people actually do is not worth shipping, even off by default, so it seemed better to fix it than to document it. **What changed.** A `CometCachedBatch` now stores one compressed Arrow IPC stream per column instead of one stream covering all of them, and `convertCachedBatchToColumnarBatch` decodes only the columns it was asked for. Two related cases fell out of the same change: an empty selection now stays empty rather than expanding to every column, and the scan asks for a single cheap column rather than the whole cache schema when a query needs only the row count. That last one is why `count(*)` was the worst case rather than the best: `CometInMemoryTableScanExec` widened an empty projection to the full schema, because the native plan needs a non-empty scan schema, and once the serializer decodes exactly what it is asked for that widening meant decoding everything. One column satisfies the native planner just as well. Per-column sizes are also known now, so the fifth statistics field Spark reserves for them holds the real value instead of zero. **Results**, 5M rows and 6 columns, min of 5 runs after a discarded warm-up, against a real `DefaultCachedBatch` baseline in a separate session: | Read shape | Spark cache | Comet before | Comet 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 | | Materialize | 3497 ms | 1563 ms | 1563 ms | | Footprint | 309 MiB | 60 MiB | 61 MiB | Narrow and medium projections now beat Spark's cache by 1.7x to 1.8x rather than losing to it. `count(*)` is at parity. A full-width projection is still 1.2x behind, which is decode and conversion cost rather than anything to do with projection. **The cost**, since I said I would measure it rather than assume it. Per-column framing adds 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 columns. At 60 columns the cached relation is still 107 MiB against Spark's 2369 MiB, so I took the tradeoff, but it is a real one and it scales with width, so it is worth knowing about. **What is still slower.** Reads that feed Spark operators rather than Comet ones remain 1.7x to 2.5x behind Spark's cache. Those pay a row conversion that Spark's format avoids with a codegen'd path over its own layout. That is a different cost from the projection problem and I have not touched it here. @sandugood this should help your pipeline on the read side, but that fallback gap is exactly the shape you were running, so I would not promise it closes the whole 10-to-15-minute difference. If you do retest, the two things worth separating are cache materialization and the repeated reads, since they now move in opposite directions. **Testing.** Five new tests in `CometInMemoryCacheSuite`. The two that matter corrupt the column streams a read must not touch, so they assert on what was actually decoded rather than on timings; I checked both fail against the previous single-stream format rather than passing for free. All 26 tests in the suite pass, along with `UtilsSuite`, `CometNativeColumnarToRowSuite` and `NativeUtilSuite`. @peterxcli this changes the cached payload and the read path, so your approval no longer covers it. Sorry to move the target after you had signed off. The lifetime handling around the readers is the part I would most want a second pair of eyes on: a cached batch is now several independent Arrow streams, so the `TaskCompletionListener` you asked for has to release all the readers for the batch in flight rather than one. -- 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]
