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,12 @@ default boolean matchReplicationScope(boolean enabled) {
     }
     return !enabled;
   }
+
+  /**
+   * Checks whether row caching is enabled for this table. Note that row caching is applied at the
+   * entire row level, not at the column family level.
+   * @return {@code true} if row caching is enabled, {@code false} if disabled, or {@code null} if
+   *         not explicitly set
+   */
+  Boolean getRowCacheEnabled();
 }

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));
@@ -236,6 +245,7 @@ public class TableDescriptorBuilder {
     DEFAULT_VALUES.put(PRIORITY, String.valueOf(DEFAULT_PRIORITY));
     // Setting ERASURE_CODING_POLICY to NULL so that it is not considered as metadata
     DEFAULT_VALUES.put(ERASURE_CODING_POLICY, String.valueOf(DEFAULT_ERASURE_CODING_POLICY));
+    DEFAULT_VALUES.put(ROW_CACHE_ENABLED, String.valueOf(DEFAULT_ROW_CACHE_ENABLED));
     DEFAULT_VALUES.keySet().stream().map(s -> new Bytes(Bytes.toBytes(s)))
       .forEach(RESERVED_KEYWORDS::add);
     RESERVED_KEYWORDS.add(IS_META_KEY);
@@ -565,6 +575,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 +1525,16 @@ public Optional<String> getRegionServerGroup() {
         return Optional.empty();
       }
     }
+
+    @Override
+    public Boolean getRowCacheEnabled() {
+      Bytes value = getValue(ROW_CACHE_ENABLED_KEY);
+      return value == null ? null : Boolean.valueOf(Bytes.toString(value.get()));
+    }
+
+    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

@@ -1017,6 +1017,24 @@ public enum OperationStatusCode {
 
   public static final float HFILE_BLOCK_CACHE_SIZE_DEFAULT = 0.4f;
 
+  /**
+   * Configuration key for the size of the row cache
+   */
+  public static final String ROW_CACHE_SIZE_KEY = "row.cache.size";
+  public static final float ROW_CACHE_SIZE_DEFAULT = 0.0f;
+
+  /**
+   * Configuration key for the evict the row cache on close
+   */
+  public static final String ROW_CACHE_EVICT_ON_CLOSE_KEY = "row.cache.evictOnClose";
+  public static final boolean ROW_CACHE_EVICT_ON_CLOSE_DEFAULT = false;
+
+  /**
+   * Configuration key for the row cache enabled
+   */
+  public static final String ROW_CACHE_ENABLED_KEY = "row.cache.enabled";
+  public static final boolean ROW_CACHE_ENABLED_DEFAULT = false;
+
   /**
    * Configuration key for the memory size of the block cache
    */

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();
     }
   }
 
@@ -796,8 +796,8 @@ public void write(ByteBuffer buf, int offset) {
 
     @Override
     public ExtendedCell deepClone() {
-      // This is not used in actual flow. Throwing UnsupportedOperationException
-      throw new UnsupportedOperationException();
+      // To cache row, we need to deep clone it
+      return super.deepClone();
     }
   }
 

hbase-common/src/main/resources/hbase-default.xml

@@ -1027,6 +1027,12 @@ possible configurations would overwhelm and obscure the important.
       This configuration allows setting an absolute memory size instead of a percentage of the maximum heap.
       Takes precedence over hfile.block.cache.size if both are specified.</description>
   </property>
+  <property>
+    <name>row.cache.size</name>
+    <value>0.0</value>
+    <description>Percentage of maximum heap (-Xmx setting) to allocate to row cache.
+      Default of 0.0; it means the row cache is disabled.</description>
+  </property>
   <property>
       <name>hfile.block.index.cacheonwrite</name>
       <value>false</value>

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

@@ -430,6 +430,13 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
   String L2_CACHE_HIT_RATIO_DESC = "L2 cache hit ratio.";
   String L2_CACHE_MISS_RATIO = "l2CacheMissRatio";
   String L2_CACHE_MISS_RATIO_DESC = "L2 cache miss ratio.";
+
+  String ROW_CACHE_HIT_COUNT = "rowCacheHitCount";
+  String ROW_CACHE_MISS_COUNT = "rowCacheMissCount";
+  String ROW_CACHE_EVICTED_ROW_COUNT = "rowCacheEvictedRowCount";
+  String ROW_CACHE_SIZE = "rowCacheSize";
+  String ROW_CACHE_COUNT = "rowCacheCount";
+
   String RS_START_TIME_NAME = "regionServerStartTime";
   String ZOOKEEPER_QUORUM_NAME = "zookeeperQuorum";
   String SERVER_NAME_NAME = "serverName";

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

