HBASE-29427 Merge all commits related to custom tiering into the feature branch (apache#7124)

This is the whole custom tiering implementation and involves the following individual works:

* HBASE-29412 Extend date tiered compaction to allow for tiering by values other than cell timestamp
* HBASE-29413 Implement a custom qualifier tiered compaction
* HBASE-29414 Refactor DataTieringManager to make priority logic pluggable
* HBASE-29422 Implement selectMinorCompation in CustomCellDateTieredCompactionPolicy
* HBASE-29424 Implement configuration validation for custom tiering compactions
* HBASE-29425 Refine and polish code
* HBASE-29426 Define a tiering value provider and refactor custom tiered compaction related classes
* HBASE-28463 Rebase time based priority branch (HBASE-28463) with latest master (and fix conflicts)

Co-authored-by: Janardhan Hungund <[email protected]>

Signed-off-by: Tak Lon (Stephen) Wu <[email protected]>
Change-Id: Ib10409b23a8cb735af5210e5ae4fc843b04b2d10

Author: wchevreuil

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

@@ -36,4 +36,6 @@ public final class TagType {
   // String based tag type used in replication
   public static final byte STRING_VIS_TAG_TYPE = (byte) 7;
   public static final byte TTL_TAG_TYPE = (byte) 8;
+  // tag with the custom cell tiering value for the row
+  public static final byte CELL_VALUE_TIERING_TAG_TYPE = (byte) 9;
 }

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

@@ -23,7 +23,6 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.conf.ConfigurationObserver;
-import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -199,13 +198,13 @@ default Optional<Boolean> shouldCacheFile(HFileInfo hFileInfo, Configuration con
    * not be overridden by all implementing classes. In such cases, the returned Optional will be
    * empty. For subclasses implementing this logic, the returned Optional would contain the boolean
    * value reflecting if the passed block should indeed be cached.
-   * @param key              The key representing the block to check if it should be cached.
-   * @param timeRangeTracker the time range tracker containing the timestamps
-   * @param conf             The configuration object to use for determining caching behavior.
+   * @param key          The key representing the block to check if it should be cached.
+   * @param maxTimeStamp The maximum timestamp for the block to check if it should be cached.
+   * @param conf         The configuration object to use for determining caching behavior.
    * @return An empty Optional if this method is not supported; otherwise, the returned Optional
    *         contains the boolean value indicating if the block should be cached.
    */
-  default Optional<Boolean> shouldCacheBlock(BlockCacheKey key, TimeRangeTracker timeRangeTracker,
+  default Optional<Boolean> shouldCacheBlock(BlockCacheKey key, long maxTimeStamp,
     Configuration conf) {
     return Optional.empty();
   }

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

@@ -282,7 +282,8 @@ public boolean shouldCacheBlockOnRead(BlockCategory category) {
   public boolean shouldCacheBlockOnRead(BlockCategory category, HFileInfo hFileInfo,
     Configuration conf) {
     Optional<Boolean> cacheFileBlock = Optional.of(true);
-    if (getBlockCache().isPresent()) {
+    // For DATA blocks only, if BuckeCache is in use, we don't need to cache block again
+    if (getBlockCache().isPresent() && category.equals(BlockCategory.DATA)) {
       Optional<Boolean> result = getBlockCache().get().shouldCacheFile(hFileInfo, conf);
       if (result.isPresent()) {
         cacheFileBlock = result;

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

@@ -26,7 +26,6 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.io.HeapSize;
 import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
-import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -484,10 +483,10 @@ public Optional<Boolean> shouldCacheFile(HFileInfo hFileInfo, Configuration conf
   }
 
   @Override
-  public Optional<Boolean> shouldCacheBlock(BlockCacheKey key, TimeRangeTracker timeRangeTracker,
+  public Optional<Boolean> shouldCacheBlock(BlockCacheKey key, long maxTimeStamp,
     Configuration conf) {
-    return combineCacheResults(l1Cache.shouldCacheBlock(key, timeRangeTracker, conf),
-      l2Cache.shouldCacheBlock(key, timeRangeTracker, conf));
+    return combineCacheResults(l1Cache.shouldCacheBlock(key, maxTimeStamp, conf),
+      l2Cache.shouldCacheBlock(key, maxTimeStamp, conf));
   }
 
   private Optional<Boolean> combineCacheResults(Optional<Boolean> result1,

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

@@ -43,6 +43,7 @@
 import org.apache.hadoop.hbase.ipc.RpcServer;
 import org.apache.hadoop.hbase.regionserver.CellSink;
 import org.apache.hadoop.hbase.regionserver.ShipperListener;
+import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
 import org.apache.hadoop.hbase.util.BloomFilterWriter;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.FSUtils;
@@ -217,6 +218,12 @@ public interface Writer extends Closeable, CellSink, ShipperListener {
      */
     void appendTrackedTimestampsToMetadata() throws IOException;
 
+    /**
+     * Add Custom cell timestamp to Metadata
+     */
+    public void appendCustomCellTimestampsToMetadata(TimeRangeTracker timeRangeTracker)
+      throws IOException;
+
     /** Returns the path to this {@link HFile} */
     Path getPath();
 

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

@@ -1397,7 +1397,7 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize, final bo
           cacheConf.getBlockCache().ifPresent(cache -> {
             LOG.debug("Skipping decompression of block {} in prefetch", cacheKey);
             // Cache the block if necessary
-            if (cacheBlock && cacheConf.shouldCacheBlockOnRead(category)) {
+            if (cacheBlock && cacheOnRead) {
               cache.cacheBlock(cacheKey, blockNoChecksum, cacheConf.isInMemory(), cacheOnly);
             }
           });
@@ -1411,7 +1411,7 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize, final bo
         HFileBlock unpackedNoChecksum = BlockCacheUtil.getBlockForCaching(cacheConf, unpacked);
         // Cache the block if necessary
         cacheConf.getBlockCache().ifPresent(cache -> {
-          if (cacheBlock && cacheConf.shouldCacheBlockOnRead(category)) {
+          if (cacheBlock && cacheOnRead) {
             // Using the wait on cache during compaction and prefetching.
             cache.cacheBlock(cacheKey,
               cacheCompressed

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.io.hfile;
 
 import static org.apache.hadoop.hbase.io.hfile.BlockCompressedSizePredicator.MAX_BLOCK_SIZE_UNCOMPRESSED;
+import static org.apache.hadoop.hbase.regionserver.CustomTieringMultiFileWriter.CUSTOM_TIERING_TIME_RANGE;
 import static org.apache.hadoop.hbase.regionserver.HStoreFile.EARLIEST_PUT_TS;
 import static org.apache.hadoop.hbase.regionserver.HStoreFile.TIMERANGE_KEY;
 
@@ -29,6 +30,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Supplier;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
@@ -127,6 +129,12 @@ public class HFileWriterImpl implements HFile.Writer {
   /** Cache configuration for caching data on write. */
   protected final CacheConfig cacheConf;
 
+  public void setTimeRangeTrackerForTiering(Supplier<TimeRangeTracker> timeRangeTrackerForTiering) {
+    this.timeRangeTrackerForTiering = timeRangeTrackerForTiering;
+  }
+
+  private Supplier<TimeRangeTracker> timeRangeTrackerForTiering;
+
   /**
    * Name for this object used when logging or in toString. Is either the result of a toString on
    * stream or else name of passed file Path.
@@ -186,7 +194,9 @@ public HFileWriterImpl(final Configuration conf, CacheConfig cacheConf, Path pat
     this.path = path;
     this.name = path != null ? path.getName() : outputStream.toString();
     this.hFileContext = fileContext;
+    // TODO: Move this back to upper layer
     this.timeRangeTracker = TimeRangeTracker.create(TimeRangeTracker.Type.NON_SYNC);
+    this.timeRangeTrackerForTiering = () -> this.timeRangeTracker;
     DataBlockEncoding encoding = hFileContext.getDataBlockEncoding();
     if (encoding != DataBlockEncoding.NONE) {
       this.blockEncoder = new HFileDataBlockEncoderImpl(encoding);
@@ -588,7 +598,8 @@ private BlockCacheKey buildCacheBlockKey(long offset, BlockType blockType) {
   }
 
   private boolean shouldCacheBlock(BlockCache cache, BlockCacheKey key) {
-    Optional<Boolean> result = cache.shouldCacheBlock(key, timeRangeTracker, conf);
+    Optional<Boolean> result =
+      cache.shouldCacheBlock(key, timeRangeTrackerForTiering.get().getMax(), conf);
     return result.orElse(true);
   }
 
@@ -899,12 +910,19 @@ public void appendTrackedTimestampsToMetadata() throws IOException {
     appendFileInfo(EARLIEST_PUT_TS, Bytes.toBytes(earliestPutTs));
   }
 
+  public void appendCustomCellTimestampsToMetadata(TimeRangeTracker timeRangeTracker)
+    throws IOException {
+    // TODO: The StoreFileReader always converts the byte[] to TimeRange
+    // via TimeRangeTracker, so we should write the serialization data of TimeRange directly.
+    appendFileInfo(CUSTOM_TIERING_TIME_RANGE, TimeRangeTracker.toByteArray(timeRangeTracker));
+  }
+
   /**
    * Record the earliest Put timestamp. If the timeRangeTracker is not set, update TimeRangeTracker
    * to include the timestamp of this key
    */
   private void trackTimestamps(final ExtendedCell cell) {
-    if (Cell.Type.Put == cell.getType()) {
+    if (KeyValue.Type.Put == KeyValue.Type.codeToType(cell.getTypeByte())) {
       earliestPutTs = Math.min(earliestPutTs, cell.getTimestamp());
     }
     timeRangeTracker.includeTimestamp(cell);

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

@@ -86,7 +86,6 @@
 import org.apache.hadoop.hbase.regionserver.DataTieringManager;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
-import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.IdReadWriteLock;
@@ -1182,8 +1181,9 @@ void freeSpace(final String why) {
           }
         }
 
-        if (bytesFreed < bytesToFreeWithExtra &&
-          coldFiles != null && coldFiles.containsKey(bucketEntryWithKey.getKey().getHfileName())
+        if (
+          bytesFreed < bytesToFreeWithExtra && coldFiles != null
+            && coldFiles.containsKey(bucketEntryWithKey.getKey().getHfileName())
         ) {
           int freedBlockSize = bucketEntryWithKey.getValue().getLength();
           if (evictBlockIfNoRpcReferenced(bucketEntryWithKey.getKey())) {
@@ -2446,10 +2446,10 @@ public Optional<Boolean> shouldCacheFile(HFileInfo hFileInfo, Configuration conf
   }
 
   @Override
-  public Optional<Boolean> shouldCacheBlock(BlockCacheKey key, TimeRangeTracker timeRangeTracker,
+  public Optional<Boolean> shouldCacheBlock(BlockCacheKey key, long maxTimestamp,
     Configuration conf) {
     DataTieringManager dataTieringManager = DataTieringManager.getInstance();
-    if (dataTieringManager != null && !dataTieringManager.isHotData(timeRangeTracker, conf)) {
+    if (dataTieringManager != null && !dataTieringManager.isHotData(maxTimestamp, conf)) {
       LOG.debug("Data tiering is enabled for file: '{}' and it is not hot data",
         key.getHfileName());
       return Optional.of(false);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java

@@ -36,6 +36,7 @@
 import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
 import org.apache.hadoop.hbase.procedure2.ProcedureSuspendedException;
 import org.apache.hadoop.hbase.procedure2.ProcedureUtil;
+import org.apache.hadoop.hbase.regionserver.compactions.CustomCellTieredUtils;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerValidationUtils;
 import org.apache.hadoop.hbase.replication.ReplicationException;
@@ -296,6 +297,8 @@ private boolean prepareCreate(final MasterProcedureEnv env) throws IOException {
     StoreFileTrackerValidationUtils.checkForCreateTable(env.getMasterConfiguration(),
       tableDescriptor);
 
+    CustomCellTieredUtils.checkForModifyTable(tableDescriptor);
+
     return true;
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
 import org.apache.hadoop.hbase.master.zksyncer.MetaLocationSyncer;
 import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
+import org.apache.hadoop.hbase.regionserver.compactions.CustomCellTieredUtils;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerValidationUtils;
 import org.apache.hadoop.hbase.replication.ReplicationException;
 import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
@@ -449,6 +450,7 @@ private void prepareModify(final MasterProcedureEnv env) throws IOException {
     // check for store file tracker configurations
     StoreFileTrackerValidationUtils.checkForModifyTable(env.getMasterConfiguration(),
       unmodifiedTableDescriptor, modifiedTableDescriptor, !isTableEnabled(env));
+    CustomCellTieredUtils.checkForModifyTable(modifiedTableDescriptor);
   }
 
   /**