andygrove commented on code in PR #5869:
URL: https://github.com/apache/datafusion-comet/pull/5869#discussion_r4238240496
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchIpc.scala:
##########
@@ -524,6 +569,45 @@ private[comet] object CachedBatchIpc {
buffers.add(packed)
}
+ deltaColumns.foreach { case (index, column) =>
+ val buffer = batch.getBuffers.get(index)
+ val packed = buffers.get(index)
+ if (batch.getLength > 0) {
+ val deltas = allocator.buffer(buffer.writerIndex())
+ try {
+ var previous = 0L
+ var smallDeltas = 0
+ var row = 0
+ while (row < batch.getLength) {
+ val value = buffer.getLong(row * 8L)
+ val delta = value - previous
+ deltas.setLong(row * 8L, delta)
+ if (delta == delta.toInt.toLong) smallDeltas += 1
+ previous = value
+ row += 1
+ }
+ deltas.writerIndex(buffer.writerIndex())
+ // Skip a second compression for full-width irregular longs. The
size comparison
+ // still rejects poorly compressing deltas from narrower
distributions.
+ if (smallDeltas.toLong * 2 >= batch.getLength) {
Review Comment:
This check sends every column whose values fit in an `int` through a second
zstd pass, whether or not the deltas end up smaller. I cached a 5M-row relation
of three such `bigint` columns: uniform below 1e6, uniform below 2^31, and a 1
ms step with up to 1 ms of jitter. With the option on, none of the 512 batches
kept deltas for any column and the footprint stayed at 46.9 MiB. The build went
from 210 ms to 299 ms, median of 7, and every run with the option on was slower
than every run without it. Per 10,000-row column, uniform values below 1000
took 328 us to serialize instead of 169 us. Their deltas compress to 1.38 times
the size of the values, so they can never pass the bar.
Could the loop also compare how wide the deltas are with how wide the values
are, and only try the second pass when the deltas are clearly narrower? For
example, summing `64 - numberOfLeadingZeros` of each value's zigzag form, and
trying only when the deltas come out at least four bits narrower on average.
That skips every uniform case I tried and still tries all the sequential,
constant-step and cyclic ones.
##########
docs/source/user-guide/latest/in-memory-cache.md:
##########
@@ -68,6 +68,14 @@ A relation whose cached plan records observed metrics, from
`Dataset.observe`, i
Comet's format but is scanned by Spark's `InMemoryTableScanExec`, because
Spark collects those
metrics only through that scan.
+Set `spark.comet.exec.inMemoryCache.deltaEncoding.enabled=true` to try delta
encoding
+for top-level `bigint` columns in compressed caches. A column uses deltas only
when
+its compressed data buffer is over 25% smaller than the plain representation.
Review Comment:
The code compares the compressed deltas with the compressed values
(`encoded.writerIndex() < packed.writerIndex() * 3 / 4`), not with the plain
representation, so this sentence describes a different threshold. The
`CometConf` doc string has the same gap, since it only says the delta buffer is
over 25% smaller. Could both say the comparison is between the two compressed
forms?
It would also help to give numbers here, the way the codec table under
Storage format does, perhaps from a delta arm in `runCodecBenchmark`. That
benchmark's relation is a good case for this option. I measured 54.8 MiB
without deltas and 44.2 MiB with them. `id`, `k` and `v` went from 6.3, 0.9 and
6.5 MiB to 1.3, 0.4 and 1.5 MiB, and reads were no slower. The build cost on
columns that are not sequential is worth stating too.
##########
spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CometCachedBatchHelper.scala:
##########
@@ -122,8 +125,9 @@ object CometCachedBatchHelper {
batch: ColumnarBatch,
codec: CompressionCodec,
allocator: BufferAllocator,
- chunkSize: Int = 1024 * 1024): ChunkedByteBuffer =
- CachedBatchIpc.serialize(batch, codec, allocator, chunkSize)._1
+ chunkSize: Int = 1024 * 1024,
+ deltaEncoding: Boolean = false): ChunkedByteBuffer =
+ CachedBatchIpc.serialize(batch, codec, allocator, chunkSize,
deltaEncoding)._1
Review Comment:
With `deltaEncoding = true` this returns the payload without its flags, and
`cachedBatch` below then builds a batch that has none. A test that round trips
through these and `load` would silently read the deltas back as values. The doc
on `cachedBatch` says it is the batch the writer would have stored, which no
longer holds. Could `serialize` return the flags as well, or the `CachedBatch`
itself?
--
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]
