HBASE-29585 Add row-level cache for the get operation

hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java

@@ -316,4 +316,11 @@ default boolean matchReplicationScope(boolean enabled) {
     }
     return !enabled;
   }
+
+  /**
+   * Checks whether row caching is enabled for this table. Note that row caching applies only at the
+   * entire row level, not at the column family level.
+   * @return {@code true} if row cache is enabled, otherwise {@code false}
+   */
+  boolean isRowCacheEnabled();
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java

@@ -227,6 +227,15 @@ public class TableDescriptorBuilder {
   private final static Map<String, String> DEFAULT_VALUES = new HashMap<>();
   private final static Set<Bytes> RESERVED_KEYWORDS = new HashSet<>();
 
+  /**
+   * Used by HBase Shell interface to access this metadata attribute which denotes if the row cache
+   * is enabled.
+   */
+  @InterfaceAudience.Private
+  public static final String ROW_CACHE_ENABLED = "ROW_CACHE_ENABLED";
+  private static final Bytes ROW_CACHE_ENABLED_KEY = new Bytes(Bytes.toBytes(ROW_CACHE_ENABLED));
+  private static final boolean DEFAULT_ROW_CACHE_ENABLED = false;
+
   static {
     DEFAULT_VALUES.put(MAX_FILESIZE, String.valueOf(HConstants.DEFAULT_MAX_FILE_SIZE));
     DEFAULT_VALUES.put(READONLY, String.valueOf(DEFAULT_READONLY));
@@ -565,6 +574,11 @@ public TableDescriptor build() {
     return new ModifyableTableDescriptor(desc);
   }
 
+  public TableDescriptorBuilder setRowCacheEnabled(boolean rowCacheEnabled) {
+    desc.setRowCacheEnabled(rowCacheEnabled);
+    return this;
+  }
+
   private static final class ModifyableTableDescriptor
     implements TableDescriptor, Comparable<ModifyableTableDescriptor> {
 
@@ -1510,6 +1524,15 @@ public Optional<String> getRegionServerGroup() {
         return Optional.empty();
       }
     }
+
+    @Override
+    public boolean isRowCacheEnabled() {
+      return getOrDefault(ROW_CACHE_ENABLED_KEY, Boolean::valueOf, DEFAULT_ROW_CACHE_ENABLED);
+    }
+
+    public ModifyableTableDescriptor setRowCacheEnabled(boolean enabled) {
+      return setValue(ROW_CACHE_ENABLED_KEY, Boolean.toString(enabled));
+    }
   }
 
   /**

hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java

@@ -1033,6 +1033,14 @@ public enum OperationStatusCode {
   public static final long HBASE_CLIENT_SCANNER_ONHEAP_BLOCK_CACHE_FIXED_SIZE_DEFAULT =
     32 * 1024 * 1024L;
 
+  /**
+   * Configuration key for the minimum number of HFiles required to activate the Row Cache. If the
+   * number of HFiles is less than this value, the Row Cache does not operate even if it is enabled
+   * at the table level.
+   */
+  public static final String ROW_CACHE_ACTIVATE_MIN_HFILES_KEY = "row.cache.activate.min.hfiles";
+  public static final int ROW_CACHE_ACTIVATE_MIN_HFILES_DEFAULT = 2;
+
   /**
    * Configuration key for setting pread must read both necessaryLen and extraLen, default is
    * disabled. This is an optimized flag for reading HFile from blob storage.

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

@@ -547,8 +547,8 @@ public void setTimestamp(byte[] ts) throws IOException {
 
     @Override
     public ExtendedCell deepClone() {
-      // This is not used in actual flow. Throwing UnsupportedOperationException
-      throw new UnsupportedOperationException();
+      // To garbage collect the objects referenced by this cell, we need to deep clone it
+      return ExtendedCell.super.deepClone();
     }
   }
 

hbase-common/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockType.java

@@ -80,6 +80,14 @@ public int getId() {
   /** Fixed file trailer, both versions (always just a magic string) */
   TRAILER("TRABLK\"$", BlockCategory.META),
 
+  // Pseudo block
+
+  /**
+   * Cells of a row for row cache. This is a pseudo block type. It only exists to share the
+   * BlockCache interface.
+   */
+  ROW_CELLS("ROWCELLS", BlockCategory.ROW),
+
   // Legacy blocks
 
   /** Block index magic string in version 1 */
@@ -91,6 +99,7 @@ public enum BlockCategory {
     INDEX,
     BLOOM,
     ALL_CATEGORIES,
+    ROW,
     UNKNOWN;
 
     /**

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerSource.java

@@ -386,6 +386,7 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
   String BLOCK_CACHE_GENERAL_BLOOM_META_MISS_COUNT = "blockCacheGeneralBloomMetaMissCount";
   String BLOCK_CACHE_DELETE_FAMILY_BLOOM_MISS_COUNT = "blockCacheDeleteFamilyBloomMissCount";
   String BLOCK_CACHE_TRAILER_MISS_COUNT = "blockCacheTrailerMissCount";
+  String BLOCK_CACHE_ROW_MISS_COUNT = "blockCacheRowMissCount";
   String BLOCK_CACHE_DATA_HIT_COUNT = "blockCacheDataHitCount";
   String BLOCK_CACHE_ENCODED_DATA_HIT_COUNT = "blockCacheEncodedDataHitCount";
   String BLOCK_CACHE_LEAF_INDEX_HIT_COUNT = "blockCacheLeafIndexHitCount";
@@ -397,6 +398,7 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
   String BLOCK_CACHE_GENERAL_BLOOM_META_HIT_COUNT = "blockCacheGeneralBloomMetaHitCount";
   String BLOCK_CACHE_DELETE_FAMILY_BLOOM_HIT_COUNT = "blockCacheDeleteFamilyBloomHitCount";
   String BLOCK_CACHE_TRAILER_HIT_COUNT = "blockCacheTrailerHitCount";
+  String BLOCK_CACHE_ROW_HIT_COUNT = "blockCacheRowHitCount";
   String L1_CACHE_FREE_SIZE = "l1CacheFreeSize";
   String L1_CACHE_FREE_SIZE_DESC = "Amount of free bytes in the L1 cache";
   String L1_CACHE_SIZE = "l1CacheSize";

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerSourceImpl.java

@@ -436,6 +436,7 @@ public void getMetrics(MetricsCollector metricsCollector, boolean all) {
         .addCounter(Interns.info(BLOCK_CACHE_DELETE_FAMILY_BLOOM_MISS_COUNT, ""),
           rsWrap.getDeleteFamilyBloomMissCount())
         .addCounter(Interns.info(BLOCK_CACHE_TRAILER_MISS_COUNT, ""), rsWrap.getTrailerMissCount())
+        .addCounter(Interns.info(BLOCK_CACHE_ROW_MISS_COUNT, ""), rsWrap.getRowMissCount())
         .addCounter(Interns.info(BLOCK_CACHE_DATA_HIT_COUNT, ""), rsWrap.getDataHitCount())
         .addCounter(Interns.info(BLOCK_CACHE_LEAF_INDEX_HIT_COUNT, ""),
           rsWrap.getLeafIndexHitCount())
@@ -452,6 +453,7 @@ public void getMetrics(MetricsCollector metricsCollector, boolean all) {
         .addCounter(Interns.info(BLOCK_CACHE_DELETE_FAMILY_BLOOM_HIT_COUNT, ""),
           rsWrap.getDeleteFamilyBloomHitCount())
         .addCounter(Interns.info(BLOCK_CACHE_TRAILER_HIT_COUNT, ""), rsWrap.getTrailerHitCount())
+        .addCounter(Interns.info(BLOCK_CACHE_ROW_HIT_COUNT, ""), rsWrap.getRowHitCount())
         .addCounter(Interns.info(UPDATES_BLOCKED_TIME, UPDATES_BLOCKED_DESC),
           rsWrap.getUpdatesBlockedTime())
         .addCounter(Interns.info(FLUSHED_CELLS, FLUSHED_CELLS_DESC), rsWrap.getFlushedCellsCount())

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerWrapper.java

@@ -615,6 +615,8 @@ public interface MetricsRegionServerWrapper {
 
   long getTrailerMissCount();
 
+  long getRowMissCount();
+
   long getDataHitCount();
 
   long getLeafIndexHitCount();
@@ -635,6 +637,8 @@ public interface MetricsRegionServerWrapper {
 
   long getTrailerHitCount();
 
+  long getRowHitCount();
+
   long getTotalRowActionRequestCount();
 
   long getByteBuffAllocatorHeapAllocationBytes();

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

@@ -91,6 +91,7 @@ public class CacheStats {
   private final LongAdder generalBloomMetaMissCount = new LongAdder();
   private final LongAdder deleteFamilyBloomMissCount = new LongAdder();
   private final LongAdder trailerMissCount = new LongAdder();
+  private final LongAdder rowMissCount = new LongAdder();
 
   private final LongAdder dataHitCount = new LongAdder();
   private final LongAdder leafIndexHitCount = new LongAdder();
@@ -102,6 +103,7 @@ public class CacheStats {
   private final LongAdder generalBloomMetaHitCount = new LongAdder();
   private final LongAdder deleteFamilyBloomHitCount = new LongAdder();
   private final LongAdder trailerHitCount = new LongAdder();
+  private final LongAdder rowHitCount = new LongAdder();
 
   // Executor for periodic cache stats rolling
   private ScheduledExecutorService metricsRollerScheduler;
@@ -219,6 +221,9 @@ public void miss(boolean caching, boolean primary, BlockType type) {
       case TRAILER:
         trailerMissCount.increment();
         break;
+      case ROW_CELLS:
+        rowMissCount.increment();
+        break;
       default:
         // If there's a new type that's fine
         // Ignore it for now. This is metrics don't exception.
@@ -266,6 +271,9 @@ public void hit(boolean caching, boolean primary, BlockType type) {
       case TRAILER:
         trailerHitCount.increment();
         break;
+      case ROW_CELLS:
+        rowHitCount.increment();
+        break;
       default:
         // If there's a new type that's fine
         // Ignore it for now. This is metrics don't exception.
@@ -376,6 +384,14 @@ public long getTrailerHitCount() {
     return trailerHitCount.sum();
   }
 
+  public long getRowHitCount() {
+    return rowHitCount.sum();
+  }
+
+  public long getRowMissCount() {
+    return rowMissCount.sum();
+  }
+
   public long getRequestCount() {
     return getHitCount() + getMissCount();
   }

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

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.io.hfile;
+
+import java.util.Arrays;
+import java.util.Objects;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.ClassSize;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * Cache Key for use with implementations of {@link BlockCache}
+ */
+@InterfaceAudience.Private
+public class RowCacheKey extends BlockCacheKey {
+  private static final long serialVersionUID = -686874540957524887L;
+  public static final long FIXED_OVERHEAD = ClassSize.estimateBase(RowCacheKey.class, false);
+
+  private final byte[] rowKey;
+  // Row cache keys should not be evicted on close, since the cache may contain many entries and
+  // eviction would be slow. Instead, the region’s rowCacheSeqNum is used to generate new keys that
+  // ignore the existing cache when the region is reopened or bulk-loaded.
+  private final long rowCacheSeqNum;
+
+  public RowCacheKey(HRegion region, byte[] rowKey) {
+    super(region.getRegionInfo().getEncodedName(), 0, region.getRegionInfo().getReplicaId() == 0,
+      BlockType.ROW_CELLS);
+
+    this.rowKey = Objects.requireNonNull(rowKey, "rowKey cannot be null");
+    this.rowCacheSeqNum = region.getRowCacheSeqNum();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    if (!super.equals(o)) return false;
+    RowCacheKey that = (RowCacheKey) o;
+    return rowCacheSeqNum == that.rowCacheSeqNum && Arrays.equals(rowKey, that.rowKey);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(super.hashCode(), Arrays.hashCode(rowKey), Long.hashCode(rowCacheSeqNum));
+  }
+
+  @Override
+  public String toString() {
+    return super.toString() + '_' + Bytes.toStringBinary(this.rowKey) + '_' + rowCacheSeqNum;
+  }
+
+  @Override
+  public long heapSize() {
+    return FIXED_OVERHEAD + ClassSize.align(rowKey.length);
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -65,6 +65,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -433,6 +434,11 @@ public MetricsTableRequests getMetricsTableRequests() {
    */
   private long openSeqNum = HConstants.NO_SEQNUM;
 
+  /**
+   * Basically the same as openSeqNum, but it is updated when bulk load is done.
+   */
+  private final AtomicLong rowCacheSeqNum = new AtomicLong(HConstants.NO_SEQNUM);
+
   /**
    * The default setting for whether to enable on-demand CF loading for scan requests to this
    * region. Requests can override it.
@@ -7868,6 +7874,7 @@ private HRegion openHRegion(final CancelableProgressable reporter) throws IOExce
       LOG.debug("checking classloading for " + this.getRegionInfo().getEncodedName());
       TableDescriptorChecker.checkClassLoading(cConfig, htableDescriptor);
       this.openSeqNum = initialize(reporter);
+      this.rowCacheSeqNum.set(this.openSeqNum);
       this.mvcc.advanceTo(openSeqNum);
       // The openSeqNum must be increased every time when a region is assigned, as we rely on it to
       // determine whether a region has been successfully reopened. So here we always write open
@@ -8696,6 +8703,17 @@ public long getOpenSeqNum() {
     return this.openSeqNum;
   }
 
+  public long getRowCacheSeqNum() {
+    return this.rowCacheSeqNum.get();
+  }
+
+  /**
+   * This is used to invalidate the entire row cache after bulk loading.
+   */
+  public void increaseRowCacheSeqNum() {
+    this.rowCacheSeqNum.incrementAndGet();
+  }
+
   @Override
   public Map<byte[], Long> getMaxStoreSeqId() {
     return this.maxSeqIdInStores;

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -3408,6 +3408,9 @@ CacheEvictionStats clearRegionBlockCache(Region region) {
       }
     }
 
+    // evict the entire row cache
+    evictedBlocks += blockCache.evictBlocksByHfileName(region.getRegionInfo().getEncodedName());
+
     return CacheEvictionStats.builder().withEvictedBlocks(evictedBlocks).build();
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerWrapperImpl.java

@@ -1144,6 +1144,11 @@ public long getTrailerMissCount() {
     return this.cacheStats != null ? this.cacheStats.getTrailerMissCount() : 0L;
   }
 
+  @Override
+  public long getRowMissCount() {
+    return this.cacheStats != null ? this.cacheStats.getRowMissCount() : 0L;
+  }
+
   @Override
   public long getDataHitCount() {
     return this.cacheStats != null ? this.cacheStats.getDataHitCount() : 0L;
@@ -1194,6 +1199,11 @@ public long getTrailerHitCount() {
     return this.cacheStats != null ? this.cacheStats.getTrailerHitCount() : 0L;
   }
 
+  @Override
+  public long getRowHitCount() {
+    return this.cacheStats != null ? this.cacheStats.getRowHitCount() : 0L;
+  }
+
   @Override
   public long getByteBuffAllocatorHeapAllocationBytes() {
     return ByteBuffAllocator.getHeapAllocationBytes(allocator, ByteBuffAllocator.HEAP);