feat: add lz4 as an opt-in codec for Comet's in-memory cache

docs/source/user-guide/latest/in-memory-cache.md

@@ -91,36 +91,47 @@ column out of six does roughly a sixth of the decompression work, and a `SELECT
 selects no columns at all, answers from the row count stored beside the payload without touching
 it.
 
-Compression defaults to `zstd`, for footprint rather than for speed. Over the same 5M-row,
-six-column relation the tables under [Performance](#performance) use — and measured by the same
-benchmark — it holds the data in a sixth of the memory and pays for that on both sides: about 40%
-longer to materialize, and, on a read wide enough to inflate everything, close to five times
-longer. A narrow projection pays far less, because it only inflates the columns it asked for.
+Three codecs are offered: `zstd`, the default, `lz4` and `none`. Measured by the same benchmark
+over the same 5M-row, six-column relation the tables under [Performance](#performance) use (Apple
+M3 Ultra, JDK 17, Spark 4.1, release build):
 
 | Codec  | Materialize | Footprint | Read 1 of 6 | Read 6 of 6 |
 | ------ | ----------: | --------: | ----------: | ----------: |
-| `zstd` |     1507 ms |    55 MiB |       45 ms |      295 ms |
-| `none` |     1081 ms |   315 MiB |       35 ms |       64 ms |
+| `zstd` |     1232 ms |    51 MiB |       36 ms |      248 ms |
+| `lz4`  |     1115 ms |   145 MiB |       33 ms |      110 ms |
+| `none` |      857 ms |   315 MiB |       31 ms |       54 ms |
+
+`zstd` is the default, for footprint rather than for speed. It holds the data in a sixth of the
+memory `none` takes and pays for that on both sides: about 45% longer to materialize, and, on a read
+wide enough to inflate everything, more than four times longer. A narrow projection pays far less,
+because it only inflates the columns it asked for.
+
+`lz4` decompresses several times faster than `zstd`, so a read of every column takes less than half
+as long, at nearly three times the footprint. It is written with lz4-java, the JNI-accelerated
+library behind `spark.io.compression.codec`, in Arrow's standard `LZ4_FRAME` format. Arrow's own
+LZ4 codec is not used: it is commons-compress's pure-Java implementation, and is orders of
+magnitude slower to write than `zstd` while also producing larger output.
+
+What `lz4` saves depends more on the data than what `zstd` saves does. It holds this relation in
+less memory than Spark's own cache format does (217 MiB), but not a relation of sequential `bigint`
+columns, which Spark's format delta-encodes: each of the wide relations under
+[Limitations](#limitations) takes about 74 MiB with `lz4`, against 26 to 27 MiB in Spark's format
+and with `zstd`.
 
 `none` is the better setting for a relation that fits in memory uncompressed and is read at close
-to full width. The default is the other way round because a cache that does not fit costs more than
-one that is slower to read, and Spark's own cache format compresses by default too.
-
-Arrow's other IPC codec, LZ4, is deliberately not offered and the config rejects it. It is
-commons-compress's pure-Java implementation, unrelated to the JNI-accelerated lz4-java behind
-`spark.io.compression.codec`, and is orders of magnitude slower to write than `zstd` while also
-producing larger output.
+to full width. The default is `zstd` because a cache that does not fit costs more than one that is
+slower to read, and Spark's own cache format compresses by default too.
 
 Dictionary-encoded columns are decoded before they are stored. A payload with no schema message has
 nowhere to record either that a column is dictionary encoded or the dictionary itself.
 
 ## Configuration
 
-| Config                                                  | Default | Description                                                                                                                                    |
-| ------------------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------- |
-| `spark.comet.exec.inMemoryCache.enabled`                | `true`  | Whether to store and scan Spark's in-memory cache in Comet's format. Read at startup. Defaults to `false` on Spark 3.4.                        |
-| `spark.comet.exec.inMemoryCache.compression.codec`      | `zstd`  | Arrow IPC compression codec for cached data: `zstd` or `none`. Affects newly cached data only — a batch records the codec it was written with. |
-| `spark.comet.exec.inMemoryCache.compression.zstd.level` | `1`     | Compression level when the codec is `zstd`. Ignored otherwise.                                                                                 |
+| Config                                                  | Default | Description                                                                                                                                          |
+| ------------------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `spark.comet.exec.inMemoryCache.enabled`                | `true`  | Whether to store and scan Spark's in-memory cache in Comet's format. Read at startup. Defaults to `false` on Spark 3.4.                              |
+| `spark.comet.exec.inMemoryCache.compression.codec`      | `zstd`  | Arrow IPC compression codec for cached data: `zstd`, `lz4` or `none`. Affects newly cached data only: a batch records the codec it was written with. |
+| `spark.comet.exec.inMemoryCache.compression.zstd.level` | `1`     | Compression level when the codec is `zstd`. Ignored otherwise.                                                                                       |
 
 ## Performance
 
@@ -205,6 +216,12 @@ all of them longs, is about 10% slower under either kind of operator. That cost
 decompression. With the `none` codec, the same read is 2.7x faster than Spark's format with Comet
 operators above the scan, and 1.6x faster with a Spark operator above it.
 
+Measured on an Apple M3 Ultra, Comet's format with `lz4` is faster than Spark's in every shape
+under both kinds of operator. The three-column read is 1.4x faster than Spark's format with either
+kind of operator above the scan, where `zstd` is about 15% slower on the same machine, and the full
+projection is 2.5x faster with Comet operators above the scan and 1.7x faster with a Spark operator
+above it.
+
 Building the cache is measured the same way, from the same source, with its rows produced either by
 Comet operators or by Spark operators. Comet's format is written straight from the Arrow batches of
 Comet operators, while Spark's is always built from rows, so above Comet operators it first converts
@@ -292,6 +309,13 @@ the others. The row reader takes up to 1.6 times as long, and it is the only rea
 wider than `spark.sql.codegen.maxFields`: reading every column of relations of 100, 200 and 1500
 nullable `bigint` columns took 2.2 to 2.5 times as long as from Spark's format. A Spark operator
 above Comet's native cache scan does not pay this; see [Performance](#performance).
+
+Most of what the row reader pays is decompression, so `lz4` narrows these gaps. Measured on an
+Apple M3 Ultra, the row reader takes up to 1.25 times as long as Spark's format with `lz4`, against
+up to 1.7 times with `zstd`, and the fused reader is faster than Spark's format for every read.
+Reading every column of the wide relations takes 1.3 to 2.0 times as long as from Spark's format
+with `lz4`, against 2.1 to 2.6 times with `zstd`. `lz4` takes more memory than Spark's format for
+those relations, though; see [Storage format](#storage-format).
 [#5485](https://github.com/apache/datafusion-comet/issues/5485) tracks these gaps.
 
 Comet's serializer exists because Spark's own Arrow cache format

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -388,7 +388,7 @@ object CometConf extends ShimCometConf {
           "without the fused reader. Relations whose schema Comet's Arrow writer does not " +
           "support are always cached in Spark's default " +
           "format. Each cached batch is stored as one Arrow IPC record batch with per-buffer " +
-          "zstd compression, and a scan copies out only the buffers of the columns it " +
+          "compression, and a scan copies out only the buffers of the columns it " +
           "projected, so the unselected ones are never decompressed. Eligible Spark " +
           "whole-stage codegen consumers read cached vectors directly when vectorized cache " +
           "reading is enabled; other Spark row consumers use a reusable row buffer. Decoding " +
@@ -409,20 +409,20 @@ object CometConf extends ShimCometConf {
       .category(CATEGORY_EXEC)
       .doc(
         "The Arrow IPC compression codec used when Comet's cache serializer writes cached " +
-          "data. Unlike spark.io.compression.codec, this compresses each Arrow buffer " +
-          "separately rather than the batch as a whole, which is what lets a projected scan " +
-          "decompress only the columns it selected. zstd is the default for footprint rather " +
-          "than speed: it stores cached data in a fraction of the memory, but makes " +
-          "materializing slower, and a read slower in proportion to how many columns it " +
-          "selects. Set to none when a relation fits in memory uncompressed and is read at " +
-          "close to full width. " +
+          "data: zstd, lz4 or none. Unlike spark.io.compression.codec, this compresses each " +
+          "Arrow buffer separately rather than the batch as a whole, which is what lets a " +
+          "projected scan decompress only the columns it selected. zstd is the default for " +
+          "footprint rather than speed: it stores cached data in a fraction of the memory, but " +
+          "makes materializing slower, and a read slower in proportion to how many columns it " +
+          "selects. lz4 decompresses several times faster than zstd, which makes wide reads " +
+          "much faster, at around three times zstd's footprint. It uses lz4-java, the " +
+          "JNI-accelerated library behind spark.io.compression.codec, rather than Arrow's own " +
+          "pure-Java lz4. Set to none when a relation fits in memory uncompressed and is read " +
+          "at close to full width. " +
           "Only affects newly cached data; the codec a batch was written with is recorded in " +
-          "the batch itself and is what the read path uses. Arrow's lz4 is deliberately not " +
-          "offered: it is a pure-Java implementation, unrelated to the JNI-accelerated lz4 " +
-          "behind spark.io.compression.codec, and is orders of magnitude slower to write than " +
-          "zstd while also producing larger output.")
+          "the batch itself and is what the read path uses.")
       .stringConf
-      .checkValues(Set("none", "zstd"))
+      .checkValues(Set("none", "zstd", "lz4"))
       .createWithDefault("zstd")
 
   val COMET_EXEC_IN_MEMORY_CACHE_COMPRESSION_ZSTD_LEVEL: ConfigEntry[Int] =

spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchIpc.scala

@@ -73,24 +73,28 @@ private[comet] object CachedBatchIpc {
     // Constructed directly rather than through CompressionCodec.Factory, which ignores the level
     // and always builds a codec at zstd's default.
     case "zstd" => new ZstdCompressionCodec(zstdLevel)
-    // Arrow's other codec, LZ4_FRAME, is not offered. It is commons-compress's pure-Java LZ4 --
-    // no relation to the JNI-accelerated lz4-java behind spark.io.compression.codec -- and
-    // measures three orders of magnitude slower to write than zstd while also producing larger
-    // output, so nothing prefers it. Reads still accept it, since the factory the read path uses
-    // handles whatever codec a batch records.
+    // Arrow's LZ4_FRAME format, written with lz4-java rather than Arrow's own pure-Java codec;
+    // see Lz4FrameCompressionCodec.
+    case "lz4" => Lz4FrameCompressionCodec
     case other =>
       throw new SparkException(
         s"Unsupported Arrow compression codec for Comet's cache: $other. " +
-          "Supported values: none, zstd")
+          "Supported values: none, zstd, lz4")
   }
 
   // Decompressors are stateless and shared. Resolving one per cached batch would allocate a codec
-  // per batch on every scan, and the enum lookup walks the CodecType values each time.
+  // per batch on every scan, and the enum lookup walks the CodecType values each time. LZ4_FRAME
+  // is read with the codec that writes it rather than with Arrow's pure-Java one, which inflates
+  // through heap copies and streams.
   private val readCodecs: Map[CompressionUtil.CodecType, CompressionCodec] =
     CompressionUtil.CodecType
       .values()
       .filter(_ != CompressionUtil.CodecType.NO_COMPRESSION)
-      .map(t => t -> CommonsCompressionFactory.INSTANCE.createCodec(t))
+      .map {
+        case CompressionUtil.CodecType.LZ4_FRAME =>
+          CompressionUtil.CodecType.LZ4_FRAME -> Lz4FrameCompressionCodec
+        case t => t -> CommonsCompressionFactory.INSTANCE.createCodec(t)
+      }
       .toMap
 
   /**