From 56f64475c63449f04c0a7a9d83ac6f2ea3804fd1 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 31 Oct 2026 09:02:45 -0600 Subject: [PATCH 1/3] GH-192: Use lz4-java for LZ4 frame compression Lz4CompressionCodec compressed with commons-compress's pure-Java LZ4, which takes minutes for some 8 MiB buffers. It now compresses each buffer with lz4-java into one frame of independent blocks of up to 4 MiB, without checksums, which every LZ4 frame decoder reads, and decompresses frames whose blocks are independent with lz4-java too, directly on the buffers' memory. A frame whose blocks are linked, as Arrow C++ writes for a buffer larger than 64 KiB, or that uses a dictionary, is still decompressed with commons-compress, since lz4-java cannot decode a block that refers to earlier ones. The new decoder checks the frame's block lengths, and that they can hold the recorded decompressed length, before allocating for it, and verifies the descriptor, block and content checksums the frame has. CompressionCodecBenchmarks measures the LZ4 and ZSTD codecs. --- compression/pom.xml | 5 + compression/src/main/java/module-info.java | 1 + .../compression/Lz4CompressionCodec.java | 321 ++++++++++++++++-- .../compression/TestLz4CompressionCodec.java | 268 +++++++++++++++ performance/pom.xml | 4 + .../CompressionCodecBenchmarks.java | 127 +++++++ 6 files changed, 694 insertions(+), 32 deletions(-) create mode 100644 compression/src/test/java/org/apache/arrow/compression/TestLz4CompressionCodec.java create mode 100644 performance/src/main/java/org/apache/arrow/compression/CompressionCodecBenchmarks.java diff --git a/compression/pom.xml b/compression/pom.xml index f03fde2472..fe3ed5e05d 100644 --- a/compression/pom.xml +++ b/compression/pom.xml @@ -57,5 +57,10 @@ under the License. zstd-jni 1.5.7-25 + + at.yawk.lz4 + lz4-java + 1.12.1 + diff --git a/compression/src/main/java/module-info.java b/compression/src/main/java/module-info.java index 113a1dba9d..e438355008 100644 --- a/compression/src/main/java/module-info.java +++ b/compression/src/main/java/module-info.java @@ -24,6 +24,7 @@ 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 f268e815fe..4720e03b53 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,257 @@ 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. + * + *

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 +278,8 @@ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf compressedBu 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 +289,7 @@ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf compressedBu 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 +301,57 @@ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf compressedBu 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 0000000000..37068e696b --- /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 otherWriters() { + List 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 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 corrupt) { + return Arguments.of(name, corrupt); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("corruptions") + void testDecompressCorruptFrame(String name, Consumer 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 744df7e62c..a6491048b4 100644 --- a/performance/pom.xml +++ b/performance/pom.xml @@ -86,6 +86,10 @@ under the License. org.apache.arrow arrow-algorithm + + org.apache.arrow + arrow-compression + 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 0000000000..3d2e476a97 --- /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 +} From 5c74f308681512fa02b6a8d3b698e3b291a5da77 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 1 Nov 2026 08:18:31 -0700 Subject: [PATCH 2/3] style(compression): wrap Lz4CompressionCodec lines as spotless expects Apply the google-java-format line wrapping that the spotless check reported for Lz4CompressionCodec. No functional change. --- .../arrow/compression/Lz4CompressionCodec.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) 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 4720e03b53..18cad8dbbb 100644 --- a/compression/src/main/java/org/apache/arrow/compression/Lz4CompressionCodec.java +++ b/compression/src/main/java/org/apache/arrow/compression/Lz4CompressionCodec.java @@ -162,7 +162,10 @@ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf compressedBu int descriptorLength = (int) (at - (PREFIX + 4)); int headerChecksum = (Lz4.XXHASH.hash( - compressedBuffer.nioBuffer(PREFIX + 4, descriptorLength), 0, descriptorLength, 0) + compressedBuffer.nioBuffer(PREFIX + 4, descriptorLength), + 0, + descriptorLength, + 0) >>> 8) & 0xFF; if ((compressedBuffer.getByte(at) & 0xFF) != headerChecksum) { @@ -191,9 +194,7 @@ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf compressedBu 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); + (word & UNCOMPRESSED_BLOCK) != 0 ? n : Math.min(maxBlockSize, n * MAX_COMPRESSION_RATIO); at += n; if (blockChecksum) { if (end - at < Integer.BYTES) { @@ -232,7 +233,8 @@ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf compressedBu } if ((word & UNCOMPRESSED_BLOCK) != 0) { if (n > length - to) { - throw corrupt("the frame holds more than its recorded decompressed length of " + length); + throw corrupt( + "the frame holds more than its recorded decompressed length of " + length); } decompressedBuffer.setBytes(to, compressedBuffer, at, n); to += n; From dd77343a434cc94fc88948ba0c498be2b7fbb1e2 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 1 Nov 2026 08:18:31 -0700 Subject: [PATCH 3/3] build(tools): list lz4-java, LZ4 and xxHash in the shaded jar LICENSE arrow-tools shades lz4-java, including the native libraries it ships. lz4-java 1.12.1 pins LZ4 at v1.10.0 and builds those libraries from lz4.c, lz4hc.c and xxhash.c in LZ4's lib directory, so they contain LZ4 1.10.0 and the xxHash 0.6.5 copy there, both under the BSD 2-Clause license. Add entries for all three, with the BSD texts inline. lz4-java has no NOTICE file, so NOTICE.txt is unchanged. --- tools/src/shade/LICENSE.txt | 79 +++++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/tools/src/shade/LICENSE.txt b/tools/src/shade/LICENSE.txt index eef614e7f2..cfed4924d2 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.