@@ -452,6 +452,12 @@ 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(ROW_CACHE_HIT_COUNT, ""), rsWrap.getRowCacheHitCount())
+        .addCounter(Interns.info(ROW_CACHE_MISS_COUNT, ""), rsWrap.getRowCacheMissCount())
+        .addCounter(Interns.info(ROW_CACHE_EVICTED_ROW_COUNT, ""),
+          rsWrap.getRowCacheEvictedRowCount())
+        .addGauge(Interns.info(ROW_CACHE_SIZE, ""), rsWrap.getRowCacheSize())
+        .addGauge(Interns.info(ROW_CACHE_COUNT, ""), rsWrap.getRowCacheCount())
         .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

@@ -635,6 +635,16 @@ public interface MetricsRegionServerWrapper {
 
   long getTrailerHitCount();
 
+  long getRowCacheHitCount();
+
+  long getRowCacheMissCount();
+
+  long getRowCacheSize();
+
+  long getRowCacheCount();
+
+  long getRowCacheEvictedRowCount();
+
   long getTotalRowActionRequestCount();
 
   long getByteBuffAllocatorHeapAllocationBytes();

hbase-server/src/main/java/org/apache/hadoop/hbase/io/util/MemorySizeUtil.java

@@ -93,25 +93,29 @@ public static void validateRegionServerHeapMemoryAllocation(Configuration conf)
     }
     float memStoreFraction = getGlobalMemStoreHeapPercent(conf, false);
     float blockCacheFraction = getBlockCacheHeapPercent(conf);
+    float rowCacheFraction =
+      conf.getFloat(HConstants.ROW_CACHE_SIZE_KEY, HConstants.ROW_CACHE_SIZE_DEFAULT);
     float minFreeHeapFraction = getRegionServerMinFreeHeapFraction(conf);
 
     int memStorePercent = (int) (memStoreFraction * 100);
     int blockCachePercent = (int) (blockCacheFraction * 100);
+    int rowCachePercent = (int) (rowCacheFraction * 100);
     int minFreeHeapPercent = (int) (minFreeHeapFraction * 100);
-    int usedPercent = memStorePercent + blockCachePercent;
+    int usedPercent = memStorePercent + blockCachePercent + rowCachePercent;
     int maxAllowedUsed = 100 - minFreeHeapPercent;
 
     if (usedPercent > maxAllowedUsed) {
       throw new RuntimeException(String.format(
         "RegionServer heap memory allocation is invalid: total memory usage exceeds 100%% "
-          + "(memStore + blockCache + requiredFreeHeap). "
-          + "Check the following configuration values:%n" + "  - %s = %.2f%n" + "  - %s = %s%n"
-          + "  - %s = %s%n" + "  - %s = %s",
+          + "(memStore + blockCache + rowCache + requiredFreeHeap). "
+          + "Check the following configuration values:" + "%n  - %s = %.2f" + "%n  - %s = %s"
+          + "%n  - %s = %s" + "%n  - %s = %s" + "%n  - %s = %s",
         MEMSTORE_SIZE_KEY, memStoreFraction, HConstants.HFILE_BLOCK_CACHE_MEMORY_SIZE_KEY,
         conf.get(HConstants.HFILE_BLOCK_CACHE_MEMORY_SIZE_KEY),
         HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, conf.get(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY),
         HBASE_REGION_SERVER_FREE_HEAP_MIN_MEMORY_SIZE_KEY,
-        conf.get(HBASE_REGION_SERVER_FREE_HEAP_MIN_MEMORY_SIZE_KEY)));
+        conf.get(HBASE_REGION_SERVER_FREE_HEAP_MIN_MEMORY_SIZE_KEY), HConstants.ROW_CACHE_SIZE_KEY,
+        conf.get(HConstants.ROW_CACHE_SIZE_KEY)));
     }
   }
 
@@ -313,4 +317,15 @@ public static long getBucketCacheSize(final Configuration conf) {
     }
     return (long) (bucketCacheSize * 1024 * 1024);
   }
+
+  public static long getRowCacheSize(Configuration conf) {
+    long max = -1L;
+    final MemoryUsage usage = safeGetHeapMemoryUsage();
+    if (usage != null) {
+      max = usage.getMax();
+    }
+    float globalRowCachePercent =
+      conf.getFloat(HConstants.ROW_CACHE_SIZE_KEY, HConstants.ROW_CACHE_SIZE_DEFAULT);
+    return ((long) (max * globalRowCachePercent));
+  }
 }

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

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hbase.regionserver;
 
 import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_LOCAL;
+import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_EVICT_ON_CLOSE_DEFAULT;
+import static org.apache.hadoop.hbase.HConstants.ROW_CACHE_EVICT_ON_CLOSE_KEY;
 import static org.apache.hadoop.hbase.regionserver.HStoreFile.MAJOR_COMPACTION_KEY;
 import static org.apache.hadoop.hbase.trace.HBaseSemanticAttributes.REGION_NAMES_KEY;
 import static org.apache.hadoop.hbase.trace.HBaseSemanticAttributes.ROW_LOCK_READ_LOCK_KEY;
