HBASE-30395 Reduce heap buffer copy in decompression

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultDecodingContext.java

@@ -110,17 +110,15 @@ 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) {
             compression.returnDecompressor(decompressor);
           }
         }
       } else {
-        BlockIOUtils.readFullyWithHeapBuffer(dataInputStream, blockBufferWithoutHeader,
-          onDiskSizeWithoutHeader);
+        BlockIOUtils.readFully(dataInputStream, blockBufferWithoutHeader, onDiskSizeWithoutHeader);
       }
     } finally {
       byteBuffInputStream.close();

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).

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;
+    }
+  }
 }