diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultDecodingContext.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultDecodingContext.java index 81f8e5fa6a24..fbee2bb4d75e 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultDecodingContext.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultDecodingContext.java @@ -110,8 +110,7 @@ public void prepareDecoding(int onDiskSizeWithoutHeader, int uncompressedSizeWit } try (InputStream is = compression.createDecompressionStream(dataInputStream, decompressor, 0)) { - BlockIOUtils.readFullyWithHeapBuffer(is, blockBufferWithoutHeader, - uncompressedSizeWithoutHeader); + BlockIOUtils.readFully(is, blockBufferWithoutHeader, uncompressedSizeWithoutHeader); } } finally { if (decompressor != null) { @@ -119,8 +118,7 @@ public void prepareDecoding(int onDiskSizeWithoutHeader, int uncompressedSizeWit } } } else { - BlockIOUtils.readFullyWithHeapBuffer(dataInputStream, blockBufferWithoutHeader, - onDiskSizeWithoutHeader); + BlockIOUtils.readFully(dataInputStream, blockBufferWithoutHeader, onDiskSizeWithoutHeader); } } finally { byteBuffInputStream.close(); diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java index bf737b2e1246..67385db878d3 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/BlockIOUtils.java @@ -19,7 +19,6 @@ import static org.apache.hadoop.hbase.trace.HBaseSemanticAttributes.DIRECT_BYTES_READ_KEY; import static org.apache.hadoop.hbase.trace.HBaseSemanticAttributes.HEAP_BYTES_READ_KEY; - import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.common.AttributesBuilder; import io.opentelemetry.api.trace.Span; @@ -128,6 +127,28 @@ public static void readFully(ByteBuff buf, FSDataInputStream dis, int length) th } } + /** + * Reads fully into the destination and advances the destination position by length. + * @param in the input stream to read from + * @param out the destination {@link ByteBuff} + * @param length bytes to read + * @throws IOException if any IO error is encountered + */ + public static void readFully(InputStream in, ByteBuff out, int length) throws IOException { + if (length < 0) { + throw new IllegalArgumentException("Length must not be negative: " + length); + } + if (out.hasArray()) { + int position = out.position(); + Span span = Span.current(); + IOUtils.readFully(in, out.array(), out.arrayOffset() + position, length); + span.addEvent("BlockIOUtils.readFully", getHeapBytesReadAttributes(span, length)); + out.position(position + length); + } else { + readFullyWithHeapBuffer(in, out, length); + } + } + /** * Copying bytes from InputStream to {@link ByteBuff} by using an temporary heap byte[] (default * size is 1024 now). diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java index 795848761840..9357ca457fdc 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestBlockIOUtils.java @@ -38,15 +38,16 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; - import io.opentelemetry.api.trace.Span; import io.opentelemetry.context.Scope; import io.opentelemetry.sdk.testing.junit5.OpenTelemetryExtension; import io.opentelemetry.sdk.trace.data.SpanData; +import java.io.ByteArrayInputStream; import java.io.DataOutputStream; import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Random; import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; @@ -146,6 +147,41 @@ public void testPreadWithoutReadFullBytes() throws IOException { testPreadReadFullBytesInternal(false, EnvironmentEdgeManager.currentTime()); } + @Test + public void testReadFullyFromInputStreamUsesHeapByteBuffArray() throws IOException { + byte[] expected = Bytes.toBytes("hello world"); + byte[] destinationArray = new byte[expected.length + 7]; + ByteBuff destination = new SingleByteBuff(ByteBuffer.wrap(destinationArray)); + destination.position(7); + CountingInputStream in = new CountingInputStream(expected); + + BlockIOUtils.readFully(in, destination, expected.length); + + assertTrue(in.wasReadInto(destinationArray)); + assertEquals(1, in.getReadCount()); + assertEquals(expected.length, in.getMaxReadLength()); + assertEquals(7 + expected.length, destination.position()); + assertArrayEquals(expected, Arrays.copyOfRange(destinationArray, 7, + 7 + expected.length)); + } + + @Test + public void testReadFullyFromInputStreamUsesHeapBufferForDirectByteBuff() throws IOException { + byte[] expected = createData(3 * 1024 + 17); + ByteBuff destination = new SingleByteBuff(ByteBuffer.allocateDirect(expected.length)); + CountingInputStream in = new CountingInputStream(expected); + + BlockIOUtils.readFully(in, destination, expected.length); + + byte[] actual = new byte[expected.length]; + destination.rewind(); + destination.get(actual); + assertEquals(4, in.getReadCount()); + assertEquals(1024, in.getMaxReadLength()); + assertEquals(expected.length, destination.position()); + assertArrayEquals(expected, actual); + } + private void testPreadReadFullBytesInternal(boolean readAllBytes, long randomSeed) throws IOException { Configuration conf = TEST_UTIL.getConfiguration(); @@ -555,4 +591,45 @@ public void testByteBufferPositionedReadableEOF() throws IOException { verify(in).hasCapability(anyString()); verifyNoMoreInteractions(in); } + + private static byte[] createData(int length) { + byte[] data = new byte[length]; + for (int i = 0; i < data.length; i++) { + data[i] = (byte) (i * 31); + } + return data; + } + + private static final class CountingInputStream extends ByteArrayInputStream { + private int readCount; + private int maxReadLength; + private byte[] lastReadBuffer; + + private CountingInputStream(byte[] data) { + super(data); + } + + @Override + public synchronized int read(byte[] b, int off, int len) { + int bytesRead = super.read(b, off, len); + if (bytesRead > 0) { + readCount++; + maxReadLength = Math.max(maxReadLength, len); + lastReadBuffer = b; + } + return bytesRead; + } + + private int getReadCount() { + return readCount; + } + + private int getMaxReadLength() { + return maxReadLength; + } + + private boolean wasReadInto(byte[] buffer) { + return lastReadBuffer == buffer; + } + } }