@@ -65,6 +67,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 +436,16 @@ 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 setting for whether to enable row cache for this region.
+   */
+  private final boolean isRowCacheEnabled;
+
   /**
    * The default setting for whether to enable on-demand CF loading for scan requests to this
    * region. Requests can override it.
@@ -929,6 +942,16 @@ public HRegion(final HRegionFileSystem fs, final WAL wal, final Configuration co
 
     minBlockSizeBytes = Arrays.stream(this.htableDescriptor.getColumnFamilies())
       .mapToInt(ColumnFamilyDescriptor::getBlocksize).min().orElse(HConstants.DEFAULT_BLOCKSIZE);
+
+    this.isRowCacheEnabled = determineRowCacheEnabled();
+  }
+
+  boolean determineRowCacheEnabled() {
+    Boolean fromDescriptor = htableDescriptor.getRowCacheEnabled();
+    // The setting from TableDescriptor has higher priority than the global configuration
+    return fromDescriptor != null
+      ? fromDescriptor
+      : conf.getBoolean(HConstants.ROW_CACHE_ENABLED_KEY, HConstants.ROW_CACHE_ENABLED_DEFAULT);
   }
 
   private void setHTableSpecificConf() {
@@ -1940,6 +1963,8 @@ public Pair<byte[], Collection<HStoreFile>> call() throws IOException {
         }
       }
 
+      evictRowCache();
+
       status.setStatus("Writing region close event to WAL");
       // Always write close marker to wal even for read only table. This is not a big problem as we
       // do not write any data into the region; it is just a meta edit in the WAL file.
@@ -1980,6 +2005,22 @@ public Pair<byte[], Collection<HStoreFile>> call() throws IOException {
     }
   }
 
+  private void evictRowCache() {
+    boolean evictOnClose = getReadOnlyConfiguration().getBoolean(ROW_CACHE_EVICT_ON_CLOSE_KEY,
+      ROW_CACHE_EVICT_ON_CLOSE_DEFAULT);
+
+    if (!evictOnClose) {
+      return;
+    }
+
+    if (!(rsServices instanceof HRegionServer regionServer)) {
+      return;
+    }
+
+    RowCacheService rowCacheService = regionServer.getRSRpcServices().getRowCacheService();
+    rowCacheService.evictRowsByRegion(this);
+  }
+
   /** Wait for all current flushes and compactions of the region to complete */
   // TODO HBASE-18906. Check the usage (if any) in Phoenix and expose this or give alternate way for
   // Phoenix needs.
@@ -7881,6 +7922,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
@@ -8709,6 +8751,22 @@ public long getOpenSeqNum() {
     return this.openSeqNum;
   }
 
+  public long getRowCacheSeqNum() {
+    return this.rowCacheSeqNum.get();
+  }
+
+  @Override
+  public boolean isRowCacheEnabled() {
+    return isRowCacheEnabled;
+  }
+
+  /**
+   * This is used to invalidate the row cache of the bulk-loaded region.
+   */
+  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/MetricsRegionServerWrapperImpl.java

@@ -68,6 +68,7 @@ class MetricsRegionServerWrapperImpl implements MetricsRegionServerWrapper {
   private BlockCache l1Cache = null;
   private BlockCache l2Cache = null;
   private MobFileCache mobFileCache;
+  private final RowCache rowCache;
   private CacheStats cacheStats;
   private CacheStats l1Stats = null;
   private CacheStats l2Stats = null;
@@ -99,6 +100,8 @@ public MetricsRegionServerWrapperImpl(final HRegionServer regionServer) {
     this.regionServer = regionServer;
     initBlockCache();
     initMobFileCache();
+    RSRpcServices rsRpcServices = this.regionServer.getRSRpcServices();
+    this.rowCache = rsRpcServices == null ? null : rsRpcServices.getRowCacheService().getRowCache();
     this.excludeDatanodeManager = this.regionServer.getWalFactory().getExcludeDatanodeManager();
 
     this.period = regionServer.getConfiguration().getLong(HConstants.REGIONSERVER_METRICS_PERIOD,
@@ -1194,6 +1197,31 @@ public long getTrailerHitCount() {
     return this.cacheStats != null ? this.cacheStats.getTrailerHitCount() : 0L;
   }
 
+  @Override
+  public long getRowCacheHitCount() {
+    return this.rowCache != null ? this.rowCache.getHitCount() : 0L;
+  }
+
+  @Override
+  public long getRowCacheMissCount() {
+    return this.rowCache != null ? this.rowCache.getMissCount() : 0L;
+  }
+
+  @Override
+  public long getRowCacheSize() {
+    return this.rowCache != null ? this.rowCache.getSize() : 0L;
+  }
+
+  @Override
+  public long getRowCacheCount() {
+    return this.rowCache != null ? this.rowCache.getCount() : 0L;
+  }
+
+  @Override
+  public long getRowCacheEvictedRowCount() {
+    return this.rowCache != null ? this.rowCache.getEvictedRowCount() : 0L;
+  }
+
   @Override
   public long getByteBuffAllocatorHeapAllocationBytes() {
     return ByteBuffAllocator.getHeapAllocationBytes(allocator, ByteBuffAllocator.HEAP);