Re: [PR] perf: add opt-in delta encoding for cached longs [datafusion-comet]

["andygrove (via GitHub)" <gi...@apache.org>]

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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org