andygrove commented on PR #5051: URL: https://github.com/apache/datafusion-comet/pull/5051#issuecomment-5432350051
> It looks similar to https://github.com/apache/spark/pull/56334 too although seems in Comet we store compressed IPC stream and can decode only selected columns. Thanks @sunchao, that is a useful pointer and I had not seen it. I went through it and filed #5487 with the comparison so it does not get lost. A few things worth saying here. **On the version.** It is in `branch-4.3` and `master` only, not 4.0, 4.1 or 4.2, so it is not available in any Spark version Comet supports today. It does not remove the need for Comet's serializer, but it is a more mature implementation of the same idea and several of its decisions are better than mine. **On the difference you spotted**, which is real but slightly the other way round from how it sounds. Both formats decode only the selected columns. Comet gets there by splitting the payload into one compressed stream per column, which is the change I pushed a few commits ago. Spark keeps a single RecordBatch and instead parses the IPC message flatbuffer, which lists every buffer's offset and length within the body, copying out only the byte ranges belonging to the selected columns and letting `VectorLoader.load` decompress just those. That leans on Arrow's native per-buffer compression rather than wrapping the whole stream in a Spark `CompressionCodec` as Comet does. Spark's is the better design point. Per-column streams pay framing per column per batch and give up cross-column compression: measured against the previous single-stream format, footprint grows 2.5% at 6 columns and 32% at 60. Spark's approach gets the same projection-proportional decode with none of that. Part of the gap is simply that their `ArrowCachedBatch` carries no Schema message at all, reconstructing it from the relation's attributes on read, where Comet writes a full stream schema per column per batch, roughly 30,000 of them in that 60-column test. I have not changed course here, since the buffer-span reader is about 120 lines of fairly intricate code and belongs in its own PR rather than bolted onto this one, but it is written up as the first thing to reconsider. **Two smaller things from their implementation**, both in #5487: they register `ArrowCachedBatch` for Kryo, and Comet has no Kryo registrator at all, so `CometCachedBatch` would fail under `spark.kryo.registrationRequired=true`. And their row read path builds typed column readers once and writes straight into an `UnsafeRowWriter` instead of going through a generic row iterator, which is a concrete candidate for #5485, where I have measurements but no established cause. **One point of convergence worth noting**, since it suggests neither of us is off in the weeds: they landed the same empty-projection optimization I did, emitting row counts without touching the payload so `count(*)` stops decoding everything, and they landed it as a follow-up in SPARK-58390 about a week after the main PR. Same gap, found in the same order. @viirya if you have a moment, the question I would most value your view on is whether the buffer-span projection reader was worth its complexity in practice, or whether you would take the simpler split-by-column route knowing what you know now. I also checked our collation handling against yours and there is no bug on our side, but you do prune collated string columns where we decline to: a collated `StringType` does not match Comet's `case StringType` in the bounds check, so those columns get null bounds and no predicate pushdown. Filed as part of #5487 rather than fixed here. -- 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]
