This is an automated email from the ASF dual-hosted git repository.
jbonofre pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-java.git
The following commit(s) were added to refs/heads/main by this push:
new 2ab1b023d GH-192: Make LZ4 frame compression and decompression at
least 10x faster with lz4-java (#1420)
2ab1b023d is described below
commit 2ab1b023dc97c60bcb34ef10f7268b41d7a2de75
Author: Andy Grove <[email protected]>
AuthorDate: Mon Nov 2 11:45:16 2026 -0700
GH-192: Make LZ4 frame compression and decompression at least 10x faster
with lz4-java (#1420)
## What's Changed
`Lz4CompressionCodec` uses commons-compress's pure-Java LZ4, which is
slow enough that compressing an 8 MiB buffer of sequential `int64`
values takes several minutes (#192). This changes the codec to use
[lz4-java](https://github.com/yawkat/lz4-java), the JNI bindings to
liblz4 that Spark, Kafka and others use, while keeping every frame it
could read before readable.
**Compression.** Each buffer is written as one LZ4 frame of independent
blocks of up to 4 MiB, without checksums, compressed with lz4-java
directly from the buffer's memory. Every LZ4 frame decoder reads frames
with independent blocks: liblz4 (Arrow C++, pyarrow), `lz4_flex`
(arrow-rs), lz4-java, and commons-compress.
**Decompression.** A frame with independent blocks is decompressed with
lz4-java directly into the output buffer. That covers frames from this
codec, from arrow-rs, and from the previous version of this codec. The
new decoder accepts any block size, block and content checksums (which
it verifies), and a content size. It checks the frame's block lengths,
and that they can hold the recorded decompressed length, before
allocating for it.
A frame whose blocks are linked, or that uses a dictionary, still goes
through commons-compress. lz4-java cannot decode a block that refers to
earlier blocks, and Arrow C++ writes linked blocks: liblz4's default
`LZ4F_preferences_t` is `LZ4F_blockLinked`, so pyarrow 26 writes every
buffer larger than 64 KiB as linked 64 KiB blocks. The LZ4 files in
arrow-testing are small enough that liblz4 writes them as single,
independent blocks, so the integration tests would not catch this.
`TestLz4CompressionCodec` covers linked frames written by
commons-compress instead.
**Dependency.** This adds `at.yawk.lz4:lz4-java` 1.12.1, the maintained
fork of `org.lz4:lz4-java`. The original artifact's last code release,
1.8.0, has several published advisories (for example CVE-2025-12183 and
CVE-2025-66566), and its 1.8.1 is a relocation POM that points to the
fork. Spark moved to the fork in 4.2. Both artifacts use the
`net.jpountz` packages and the `org.lz4.java` module name, and the codec
uses only APIs present in both, so an application that pins
`org.lz4:lz4-java` instead still works. The codec asks for lz4-java's
JNI implementation and falls back to its Java one only when the native
library cannot load, because `LZ4Factory.fastestInstance()` settles for
Java whenever lz4-java is not loaded by the system class loader.
### Performance
`CompressionCodecBenchmarks` (new, in `performance`) on an Apple M3
Ultra, JDK 17, 8 MiB buffers:
| Benchmark | LZ4_FRAME | ZSTD |
| --- | ---: | ---: |
| compress, sequential `int64` | 8.5 ms | 12.5 ms |
| compress, random decimal text | 9.9 ms | 44.6 ms |
| decompress, sequential `int64` | 2.1 ms | 4.8 ms |
| decompress, random decimal text | 1.9 ms | 8.5 ms |
The previous codec is too slow for the same JMH run. From a simple
timing harness on the same machine:
| Input | Previous compress | New compress | Previous decompress | New
decompress |
| --- | ---: | ---: | ---: | ---: |
| 8 MiB sequential `int64` | 278 s | 7.8 ms | 93.5 ms | 2.2 ms |
| 8 MiB of `i % 1000` as text | 17.9 ms | 0.3 ms | 3.3 ms | 0.3 ms |
| 8 MiB random bytes | 243 ms | 0.5 ms | (stored) | (stored) |
| 1 MiB random decimal text | 38.5 s | 1.4 ms | 33.3 ms | 0.3 ms |
### Compression ratio
On the inputs above, the ratio is unchanged except for random decimal
text, where commons-compress's more thorough match search reached 1.61
and lz4-java's fast compressor reaches 1.10. The new output matches what
Arrow C++ writes by default: pyarrow's LZ4 codec also reaches 1.10 on
the same kind of data, and 1.63 at compression level 9.
`CommonsCompressionFactory.createCodec(LZ4_FRAME, level)` ignores the
level today. Mapping it to lz4-java's high-compression mode would give
users who want the better ratio a way to ask for it; I left that out to
keep this change focused, and can add it here or in a follow-up.
### Testing
- The existing `TestCompressionCodec` and
`TestArrowReaderWriterWithCompression` pass unchanged, including the
decompressed-length check from #1207.
- `TestLz4CompressionCodec` (new):
- round-trips buffers across block boundaries, including one whose first
block is incompressible and stored, and checks that commons-compress and
lz4-java's `LZ4FrameInputStream` both read the frames this codec writes;
- decompresses frames written by commons-compress for every block size,
with linked and independent blocks, with and without checksums, and a
frame written by lz4-java with a content size;
- rejects corrupt frames (descriptor, block and content checksums, a
missing end mark, blocks that run past the frame or exceed the maximum
block size, undecodable blocks, a wrong content size, and recorded
lengths that are too long or too short), and releases everything it
allocated.
- I also decoded pyarrow 26 output (buffers over 64 KiB, linked blocks)
by hand to confirm that lz4-java rejects linked frames and
commons-compress reads them, which is why the fallback stays.
### AI usage
The code, tests and benchmark in this PR were written with an AI coding
assistant. The frame behavior described above (pyarrow's linked blocks,
lz4-java's rejection of them) and the benchmark numbers come from
running them.
Closes #192.
---
compression/pom.xml | 5 +
compression/src/main/java/module-info.java | 1 +
.../arrow/compression/Lz4CompressionCodec.java | 323 +++++++++++++++++++--
.../arrow/compression/TestLz4CompressionCodec.java | 268 +++++++++++++++++
performance/pom.xml | 4 +
.../compression/CompressionCodecBenchmarks.java | 127 ++++++++
tools/src/shade/LICENSE.txt | 79 +++++
7 files changed, 775 insertions(+), 32 deletions(-)
diff --git a/compression/pom.xml b/compression/pom.xml
index f03fde247..fe3ed5e05 100644
--- a/compression/pom.xml
+++ b/compression/pom.xml
@@ -57,5 +57,10 @@ under the License.
<artifactId>zstd-jni</artifactId>
<version>1.5.7-25</version>
</dependency>
+ <dependency>
+ <groupId>at.yawk.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ <version>1.12.1</version>
+ </dependency>
</dependencies>
</project>
diff --git a/compression/src/main/java/module-info.java
b/compression/src/main/java/module-info.java
index 113a1dba9..e43835500 100644
--- a/compression/src/main/java/module-info.java
+++ b/compression/src/main/java/module-info.java
@@ -24,6 +24,7 @@ module org.apache.arrow.compression {
requires org.apache.arrow.memory.core;
requires org.apache.arrow.vector;
requires org.apache.commons.compress;
+ requires org.lz4.java;
// Also defined under META-INF/services to support non-modular applications
provides CompressionCodec.Factory with
diff --git
a/compression/src/main/java/org/apache/arrow/compression/Lz4CompressionCodec.java
b/compression/src/main/java/org/apache/arrow/compression/Lz4CompressionCodec.java
index f268e815f..18cad8dbb 100644
---
a/compression/src/main/java/org/apache/arrow/compression/Lz4CompressionCodec.java
+++
b/compression/src/main/java/org/apache/arrow/compression/Lz4CompressionCodec.java
@@ -20,47 +20,259 @@ import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.io.OutputStream;
+import net.jpountz.lz4.LZ4Compressor;
+import net.jpountz.lz4.LZ4Exception;
+import net.jpountz.lz4.LZ4Factory;
+import net.jpountz.lz4.LZ4SafeDecompressor;
+import net.jpountz.xxhash.XXHash32;
+import net.jpountz.xxhash.XXHashFactory;
import org.apache.arrow.memory.ArrowBuf;
import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.util.MemoryUtil;
import org.apache.arrow.util.Preconditions;
import org.apache.arrow.vector.compression.AbstractCompressionCodec;
import org.apache.arrow.vector.compression.CompressionUtil;
import
org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorInputStream;
-import
org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorOutputStream;
import org.apache.commons.compress.utils.IOUtils;
-/** Compression codec for the LZ4 algorithm. */
+/**
+ * Compression codec for the LZ4 frame format.
+ *
+ * <p>Each buffer is written as one frame of independent blocks of up to 4
MiB, without checksums,
+ * which every LZ4 frame decoder reads. Compression, and the decompression of
frames whose blocks
+ * are independent, use lz4-java on the buffers' memory directly. A frame
whose blocks are linked,
+ * as Arrow C++ writes for a buffer larger than 64 KiB, or that uses a
dictionary, is decompressed
+ * with commons-compress, since lz4-java cannot decode a block that refers to
earlier ones.
+ */
public class Lz4CompressionCodec extends AbstractCompressionCodec {
+ private static final long PREFIX =
CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH;
+
+ private static final int MAGIC = 0x184D2204;
+
+ // The frame descriptor this codec writes: FLG (version 01, independent
blocks), BD (4 MiB
+ // maximum block size) and HC, the second byte of the xxHash32 of FLG and BD
with a zero seed.
+ private static final byte WRITE_FLG = 0x60;
+ private static final byte WRITE_BD = 0x70;
+ private static final byte WRITE_HC = 0x73;
+ private static final int WRITE_BLOCK_SIZE = 4 * 1024 * 1024;
+
+ private static final int FLG_VERSION_MASK = 0xC0;
+ private static final int FLG_VERSION = 0x40;
+ private static final int FLG_INDEPENDENT_BLOCKS = 0x20;
+ private static final int FLG_BLOCK_CHECKSUM = 0x10;
+ private static final int FLG_CONTENT_SIZE = 0x08;
+ private static final int FLG_CONTENT_CHECKSUM = 0x04;
+ private static final int FLG_RESERVED = 0x02;
+ private static final int FLG_DICTIONARY_ID = 0x01;
+ private static final int BD_RESERVED = 0x8F;
+
+ // A block length's high bit marks a block stored uncompressed. A zero
length ends the frame.
+ private static final int UNCOMPRESSED_BLOCK = 0x80000000;
+
+ // An LZ4 block cannot decompress to more than 255 times its compressed size.
+ private static final long MAX_COMPRESSION_RATIO = 255;
+
@Override
protected ArrowBuf doCompress(BufferAllocator allocator, ArrowBuf
uncompressedBuffer) {
- Preconditions.checkArgument(
- uncompressedBuffer.writerIndex() <= Integer.MAX_VALUE,
- "The uncompressed buffer size exceeds the integer limit %s.",
- Integer.MAX_VALUE);
+ LZ4Compressor compressor = Lz4.COMPRESSOR;
+ long length = uncompressedBuffer.writerIndex();
+ long fullBlocks = length / WRITE_BLOCK_SIZE;
+ int lastBlock = (int) (length % WRITE_BLOCK_SIZE);
+ long bound =
+ PREFIX
+ + 7
+ + fullBlocks * (Integer.BYTES +
compressor.maxCompressedLength(WRITE_BLOCK_SIZE))
+ + (lastBlock > 0 ? Integer.BYTES +
compressor.maxCompressedLength(lastBlock) : 0)
+ + Integer.BYTES;
+ ArrowBuf frame = allocator.buffer(bound);
+ try {
+ setIntLE(frame, PREFIX, MAGIC);
+ frame.setByte(PREFIX + 4, WRITE_FLG);
+ frame.setByte(PREFIX + 5, WRITE_BD);
+ frame.setByte(PREFIX + 6, WRITE_HC);
+ long at = PREFIX + 7;
+ for (long from = 0; from < length; ) {
+ int n = (int) Math.min(WRITE_BLOCK_SIZE, length - from);
+ int room = compressor.maxCompressedLength(n);
+ int packed =
+ compressor.compress(
+ uncompressedBuffer.nioBuffer(from, n),
+ 0,
+ n,
+ frame.nioBuffer(at + Integer.BYTES, room),
+ 0,
+ room);
+ if (packed < n) {
+ setIntLE(frame, at, packed);
+ at += Integer.BYTES + packed;
+ } else {
+ setIntLE(frame, at, n | UNCOMPRESSED_BLOCK);
+ frame.setBytes(at + Integer.BYTES, uncompressedBuffer, from, n);
+ at += Integer.BYTES + n;
+ }
+ from += n;
+ }
+ setIntLE(frame, at, 0);
+ frame.writerIndex(at + Integer.BYTES);
+ return frame;
+ } catch (RuntimeException | Error e) {
+ frame.close();
+ throw e;
+ }
+ }
- byte[] inBytes = new byte[(int) uncompressedBuffer.writerIndex()];
- uncompressedBuffer.getBytes(/* index= */ 0, inBytes);
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- try (InputStream in = new ByteArrayInputStream(inBytes);
- OutputStream out = new FramedLZ4CompressorOutputStream(baos)) {
- IOUtils.copy(in, out);
- } catch (IOException e) {
- throw new RuntimeException(e);
+ @Override
+ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf
compressedBuffer) {
+ long length = readUncompressedLength(compressedBuffer);
+ long end = compressedBuffer.writerIndex();
+ if (end - PREFIX < 7 || getIntLE(compressedBuffer, PREFIX) != MAGIC) {
+ return decompressWithCommonsCompress(allocator, compressedBuffer);
+ }
+ int flg = compressedBuffer.getByte(PREFIX + 4) & 0xFF;
+ int bd = compressedBuffer.getByte(PREFIX + 5) & 0xFF;
+ int blockSizeId = (bd >>> 4) & 0x07;
+ boolean contentChecksum = (flg & FLG_CONTENT_CHECKSUM) != 0;
+ if ((flg & FLG_VERSION_MASK) != FLG_VERSION
+ || (flg & (FLG_RESERVED | FLG_DICTIONARY_ID)) != 0
+ || (flg & FLG_INDEPENDENT_BLOCKS) == 0
+ || (bd & BD_RESERVED) != 0
+ || blockSizeId < 4
+ || (contentChecksum && length > Integer.MAX_VALUE)) {
+ return decompressWithCommonsCompress(allocator, compressedBuffer);
}
+ boolean blockChecksum = (flg & FLG_BLOCK_CHECKSUM) != 0;
+ int maxBlockSize = 1 << (8 + 2 * blockSizeId);
- byte[] outBytes = baos.toByteArray();
+ long at = PREFIX + 6;
+ if ((flg & FLG_CONTENT_SIZE) != 0) {
+ if (end - at < Long.BYTES + 1) {
+ throw corrupt("the frame descriptor is cut off");
+ }
+ long contentSize = getLongLE(compressedBuffer, at);
+ if (contentSize != length) {
+ throw corrupt(
+ "the frame records a content size of "
+ + contentSize
+ + " bytes, but the buffer's decompressed length is "
+ + length);
+ }
+ at += Long.BYTES;
+ }
+ int descriptorLength = (int) (at - (PREFIX + 4));
+ int headerChecksum =
+ (Lz4.XXHASH.hash(
+ compressedBuffer.nioBuffer(PREFIX + 4, descriptorLength),
+ 0,
+ descriptorLength,
+ 0)
+ >>> 8)
+ & 0xFF;
+ if ((compressedBuffer.getByte(at) & 0xFF) != headerChecksum) {
+ throw corrupt("the frame descriptor's checksum does not match");
+ }
+ at += 1;
- ArrowBuf compressedBuffer =
- allocator.buffer(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH +
outBytes.length);
- compressedBuffer.setBytes(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH,
outBytes);
- compressedBuffer.writerIndex(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH +
outBytes.length);
- return compressedBuffer;
+ // Check the blocks' lengths, and that they can hold the decompressed
length, before
+ // allocating for it.
+ long blocksStart = at;
+ long maxDecompressedLength = 0;
+ while (true) {
+ if (end - at < Integer.BYTES) {
+ throw corrupt("the frame ends before its end mark");
+ }
+ int word = getIntLE(compressedBuffer, at);
+ at += Integer.BYTES;
+ if (word == 0) {
+ break;
+ }
+ int n = word & ~UNCOMPRESSED_BLOCK;
+ if (n > maxBlockSize) {
+ throw corrupt("a block of " + n + " bytes is larger than the frame's
maximum block size");
+ }
+ if (n > end - at) {
+ throw corrupt("a block of " + n + " bytes runs past the end of the
frame");
+ }
+ maxDecompressedLength +=
+ (word & UNCOMPRESSED_BLOCK) != 0 ? n : Math.min(maxBlockSize, n *
MAX_COMPRESSION_RATIO);
+ at += n;
+ if (blockChecksum) {
+ if (end - at < Integer.BYTES) {
+ throw corrupt("the frame ends before a block's checksum");
+ }
+ at += Integer.BYTES;
+ }
+ }
+ if (contentChecksum && end - at < Integer.BYTES) {
+ throw corrupt("the frame ends before its content checksum");
+ }
+ if (length < 0 || length > maxDecompressedLength) {
+ throw corrupt(
+ "the frame decompresses to at most "
+ + maxDecompressedLength
+ + " bytes, not its recorded decompressed length of "
+ + length);
+ }
+
+ ArrowBuf decompressedBuffer = allocator.buffer(length);
+ try {
+ LZ4SafeDecompressor decompressor = Lz4.DECOMPRESSOR;
+ at = blocksStart;
+ long to = 0;
+ while (true) {
+ int word = getIntLE(compressedBuffer, at);
+ at += Integer.BYTES;
+ if (word == 0) {
+ break;
+ }
+ int n = word & ~UNCOMPRESSED_BLOCK;
+ if (blockChecksum
+ && Lz4.XXHASH.hash(compressedBuffer.nioBuffer(at, n), 0, n, 0)
+ != getIntLE(compressedBuffer, at + n)) {
+ throw corrupt("a block's checksum does not match");
+ }
+ if ((word & UNCOMPRESSED_BLOCK) != 0) {
+ if (n > length - to) {
+ throw corrupt(
+ "the frame holds more than its recorded decompressed length of
" + length);
+ }
+ decompressedBuffer.setBytes(to, compressedBuffer, at, n);
+ to += n;
+ } else {
+ int room = (int) Math.min(maxBlockSize, length - to);
+ try {
+ to +=
+ decompressor.decompress(
+ compressedBuffer.nioBuffer(at, n),
+ 0,
+ n,
+ decompressedBuffer.nioBuffer(to, room),
+ 0,
+ room);
+ } catch (LZ4Exception e) {
+ throw corrupt("a block does not decompress within " + room + "
bytes", e);
+ }
+ }
+ at += n + (blockChecksum ? Integer.BYTES : 0);
+ }
+ if (to != length) {
+ throw wrongLength(length, to);
+ }
+ if (contentChecksum
+ && Lz4.XXHASH.hash(decompressedBuffer.nioBuffer(0, (int) length), 0,
(int) length, 0)
+ != getIntLE(compressedBuffer, at)) {
+ throw corrupt("the frame's content checksum does not match");
+ }
+ decompressedBuffer.writerIndex(length);
+ return decompressedBuffer;
+ } catch (RuntimeException | Error e) {
+ decompressedBuffer.close();
+ throw e;
+ }
}
- @Override
- protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf
compressedBuffer) {
+ private ArrowBuf decompressWithCommonsCompress(
+ BufferAllocator allocator, ArrowBuf compressedBuffer) {
Preconditions.checkArgument(
compressedBuffer.writerIndex() <= Integer.MAX_VALUE,
"The compressed buffer size exceeds the integer limit %s",
@@ -68,10 +280,8 @@ public class Lz4CompressionCodec extends
AbstractCompressionCodec {
long decompressedLength = readUncompressedLength(compressedBuffer);
- byte[] inBytes =
- new byte
- [(int) (compressedBuffer.writerIndex() -
CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH)];
- compressedBuffer.getBytes(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH,
inBytes);
+ byte[] inBytes = new byte[(int) (compressedBuffer.writerIndex() - PREFIX)];
+ compressedBuffer.getBytes(PREFIX, inBytes);
ByteArrayOutputStream out = new ByteArrayOutputStream((int)
decompressedLength);
try (InputStream in = new FramedLZ4CompressorInputStream(new
ByteArrayInputStream(inBytes))) {
IOUtils.copy(in, out);
@@ -81,11 +291,7 @@ public class Lz4CompressionCodec extends
AbstractCompressionCodec {
byte[] outBytes = out.toByteArray();
if (outBytes.length != decompressedLength) {
- throw new RuntimeException(
- "Expected != actual decompressed length: "
- + decompressedLength
- + " != "
- + outBytes.length);
+ throw wrongLength(decompressedLength, outBytes.length);
}
ArrowBuf decompressedBuffer = allocator.buffer(outBytes.length);
decompressedBuffer.setBytes(/* index= */ 0, outBytes);
@@ -97,4 +303,57 @@ public class Lz4CompressionCodec extends
AbstractCompressionCodec {
public CompressionUtil.CodecType getCodecType() {
return CompressionUtil.CodecType.LZ4_FRAME;
}
+
+ private static RuntimeException corrupt(String reason) {
+ return new IllegalArgumentException("Corrupt LZ4 frame: " + reason);
+ }
+
+ private static RuntimeException corrupt(String reason, Throwable cause) {
+ return new IllegalArgumentException("Corrupt LZ4 frame: " + reason, cause);
+ }
+
+ private static RuntimeException wrongLength(long expected, long actual) {
+ return new RuntimeException(
+ "Expected != actual decompressed length: " + expected + " != " +
actual);
+ }
+
+ private static int getIntLE(ArrowBuf buf, long index) {
+ int value = buf.getInt(index);
+ return MemoryUtil.LITTLE_ENDIAN ? value : Integer.reverseBytes(value);
+ }
+
+ private static long getLongLE(ArrowBuf buf, long index) {
+ long value = buf.getLong(index);
+ return MemoryUtil.LITTLE_ENDIAN ? value : Long.reverseBytes(value);
+ }
+
+ private static void setIntLE(ArrowBuf buf, long index, int value) {
+ buf.setInt(index, MemoryUtil.LITTLE_ENDIAN ? value :
Integer.reverseBytes(value));
+ }
+
+ /** lz4-java's implementations, loaded on first use. */
+ private static final class Lz4 {
+ // fastestInstance() picks the JNI implementations only when lz4-java was
loaded by the system
+ // class loader, and otherwise settles for the Java ones, so ask for the
JNI ones directly.
+ static final LZ4Factory FACTORY = lz4Factory();
+ static final LZ4Compressor COMPRESSOR = FACTORY.fastCompressor();
+ static final LZ4SafeDecompressor DECOMPRESSOR = FACTORY.safeDecompressor();
+ static final XXHash32 XXHASH = xxHashFactory().hash32();
+
+ private static LZ4Factory lz4Factory() {
+ try {
+ return LZ4Factory.nativeInstance();
+ } catch (LinkageError | RuntimeException e) {
+ return LZ4Factory.fastestJavaInstance();
+ }
+ }
+
+ private static XXHashFactory xxHashFactory() {
+ try {
+ return XXHashFactory.nativeInstance();
+ } catch (LinkageError | RuntimeException e) {
+ return XXHashFactory.fastestJavaInstance();
+ }
+ }
+ }
}
diff --git
a/compression/src/test/java/org/apache/arrow/compression/TestLz4CompressionCodec.java
b/compression/src/test/java/org/apache/arrow/compression/TestLz4CompressionCodec.java
new file mode 100644
index 000000000..37068e696
--- /dev/null
+++
b/compression/src/test/java/org/apache/arrow/compression/TestLz4CompressionCodec.java
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.arrow.compression;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+import net.jpountz.lz4.LZ4FrameInputStream;
+import net.jpountz.lz4.LZ4FrameOutputStream;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.compression.CompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+import
org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorInputStream;
+import
org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorOutputStream;
+import
org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorOutputStream.BlockSize;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/** Test cases for {@link Lz4CompressionCodec}'s LZ4 frames. */
+class TestLz4CompressionCodec {
+ private static final int MIB = 1024 * 1024;
+
+ // Long enough for several 64 KiB blocks.
+ private static final int OTHER_WRITER_LENGTH = 200 * 1024;
+
+ private final CompressionCodec codec = new Lz4CompressionCodec();
+ private BufferAllocator allocator;
+
+ @BeforeEach
+ void init() {
+ allocator = new RootAllocator(Integer.MAX_VALUE);
+ }
+
+ @AfterEach
+ void terminate() {
+ allocator.close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 1000, 64 * 1024 + 1, 4 * MIB, 4 * MIB + 1, 9 * MIB})
+ void testRoundTrip(int length) throws IOException {
+ checkRoundTrip(compressible(length));
+ }
+
+ @Test
+ void testRoundTripWithStoredBlock() throws IOException {
+ // The first 4 MiB block is incompressible, so it is stored, while the
rest compresses.
+ byte[] data = new byte[9 * MIB];
+ byte[] noise = new byte[4 * MIB];
+ new Random(42).nextBytes(noise);
+ System.arraycopy(noise, 0, data, 0, noise.length);
+ checkRoundTrip(data);
+ }
+
+ /** Round-trips {@code data}, and checks that other decoders read the frame
it compresses to. */
+ private void checkRoundTrip(byte[] data) throws IOException {
+ ArrowBuf compressed = codec.compress(allocator, toArrowBuf(data));
+ // A buffer that does not compress is sent as it is, without a frame.
+ if (compressed.getLong(0) != CompressionUtil.NO_COMPRESSION_LENGTH) {
+ byte[] frame = new byte[(int) (compressed.writerIndex() - Long.BYTES)];
+ compressed.getBytes(Long.BYTES, frame);
+ try (InputStream in = new FramedLZ4CompressorInputStream(new
ByteArrayInputStream(frame))) {
+ assertArrayEquals(data, in.readAllBytes());
+ }
+ try (InputStream in = new LZ4FrameInputStream(new
ByteArrayInputStream(frame))) {
+ assertArrayEquals(data, in.readAllBytes());
+ }
+ }
+ assertArrayEquals(data, decompress(compressed));
+ }
+
+ static Stream<Arguments> otherWriters() {
+ List<Arguments> writers = new ArrayList<>();
+ for (BlockSize blockSize : BlockSize.values()) {
+ for (boolean linked : new boolean[] {false, true}) {
+ for (boolean checksums : new boolean[] {false, true}) {
+ FramedLZ4CompressorOutputStream.Parameters parameters =
+ new FramedLZ4CompressorOutputStream.Parameters(
+ blockSize, checksums, checksums, linked);
+ writers.add(
+ Arguments.of(
+ "commons-compress " + parameters,
+ (Writer) out -> new FramedLZ4CompressorOutputStream(out,
parameters)));
+ }
+ }
+ }
+ writers.add(
+ Arguments.of(
+ "lz4-java with content size and block checksums",
+ (Writer)
+ out ->
+ new LZ4FrameOutputStream(
+ out,
+ LZ4FrameOutputStream.BLOCKSIZE.SIZE_64KB,
+ OTHER_WRITER_LENGTH,
+ LZ4FrameOutputStream.FLG.Bits.BLOCK_INDEPENDENCE,
+ LZ4FrameOutputStream.FLG.Bits.BLOCK_CHECKSUM,
+ LZ4FrameOutputStream.FLG.Bits.CONTENT_SIZE)));
+ return writers.stream();
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("otherWriters")
+ void testDecompressOtherWriters(String name, Writer writer) throws
IOException {
+ byte[] data = repetitive(OTHER_WRITER_LENGTH);
+ assertArrayEquals(data, decompress(frameOf(data, writer)));
+ }
+
+ static Stream<Arguments> corruptions() {
+ // Offsets are into the buffer: the 8-byte length prefix, then the frame,
whose descriptor
+ // is FLG, BD and HC after the 4-byte magic number. The first block's
length follows at 15.
+ return Stream.of(
+ corruption("a wrong descriptor checksum", buf -> buf.setByte(14,
buf.getByte(14) ^ 1)),
+ corruption("no end mark", buf -> buf.writerIndex(buf.writerIndex() -
Integer.BYTES)),
+ corruption("a block that runs past the frame", buf -> buf.setInt(15, 1
<< 20)),
+ corruption("a block larger than the maximum", buf -> buf.setInt(15, 5
* MIB)),
+ corruption("compressed bytes that do not decode", buf ->
buf.setLong(19, -1L)),
+ corruption("a recorded length that the frame cannot hold", buf ->
buf.setLong(0, MIB)),
+ corruption("a recorded length that is too short", buf ->
buf.setLong(0, 1000)));
+ }
+
+ private static Arguments corruption(String name, Consumer<ArrowBuf> corrupt)
{
+ return Arguments.of(name, corrupt);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("corruptions")
+ void testDecompressCorruptFrame(String name, Consumer<ArrowBuf> corrupt) {
+ ArrowBuf compressed = codec.compress(allocator,
toArrowBuf(compressible(4096)));
+ corrupt.accept(compressed);
+ assertThrows(RuntimeException.class, () -> codec.decompress(allocator,
compressed));
+ compressed.close();
+ }
+
+ @Test
+ void testDecompressWrongChecksums() throws IOException {
+ byte[] data = compressible(4096);
+ Writer writer =
+ out ->
+ new FramedLZ4CompressorOutputStream(
+ out,
+ new FramedLZ4CompressorOutputStream.Parameters(BlockSize.K64,
true, true, false));
+ // The frame ends with a block's checksum, the end mark and the content
checksum.
+ ArrowBuf block = frameOf(data, writer);
+ block.setByte(block.writerIndex() - 12, block.getByte(block.writerIndex()
- 12) ^ 1);
+ RuntimeException e =
+ assertThrows(RuntimeException.class, () -> codec.decompress(allocator,
block));
+ assertTrue(e.getMessage().contains("block's checksum"), e.getMessage());
+ block.close();
+
+ ArrowBuf content = frameOf(data, writer);
+ content.setByte(content.writerIndex() - 1,
content.getByte(content.writerIndex() - 1) ^ 1);
+ e = assertThrows(RuntimeException.class, () -> codec.decompress(allocator,
content));
+ assertTrue(e.getMessage().contains("content checksum"), e.getMessage());
+ content.close();
+ }
+
+ @Test
+ void testDecompressWrongContentSize() throws IOException {
+ byte[] data = compressible(4096);
+ ArrowBuf compressed =
+ frameOf(
+ data,
+ out ->
+ new LZ4FrameOutputStream(
+ out,
+ LZ4FrameOutputStream.BLOCKSIZE.SIZE_64KB,
+ data.length,
+ LZ4FrameOutputStream.FLG.Bits.BLOCK_INDEPENDENCE,
+ LZ4FrameOutputStream.FLG.Bits.CONTENT_SIZE));
+ compressed.setLong(0, data.length - 1);
+ RuntimeException e =
+ assertThrows(RuntimeException.class, () -> codec.decompress(allocator,
compressed));
+ assertTrue(e.getMessage().contains("content size"), e.getMessage());
+ compressed.close();
+ }
+
+ /** Opens a stream that writes an LZ4 frame. */
+ @FunctionalInterface
+ interface Writer {
+ OutputStream open(OutputStream out) throws IOException;
+ }
+
+ /** {@code data} compressed by {@code writer}, behind Arrow's
uncompressed-length prefix. */
+ private ArrowBuf frameOf(byte[] data, Writer writer) throws IOException {
+ ByteArrayOutputStream frame = new ByteArrayOutputStream();
+ try (OutputStream out = writer.open(frame)) {
+ out.write(data);
+ }
+ byte[] bytes = frame.toByteArray();
+ ArrowBuf buf = allocator.buffer(Long.BYTES + bytes.length);
+ buf.setLong(0, data.length);
+ buf.setBytes(Long.BYTES, bytes);
+ buf.writerIndex(Long.BYTES + bytes.length);
+ return buf;
+ }
+
+ private byte[] decompress(ArrowBuf compressed) {
+ try (ArrowBuf decompressed = codec.decompress(allocator, compressed)) {
+ byte[] bytes = new byte[(int) decompressed.writerIndex()];
+ decompressed.getBytes(0, bytes);
+ return bytes;
+ }
+ }
+
+ private ArrowBuf toArrowBuf(byte[] data) {
+ ArrowBuf buf = allocator.buffer(data.length);
+ buf.setBytes(0, data);
+ buf.writerIndex(data.length);
+ return buf;
+ }
+
+ /**
+ * Decimal numbers that repeat every thousand, for commons-compress, whose
compressor takes
+ * seconds per 100 KiB of input with many short matches, such as {@link
#compressible}'s.
+ */
+ private static byte[] repetitive(int length) {
+ StringBuilder text = new StringBuilder(length + 8);
+ for (int i = 0; text.length() < length; i++) {
+ text.append(i % 1000).append(',');
+ }
+ return Arrays.copyOf(text.toString().getBytes(StandardCharsets.US_ASCII),
length);
+ }
+
+ /** Random decimal numbers as text, which compress well but not trivially. */
+ private static byte[] compressible(int length) {
+ Random random = new Random(length);
+ StringBuilder text = new StringBuilder(length + 8);
+ while (text.length() < length) {
+ text.append(random.nextInt(100_000)).append(',');
+ }
+ return Arrays.copyOf(text.toString().getBytes(StandardCharsets.US_ASCII),
length);
+ }
+}
diff --git a/performance/pom.xml b/performance/pom.xml
index 744df7e62..a6491048b 100644
--- a/performance/pom.xml
+++ b/performance/pom.xml
@@ -86,6 +86,10 @@ under the License.
<groupId>org.apache.arrow</groupId>
<artifactId>arrow-algorithm</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-compression</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git
a/performance/src/main/java/org/apache/arrow/compression/CompressionCodecBenchmarks.java
b/performance/src/main/java/org/apache/arrow/compression/CompressionCodecBenchmarks.java
new file mode 100644
index 000000000..3d2e476a9
--- /dev/null
+++
b/performance/src/main/java/org/apache/arrow/compression/CompressionCodecBenchmarks.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.arrow.compression;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.compression.CompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/** Benchmarks for the {@link CompressionCodec}s of {@link
CommonsCompressionFactory}. */
+@State(Scope.Benchmark)
+public class CompressionCodecBenchmarks {
+ // checkstyle:off: MissingJavadocMethod
+
+ private static final int BUFFER_LENGTH = 8 * 1024 * 1024;
+
+ @Param({"LZ4_FRAME", "ZSTD"})
+ public CompressionUtil.CodecType codecType;
+
+ /** {@code int64}: little-endian longs that count up. {@code text}: random
decimal numbers. */
+ @Param({"int64", "text"})
+ public String data;
+
+ private BufferAllocator allocator;
+
+ private CompressionCodec codec;
+
+ private ArrowBuf uncompressed;
+
+ private ArrowBuf compressed;
+
+ /** Setup benchmarks. */
+ @Setup(Level.Trial)
+ public void prepare() {
+ allocator = new RootAllocator();
+ codec = CommonsCompressionFactory.INSTANCE.createCodec(codecType);
+ uncompressed = allocator.buffer(BUFFER_LENGTH);
+ if (data.equals("int64")) {
+ for (int i = 0; i < BUFFER_LENGTH / Long.BYTES; i++) {
+ uncompressed.setLong((long) i * Long.BYTES, i);
+ }
+ } else {
+ Random random = new Random(0);
+ StringBuilder text = new StringBuilder(BUFFER_LENGTH + 8);
+ while (text.length() < BUFFER_LENGTH) {
+ text.append(random.nextInt(100_000)).append(',');
+ }
+ uncompressed.setBytes(
+ 0, text.substring(0,
BUFFER_LENGTH).getBytes(StandardCharsets.US_ASCII));
+ }
+ uncompressed.writerIndex(BUFFER_LENGTH);
+ compressed = compress();
+ }
+
+ /** Tear down benchmarks. */
+ @TearDown(Level.Trial)
+ public void tearDown() {
+ compressed.close();
+ uncompressed.close();
+ allocator.close();
+ }
+
+ private ArrowBuf compress() {
+ // compress() releases its input, so give it a reference of its own.
+ uncompressed.getReferenceManager().retain();
+ return codec.compress(allocator, uncompressed);
+ }
+
+ @Benchmark
+ @BenchmarkMode(Mode.AverageTime)
+ @OutputTimeUnit(TimeUnit.MILLISECONDS)
+ public void compressBenchmark() {
+ compress().close();
+ }
+
+ @Benchmark
+ @BenchmarkMode(Mode.AverageTime)
+ @OutputTimeUnit(TimeUnit.MILLISECONDS)
+ public void decompressBenchmark() {
+ // decompress() releases its input, so give it a reference of its own.
+ compressed.getReferenceManager().retain();
+ codec.decompress(allocator, compressed).close();
+ }
+
+ public static void main(String[] args) throws RunnerException {
+ Options opt =
+ new OptionsBuilder()
+ .include(CompressionCodecBenchmarks.class.getSimpleName())
+ .forks(1)
+ .build();
+
+ new Runner(opt).run();
+ }
+ // checkstyle:on: MissingJavadocMethod
+}
diff --git a/tools/src/shade/LICENSE.txt b/tools/src/shade/LICENSE.txt
index eef614e7f..cfed4924d 100644
--- a/tools/src/shade/LICENSE.txt
+++ b/tools/src/shade/LICENSE.txt
@@ -329,3 +329,82 @@ License text:
| LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
| OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
| WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+--------------------------------------------------------------------------------
+
+This binary artifact contains lz4-java 1.12.1.
+
+Copyright: Copyright 2020 Adrien Grand and the lz4-java contributors.
+Home page: https://github.com/yawkat/lz4-java
+License: https://www.apache.org/licenses/LICENSE-2.0
+
+--------------------------------------------------------------------------------
+
+This binary artifact contains LZ4 1.10.0, in the native libraries of lz4-java.
+
+Copyright: Copyright (c) 2011-2020, Yann Collet
+Home page: https://github.com/lz4/lz4
+License: BSD 2-Clause License
+License text:
+
+| LZ4 Library
+| Copyright (c) 2011-2020, Yann Collet
+| All rights reserved.
+|
+| Redistribution and use in source and binary forms, with or without
modification,
+| are permitted provided that the following conditions are met:
+|
+| * Redistributions of source code must retain the above copyright notice, this
+| list of conditions and the following disclaimer.
+|
+| * Redistributions in binary form must reproduce the above copyright notice,
this
+| list of conditions and the following disclaimer in the documentation and/or
+| other materials provided with the distribution.
+|
+| THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND
+| ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+| WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+| DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR
+| ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES
+| (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+| LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON
+| ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+| (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+| SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+--------------------------------------------------------------------------------
+
+This binary artifact contains xxHash 0.6.5, in the native libraries of
lz4-java.
+
+Copyright: Copyright (C) 2012-2016, Yann Collet
+Home page: https://github.com/Cyan4973/xxHash
+License: BSD 2-Clause License
+License text:
+
+| xxHash - Extremely Fast Hash algorithm
+| Copyright (C) 2012-2016, Yann Collet.
+|
+| BSD 2-Clause License (http://www.opensource.org/licenses/bsd-license.php)
+|
+| Redistribution and use in source and binary forms, with or without
+| modification, are permitted provided that the following conditions are
+| met:
+|
+| * Redistributions of source code must retain the above copyright
+| notice, this list of conditions and the following disclaimer.
+| * Redistributions in binary form must reproduce the above
+| copyright notice, this list of conditions and the following disclaimer
+| in the documentation and/or other materials provided with the
+| distribution.
+|
+| THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+| "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+| LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+| A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+| OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+| SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+| LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+| DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+| THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+| (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+| OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
