apacheGH-3522: Use pack32Values fast path in RLE hybrid encoder

Buffer four 8-value groups (32 values) before packing, then use the
packer's pack32Values entry point instead of calling pack8Values four
separate times. This reduces per-group overhead and enables the packer's
optimized 32-value code path.

When fewer than 32 values remain at run end, flushBitPackedValues falls
back to per-8 packing via pack8Values.

All 573 parquet-column tests pass.

Author: iemejia

parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridEncoder.java

@@ -74,6 +74,11 @@ public class RunLengthBitPackingHybridEncoder implements AutoCloseable {
    */
   private final byte[] packBuffer;
 
+  /**
+   * Buffer four 8-value groups so we can use the packer's 32-value fast path.
+   */
+  private final int[] bitPackedValuesBuffer;
+
   /**
    * Previous value written, used to detect repeated values
    */
@@ -98,6 +103,8 @@ public class RunLengthBitPackingHybridEncoder implements AutoCloseable {
    */
   private int bitPackedGroupCount;
 
+  private int numBitPackedValues;
+
   /**
    * A "pointer" to a single byte in baos,
    * which we use as our bit-packed-header. It's really
@@ -125,7 +132,8 @@ public RunLengthBitPackingHybridEncoder(
 
     this.bitWidth = bitWidth;
     this.baos = new CapacityByteArrayOutputStream(initialCapacity, pageSize, allocator);
-    this.packBuffer = new byte[bitWidth];
+    this.packBuffer = new byte[bitWidth * 4];
+    this.bitPackedValuesBuffer = new int[32];
     this.bufferedValues = new int[8];
     this.packer = Packer.LITTLE_ENDIAN.newBytePacker(bitWidth);
     reset(false);
@@ -139,6 +147,7 @@ private void reset(boolean resetBaos) {
     this.numBufferedValues = 0;
     this.repeatCount = 0;
     this.bitPackedGroupCount = 0;
+    this.numBitPackedValues = 0;
     this.bitPackedRunHeaderPointer = -1;
     this.toBytesCalled = false;
   }
@@ -196,8 +205,9 @@ private void writeOrAppendBitPackedRun() throws IOException {
       bitPackedRunHeaderPointer = baos.getCurrentIndex();
     }
 
-    packer.pack8Values(bufferedValues, 0, packBuffer, 0);
-    baos.write(packBuffer);
+    System.arraycopy(bufferedValues, 0, bitPackedValuesBuffer, numBitPackedValues, 8);
+    numBitPackedValues += 8;
+    flushBitPackedValuesIfFull();
 
     // empty the buffer, they've all been written
     numBufferedValues = 0;
@@ -209,6 +219,34 @@ private void writeOrAppendBitPackedRun() throws IOException {
     ++bitPackedGroupCount;
   }
 
+  private void flushBitPackedValuesIfFull() {
+    if (numBitPackedValues == bitPackedValuesBuffer.length) {
+      packer.pack32Values(bitPackedValuesBuffer, 0, packBuffer, 0);
+      baos.write(packBuffer, 0, bitWidth * 4);
+      numBitPackedValues = 0;
+    }
+  }
+
+  private void flushBitPackedValues() {
+    if (numBitPackedValues == 0) {
+      return;
+    }
+
+    if (numBitPackedValues == bitPackedValuesBuffer.length) {
+      packer.pack32Values(bitPackedValuesBuffer, 0, packBuffer, 0);
+      baos.write(packBuffer, 0, bitWidth * 4);
+    } else {
+      int outPos = 0;
+      for (int inPos = 0; inPos < numBitPackedValues; inPos += 8) {
+        packer.pack8Values(bitPackedValuesBuffer, inPos, packBuffer, outPos);
+        outPos += bitWidth;
+      }
+      baos.write(packBuffer, 0, outPos);
+    }
+
+    numBitPackedValues = 0;
+  }
+
   /**
    * If we are currently writing a bit-packed-run, update the
    * bit-packed-header and consider this run to be over
@@ -221,6 +259,8 @@ private void endPreviousBitPackedRun() {
       return;
     }
 
+    flushBitPackedValues();
+
     // create bit-packed-header, which needs to fit in 1 byte
     byte bitPackHeader = (byte) ((bitPackedGroupCount << 1) | 1);