Fix compression codec config gaps from Hadoop bypass: honor ZSTD bufferPool, route GZIP in DirectCodecFactory

Respect parquet.compression.codec.zstd.bufferPool.enabled in the optimized
ZstdBytesCompressor/Decompressor (was hardcoded to RecyclingBufferPool).
Route GZIP decompression through the optimized path in DirectCodecFactory
instead of falling back to the Hadoop codec pool. Remove dead GZIP/ZSTD
branches from cacheKey(). Document ISA-L native library bypass in GZIP
Javadocs. Replace obsolete Hadoop codec caching tests with end-to-end
compression level verification tests.

Author: iemejia

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java

@@ -18,6 +18,8 @@
  */
 package org.apache.parquet.hadoop;
 
+import com.github.luben.zstd.BufferPool;
+import com.github.luben.zstd.NoPool;
 import com.github.luben.zstd.RecyclingBufferPool;
 import com.github.luben.zstd.ZstdInputStreamNoFinalizer;
 import com.github.luben.zstd.ZstdOutputStreamNoFinalizer;
@@ -285,12 +287,17 @@ protected BytesCompressor createCompressor(CompressionCodecName codecName) {
       case SNAPPY:
         return new SnappyBytesCompressor();
       case ZSTD:
+        BufferPool zstdCompressPool = conf.getBoolean(
+            ZstandardCodec.PARQUET_COMPRESS_ZSTD_BUFFERPOOL_ENABLED,
+            ZstandardCodec.DEFAULT_PARQUET_COMPRESS_ZSTD_BUFFERPOOL_ENABLED)
+            ? RecyclingBufferPool.INSTANCE : NoPool.INSTANCE;
         return new ZstdBytesCompressor(
             conf.getInt(
                 ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL, ZstandardCodec.DEFAULT_PARQUET_COMPRESS_ZSTD_LEVEL),
             conf.getInt(
                 ZstandardCodec.PARQUET_COMPRESS_ZSTD_WORKERS, ZstandardCodec.DEFAULTPARQUET_COMPRESS_ZSTD_WORKERS),
-            pageSize);
+            pageSize,
+            zstdCompressPool);
       case LZ4_RAW:
         return new Lz4RawBytesCompressor();
       case GZIP:
