GH-3466: Improve RunLengthBitPackingHybridDecoder.readNext to avoid per-call buffer allocation and DataInputStream wrapping

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

@@ -123,11 +123,15 @@ private void readNext() throws IOException {
         LOG.debug("reading {} values BIT PACKED", currentCount);
         if (currentBuffer.length < currentCount) {
           currentBuffer = new int[currentCount];
+        } else {
+          Arrays.fill(currentBuffer, currentCount, currentBuffer.length, 0);
         }
         currentBufferPosition = 0;
         int bytesRequired = numGroups * bitWidth;
         if (packedBytes.length < bytesRequired) {
           packedBytes = new byte[bytesRequired];
+        } else {
+          Arrays.fill(packedBytes, bytesRequired, packedBytes.length, (byte) 0);
         }
         // 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);

parquet-column/src/test/java/org/apache/parquet/column/values/rle/TestRunLengthBitPackingHybridEncoder.java

@@ -22,6 +22,7 @@
 
 import java.io.ByteArrayInputStream;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import org.apache.parquet.bytes.BytesUtils;
 import org.apache.parquet.bytes.DirectByteBufferAllocator;
@@ -352,6 +353,49 @@ public void testTruncatedPackedRunClearsReusedBytes() throws Exception {
     assertThat(decoder.readInt()).isEqualTo(0);
   }
 
+  @Test
+  public void testTruncatedPackedRunAfterFullPackedRunDoesNotReuseStaleBytes() throws Exception {
+    int bitWidth = 3;
+    BytePacker packer = Packer.LITTLE_ENDIAN.newBytePacker(bitWidth);
+
+    int[] firstRunValues = new int[8];
+    Arrays.fill(firstRunValues, 7);
+    byte[] firstRunPacked = new byte[bitWidth];
+    packer.pack8Values(firstRunValues, 0, firstRunPacked, 0);
+
+    int[] secondRunValues = {1, 2, 3, 4, 5, 6, 7, 0};
+    byte[] secondRunPacked = new byte[bitWidth];
+    packer.pack8Values(secondRunValues, 0, secondRunPacked, 0);
+
+    byte[] encoded = {
+      (byte) ((1 << 1) | 1),
+      firstRunPacked[0],
+      firstRunPacked[1],
+      firstRunPacked[2],
+      (byte) ((1 << 1) | 1),
+      secondRunPacked[0]
+    };
+
+    RunLengthBitPackingHybridDecoder decoder =
+        new RunLengthBitPackingHybridDecoder(bitWidth, new ByteArrayInputStream(encoded));
+
+    for (int ignored = 0; ignored < 8; ignored++) {
+      assertThat(decoder.readInt()).isEqualTo(7);
+    }
+
+    int[] actualSecondRun = new int[8];
+    for (int i = 0; i < 8; i++) {
+      actualSecondRun[i] = decoder.readInt();
+    }
+
+    byte[] expectedSecondPacked = new byte[bitWidth];
+    expectedSecondPacked[0] = secondRunPacked[0];
+    int[] expectedSecondRun = new int[8];
+    packer.unpack8Values(expectedSecondPacked, 0, expectedSecondRun, 0);
+
+    assertThat(actualSecondRun).containsExactly(expectedSecondRun);
+  }
+
   private static List<Integer> unpack(int bitWidth, int numValues, ByteArrayInputStream is) throws Exception {
 
     BytePacker packer = Packer.LITTLE_ENDIAN.newBytePacker(bitWidth);