GH-3466 Fix stale packed-byte tail corruption in RLE/bit-pack decoder

Author: arouel

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

@@ -18,9 +18,9 @@
  */
 package org.apache.parquet.column.values.rle;
 
-import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Arrays;
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.bytes.BytesUtils;
 import org.apache.parquet.column.values.bitpacking.BytePacker;
@@ -101,10 +101,10 @@ private void readNext() throws IOException {
         if (packedBytes == null || packedBytes.length < bytesNeeded) {
           packedBytes = new byte[bytesNeeded];
         }
-        // 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());
-        readFully(in, packedBytes, bytesToRead);
+        int bytesRead = in.readNBytes(packedBytes, 0, bytesNeeded);
+        if (bytesRead < bytesNeeded) {
+          Arrays.fill(packedBytes, bytesRead, bytesNeeded, (byte) 0);
+        }
         for (int valueIndex = 0, byteIndex = 0;
             valueIndex < currentCount;
             valueIndex += 8, byteIndex += bitWidth) {
@@ -116,15 +116,4 @@ private void readNext() throws IOException {
     }
   }
 
-  private static void readFully(InputStream in, byte[] buf, int len) throws IOException {
-    int offset = 0;
-    while (offset < len) {
-      int read = in.read(buf, offset, len - offset);
-      if (read < 0) {
-        throw new EOFException(
-            "Unexpected end of stream: still needed " + (len - offset) + " bytes");
-      }
-      offset += read;
-    }
-  }
 }

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

@@ -19,9 +19,11 @@
 package org.apache.parquet.column.values.rle;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertArrayEquals;
 
 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;
@@ -298,6 +300,49 @@ public void testGroupBoundary() throws Exception {
     assertEquals(stream.available(), 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++) {
+      assertEquals(7, decoder.readInt());
+    }
+
+    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);
+
+    assertArrayEquals(expectedSecondRun, actualSecondRun);
+  }
+
   private static List<Integer> unpack(int bitWidth, int numValues, ByteArrayInputStream is) throws Exception {
 
     BytePacker packer = Packer.LITTLE_ENDIAN.newBytePacker(bitWidth);