Add BROTLI direct bypass in DirectCodecFactory using jbrotli JNI

Bypass the Hadoop BrotliCodec/stream wrapper for BROTLI compression and
decompression by using org.meteogroup.jbrotli's native JNI bindings directly
with ByteBuffer support via reflection (brotli-codec remains runtime scope).
This eliminates intermediate buffer copies and the BrotliStreamCompressor
state machine overhead.

Changes:
- DirectCodecFactory: Add BrotliDirectCompressor (quality=1, matching Hadoop
  default) and BrotliDirectDecompressor using one-shot jbrotli API via reflection
- Load native library eagerly with graceful fallback to Hadoop codec path
- CompressionBenchmark: Switch from heap CodecFactory to DirectCodecFactory
  to benchmark the actual production code path

Results at 64KB page size:
- Compress: 6,746 -> 9,662 ops/s (1.43x speedup)
- Decompress: 2,534 -> 2,786 ops/s (1.10x speedup)

Author: iemejia

parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/CompressionBenchmark.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
 import org.apache.parquet.compression.CompressionCodecFactory;
 import org.apache.parquet.hadoop.CodecFactory;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
@@ -45,8 +46,9 @@
  *
  * <p>Measures the performance of {@link CompressionCodecFactory.BytesInputCompressor}
  * and {@link CompressionCodecFactory.BytesInputDecompressor} for each supported codec,
- * using the heap-based {@link CodecFactory} path. Input data is generated to approximate
- * realistic Parquet page content (a mix of sequential, repeated, and random byte patterns).
+ * using the direct-memory {@link CodecFactory} path (same as actual Parquet file I/O).
+ * Input data is generated to approximate realistic Parquet page content (a mix of
+ * sequential, repeated, and random byte patterns).
  *
  * <p>This benchmark isolates the codec hot path from file I/O, encoding, and other
  * Parquet overhead, making it ideal for measuring compression-specific optimizations.
