andygrove opened a new pull request, #6859: URL: https://github.com/apache/datafusion-comet/pull/6859
## Which issue does this PR close? Part of #5485. ## Rationale for this change #5485 tracks Spark operators reading Comet's cache format more slowly than Spark's own format. Measuring each codec shows that what remains of that gap is almost all decompression: with `none`, every read of the benchmark's six-column relation matches or beats Spark's format. The only compressing codec Comet offers is `zstd`, because Arrow's own LZ4 codec is commons-compress's pure-Java implementation, which is far slower than `zstd`. lz4-java, the JNI library behind `spark.io.compression.codec`, is on the classpath of every Spark version Comet supports (`org.lz4` 1.8.0 on 3.4 to 4.1, the `at.yawk.lz4` fork 1.11.0 on 4.2) and decompresses several times faster than `zstd`. From `CometInMemoryCacheBenchmark` on an Apple M3 Ultra (JDK 17, Spark 4.1, release build), over 5M rows of three long and three string columns. Each cell is the read's time relative to Spark's own cache format, so below 1 is faster: | Read | `zstd`, 1 / 3 / 6 columns | `lz4`, 1 / 3 / 6 columns | | --- | --- | --- | | Native scan, Comet operators above (AQE) | 0.69 / 1.17 / 0.91 | 0.63 / 0.70 / 0.39 | | Native scan, Spark operator above (AQE) | 0.69 / 1.14 / 1.10 | 0.62 / 0.69 / 0.59 | | Spark's scan, fused reader | 0.67 / 1.19 / 0.95 | 0.58 / 0.73 / 0.51 | | Spark's scan, row reader | 1.15 / 1.71 / 1.48 | 1.08 / 1.25 / 1.02 | Reading every column of relations of 100 / 200 / 1500 nullable `bigint` columns through the row reader takes 2.08 / 2.10 / 2.60 times as long as from Spark's format with `zstd`, and 1.26 / 1.37 / 1.98 times with `lz4`. | Codec | Materialize | Footprint | Read 6 of 6 | | --- | ---: | ---: | ---: | | `zstd` | 1232 ms | 51 MiB | 248 ms | | `lz4` | 1115 ms | 145 MiB | 110 ms | | `none` | 857 ms | 315 MiB | 54 ms | Spark's own format holds the same relation in 217 MiB. These are from one run; two earlier exploratory runs of the same comparison agreed within a few percent. It is opt-in rather than the new default because of footprint. `lz4` takes about three times `zstd`'s memory, and on sequential longs, which Spark's format delta-encodes, more than Spark's own format: the wide relations above take about 74 MiB with `lz4`, against 26 to 27 MiB in Spark's format and with `zstd`. `lz4` combined with delta-encoded longs (#5869) is the candidate to revisit the default with. At 1500 columns even `none` took 1.25 times as long as Spark's format in an exploratory run, so the rest of that gap is per-column decoding rather than the codec, and stays with #5485. ## What changes are included in this PR? - `Lz4FrameCompressionCodec`: Arrow's `LZ4_FRAME` codec on lz4-java. Each buffer is one standard LZ4 frame (independent blocks of up to 4 MiB, no checksums) behind Arrow's uncompressed-length prefix, compressed and decompressed with lz4-java's block API directly on the off-heap buffers. It asks for lz4-java's JNI implementation explicitly, because `LZ4Factory.fastestInstance()` silently settles for the Java one when lz4-java is not loaded by the system class loader (as under `mvn exec:java`), and falls back to Java, with a warning, only when the native library cannot load. Comet turns off Arrow's bounds checks, so the reader checks every length a frame records before reading through it. - `spark.comet.exec.inMemoryCache.compression.codec` accepts `lz4`, and the read path decodes `LZ4_FRAME` batches with this codec rather than Arrow's commons-compress one. - User guide: the codec table gains `lz4`, with when to choose it and its footprint caveat, and Performance and Limitations say what it changes for reads against Spark's format. - `CometInMemoryCacheBenchmark` measures `lz4` in the codec, Spark-operator, wide-relation and AQE cases, and prints the footprint of each cached copy. ## How are these changes tested? New and extended tests in `CometInMemoryCacheSuite`: - The codec round-trip test also covers `lz4`: full and projected reads through the native scan and through Spark's reader, a row count, and batch pruning. - What the codec writes decodes with lz4-java's own `LZ4FrameInputStream`, for a one-block buffer and for a three-block buffer whose first block is incompressible and stored verbatim. - Corrupt frames (not a frame, cut off inside the header or before the end mark, a block that runs past the end, decoding to less or more than the recorded length) fail with a `SparkException` naming LZ4 and release everything they allocated. - The test that a column failing to decode releases its vectors also runs under `lz4`. The suite passes on Spark 4.1 and on Spark 4.2, which ships the `at.yawk.lz4` fork, and the Spark 3.5 strict-warnings build and scalafix check pass. -- 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]