@@ -310,7 +317,11 @@ protected BytesDecompressor createDecompressor(CompressionCodecName codecName) {
       case SNAPPY:
         return new SnappyBytesDecompressor();
       case ZSTD:
-        return new ZstdBytesDecompressor();
+        BufferPool zstdDecompressPool = conf.getBoolean(
+            ZstandardCodec.PARQUET_COMPRESS_ZSTD_BUFFERPOOL_ENABLED,
+            ZstandardCodec.DEFAULT_PARQUET_COMPRESS_ZSTD_BUFFERPOOL_ENABLED)
+            ? RecyclingBufferPool.INSTANCE : NoPool.INSTANCE;
+        return new ZstdBytesDecompressor(zstdDecompressPool);
       case LZ4_RAW:
         return new Lz4RawBytesDecompressor();
       case GZIP:
@@ -356,15 +367,9 @@ protected CompressionCodec getCodec(CompressionCodecName codecName) {
   private String cacheKey(CompressionCodecName codecName) {
     String level = null;
     switch (codecName) {
-      case GZIP:
-        level = conf.get("zlib.compress.level");
-        break;
       case BROTLI:
         level = conf.get("compression.brotli.quality");
         break;
-      case ZSTD:
-        level = conf.get("parquet.compression.codec.zstd.level");
-        break;
       default:
         // compression level is not supported; ignore it
     }
@@ -472,28 +477,32 @@ public void release() {}
   /**
    * Compresses using zstd-jni's {@link ZstdOutputStreamNoFinalizer} directly,
    * bypassing the Hadoop codec framework ({@code ZstandardCodec}, {@code CodecPool},
-   * {@code CompressionOutputStream} wrapper). Uses {@link RecyclingBufferPool} for the
-   * internal 128KB output buffer, matching the streaming API's natural buffer size.
+   * {@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) {
+    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, RecyclingBufferPool.INSTANCE, level)) {
+          new ZstdOutputStreamNoFinalizer(compressedOutBuffer, bufferPool, level)) {
         if (workers > 0) {
           zos.setWorkers(workers);
         }
@@ -515,16 +524,23 @@ public void release() {
 
   /**
    * Decompresses using zstd-jni's {@link ZstdInputStreamNoFinalizer} directly,
-   * bypassing the Hadoop codec framework. Uses {@link RecyclingBufferPool} for internal
-   * buffers, matching the streaming decompression path. Reads the full decompressed output
-   * in a single pass via {@link InputStream#readNBytes(int)}.
+   * 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(), RecyclingBufferPool.INSTANCE)) {
+          new ZstdInputStreamNoFinalizer(bytes.toInputStream(), bufferPool)) {
         byte[] output = new byte[decompressedSize];
         int offset = 0;
         while (offset < decompressedSize) {
@@ -546,7 +562,7 @@ public void decompress(ByteBuffer input, int compressedSize, ByteBuffer output,
       input.get(inputBytes);
       ByteArrayInputStream bais = new ByteArrayInputStream(inputBytes);
       try (ZstdInputStreamNoFinalizer zis =
-          new ZstdInputStreamNoFinalizer(bais, RecyclingBufferPool.INSTANCE)) {
+          new ZstdInputStreamNoFinalizer(bais, bufferPool)) {
         byte[] outputBytes = new byte[decompressedSize];
         int offset = 0;
         while (offset < decompressedSize) {
@@ -659,6 +675,13 @@ public void release() {}
    * 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 {
     private final Deflater deflater;
@@ -714,6 +737,11 @@ public void release() {
    * bypassing Hadoop's GzipCodec and the stream overhead of
    * {@link java.util.zip.GZIPInputStream}. Skips the GZIP header, inflates
    * into the output buffer, and verifies the CRC32 + size trailer.
+   *
+   * <p>Note: this implementation always uses Java's built-in {@link Inflater}
+   * (java.util.zip / JDK zlib). It does <em>not</em> use Hadoop native libraries,
+   * so hardware-accelerated decompression via Intel ISA-L will not be used even if
+   * the native libraries are installed.
    */
   static class GzipBytesDecompressor extends BytesDecompressor {
     private final Inflater inflater = new Inflater(true);

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java

@@ -105,8 +105,6 @@ protected BytesCompressor createCompressor(final CompressionCodecName codecName)
         return new ZstdCompressor();
       case LZ4_RAW:
         return new Lz4RawCompressor();
-        // todo: create class similar to the SnappyCompressor for zlib and exclude it as
-        // snappy is above since it also generates allocateDirect calls.
       default:
         return super.createCompressor(codecName);
     }
@@ -121,6 +119,9 @@ protected BytesDecompressor createDecompressor(final CompressionCodecName codecN
         return new ZstdDecompressor();
       case LZ4_RAW:
         return new Lz4RawDecompressor();
+      case GZIP:
+      case UNCOMPRESSED:
+        return super.createDecompressor(codecName);
       default:
         CompressionCodec codec = getCodec(codecName);
         if (codec == null) {

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java

@@ -28,7 +28,6 @@
 import java.util.Random;
 import java.util.Set;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.parquet.bytes.ByteBufferAllocator;
 import org.apache.parquet.bytes.ByteBufferReleaser;
 import org.apache.parquet.bytes.BytesInput;
@@ -235,53 +234,63 @@ public void compressionCodecs() {
     }
   }
 
-  static class PublicCodecFactory extends CodecFactory {
-    // To make getCodec public
+  @Test
+  public void compressionLevelGzip() throws IOException {
+    Configuration config_zlib_1 = new Configuration();
+    config_zlib_1.set("zlib.compress.level", "1");
 
-    public PublicCodecFactory(Configuration configuration, int pageSize) {
-      super(configuration, pageSize);
-    }
+    Configuration config_zlib_9 = new Configuration();
+    config_zlib_9.set("zlib.compress.level", "9");
 
-    public org.apache.hadoop.io.compress.CompressionCodec getCodec(CompressionCodecName name) {
-      return super.getCodec(name);
-    }
-  }
+    // Generate compressible data so different levels produce different sizes
+    byte[] data = new byte[64 * 1024];
+    new Random(42).nextBytes(data);
 
-  @Test
-  public void cachingKeysGzip() {
-    Configuration config_zlib_2 = new Configuration();
-    config_zlib_2.set("zlib.compress.level", "2");
+    final CodecFactory codecFactory_1 = new CodecFactory(config_zlib_1, pageSize);
+    final CodecFactory codecFactory_9 = new CodecFactory(config_zlib_9, pageSize);
 
-    Configuration config_zlib_5 = new Configuration();
-    config_zlib_5.set("zlib.compress.level", "5");
+    BytesInputCompressor compressor_1 = codecFactory_1.getCompressor(CompressionCodecName.GZIP);
+    BytesInputCompressor compressor_9 = codecFactory_9.getCompressor(CompressionCodecName.GZIP);
 
-    final CodecFactory codecFactory_2 = new PublicCodecFactory(config_zlib_2, pageSize);
-    final CodecFactory codecFactory_5 = new PublicCodecFactory(config_zlib_5, pageSize);
+    long size_1 = compressor_1.compress(BytesInput.from(data)).size();
+    long size_9 = compressor_9.compress(BytesInput.from(data)).size();
 
-    CompressionCodec codec_2_1 = codecFactory_2.getCodec(CompressionCodecName.GZIP);
-    CompressionCodec codec_2_2 = codecFactory_2.getCodec(CompressionCodecName.GZIP);
-    CompressionCodec codec_5_1 = codecFactory_5.getCodec(CompressionCodecName.GZIP);
+    // Level 9 should produce smaller (or equal) output than level 1
+    Assert.assertTrue(
+        "Expected level 9 (" + size_9 + ") <= level 1 (" + size_1 + ")",
+        size_9 <= size_1);
 
-    Assert.assertEquals(codec_2_1, codec_2_2);
-    Assert.assertNotEquals(codec_2_1, codec_5_1);
+    codecFactory_1.release();
+    codecFactory_9.release();
   }
 
   @Test
-  public void cachingKeysZstd() {
-    Configuration config_zstd_2 = new Configuration();
-    config_zstd_2.set("parquet.compression.codec.zstd.level", "2");
+  public void compressionLevelZstd() throws IOException {
+    Configuration config_zstd_1 = new Configuration();
+    config_zstd_1.set("parquet.compression.codec.zstd.level", "1");
+
+    Configuration config_zstd_19 = new Configuration();
+    config_zstd_19.set("parquet.compression.codec.zstd.level", "19");
+
+    // Generate compressible data so different levels produce different sizes
+    byte[] data = new byte[64 * 1024];
+    new Random(42).nextBytes(data);
+
+    final CodecFactory codecFactory_1 = new CodecFactory(config_zstd_1, pageSize);
+    final CodecFactory codecFactory_19 = new CodecFactory(config_zstd_19, pageSize);
 
-    Configuration config_zstd_5 = new Configuration();
-    config_zstd_5.set("parquet.compression.codec.zstd.level", "5");
+    BytesInputCompressor compressor_1 = codecFactory_1.getCompressor(CompressionCodecName.ZSTD);
+    BytesInputCompressor compressor_19 = codecFactory_19.getCompressor(CompressionCodecName.ZSTD);
 
-    final CodecFactory codecFactory_2 = new PublicCodecFactory(config_zstd_2, pageSize);
-    final CodecFactory codecFactory_5 = new PublicCodecFactory(config_zstd_5, pageSize);
+    long size_1 = compressor_1.compress(BytesInput.from(data)).size();
+    long size_19 = compressor_19.compress(BytesInput.from(data)).size();
 
-    CompressionCodec codec_2_1 = codecFactory_2.getCodec(CompressionCodecName.ZSTD);
-    CompressionCodec codec_2_2 = codecFactory_2.getCodec(CompressionCodecName.ZSTD);
-    CompressionCodec codec_5_1 = codecFactory_5.getCodec(CompressionCodecName.ZSTD);
+    // Level 19 should produce smaller (or equal) output than level 1
+    Assert.assertTrue(
+        "Expected level 19 (" + size_19 + ") <= level 1 (" + size_1 + ")",
+        size_19 <= size_1);
 
-    Assert.assertEquals(codec_2_1, codec_2_2);
-    Assert.assertNotEquals(codec_2_1, codec_5_1);
+    codecFactory_1.release();
+    codecFactory_19.release();
   }
 }