@@ -79,7 +81,7 @@ public void setup() throws IOException {
     decompressedSize = uncompressedData.length;
 
     Configuration conf = new Configuration();
-    factory = new CodecFactory(conf, pageSize);
+    factory = CodecFactory.createDirectCodecFactory(conf, DirectByteBufferAllocator.getInstance(), pageSize);
     CompressionCodecName codecName = CompressionCodecName.valueOf(codec);
 
     compressor = factory.getCompressor(codecName);

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

@@ -21,6 +21,7 @@
 import com.github.luben.zstd.ZstdCompressCtx;
 import com.github.luben.zstd.ZstdDecompressCtx;
 import java.io.IOException;
+import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
@@ -61,6 +62,14 @@ class DirectCodecFactory extends CodecFactory implements AutoCloseable {
   private static final Method DECOMPRESS_METHOD;
   private static final Method CREATE_DIRECT_DECOMPRESSOR_METHOD;
 
+  // Brotli JNI bypass via reflection (brotli-codec is a runtime-only dependency)
+  private static final boolean BROTLI_NATIVE_AVAILABLE;
+  private static final Method BROTLI_DECOMPRESS_METHOD; // BrotliDeCompressor.deCompress(ByteBuffer, ByteBuffer)
+  private static final Method BROTLI_COMPRESS_METHOD; // BrotliCompressor.compress(Parameter, ByteBuffer, ByteBuffer)
+  private static final Constructor<?> BROTLI_DECOMPRESSOR_CTOR; // BrotliDeCompressor()
+  private static final Constructor<?> BROTLI_COMPRESSOR_CTOR; // BrotliCompressor()
+  private static final Object BROTLI_COMPRESS_PARAMETER; // Brotli.Parameter instance (quality=1)
+
   static {
     Class<?> tempClass = null;
     Method tempCreateMethod = null;
@@ -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);
+
+      // BrotliCompressor: no-arg ctor + compress(Parameter, ByteBuffer, ByteBuffer) -> int
+      Class<?> compClass = Class.forName("org.meteogroup.jbrotli.BrotliCompressor");
+      Class<?> paramClass = Class.forName("org.meteogroup.jbrotli.Brotli$Parameter");
+      Class<?> modeClass = Class.forName("org.meteogroup.jbrotli.Brotli$Mode");
+      brotliCompressorCtor = compClass.getConstructor();
+      brotliCompress = compClass.getMethod("compress", paramClass, ByteBuffer.class, ByteBuffer.class);
+
+      // Create Parameter(Mode.GENERIC, quality=1, lgwin=22, lgblock=0)
+      Object genericMode = modeClass.getField("GENERIC").get(null);
+      Constructor<?> paramCtor = paramClass.getConstructor(modeClass, int.class, int.class, int.class);
+      brotliParam = paramCtor.newInstance(genericMode, 1, 22, 0);
+
+      brotliLoaded = true;
+    } catch (Throwable t) {
+      LOG.debug("Brotli native library not available, falling back to Hadoop codec", t);
+    }
+    BROTLI_NATIVE_AVAILABLE = brotliLoaded;
+    BROTLI_DECOMPRESS_METHOD = brotliDecompress;
+    BROTLI_COMPRESS_METHOD = brotliCompress;
+    BROTLI_DECOMPRESSOR_CTOR = brotliDecompressorCtor;
+    BROTLI_COMPRESSOR_CTOR = brotliCompressorCtor;
+    BROTLI_COMPRESS_PARAMETER = brotliParam;
   }
 
   /**
@@ -105,6 +154,11 @@ protected BytesCompressor createCompressor(final CompressionCodecName codecName)
         return new ZstdCompressor();
       case LZ4_RAW:
         return new Lz4RawCompressor();
+      case BROTLI:
+        if (BROTLI_NATIVE_AVAILABLE) {
+          return new BrotliDirectCompressor();
+        }
+        return super.createCompressor(codecName);
       default:
         return super.createCompressor(codecName);
     }
@@ -119,6 +173,11 @@ protected BytesDecompressor createDecompressor(final CompressionCodecName codecN
         return new ZstdDecompressor();
       case LZ4_RAW:
         return new Lz4RawDecompressor();
+      case BROTLI:
+        if (BROTLI_NATIVE_AVAILABLE) {
+          return new BrotliDirectDecompressor();
+        }
+        // fall through to default Hadoop codec path
       case GZIP:
       case UNCOMPRESSED:
         return super.createDecompressor(codecName);
@@ -491,6 +550,83 @@ void closeCompressor() {
     }
   }
 
+  /**
+   * Direct-memory Brotli decompressor using jbrotli's native JNI bindings via reflection,
+   * bypassing the Hadoop BrotliCodec/stream wrapper overhead.
+   */
+  private class BrotliDirectDecompressor extends BaseDecompressor {
+    private final Object decompressor;
+
+    BrotliDirectDecompressor() {
+      try {
+        this.decompressor = BROTLI_DECOMPRESSOR_CTOR.newInstance();
+      } catch (ReflectiveOperationException e) {
+        throw new DirectCodecPool.ParquetCompressionCodecException("Failed to create Brotli decompressor", e);
+      }
+    }
+
+    @Override
+    int decompress(ByteBuffer input, ByteBuffer output) throws IOException {
+      try {
+        return (int) BROTLI_DECOMPRESS_METHOD.invoke(decompressor, input, output);
+      } catch (InvocationTargetException e) {
+        throw new IOException("Brotli decompression failed", e.getCause());
+      } catch (IllegalAccessException e) {
+        throw new IOException("Brotli decompression failed", e);
+      }
+    }
+
+    @Override
+    void closeDecompressor() {
+      // no-op: BrotliDeCompressor has no resources to release
+    }
+  }
+
+  /**
+   * Direct-memory Brotli compressor using jbrotli's native JNI bindings via reflection,
+   * bypassing the Hadoop BrotliCodec/stream wrapper overhead.
+   * Uses quality=1 by default (fast compression, matching Hadoop's BrotliCompressor default).
+   */
+  private class BrotliDirectCompressor extends BaseCompressor {
+    private final Object compressor;
+
+    BrotliDirectCompressor() {
+      try {
+        this.compressor = BROTLI_COMPRESSOR_CTOR.newInstance();
+      } catch (ReflectiveOperationException e) {
+        throw new DirectCodecPool.ParquetCompressionCodecException("Failed to create Brotli compressor", e);
+      }
+    }
+
+    @Override
+    public CompressionCodecName getCodecName() {
+      return CompressionCodecName.BROTLI;
+    }
+
+    @Override
+    int maxCompressedSize(int size) {
+      // Brotli worst case: input size + (input size >> 2) + 1K overhead for small inputs
+      // This is a conservative upper bound matching the Brotli spec
+      return size + (size >> 2) + 1024;
+    }
+
+    @Override
+    int compress(ByteBuffer input, ByteBuffer output) throws IOException {
+      try {
+        return (int) BROTLI_COMPRESS_METHOD.invoke(compressor, BROTLI_COMPRESS_PARAMETER, input, output);
+      } catch (InvocationTargetException e) {
+        throw new IOException("Brotli compression failed", e.getCause());
+      } catch (IllegalAccessException e) {
+        throw new IOException("Brotli compression failed", e);
+      }
+    }
+
+    @Override
+    void closeCompressor() {
+      // no-op: BrotliCompressor has no resources to release
+    }
+  }
+
   /**
    * @deprecated Use {@link CodecFactory#NO_OP_COMPRESSOR} instead
    */