apacheGH-3522: Use pack32Values and cache packers in delta writers

Delta binary packing writers encoded each 32-value mini block with four
pack8Values calls despite the generated packers exposing pack32Values.
Switch both int and long writers to the 32-value entry point so each
mini block is packed in a single call.

Also cache BytePacker/BytePackerForLong instances in per-instance arrays
indexed by bit width, avoiding the factory dispatch per miniblock flush.
This resolves the TODO at DeltaBinaryPackingValuesWriterForLong.java:119.

Combined benchmark for both par12 commits (decode batch unpack32 + encode
pack32Values + cached packers):

IntEncodingBenchmark (100k INT32 values, JMH -wi 3 -i 5 -f 1):

  Benchmark     Pattern           Before (ops/s)   After (ops/s)   Improvement
  decodeDelta   SEQUENTIAL          903,892,292   1,096,895,285    +21% (1.21x)
  decodeDelta   RANDOM              364,659,977     410,632,530    +13% (1.13x)
  decodeDelta   LOW_CARDINALITY     581,649,861     676,449,008    +16% (1.16x)
  decodeDelta   HIGH_CARDINALITY    370,718,831     506,116,434    +37% (1.37x)
  encodeDelta   SEQUENTIAL          556,155,088     558,868,426    flat
  encodeDelta   RANDOM              360,327,834     376,594,239    +5%
  encodeDelta   LOW_CARDINALITY     412,396,181     434,569,306    +5%
  encodeDelta   HIGH_CARDINALITY    335,702,852     345,528,410    +3%

The decode path shows the larger gains because it eliminates per-miniblock
ByteBuffer.slice() allocations and uses the faster byte[] unpack path.
Encode gains are modest because pack32Values is structurally similar to 4x
pack8Values at this optimization level.

All 573 parquet-column tests pass.

Author: iemejia

parquet-column/src/main/java/org/apache/parquet/column/values/delta/DeltaBinaryPackingValuesWriterForInteger.java

@@ -60,6 +60,12 @@ public class DeltaBinaryPackingValuesWriterForInteger extends DeltaBinaryPacking
    */
   private int minDeltaInCurrentBlock = Integer.MAX_VALUE;
 
+  /**
+   * Cache of BytePacker instances indexed by bit width [0, 32].
+   * Avoids a factory dispatch + array load per miniblock flush.
+   */
+  private final BytePacker[] packerCache = new BytePacker[MAX_BITWIDTH + 1];
+
   public DeltaBinaryPackingValuesWriterForInteger(int slabSize, int pageSize, ByteBufferAllocator allocator) {
     this(DEFAULT_NUM_BLOCK_VALUES, DEFAULT_NUM_MINIBLOCKS, slabSize, pageSize, allocator);
   }
@@ -116,18 +122,16 @@ private void flushBlockBuffer() {
     for (int i = 0; i < miniBlocksToFlush; i++) {
       // writing i th miniblock
       int currentBitWidth = bitWidths[i];
-      int blockOffset = 0;
-      BytePacker packer = Packer.LITTLE_ENDIAN.newBytePacker(currentBitWidth);
-      int miniBlockStart = i * config.miniBlockSizeInValues;
-      for (int j = miniBlockStart; j < (i + 1) * config.miniBlockSizeInValues; j += 8) { // 8 values per pack
-        // mini block is atomic in terms of flushing
-        // This may write more values when reach to the end of data writing to last mini block,
-        // since it may not be aligned to miniblock,
-        // but doesn't matter. The reader uses total count to see if reached the end.
-        packer.pack8Values(deltaBlockBuffer, j, miniBlockByteBuffer, blockOffset);
-        blockOffset += currentBitWidth;
+      BytePacker packer = packerCache[currentBitWidth];
+      if (packer == null) {
+        packer = Packer.LITTLE_ENDIAN.newBytePacker(currentBitWidth);
+        packerCache[currentBitWidth] = packer;
       }
-      baos.write(miniBlockByteBuffer, 0, blockOffset);
+      int miniBlockStart = i * config.miniBlockSizeInValues;
+      // Mini blocks are always flushed as full 32-value groups in the current format.
+      // Use the packer's 32-value entry point to avoid four pack8Values calls per miniblock.
+      packer.pack32Values(deltaBlockBuffer, miniBlockStart, miniBlockByteBuffer, 0);
+      baos.write(miniBlockByteBuffer, 0, currentBitWidth * 4);
     }
 
     minDeltaInCurrentBlock = Integer.MAX_VALUE;

parquet-column/src/main/java/org/apache/parquet/column/values/delta/DeltaBinaryPackingValuesWriterForLong.java

@@ -60,6 +60,12 @@ public class DeltaBinaryPackingValuesWriterForLong extends DeltaBinaryPackingVal
    */
   private long minDeltaInCurrentBlock = Long.MAX_VALUE;
 
+  /**
+   * Cache of BytePackerForLong instances indexed by bit width [0, 64].
+   * Avoids a factory dispatch + array load per miniblock flush.
+   */
+  private final BytePackerForLong[] packerCache = new BytePackerForLong[MAX_BITWIDTH + 1];
+
   public DeltaBinaryPackingValuesWriterForLong(int slabSize, int pageSize, ByteBufferAllocator allocator) {
     this(DEFAULT_NUM_BLOCK_VALUES, DEFAULT_NUM_MINIBLOCKS, slabSize, pageSize, allocator);
   }
@@ -116,20 +122,16 @@ private void flushBlockBuffer() {
     for (int i = 0; i < miniBlocksToFlush; i++) {
       // writing i th miniblock
       int currentBitWidth = bitWidths[i];
-      int blockOffset = 0;
-      // TODO: should this cache the packer?
-      BytePackerForLong packer = Packer.LITTLE_ENDIAN.newBytePackerForLong(currentBitWidth);
-      int miniBlockStart = i * config.miniBlockSizeInValues;
-      // pack values into the miniblock buffer, 8 at a time to get exactly currentBitWidth bytes
-      for (int j = miniBlockStart; j < (i + 1) * config.miniBlockSizeInValues; j += 8) {
-        // mini block is atomic in terms of flushing
-        // This may write more values when reach to the end of data writing to last mini block,
-        // since it may not be aligned to miniblock,
-        // but doesn't matter. The reader uses total count to see if reached the end.
-        packer.pack8Values(deltaBlockBuffer, j, miniBlockByteBuffer, blockOffset);
-        blockOffset += currentBitWidth;
+      BytePackerForLong packer = packerCache[currentBitWidth];
+      if (packer == null) {
+        packer = Packer.LITTLE_ENDIAN.newBytePackerForLong(currentBitWidth);
+        packerCache[currentBitWidth] = packer;
       }
-      baos.write(miniBlockByteBuffer, 0, blockOffset);
+      int miniBlockStart = i * config.miniBlockSizeInValues;
+      // Mini blocks are always flushed as full 32-value groups in the current format.
+      // Use the packer's 32-value entry point to avoid four pack8Values calls per miniblock.
+      packer.pack32Values(deltaBlockBuffer, miniBlockStart, miniBlockByteBuffer, 0);
+      baos.write(miniBlockByteBuffer, 0, currentBitWidth * 4);
     }
 
     minDeltaInCurrentBlock = Long.MAX_VALUE;