GH-3716: Fix data corruption in ByteBufferBackedBinary.getBytes() for non-array-backed buffers

parquet-column/src/main/java/org/apache/parquet/io/api/Binary.java

@@ -442,16 +442,14 @@ public String toStringUsingUTF8() {
       if (value.hasArray()) {
         ret = new String(value.array(), value.arrayOffset() + offset, length, StandardCharsets.UTF_8);
       } else {
-        int limit = value.limit();
-        value.limit(offset + length);
-        int position = value.position();
-        value.position(offset);
-        // no corresponding interface to read a subset of a buffer, would have to slice it
-        // which creates another ByteBuffer object or do what is done here to adjust the
-        // limit/offset and set them back after
-        ret = StandardCharsets.UTF_8.decode(value).toString();
-        value.limit(limit);
-        value.position(position);
+        // Duplicate before adjusting position/limit so we never mutate the shared
+        // buffer's own position: readBytes() may have already advanced it past
+        // this value's range (e.g. lazily-consumed values in a repeated field),
+        // and limit(offset + length) would otherwise silently clamp it backwards.
+        ByteBuffer duplicate = value.duplicate();
+        duplicate.position(offset);
+        duplicate.limit(offset + length);
+        ret = StandardCharsets.UTF_8.decode(duplicate).toString();
       }
 
       return ret;
@@ -475,13 +473,18 @@ public void writeTo(OutputStream out) throws IOException {
     public byte[] getBytes() {
       byte[] bytes = new byte[length];
 
-      int limit = value.limit();
-      value.limit(offset + length);
-      int position = value.position();
-      value.position(offset);
-      value.get(bytes);
-      value.limit(limit);
-      value.position(position);
+      if (value.hasArray()) {
+        System.arraycopy(value.array(), value.arrayOffset() + offset, bytes, 0, length);
+      } else {
+        // Duplicate before adjusting position/limit so we never mutate the shared
+        // buffer's own position: readBytes() may have already advanced it past
+        // this value's range (e.g. lazily-consumed values in a repeated field),
+        // and limit(offset + length) would otherwise silently clamp it backwards.
+        ByteBuffer duplicate = value.duplicate();
+        duplicate.position(offset);
+        duplicate.limit(offset + length);
+        duplicate.get(bytes);
+      }
       if (!isBackingBytesReused) { // backing buffer might change
         cachedBytes = bytes;
       }

parquet-column/src/test/java/org/apache/parquet/column/values/plain/TestFixedLenByteArrayPlainValuesWriterReader.java

@@ -160,4 +160,64 @@ public void testEmptyPage() throws IOException {
       // Should not throw
     }
   }
+
+  // ---- Lazy getBytes() must not corrupt the shared page buffer's live position ----
+
+  // Regression test: getBytes() on an old value used to shift the buffer's shared
+  // read position backwards, so the next readBytes() call returned stale data.
+  @Test
+  public void testLazyGetBytesDoesNotCorruptSubsequentReadsDirectBuffer() throws IOException {
+    Binary[] expected = {fixedBinary(0), fixedBinary(50), fixedBinary(100), fixedBinary(150)};
+    ByteBuffer direct = writeToDirectBuffer(expected);
+
+    FixedLenByteArrayPlainValuesReader reader = new FixedLenByteArrayPlainValuesReader(FIXED_LEN);
+    reader.initFromPage(expected.length, ByteBufferInputStream.wrap(direct));
+
+    // "row 0": read two values, then materialize them lazily -- getBytes() on the
+    // first value happens after the shared buffer's position has already moved past it.
+    Binary row0v0 = reader.readBytes();
+    Binary row0v1 = reader.readBytes();
+    assertThat(row0v0.getBytes()).as("row0 value 0").isEqualTo(expected[0].getBytes());
+    assertThat(row0v1.getBytes()).as("row0 value 1").isEqualTo(expected[1].getBytes());
+
+    // "row 1": the corruption from materializing row 0 above must not affect this.
+    Binary row1v0 = reader.readBytes();
+    Binary row1v1 = reader.readBytes();
+    assertThat(row1v0.getBytes()).as("row1 value 0").isEqualTo(expected[2].getBytes());
+    assertThat(row1v1.getBytes()).as("row1 value 1").isEqualTo(expected[3].getBytes());
+  }
+
+  // Same scenario using toStringUsingUTF8(), which shares the buggy non-array-backed
+  // path with getBytes().
+  @Test
+  public void testLazyToStringUsingUTF8DoesNotCorruptSubsequentReadsDirectBuffer() throws IOException {
+    Binary[] expected = {fixedBinary(0), fixedBinary(50), fixedBinary(100), fixedBinary(150)};
+    ByteBuffer direct = writeToDirectBuffer(expected);
+
+    FixedLenByteArrayPlainValuesReader reader = new FixedLenByteArrayPlainValuesReader(FIXED_LEN);
+    reader.initFromPage(expected.length, ByteBufferInputStream.wrap(direct));
+
+    Binary row0v0 = reader.readBytes();
+    Binary row0v1 = reader.readBytes();
+    assertThat(row0v0.toStringUsingUTF8()).as("row0 value 0").isEqualTo(expected[0].toStringUsingUTF8());
+    assertThat(row0v1.toStringUsingUTF8()).as("row0 value 1").isEqualTo(expected[1].toStringUsingUTF8());
+
+    Binary row1v0 = reader.readBytes();
+    Binary row1v1 = reader.readBytes();
+    assertThat(row1v0.getBytes()).as("row1 value 0").isEqualTo(expected[2].getBytes());
+    assertThat(row1v1.getBytes()).as("row1 value 1").isEqualTo(expected[3].getBytes());
+  }
+
+  private ByteBuffer writeToDirectBuffer(Binary[] values) throws IOException {
+    try (FixedLenByteArrayPlainValuesWriter writer = newWriter()) {
+      for (Binary v : values) {
+        writer.writeBytes(v);
+      }
+      byte[] pageBytes = writer.getBytes().toByteArray();
+      ByteBuffer direct = ByteBuffer.allocateDirect(pageBytes.length);
+      direct.put(pageBytes);
+      direct.flip();
+      return direct;
+    }
+  }
 }