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