GH-3522: Reuse intermediate buffers in RunLengthBitPackingHybridDecoder PACKED path (~22% throughput on dictionary-id decode)

[iemejia]

Summary

Closes #3522.

RunLengthBitPackingHybridDecoder allocates a new int[] and byte[] on every PACKED run during decode. The code itself flagged this with a // TODO: reuse a buffer comment. This PR resolves the TODO by reusing the buffers across runs within the same decoder instance, growing them lazily only when a larger run is encountered.

Also adds a currentBufferLength field to track the logical active-region length in packedValuesBuffer (since packedValuesBuffer.length may now exceed the current run's size after a prior larger run grew it).

Benchmark

RleDictionaryIndexDecodingBenchmark (added in #3512) isolates the RLE/bit-packed dictionary-id decode path. 100k INT32 dictionary IDs, BIT_WIDTH=10, JMH -wi 5 -i 10 -f 2 (20 measurement iterations):

Pattern	master (ops/s)	optimized (ops/s)	Improvement
SEQUENTIAL	93,061,521	113,856,860	+22.3%
RANDOM	92,929,824	114,238,638	+22.9%
LOW_CARDINALITY	92,813,229	115,271,347	+24.2%

End-to-end FileReadBenchmark sees a much smaller ~2% improvement because RLE decoding is only one of many pipeline stages; the isolated micro-benchmark shows the true magnitude on the affected code path.

Validation

• parquet-column: 573 tests pass
• TestRunLengthBitPackingHybridEncoder: 9 tests pass (these round-trip values through the decoder)
• Built with -Dspotless.check.skip=true -Drat.skip=true -Djapicmp.skip=true

Scope

17 LOC change to a single file. Self-contained and obviously correct (resolves the existing TODO).

Related

Part of the focused performance PR series from https://github.com/iemejia/parquet-perf. The companion ByteStreamSplit writer/reader changes from the same source commit (ba52f82c3) have already been submitted as #3504 and #3506.

How to reproduce

The benchmark is added in #3512. Once that lands, reproduce with:

./mvnw clean package -pl parquet-benchmarks -DskipTests \
    -Dspotless.check.skip=true -Drat.skip=true -Djapicmp.skip=true
java -jar parquet-benchmarks/target/parquet-benchmarks.jar \
    'RleDictionaryIndexDecodingBenchmark' -wi 5 -i 10 -f 2

[arouel]

Have you seen the PR #3467 which addresses the same issue? I would prefer if we help each other in reviewing instead of creating more PRs of the same kind.

[Fokko]

Thanks for fixing this outstanding TODO @iemejia 🙌

I left one comment because I think it would good to update to the non-deprecated API of parquet-encoding while at it. WDYT?

[Fokko]

I think it is on purpose to not close this resource, should we add a comment to it?

[Fokko]

I noticed that unpack8Values(final byte[] input, ...) is deprecated, should we switch to the ByteBuffer overload while we're reworking this?

diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java b/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java
index da9db00c7..14ec1a87d 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java
@@ -21,6 +21,7 @@ package org.apache.parquet.column.values.rle;
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.bytes.BytesUtils;
 import org.apache.parquet.column.values.bitpacking.BytePacker;
@@ -52,7 +53,7 @@ public class RunLengthBitPackingHybridDecoder {
 
   // Reusable buffers to avoid per-run allocation in PACKED mode
   private int[] packedValuesBuffer = new int[0];
-  private byte[] packedBytesBuffer = new byte[0];
+  private ByteBuffer packedBytesBuffer = ByteBuffer.allocate(0);
 
   public RunLengthBitPackingHybridDecoder(int bitWidth, InputStream in) {
     LOG.debug("decoding bitWidth {}", bitWidth);
@@ -102,13 +103,13 @@ public class RunLengthBitPackingHybridDecoder {
         }
         currentBuffer = packedValuesBuffer;
         int bytesRequired = numGroups * bitWidth;
-        if (packedBytesBuffer.length < bytesRequired) {
-          packedBytesBuffer = new byte[bytesRequired];
+        if (packedBytesBuffer.capacity() < bytesRequired) {
+          packedBytesBuffer = ByteBuffer.allocate(bytesRequired);
         }
         // At the end of the file RLE data though, there might not be that many bytes left.
         int bytesToRead = (int) Math.ceil(currentCount * bitWidth / 8.0);
         bytesToRead = Math.min(bytesToRead, in.available());
-        new DataInputStream(in).readFully(packedBytesBuffer, 0, bytesToRead);
+        new DataInputStream(in).readFully(packedBytesBuffer.array(), 0, bytesToRead);
         for (int valueIndex = 0, byteIndex = 0;
             valueIndex < currentCount;
             valueIndex += 8, byteIndex += bitWidth) {