Option 2: Data Tiering integration into cache evictions.

This code change is only for the purpose of sharing the idea about implementation.
The idea here is to only rely only the file names to handling the data tiering.

DO NOT SUBTMIT

Still TODO:
1. Finalise design.
2. Unit Tests.

Change-Id: I4b0a9fbffb274fb80dcc7e1303985901e4da22f2

Author: jhungund

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

@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -76,6 +77,7 @@
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.nio.RefCnt;
 import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
+import org.apache.hadoop.hbase.regionserver.DataTieringManager;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.IdReadWriteLock;
@@ -935,6 +937,27 @@ void freeSpace(final String why) {
     }
     try {
       freeInProgress = true;
+
+      // Fetch the file names from the backing map.
+      // Use the hash-map for efficient lookups
+      Map<String, String> fileNames = new HashMap<>();
+      for (Map.Entry<BlockCacheKey, BucketEntry> bucketEntryWithKey : backingMap.entrySet()) {
+        String fileName = bucketEntryWithKey.getKey().getHfileName();
+        if (!fileNames.containsKey(fileName)) {
+          // We are only interested in the keys.
+          fileNames.put(fileName, null);
+        }
+      }
+
+      // Check the list of files to determine the cold files which can be readily evicted.
+      List<String> coldFiles =
+        DataTieringManager.getInstance().getColdFileList(new ArrayList<>(fileNames.keySet()));
+      if (coldFiles != null) {
+        for(String fileName : coldFiles) {
+          evictBlocksByHfileName(fileName);
+        }
+      }
+
       long bytesToFreeWithoutExtra = 0;
       // Calculate free byte for each bucketSizeinfo
       StringBuilder msgBuffer = LOG.isDebugEnabled() ? new StringBuilder() : null;

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

@@ -17,15 +17,20 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.OptionalLong;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -219,4 +224,44 @@ private long getDataTieringHotDataAge(Configuration conf) {
     return Long.parseLong(
       conf.get(DATATIERING_HOT_DATA_AGE_KEY, String.valueOf(DEFAULT_DATATIERING_HOT_DATA_AGE)));
   }
+
+  /*
+   * This API takes the names of files as input and returns a subset of these file names
+   * that are cold.
+   * @parameter inputFileNames: Input list of file names
+   * @return List of names of files that are cold as per data-tiering logic.
+   */
+  public List<String> getColdFileList(List<String> inputFileNames) {
+    List<String> coldFileList = new ArrayList<>();
+    for (HRegion r : this.onlineRegions.values()) {
+      for (HStore hStore : r.getStores()) {
+        Configuration conf = hStore.getReadOnlyConfiguration();
+        if (getDataTieringType(conf) != DataTieringType.TIME_RANGE) {
+          // Data-Tiering not enabled for the store. Just skip it.
+          continue;
+        }
+        Long hotDataAge = getDataTieringHotDataAge(conf);
+
+        for (HStoreFile hStoreFile : hStore.getStorefiles()) {
+          String hFileName =
+            hStoreFile.getFileInfo().getHFileInfo().getHFileContext().getHFileName();
+          if(inputFileNames.contains(hFileName)) {
+            OptionalLong maxTimestamp = hStoreFile.getMaximumTimestamp();
+            if (!maxTimestamp.isPresent()) {
+              // We could throw from here, But we are already in the critical code-path
+              // of freeing space. Hence, we can ignore that file for now
+              // Or do we want to include it?
+              continue;
+            }
+            long currentTimestamp = EnvironmentEdgeManager.getDelegate().currentTime();
+            long fileAge = currentTimestamp - maxTimestamp.getAsLong();
+            if (fileAge > hotDataAge) {
+              coldFileList.add(hFileName);
+            }
+          }
+        }
+      }
+    }
+    return coldFileList;
+  }
 }