diff --git a/parquet-column/src/main/java/org/apache/parquet/io/api/Binary.java b/parquet-column/src/main/java/org/apache/parquet/io/api/Binary.java index 3160f091e7..58961256fc 100644 --- a/parquet-column/src/main/java/org/apache/parquet/io/api/Binary.java +++ b/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; } diff --git a/parquet-column/src/test/java/org/apache/parquet/column/values/plain/TestFixedLenByteArrayPlainValuesWriterReader.java b/parquet-column/src/test/java/org/apache/parquet/column/values/plain/TestFixedLenByteArrayPlainValuesWriterReader.java index d86eefbe7a..66db13c10b 100644 --- a/parquet-column/src/test/java/org/apache/parquet/column/values/plain/TestFixedLenByteArrayPlainValuesWriterReader.java +++ b/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; + } + } }