You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@parquet.apache.org by "iemejia (via GitHub)" <gi...@apache.org> on 2026/05/11 15:11:27 UTC
[PR] Optimize compression hot paths: bypass Hadoop codec abstraction for Snappy, LZ4_RAW, ZSTD, and GZIP [parquet-java]
iemejia opened a new pull request, #3555:
URL: https://github.com/apache/parquet-java/pull/3555
## Summary
Bypass the Hadoop `Compressor`/`Decompressor`/`CodecPool` abstraction layer in `CodecFactory` and `DirectCodecFactory`, calling native compression libraries directly. This eliminates per-page stream creation, intermediate buffer copies, and codec pool synchronization for all four supported codecs.
### What changes
- **Snappy**: Replace `CodecPool` + `SnappyCompressor` (which copies heap→direct→heap) with a single `Snappy.compress(byte[], byte[])` / `Snappy.uncompress(byte[], byte[])` JNI call and a reusable output buffer.
- **LZ4_RAW**: Replace `NonBlockedCompressor` (which allocates direct ByteBuffers and copies heap↔direct twice per call) with heap `ByteBuffer.wrap()` and direct airlift LZ4 compress/decompress — zero intermediate copies.
- **ZSTD**: Replace `ZstdCompressorStream` with `ZstdOutputStreamNoFinalizer` (avoids finalizer registration) and cache the ZSTD level / buffer pool configuration reads per compressor instance instead of re-reading `Configuration` on each page.
- **GZIP**: Replace Hadoop's `GzipCodec` (which wraps Java's `Deflater`/`Inflater` in stream abstractions) with direct `Deflater`/`Inflater` usage, reusing instances via `reset()` and managing GZIP headers/trailers manually.
- **Benchmark**: Update `CompressionBenchmark` page sizes from `{8KB, 64KB, 256KB}` to `{64KB, 128KB, 256KB, 1MB}` to reflect real-world Parquet page sizes (most pages are 64-256KB due to the 20K row-count limit from PARQUET-1414; only wide string/binary columns hit the 1MB size limit).
### Benchmark results (ops/s, higher is better)
#### Compression
| Codec | Page Size | Master | Branch | Delta |
|-------|-----------|-------:|-------:|------:|
| SNAPPY | 64 KB | 53,979 | 60,799 | **+12.6%** |
| SNAPPY | 128 KB | 27,764 | 30,524 | **+9.9%** |
| SNAPPY | 256 KB | 13,549 | 14,648 | **+8.1%** |
| SNAPPY | 1 MB | 2,445 | 2,675 | **+9.4%** |
| LZ4_RAW | 1 MB | 1,961 | 2,191 | **+11.7%** |
| LZ4_RAW | 64-256 KB | — | — | within noise (-1 to -4%) |
| ZSTD | all sizes | — | — | within noise |
| GZIP | all sizes | — | — | within noise |
#### Decompression
| Codec | Page Size | Master | Branch | Delta |
|-------|-----------|-------:|-------:|------:|
| LZ4_RAW | 64 KB | 80,415 | 118,358 | **+47.2%** |
| LZ4_RAW | 128 KB | 40,615 | 59,620 | **+46.8%** |
| LZ4_RAW | 256 KB | 19,888 | 29,914 | **+50.4%** |
| LZ4_RAW | 1 MB | 4,628 | 7,517 | **+62.4%** |
| SNAPPY | 64 KB | 60,928 | 67,224 | **+10.3%** |
| SNAPPY | 128 KB | 29,919 | 33,457 | **+11.8%** |
| SNAPPY | 256 KB | 14,431 | 15,912 | **+10.3%** |
| SNAPPY | 1 MB | 3,140 | 3,540 | **+12.7%** |
| ZSTD | 64 KB | 32,042 | 35,750 | **+11.6%** |
| ZSTD | 128 KB | 19,447 | 21,800 | **+12.1%** |
| ZSTD | 256 KB | 9,495 | 10,759 | **+13.3%** |
| ZSTD | 1 MB | 2,155 | 2,409 | **+11.8%** |
| GZIP | 128 KB | 4,101 | 4,536 | **+10.6%** |
| GZIP | 256 KB | 1,736 | 1,891 | **+8.9%** |
| GZIP | 1 MB | 406 | 442 | **+9.1%** |
JMH config: JDK 25.0.3 Temurin, 1 fork, 2 warmup × 1s, 3 measurement × 2s.
### Why LZ4_RAW decompression gains are largest
`NonBlockedDecompressor` performs two full data copies per operation — heap byte[] → direct ByteBuffer on input, direct ByteBuffer → heap byte[] on output — plus direct buffer allocation and synchronized access. The bypass eliminates both copies by using `ByteBuffer.wrap()` on heap arrays, letting airlift's LZ4 decompress directly between heap buffers.
### Why ZSTD compression gains are minimal
`ZstandardCodec` already returns `null` from `createCompressor()`/`createDecompressor()` and delegates directly to `zstd-jni` streams. The Hadoop abstraction overhead was already bypassed at the codec level. The branch adds finalizer avoidance (`NoFinalizer` variants) and caches configuration reads, which helps decompression but leaves compression within noise.
### Alternative considered: modify codecs instead of CodecFactory
We evaluated modifying `SnappyCodec` and `Lz4RawCodec` to follow the `ZstandardCodec` pattern (return `null` from `createCompressor()`, use custom stream wrappers). This approach was **25-50% slower** than the `CodecFactory` bypass for Snappy/LZ4 and even **20-47% slower than master**. The per-call stream creation, `ByteArrayOutputStream` buffering, and lack of buffer reuse dominate for memory-bandwidth-bound codecs where the actual compression takes only 8-65 microseconds.
### Files changed
- `CodecFactory.java`: Bypass compressor/decompressor with codec-specific inner classes (`SnappyBytesCompressor`, `Lz4RawBytesCompressor`, `ZstdBytesCompressor`, `GzipBytesCompressor` + matching decompressors)
- `DirectCodecFactory.java`: Bypass for direct `ByteBuffer` path (Snappy, LZ4_RAW, ZSTD)
- `BytesInput.java`: Add `ByteBufferBackedOutputStream` to avoid `toByteArray()` copies
- `CompressionBenchmark.java`: Realistic page sizes + JMH annotation processor fix for Java 17+
- `TestDirectCodecFactory.java`: Updated tests for bypass path
--
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: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] Optimize compression hot paths: bypass Hadoop codec abstraction for Snappy, LZ4_RAW, ZSTD, and GZIP [parquet-java]
Posted by "iemejia (via GitHub)" <gi...@apache.org>.
iemejia commented on PR #3555:
URL: https://github.com/apache/parquet-java/pull/3555#issuecomment-4472736090
Closing in favor of #3570.
I initially submitted a series of small, focused PRs thinking they'd be easier to review. In practice the sheer number (~16 PRs, with more pending) made things harder to follow — even for me. I've regrouped the changes by encoding type / performance area so that each PR is self-contained with its own benchmarks and test coverage, which should make review and performance analysis much more straightforward.
Apologies for the churn. If you've been reviewing this PR, please continue the discussion on #3570 which supersedes it. Thank you.
--
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: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] Optimize compression hot paths: bypass Hadoop codec abstraction for Snappy, LZ4_RAW, ZSTD, and GZIP [parquet-java]
Posted by "iemejia (via GitHub)" <gi...@apache.org>.
iemejia closed pull request #3555: Optimize compression hot paths: bypass Hadoop codec abstraction for Snappy, LZ4_RAW, ZSTD, and GZIP
URL: https://github.com/apache/parquet-java/pull/3555
--
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: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org
Re: [PR] Optimize compression hot paths: bypass Hadoop codec abstraction for Snappy, LZ4_RAW, ZSTD, and GZIP [parquet-java]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #3555:
URL: https://github.com/apache/parquet-java/pull/3555#discussion_r3240973528
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java:
##########
@@ -367,4 +418,447 @@ public abstract void decompress(ByteBuffer input, int compressedSize, ByteBuffer
public abstract void release();
}
+
+ // ---- Optimized Snappy compressor/decompressor using direct JNI calls ----
+
+ /**
+ * Compresses using Snappy's byte-array JNI API directly, bypassing the Hadoop
+ * stream abstraction. This avoids intermediate direct ByteBuffer copies and
+ * reduces the compression to a single native call per page.
+ */
+ static class SnappyBytesCompressor extends BytesCompressor {
+ private byte[] outputBuffer;
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ byte[] input = bytes.toByteArray();
+ int maxLen = Snappy.maxCompressedLength(input.length);
+ if (outputBuffer == null || outputBuffer.length < maxLen) {
+ outputBuffer = new byte[maxLen];
+ }
+ int compressed = Snappy.compress(input, 0, input.length, outputBuffer, 0);
+ return BytesInput.from(outputBuffer, 0, compressed);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.SNAPPY;
+ }
+
+ @Override
+ public void release() {
+ outputBuffer = null;
+ }
+ }
+
+ /**
+ * Decompresses using Snappy's byte-array JNI API directly.
+ */
+ static class SnappyBytesDecompressor extends BytesDecompressor {
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ byte[] input = bytes.toByteArray();
+ byte[] output = new byte[decompressedSize];
+ Snappy.uncompress(input, 0, input.length, output, 0);
+ return BytesInput.from(output);
+ }
+
+ @Override
+ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize)
+ throws IOException {
+ byte[] inputBytes = new byte[compressedSize];
+ input.get(inputBytes);
+ byte[] outputBytes = new byte[decompressedSize];
+ Snappy.uncompress(inputBytes, 0, compressedSize, outputBytes, 0);
+ output.put(outputBytes);
+ }
+
+ @Override
+ public void release() {}
+ }
+
+ // ---- Optimized ZSTD compressor/decompressor using zstd-jni streaming API directly ----
+
+ /**
+ * Compresses using zstd-jni's {@link ZstdOutputStreamNoFinalizer} directly,
+ * bypassing the Hadoop codec framework ({@code ZstandardCodec}, {@code CodecPool},
+ * {@code CompressionOutputStream} wrapper). Uses a configurable {@link BufferPool}
+ * (defaulting to {@link RecyclingBufferPool}) for the internal 128KB output buffer,
+ * matching the streaming API's natural buffer size. The buffer pool strategy is
+ * controlled by the {@code parquet.compression.codec.zstd.bufferPool.enabled} config.
+ * This avoids the overhead of Hadoop codec instantiation and compressor pool management
+ * while using the same underlying ZSTD streaming path, which is well-optimized for all
+ * input sizes including large pages (256KB+).
+ */
+ static class ZstdBytesCompressor extends BytesCompressor {
+ private final int level;
+ private final int workers;
+ private final BufferPool bufferPool;
+ private final ByteArrayOutputStream compressedOutBuffer;
+
+ ZstdBytesCompressor(int level, int workers, int pageSize, BufferPool bufferPool) {
+ this.level = level;
+ this.workers = workers;
+ this.bufferPool = bufferPool;
+ this.compressedOutBuffer = new ByteArrayOutputStream(pageSize);
+ }
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ compressedOutBuffer.reset();
+ try (ZstdOutputStreamNoFinalizer zos =
+ new ZstdOutputStreamNoFinalizer(compressedOutBuffer, bufferPool, level)) {
+ if (workers > 0) {
+ zos.setWorkers(workers);
+ }
+ bytes.writeAllTo(zos);
+ }
+ return BytesInput.from(compressedOutBuffer);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.ZSTD;
+ }
+
+ @Override
+ public void release() {
+ // ByteArrayOutputStream does not hold native resources
+ }
+ }
+
+ /**
+ * Decompresses using zstd-jni's {@link ZstdInputStreamNoFinalizer} directly,
+ * bypassing the Hadoop codec framework. Uses a configurable {@link BufferPool}
+ * for internal buffers, matching the streaming decompression path. The buffer pool
+ * strategy is controlled by the {@code parquet.compression.codec.zstd.bufferPool.enabled}
+ * config. Reads the full decompressed output in a single pass via
+ * {@link InputStream#readNBytes(int)}.
+ */
+ static class ZstdBytesDecompressor extends BytesDecompressor {
+ private final BufferPool bufferPool;
+
+ ZstdBytesDecompressor(BufferPool bufferPool) {
+ this.bufferPool = bufferPool;
+ }
+
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ try (ZstdInputStreamNoFinalizer zis = new ZstdInputStreamNoFinalizer(bytes.toInputStream(), bufferPool)) {
+ byte[] output = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = zis.read(output, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
Review Comment:
EOFException would be stricter here
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java:
##########
@@ -367,4 +418,447 @@ public abstract void decompress(ByteBuffer input, int compressedSize, ByteBuffer
public abstract void release();
}
+
+ // ---- Optimized Snappy compressor/decompressor using direct JNI calls ----
+
+ /**
+ * Compresses using Snappy's byte-array JNI API directly, bypassing the Hadoop
+ * stream abstraction. This avoids intermediate direct ByteBuffer copies and
+ * reduces the compression to a single native call per page.
+ */
+ static class SnappyBytesCompressor extends BytesCompressor {
+ private byte[] outputBuffer;
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ byte[] input = bytes.toByteArray();
+ int maxLen = Snappy.maxCompressedLength(input.length);
+ if (outputBuffer == null || outputBuffer.length < maxLen) {
+ outputBuffer = new byte[maxLen];
+ }
+ int compressed = Snappy.compress(input, 0, input.length, outputBuffer, 0);
+ return BytesInput.from(outputBuffer, 0, compressed);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.SNAPPY;
+ }
+
+ @Override
+ public void release() {
+ outputBuffer = null;
+ }
+ }
+
+ /**
+ * Decompresses using Snappy's byte-array JNI API directly.
+ */
+ static class SnappyBytesDecompressor extends BytesDecompressor {
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ byte[] input = bytes.toByteArray();
+ byte[] output = new byte[decompressedSize];
+ Snappy.uncompress(input, 0, input.length, output, 0);
+ return BytesInput.from(output);
+ }
+
+ @Override
+ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize)
+ throws IOException {
+ byte[] inputBytes = new byte[compressedSize];
+ input.get(inputBytes);
+ byte[] outputBytes = new byte[decompressedSize];
+ Snappy.uncompress(inputBytes, 0, compressedSize, outputBytes, 0);
+ output.put(outputBytes);
+ }
+
+ @Override
+ public void release() {}
+ }
+
+ // ---- Optimized ZSTD compressor/decompressor using zstd-jni streaming API directly ----
+
+ /**
+ * Compresses using zstd-jni's {@link ZstdOutputStreamNoFinalizer} directly,
+ * bypassing the Hadoop codec framework ({@code ZstandardCodec}, {@code CodecPool},
+ * {@code CompressionOutputStream} wrapper). Uses a configurable {@link BufferPool}
+ * (defaulting to {@link RecyclingBufferPool}) for the internal 128KB output buffer,
+ * matching the streaming API's natural buffer size. The buffer pool strategy is
+ * controlled by the {@code parquet.compression.codec.zstd.bufferPool.enabled} config.
+ * This avoids the overhead of Hadoop codec instantiation and compressor pool management
+ * while using the same underlying ZSTD streaming path, which is well-optimized for all
+ * input sizes including large pages (256KB+).
+ */
+ static class ZstdBytesCompressor extends BytesCompressor {
+ private final int level;
+ private final int workers;
+ private final BufferPool bufferPool;
+ private final ByteArrayOutputStream compressedOutBuffer;
+
+ ZstdBytesCompressor(int level, int workers, int pageSize, BufferPool bufferPool) {
+ this.level = level;
+ this.workers = workers;
+ this.bufferPool = bufferPool;
+ this.compressedOutBuffer = new ByteArrayOutputStream(pageSize);
+ }
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ compressedOutBuffer.reset();
+ try (ZstdOutputStreamNoFinalizer zos =
+ new ZstdOutputStreamNoFinalizer(compressedOutBuffer, bufferPool, level)) {
+ if (workers > 0) {
+ zos.setWorkers(workers);
+ }
+ bytes.writeAllTo(zos);
+ }
+ return BytesInput.from(compressedOutBuffer);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.ZSTD;
+ }
+
+ @Override
+ public void release() {
+ // ByteArrayOutputStream does not hold native resources
+ }
+ }
+
+ /**
+ * Decompresses using zstd-jni's {@link ZstdInputStreamNoFinalizer} directly,
+ * bypassing the Hadoop codec framework. Uses a configurable {@link BufferPool}
+ * for internal buffers, matching the streaming decompression path. The buffer pool
+ * strategy is controlled by the {@code parquet.compression.codec.zstd.bufferPool.enabled}
+ * config. Reads the full decompressed output in a single pass via
+ * {@link InputStream#readNBytes(int)}.
+ */
+ static class ZstdBytesDecompressor extends BytesDecompressor {
+ private final BufferPool bufferPool;
+
+ ZstdBytesDecompressor(BufferPool bufferPool) {
+ this.bufferPool = bufferPool;
+ }
+
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ try (ZstdInputStreamNoFinalizer zis = new ZstdInputStreamNoFinalizer(bytes.toInputStream(), bufferPool)) {
+ byte[] output = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = zis.read(output, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
+ "Unexpected end of ZSTD stream at offset " + offset + " of " + decompressedSize);
+ }
+ offset += read;
+ }
+ return BytesInput.from(output);
+ }
+ }
+
+ @Override
+ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize)
+ throws IOException {
+ byte[] inputBytes = new byte[compressedSize];
+ input.get(inputBytes);
+ ByteArrayInputStream bais = new ByteArrayInputStream(inputBytes);
+ try (ZstdInputStreamNoFinalizer zis = new ZstdInputStreamNoFinalizer(bais, bufferPool)) {
+ byte[] outputBytes = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = zis.read(outputBytes, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
+ "Unexpected end of ZSTD stream at offset " + offset + " of " + decompressedSize);
+ }
+ offset += read;
+ }
+ output.put(outputBytes);
+ }
+ }
+
+ @Override
+ public void release() {
+ // No persistent resources - streams are closed per call
+ }
+ }
+
+ // ---- Optimized LZ4_RAW compressor/decompressor using airlift LZ4 directly ----
+
+ /**
+ * Compresses using airlift's LZ4 compressor directly with heap ByteBuffers,
+ * bypassing the Hadoop stream abstraction and NonBlockedCompressor's direct
+ * buffer copies.
+ */
+ static class Lz4RawBytesCompressor extends BytesCompressor {
+ private final Lz4Compressor compressor = new Lz4Compressor();
+ private byte[] outputBuffer;
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ byte[] input = bytes.toByteArray();
+ int maxLen = compressor.maxCompressedLength(input.length);
+ if (outputBuffer == null || outputBuffer.length < maxLen) {
+ outputBuffer = new byte[maxLen];
+ }
+ ByteBuffer inputBuf = ByteBuffer.wrap(input);
+ ByteBuffer outputBuf = ByteBuffer.wrap(outputBuffer);
+ compressor.compress(inputBuf, outputBuf);
+ int compressedSize = outputBuf.position();
+ return BytesInput.from(outputBuffer, 0, compressedSize);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.LZ4_RAW;
+ }
+
+ @Override
+ public void release() {
+ outputBuffer = null;
+ }
+ }
+
+ /**
+ * Decompresses using airlift's LZ4 decompressor directly with heap ByteBuffers.
+ */
+ static class Lz4RawBytesDecompressor extends BytesDecompressor {
+ private final Lz4Decompressor decompressor = new Lz4Decompressor();
+
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ byte[] input = bytes.toByteArray();
+ byte[] output = new byte[decompressedSize];
+ ByteBuffer inputBuf = ByteBuffer.wrap(input);
+ ByteBuffer outputBuf = ByteBuffer.wrap(output);
+ decompressor.decompress(inputBuf, outputBuf);
+ return BytesInput.from(output);
+ }
+
+ @Override
+ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize)
+ throws IOException {
+ byte[] inputBytes = new byte[compressedSize];
+ input.get(inputBytes);
+ byte[] outputBytes = new byte[decompressedSize];
+ ByteBuffer inputBuf = ByteBuffer.wrap(inputBytes);
+ ByteBuffer outputBuf = ByteBuffer.wrap(outputBytes);
+ decompressor.decompress(inputBuf, outputBuf);
+ output.put(outputBytes);
+ }
+
+ @Override
+ public void release() {}
+ }
+
+ // ---- Optimized GZIP compressor/decompressor using Deflater/Inflater directly ----
+
+ /** GZIP magic number: 0x1f 0x8b. */
+ private static final int GZIP_MAGIC = 0x8b1f;
+
+ /** Minimal 10-byte GZIP header: magic, method=8 (deflate), flags=0, mtime=0, xfl=0, os=0. */
+ private static final byte[] GZIP_HEADER = {
+ 0x1f,
+ (byte) 0x8b, // magic
+ 0x08, // method: deflate
+ 0x00, // flags: none
+ 0x00,
+ 0x00,
+ 0x00,
+ 0x00, // mtime: not set
+ 0x00, // extra flags
+ 0x00 // OS: FAT (matches Java's GZIPOutputStream default)
+ };
+
+ /**
+ * Compresses using {@link Deflater} directly with a reusable instance,
+ * bypassing Hadoop's GzipCodec and the stream overhead of
+ * {@link java.util.zip.GZIPOutputStream}. The Deflater is kept across
+ * calls and reset via {@link Deflater#reset()}, avoiding native zlib
+ * state allocation per page. Writes a minimal GZIP header and trailer
+ * (CRC32 + original size) manually.
+ *
+ * <p>Note: this implementation always uses Java's built-in {@link Deflater}
+ * (java.util.zip / JDK zlib). It does <em>not</em> use Hadoop native libraries,
+ * so hardware-accelerated compression via Intel ISA-L will not be used even if
+ * the native libraries are installed. The overhead reduction from bypassing the
+ * Hadoop codec framework typically outweighs the ISA-L advantage for the page
+ * sizes used by Parquet.
+ */
+ static class GzipBytesCompressor extends BytesCompressor {
Review Comment:
it'd be good to verify that the hadoop gzip can decompress this and vice versa, as the regression test on this implementation
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java:
##########
@@ -76,6 +85,46 @@ class DirectCodecFactory extends CodecFactory implements AutoCloseable {
DIRECT_DECOMPRESSION_CODEC_CLASS = tempClass;
CREATE_DIRECT_DECOMPRESSOR_METHOD = tempCreateMethod;
DECOMPRESS_METHOD = tempDecompressMethod;
+
+ // Initialize Brotli JNI bypass via reflection
+ boolean brotliLoaded = false;
+ Method brotliDecompress = null;
+ Method brotliCompress = null;
+ Constructor<?> brotliDecompressorCtor = null;
+ Constructor<?> brotliCompressorCtor = null;
+ Object brotliParam = null;
+ try {
+ // Load native library
+ Class<?> loaderClass = Class.forName("org.meteogroup.jbrotli.libloader.BrotliLibraryLoader");
+ loaderClass.getMethod("loadBrotli").invoke(null);
+
+ // BrotliDeCompressor: no-arg ctor + deCompress(ByteBuffer, ByteBuffer) -> int
+ Class<?> decompClass = Class.forName("org.meteogroup.jbrotli.BrotliDeCompressor");
+ brotliDecompressorCtor = decompClass.getConstructor();
+ brotliDecompress = decompClass.getMethod("deCompress", ByteBuffer.class, ByteBuffer.class);
Review Comment:
I'd recommend using org.apache.parquet.util.DynMethods for this stuff
* versions of it in different asf projects (iceberg, hadoop...)
* uses context classloader, which may matter in some deployments
* easier!
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java:
##########
@@ -367,4 +418,447 @@ public abstract void decompress(ByteBuffer input, int compressedSize, ByteBuffer
public abstract void release();
}
+
+ // ---- Optimized Snappy compressor/decompressor using direct JNI calls ----
+
+ /**
+ * Compresses using Snappy's byte-array JNI API directly, bypassing the Hadoop
+ * stream abstraction. This avoids intermediate direct ByteBuffer copies and
+ * reduces the compression to a single native call per page.
+ */
+ static class SnappyBytesCompressor extends BytesCompressor {
+ private byte[] outputBuffer;
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ byte[] input = bytes.toByteArray();
+ int maxLen = Snappy.maxCompressedLength(input.length);
+ if (outputBuffer == null || outputBuffer.length < maxLen) {
+ outputBuffer = new byte[maxLen];
+ }
+ int compressed = Snappy.compress(input, 0, input.length, outputBuffer, 0);
+ return BytesInput.from(outputBuffer, 0, compressed);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.SNAPPY;
+ }
+
+ @Override
+ public void release() {
+ outputBuffer = null;
+ }
+ }
+
+ /**
+ * Decompresses using Snappy's byte-array JNI API directly.
+ */
+ static class SnappyBytesDecompressor extends BytesDecompressor {
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ byte[] input = bytes.toByteArray();
+ byte[] output = new byte[decompressedSize];
+ Snappy.uncompress(input, 0, input.length, output, 0);
+ return BytesInput.from(output);
+ }
+
+ @Override
+ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize)
+ throws IOException {
+ byte[] inputBytes = new byte[compressedSize];
+ input.get(inputBytes);
+ byte[] outputBytes = new byte[decompressedSize];
+ Snappy.uncompress(inputBytes, 0, compressedSize, outputBytes, 0);
+ output.put(outputBytes);
+ }
+
+ @Override
+ public void release() {}
+ }
+
+ // ---- Optimized ZSTD compressor/decompressor using zstd-jni streaming API directly ----
+
+ /**
+ * Compresses using zstd-jni's {@link ZstdOutputStreamNoFinalizer} directly,
+ * bypassing the Hadoop codec framework ({@code ZstandardCodec}, {@code CodecPool},
+ * {@code CompressionOutputStream} wrapper). Uses a configurable {@link BufferPool}
+ * (defaulting to {@link RecyclingBufferPool}) for the internal 128KB output buffer,
+ * matching the streaming API's natural buffer size. The buffer pool strategy is
+ * controlled by the {@code parquet.compression.codec.zstd.bufferPool.enabled} config.
+ * This avoids the overhead of Hadoop codec instantiation and compressor pool management
+ * while using the same underlying ZSTD streaming path, which is well-optimized for all
+ * input sizes including large pages (256KB+).
+ */
+ static class ZstdBytesCompressor extends BytesCompressor {
+ private final int level;
+ private final int workers;
+ private final BufferPool bufferPool;
+ private final ByteArrayOutputStream compressedOutBuffer;
+
+ ZstdBytesCompressor(int level, int workers, int pageSize, BufferPool bufferPool) {
+ this.level = level;
+ this.workers = workers;
+ this.bufferPool = bufferPool;
+ this.compressedOutBuffer = new ByteArrayOutputStream(pageSize);
+ }
+
+ @Override
+ public BytesInput compress(BytesInput bytes) throws IOException {
+ compressedOutBuffer.reset();
+ try (ZstdOutputStreamNoFinalizer zos =
+ new ZstdOutputStreamNoFinalizer(compressedOutBuffer, bufferPool, level)) {
+ if (workers > 0) {
+ zos.setWorkers(workers);
+ }
+ bytes.writeAllTo(zos);
+ }
+ return BytesInput.from(compressedOutBuffer);
+ }
+
+ @Override
+ public CompressionCodecName getCodecName() {
+ return CompressionCodecName.ZSTD;
+ }
+
+ @Override
+ public void release() {
+ // ByteArrayOutputStream does not hold native resources
+ }
+ }
+
+ /**
+ * Decompresses using zstd-jni's {@link ZstdInputStreamNoFinalizer} directly,
+ * bypassing the Hadoop codec framework. Uses a configurable {@link BufferPool}
+ * for internal buffers, matching the streaming decompression path. The buffer pool
+ * strategy is controlled by the {@code parquet.compression.codec.zstd.bufferPool.enabled}
+ * config. Reads the full decompressed output in a single pass via
+ * {@link InputStream#readNBytes(int)}.
+ */
+ static class ZstdBytesDecompressor extends BytesDecompressor {
+ private final BufferPool bufferPool;
+
+ ZstdBytesDecompressor(BufferPool bufferPool) {
+ this.bufferPool = bufferPool;
+ }
+
+ @Override
+ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOException {
+ try (ZstdInputStreamNoFinalizer zis = new ZstdInputStreamNoFinalizer(bytes.toInputStream(), bufferPool)) {
+ byte[] output = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = zis.read(output, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
+ "Unexpected end of ZSTD stream at offset " + offset + " of " + decompressedSize);
+ }
+ offset += read;
+ }
+ return BytesInput.from(output);
+ }
+ }
+
+ @Override
+ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output, int decompressedSize)
+ throws IOException {
+ byte[] inputBytes = new byte[compressedSize];
+ input.get(inputBytes);
+ ByteArrayInputStream bais = new ByteArrayInputStream(inputBytes);
+ try (ZstdInputStreamNoFinalizer zis = new ZstdInputStreamNoFinalizer(bais, bufferPool)) {
+ byte[] outputBytes = new byte[decompressedSize];
+ int offset = 0;
+ while (offset < decompressedSize) {
+ int read = zis.read(outputBytes, offset, decompressedSize - offset);
+ if (read < 0) {
+ throw new IOException(
Review Comment:
EOFException
--
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: issues-unsubscribe@parquet.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@parquet.apache.org
For additional commands, e-mail: issues-help@parquet.apache.org