HBASE-29398: Server side scan metrics for bytes read from FS vs Block cache vs memstore (#7163) (#7136)

Signed-off-by: Viraj Jasani <vjasani@apache.org>
Signed-off-by: Hari Krishna Dara <haridara@gmail.com>

Author: sanjeet006py

hbase-client/src/main/java/org/apache/hadoop/hbase/client/metrics/ServerSideScanMetrics.java

@@ -49,6 +49,10 @@ public void moveToNextRegion() {
     regionScanMetricsData.add(currentRegionScanMetricsData);
     currentRegionScanMetricsData.createCounter(COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME);
     currentRegionScanMetricsData.createCounter(COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(BYTES_READ_FROM_FS_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(BYTES_READ_FROM_MEMSTORE_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(BLOCK_READ_OPS_COUNT_METRIC_NAME);
   }
 
   /**
@@ -62,6 +66,12 @@ protected AtomicLong createCounter(String counterName) {
   public static final String COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME = "ROWS_SCANNED";
   public static final String COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME = "ROWS_FILTERED";
 
+  public static final String BYTES_READ_FROM_FS_METRIC_NAME = "BYTES_READ_FROM_FS";
+  public static final String BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME =
+    "BYTES_READ_FROM_BLOCK_CACHE";
+  public static final String BYTES_READ_FROM_MEMSTORE_METRIC_NAME = "BYTES_READ_FROM_MEMSTORE";
+  public static final String BLOCK_READ_OPS_COUNT_METRIC_NAME = "BLOCK_READ_OPS_COUNT";
+
   /**
    * @deprecated As of release 2.0.0, this will be removed in HBase 3.0.0
    *             (<a href="https://issues.apache.org/jira/browse/HBASE-17886">HBASE-17886</a>). Use
@@ -90,6 +100,16 @@ protected AtomicLong createCounter(String counterName) {
    */
   public final AtomicLong countOfRowsScanned = createCounter(COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME);
 
+  public final AtomicLong bytesReadFromFs = createCounter(BYTES_READ_FROM_FS_METRIC_NAME);
+
+  public final AtomicLong bytesReadFromBlockCache =
+    createCounter(BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME);
+
+  public final AtomicLong bytesReadFromMemstore =
+    createCounter(BYTES_READ_FROM_MEMSTORE_METRIC_NAME);
+
+  public final AtomicLong blockReadOpsCount = createCounter(BLOCK_READ_OPS_COUNT_METRIC_NAME);
+
   /**
    * Sets counter with counterName to passed in value, does nothing if counter does not exist. If
    * region level scan metrics are enabled then sets the value of counter for the current region

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/CompoundBloomFilter.java

@@ -20,6 +20,7 @@
 import java.io.DataInput;
 import java.io.IOException;
 import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.regionserver.BloomType;
 import org.apache.hadoop.hbase.util.BloomFilter;
@@ -120,7 +121,8 @@ private boolean containsInternal(byte[] key, int keyOffset, int keyLength, ByteB
     return result;
   }
 
-  private HFileBlock getBloomBlock(int block) {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public HFileBlock getBloomBlock(int block) {
     HFileBlock bloomBlock;
     try {
       // We cache the block and use a positional read.
@@ -218,4 +220,18 @@ public String toString() {
     return sb.toString();
   }
 
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public HFileBlockIndex.BlockIndexReader getBloomIndex() {
+    return index;
+  }
+
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public int getHashCount() {
+    return hashCount;
+  }
+
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public Hash getHash() {
+    return hash;
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/FixedFileTrailer.java

@@ -27,9 +27,11 @@
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.hbase.CellComparator;
 import org.apache.hadoop.hbase.CellComparatorImpl;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.MetaCellComparator;
 import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.monitoring.ThreadLocalServerSideScanMetrics;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -408,6 +410,11 @@ public static FixedFileTrailer readFromStream(FSDataInputStream istream, long fi
     FixedFileTrailer fft = new FixedFileTrailer(majorVersion, minorVersion);
     fft.deserialize(new DataInputStream(new ByteArrayInputStream(buf.array(),
       buf.arrayOffset() + bufferSize - trailerSize, trailerSize)));
+    boolean isScanMetricsEnabled = ThreadLocalServerSideScanMetrics.isScanMetricsEnabled();
+    if (isScanMetricsEnabled) {
+      ThreadLocalServerSideScanMetrics.addBytesReadFromFs(trailerSize);
+      ThreadLocalServerSideScanMetrics.addBlockReadOpsCount(1);
+    }
     return fft;
   }
 
@@ -640,7 +647,8 @@ static CellComparator createComparator(String comparatorClassName) throws IOExce
     }
   }
 
-  CellComparator createComparator() throws IOException {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public CellComparator createComparator() throws IOException {
     expectAtLeastMajorVersion(2);
     return createComparator(comparatorClassName);
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java

@@ -40,6 +40,7 @@
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.fs.HFileSystem;
 import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
@@ -55,6 +56,7 @@
 import org.apache.hadoop.hbase.io.encoding.HFileBlockEncodingContext;
 import org.apache.hadoop.hbase.io.hfile.trace.HFileContextAttributesBuilderConsumer;
 import org.apache.hadoop.hbase.io.util.BlockIOUtils;
+import org.apache.hadoop.hbase.monitoring.ThreadLocalServerSideScanMetrics;
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.nio.MultiByteBuff;
 import org.apache.hadoop.hbase.nio.SingleByteBuff;
@@ -403,7 +405,8 @@ private static int getOnDiskSizeWithHeader(final ByteBuff headerBuf, boolean che
    *         present) read by peeking into the next block's header; use as a hint when doing a read
    *         of the next block when scanning or running over a file.
    */
-  int getNextBlockOnDiskSize() {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public int getNextBlockOnDiskSize() {
     return nextBlockOnDiskSize;
   }
 
@@ -468,7 +471,8 @@ int getOnDiskSizeWithoutHeader() {
   }
 
   /** Returns the uncompressed size of data part (header and checksum excluded). */
-  int getUncompressedSizeWithoutHeader() {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public int getUncompressedSizeWithoutHeader() {
     return uncompressedSizeWithoutHeader;
   }
 
@@ -624,7 +628,8 @@ public String toString() {
    * Retrieves the decompressed/decrypted view of this block. An encoded block remains in its
    * encoded structure. Internal structures are shared between instances where applicable.
    */
-  HFileBlock unpack(HFileContext fileContext, FSReader reader) throws IOException {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public HFileBlock unpack(HFileContext fileContext, FSReader reader) throws IOException {
     if (!fileContext.isCompressedOrEncrypted()) {
       // TODO: cannot use our own fileContext here because HFileBlock(ByteBuffer, boolean),
       // which is used for block serialization to L2 cache, does not preserve encoding and
@@ -695,7 +700,8 @@ public boolean isUnpacked() {
    * when block is returned to the cache.
    * @return the offset of this block in the file it was read from
    */
-  long getOffset() {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public long getOffset() {
     if (offset < 0) {
       throw new IllegalStateException("HFile block offset not initialized properly");
     }
@@ -1221,7 +1227,8 @@ interface BlockWritable {
    * Iterator for reading {@link HFileBlock}s in load-on-open-section, such as root data index
    * block, meta index block, file info block etc.
    */
-  interface BlockIterator {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public interface BlockIterator {
     /**
      * Get the next block, or null if there are no more blocks to iterate.
      */
@@ -1245,7 +1252,8 @@ interface BlockIterator {
   }
 
   /** An HFile block reader with iteration ability. */
-  interface FSReader {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public interface FSReader {
     /**
      * Reads the block at the given offset in the file with the given on-disk size and uncompressed
      * size.
@@ -1720,6 +1728,7 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
       // checksums. Can change with circumstances. The below flag is whether the
       // file has support for checksums (version 2+).
       boolean checksumSupport = this.fileContext.isUseHBaseChecksum();
+      boolean isScanMetricsEnabled = ThreadLocalServerSideScanMetrics.isScanMetricsEnabled();
       long startTime = EnvironmentEdgeManager.currentTime();
       if (onDiskSizeWithHeader == -1) {
         // The caller does not know the block size. Need to get it from the header. If header was
@@ -1736,6 +1745,9 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
           headerBuf = HEAP.allocate(hdrSize);
           readAtOffset(is, headerBuf, hdrSize, false, offset, pread);
           headerBuf.rewind();
+          if (isScanMetricsEnabled) {
+            ThreadLocalServerSideScanMetrics.addBytesReadFromFs(hdrSize);
+          }
         }
         onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf, checksumSupport);
       }
@@ -1783,6 +1795,12 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
         boolean readNextHeader = readAtOffset(is, onDiskBlock,
           onDiskSizeWithHeader - preReadHeaderSize, true, offset + preReadHeaderSize, pread);
         onDiskBlock.rewind(); // in case of moving position when copying a cached header
+        if (isScanMetricsEnabled) {
+          long bytesRead =
+            (onDiskSizeWithHeader - preReadHeaderSize) + (readNextHeader ? hdrSize : 0);
+          ThreadLocalServerSideScanMetrics.addBytesReadFromFs(bytesRead);
+          ThreadLocalServerSideScanMetrics.addBlockReadOpsCount(1);
+        }
 
         // the call to validateChecksum for this block excludes the next block header over-read, so
         // no reason to delay extracting this value.

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java

@@ -34,6 +34,7 @@
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellComparator;
 import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.KeyValue.KeyOnlyKeyValue;
 import org.apache.hadoop.hbase.PrivateCellUtil;
@@ -567,7 +568,8 @@ public String toString() {
    * array of offsets to the entries within the block. This allows us to do binary search for the
    * entry corresponding to the given key without having to deserialize the block.
    */
-  static abstract class BlockIndexReader implements HeapSize {
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public static abstract class BlockIndexReader implements HeapSize {
 
     protected long[] blockOffsets;
     protected int[] blockDataSizes;
@@ -814,7 +816,8 @@ static int binarySearchNonRootIndex(Cell key, ByteBuff nonRootIndex,
      * @return the index position where the given key was found, otherwise return -1 in the case the
      *         given key is before the first key.
      */
-    static int locateNonRootIndexEntry(ByteBuff nonRootBlock, Cell key, CellComparator comparator) {
+    public static int locateNonRootIndexEntry(ByteBuff nonRootBlock, Cell key,
+      CellComparator comparator) {
       int entryIndex = binarySearchNonRootIndex(key, nonRootBlock, comparator);
 
       if (entryIndex != -1) {