diff --git a/core/src/main/java/org/apache/iceberg/io/SingleBufferInputStream.java b/core/src/main/java/org/apache/iceberg/io/SingleBufferInputStream.java index 50431faf7e95..20c0701dc3f2 100644 --- a/core/src/main/java/org/apache/iceberg/io/SingleBufferInputStream.java +++ b/core/src/main/java/org/apache/iceberg/io/SingleBufferInputStream.java @@ -115,6 +115,14 @@ public long skip(long len) { @Override public int read(ByteBuffer out) { + if (!out.hasRemaining()) { + return 0; + } + + if (!buffer.hasRemaining()) { + return -1; + } + int bytesToCopy; ByteBuffer copyBuffer; if (buffer.remaining() <= out.remaining()) { @@ -130,7 +138,6 @@ public int read(ByteBuffer out) { } out.put(copyBuffer); - out.flip(); return bytesToCopy; } diff --git a/core/src/test/java/org/apache/iceberg/io/TestByteBufferInputStreams.java b/core/src/test/java/org/apache/iceberg/io/TestByteBufferInputStreams.java index fd2bd6a9d3a5..0061e0abd156 100644 --- a/core/src/test/java/org/apache/iceberg/io/TestByteBufferInputStreams.java +++ b/core/src/test/java/org/apache/iceberg/io/TestByteBufferInputStreams.java @@ -172,6 +172,32 @@ public void testReadByte() throws Exception { checkOriginalData(); } + @Test + public void testReadByteBuffer() throws Exception { + ByteBufferInputStream stream = newStream(); + ByteBuffer destination = ByteBuffer.allocate(12); + destination.position(2); + destination.limit(10); + + assertThat(stream.read(destination)).as("Should fill the destination").isEqualTo(8); + assertThat(destination.position()).as("Should advance the destination position").isEqualTo(10); + assertThat(destination.limit()).as("Should preserve the destination limit").isEqualTo(10); + + destination.flip(); + destination.position(2); + for (int i = 0; i < 8; i += 1) { + assertThat(destination.get()).as("Byte i should be i").isEqualTo((byte) i); + } + + ByteBuffer full = ByteBuffer.allocate(0); + assertThat(stream.read(full)).as("Should read 0 bytes into a full destination").isEqualTo(0); + + stream.skipFully(stream.available()); + assertThat(stream.read(ByteBuffer.allocate(1))).as("Should return -1 at EOF").isEqualTo(-1); + + checkOriginalData(); + } + @Test @SuppressWarnings("LocalVariableName") public void testSlice() throws Exception {