Improve project construction information for tikv

assembly/client/pom.xml

@@ -81,6 +81,7 @@
                 <excludes>
                   <exclude>org.alluxio:alluxio-microbench</exclude>
                   <exclude>org.openjdk.jmh:*</exclude>
+                  <exclude>**/Log4j2Plugins.dat</exclude>
                 </excludes>
               </artifactSet>
               <filters>

assembly/server/pom.xml

@@ -86,6 +86,7 @@
                 <excludes>
                   <exclude>org.alluxio:alluxio-microbench</exclude>
                   <exclude>org.openjdk.jmh:*</exclude>
+                  <exclude>**/Log4j2Plugins.dat</exclude>
                 </excludes>
               </artifactSet>
               <filters>

core/common/pom.xml

@@ -115,6 +115,10 @@
       <groupId>org.apache.curator</groupId>
       <artifactId>curator-framework</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.thrift</groupId>
+      <artifactId>libthrift</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.zookeeper</groupId>
       <artifactId>zookeeper</artifactId>
@@ -127,6 +131,11 @@
       <groupId>org.rocksdb</groupId>
       <artifactId>rocksdbjni</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.tikv</groupId>
+      <artifactId>tikv-client-java</artifactId>
+      <version>${tikv.version}</version>
+    </dependency>
     <!-- Determine version with the table here: https://github.com/grpc/grpc-java/blob/master/SECURITY.md-->
     <dependency>
       <groupId>io.netty</groupId>

core/common/src/main/java/alluxio/conf/PropertyKey.java

@@ -3414,12 +3414,19 @@ public String toString() {
           .build();
   public static final PropertyKey MASTER_TIERED_STORE_GLOBAL_LEVEL1_ALIAS =
       stringBuilder(Name.MASTER_TIERED_STORE_GLOBAL_LEVEL1_ALIAS)
-          .setDefaultValue(Constants.MEDIUM_SSD)
+          .setDefaultValue(Constants.MEDIUM_PMEM)
           .setDescription("The name of the second highest storage tier in the entire system.")
           .setConsistencyCheckLevel(ConsistencyCheckLevel.ENFORCE)
           .setScope(Scope.MASTER)
           .build();
   public static final PropertyKey MASTER_TIERED_STORE_GLOBAL_LEVEL2_ALIAS =
+      stringBuilder(Name.MASTER_TIERED_STORE_GLOBAL_LEVEL1_ALIAS)
+          .setDefaultValue(Constants.MEDIUM_SSD)
+          .setDescription("The name of the second highest storage tier in the entire system.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.ENFORCE)
+          .setScope(Scope.MASTER)
+          .build();
+  public static final PropertyKey MASTER_TIERED_STORE_GLOBAL_LEVEL3_ALIAS =
       stringBuilder(Name.MASTER_TIERED_STORE_GLOBAL_LEVEL2_ALIAS)
           .setDefaultValue(Constants.MEDIUM_HDD)
           .setDescription("The name of the third highest storage tier in the entire system.")
@@ -3428,14 +3435,14 @@ public String toString() {
           .build();
   public static final PropertyKey MASTER_TIERED_STORE_GLOBAL_LEVELS =
       intBuilder(Name.MASTER_TIERED_STORE_GLOBAL_LEVELS)
-          .setDefaultValue(3)
+          .setDefaultValue(4)
           .setDescription("The total number of storage tiers in the system.")
           .setConsistencyCheckLevel(ConsistencyCheckLevel.ENFORCE)
           .setScope(Scope.MASTER)
           .build();
   public static final PropertyKey MASTER_TIERED_STORE_GLOBAL_MEDIUMTYPE =
       listBuilder(Name.MASTER_TIERED_STORE_GLOBAL_MEDIUMTYPE)
-          .setDefaultValue("MEM,SSD,HDD")
+          .setDefaultValue("MEM,PMEM,SSD,HDD")
           .setDescription("The list of medium types we support in the system.")
           .setConsistencyCheckLevel(ConsistencyCheckLevel.ENFORCE)
           .setScope(Scope.MASTER)
@@ -5104,6 +5111,12 @@ public String toString() {
           .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
           .setScope(Scope.WORKER)
           .build();
+  public static final PropertyKey WORKER_TIERED_STORE_LEVEL3_ALIAS =
+          new Builder(PropertyType.STRING, Template.WORKER_TIERED_STORE_LEVEL_ALIAS, 3)
+                  .setDescription("The alias of the four storage tier on this worker.")
+                  .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
+                  .setScope(Scope.WORKER)
+                  .build();
   public static final PropertyKey WORKER_TIERED_STORE_LEVEL2_DIRS_PATH =
       new Builder(PropertyType.LIST, Optional.of(","),
           Template.WORKER_TIERED_STORE_LEVEL_DIRS_PATH, 2)
@@ -7732,6 +7745,13 @@ public String toString() {
           .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
           .setScope(Scope.CLIENT)
           .build();
+  public static final PropertyKey MASTER_METASTORE_INODE_TIKV_CONNECTION =
+      stringBuilder(Name.MASTER_METASTORE_INODE_TIKV_CONNECTION)
+          .setDefaultValue("127.0.0.1:2379")
+          .setDescription("The connection of tikv for backing store.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.ENFORCE)
+          .setScope(Scope.MASTER)
+          .build();
   /**
    * @deprecated This key is used for testing. It is always deprecated.
    */
@@ -8398,6 +8418,8 @@ public static final class Name {
         "alluxio.master.tieredstore.global.level1.alias";
     public static final String MASTER_TIERED_STORE_GLOBAL_LEVEL2_ALIAS =
         "alluxio.master.tieredstore.global.level2.alias";
+    public static final String MASTER_TIERED_STORE_GLOBAL_LEVEL3_ALIAS =
+        "alluxio.master.tieredstore.global.level3.alias";
     public static final String MASTER_TIERED_STORE_GLOBAL_LEVELS =
         "alluxio.master.tieredstore.global.levels";
     public static final String MASTER_TIERED_STORE_GLOBAL_MEDIUMTYPE =
@@ -9304,6 +9326,8 @@ public static final class Name {
         "alluxio.hadoop.kerberos.keytab.login.autorenewal";
     public static final String HADOOP_CHECKSUM_COMBINE_MODE =
         "alluxio.hadoop.checksum.combine.mode";
+    public static final String MASTER_METASTORE_INODE_TIKV_CONNECTION =
+        "alluxio.master.metastore.inode.tikv.connection";
 
     private Name() {} // prevent instantiation
   }

core/common/src/main/java/alluxio/master/metastore/MetastoreType.java

@@ -16,5 +16,6 @@
  */
 public enum MetastoreType {
   HEAP,
-  ROCKS
+  ROCKS,
+  TIKV
 }

core/common/src/main/java/alluxio/master/metastore/tikv/TiKVUtils.java

@@ -0,0 +1,145 @@
+package alluxio.master.metastore.tikv;
+
+import alluxio.resource.CloseableIterator;
+import com.google.common.primitives.Longs;
+import org.tikv.kvproto.Kvrpcpb;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Iterator;
+import java.util.ListIterator;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Convenience methods for working with TiKV.
+ */
+public final class TiKVUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(TiKVUtils.class);
+
+    private TiKVUtils() {} // Utils class.
+
+
+    /**
+     * @param str a String value
+     * @param long1 a long value
+     * @param long2 a long value
+     * @return a byte array formed by writing the bytes of n followed by the bytes of str
+     */
+    public static byte[] toByteArray(String str, long long1, long long2) {
+        byte[] strBytes = str.getBytes();
+
+        byte[] key = new byte[strBytes.length + 2 * Longs.BYTES];
+        System.arraycopy(strBytes, 0, key, 0, strBytes.length);
+        for (int i = strBytes.length + Longs.BYTES - 1; i >= strBytes.length; i--) {
+            key[i] = (byte) (long1 & 0xffL);
+            long1 >>= Byte.SIZE;
+        }
+        for (int i = strBytes.length + 2 * Longs.BYTES - 1; i >= strBytes.length + Longs.BYTES; i--) {
+            key[i] = (byte) (long2 & 0xffL);
+            long2 >>= Byte.SIZE;
+        }
+        return key;
+    }
+
+    /**
+     * @param n a long value
+     * @param str a string value
+     * @return a byte array formed by writing the bytes of n followed by the bytes of str
+     */
+    public static byte[] toByteArray(String str, long n) {
+        byte[] strBytes = str.getBytes();
+
+        byte[] key = new byte[Longs.BYTES + strBytes.length];
+        System.arraycopy(strBytes, 0, key, 0, strBytes.length);
+        for (int i = key.length - 1; i >= strBytes.length; i--) {
+            key[i] = (byte) (n & 0xffL);
+            n >>= Byte.SIZE;
+        }
+        return key;
+    }
+
+    /**
+     * @param n a long value
+     * @param str1 a string value
+     * @param str2 a string value
+     * @return a byte array formed by writing the bytes of n followed by the bytes of str
+     */
+    public static byte[] toByteArray(String str1, long n, String str2) {
+        byte[] strBytes1 = str1.getBytes();
+        byte[] strBytes2 = str2.getBytes();
+
+        byte[] key = new byte[Longs.BYTES + strBytes1.length + strBytes2.length];
+        System.arraycopy(strBytes1, 0, key, 0, strBytes1.length);
+        for (int i = strBytes1.length + Longs.BYTES - 1; i >= strBytes1.length; i--) {
+            key[i] = (byte) (n & 0xffL);
+            n >>= Byte.SIZE;
+        }
+        System.arraycopy(strBytes2, 0, key, strBytes1.length + Longs.BYTES, strBytes2.length);
+        return key;
+    }
+
+    /**
+     * @param bytes an array of bytes
+     * @param start the place in the array to read the long from
+     * @return the long
+     */
+    public static long readLong(byte[] bytes, int start) {
+        return Longs.fromBytes(bytes[start], bytes[start + 1], bytes[start + 2], bytes[start + 3],
+                bytes[start + 4], bytes[start + 5], bytes[start + 6], bytes[start + 7]);
+    }
+
+
+    /**
+     * Used to parse current {@link ListIterator<Kvrpcpb.KvPair>} element.
+     *
+     * @param <T> return type of parser's next method
+     */
+    public interface TiKVIteratorParser<T> {
+        /**
+         * Parses and return next element.
+         *
+         * @param iter {@link ListIterator<Kvrpcpb.KvPair>} instance
+         * @return parsed value
+         * @throws Exception if parsing fails
+         */
+        T next(ListIterator<Kvrpcpb.KvPair> iter) throws Exception;
+    }
+
+    /**
+     * Used to wrap an {@link CloseableIterator} over {@link ListIterator<Kvrpcpb.KvPair>}.
+     * It seeks given iterator to first entry before returning the iterator.
+     *
+     * @param tikvIterator the tikv iterator
+     * @param parser parser to produce iterated values from tikv key-value
+     * @param <T> iterator value type
+     * @return wrapped iterator
+     */
+    public static <T> CloseableIterator<T> createCloseableIterator(
+            ListIterator<Kvrpcpb.KvPair> tikvIterator, TiKVIteratorParser<T> parser) {
+        AtomicBoolean valid = new AtomicBoolean(true);
+        Iterator<T> iter = new Iterator<T>() {
+            @Override
+            public boolean hasNext() {
+                return valid.get() && tikvIterator.hasNext();
+            }
+
+            @Override
+            public T next() {
+                try {
+                    return parser.next(tikvIterator);
+                } catch (Exception exc) {
+                    LOG.warn("Iteration aborted because of error", exc);
+                    valid.set(false);
+                    throw new RuntimeException(exc);
+                } finally {
+                    if (!tikvIterator.hasNext()) {
+                        valid.set(false);
+                    }
+                }
+            }
+        };
+
+        return CloseableIterator.noopCloseable(iter);
+    }
+
+}

core/server/common/src/main/java/alluxio/master/journal/checkpoint/CheckpointName.java

@@ -42,4 +42,5 @@ public enum CheckpointName {
   TTL_BUCKET_LIST,
   SCHEDULER,
   SNAPSHOT_ID,
+  TIKV_INODE_STORE,
 }

core/server/master/pom.xml

@@ -78,6 +78,16 @@
       <groupId>org.rocksdb</groupId>
       <artifactId>rocksdbjni</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.tikv</groupId>
+      <artifactId>tikv-client-java</artifactId>
+      <version>${tikv.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>com.jcraft</groupId>
+      <artifactId>jsch</artifactId>
+      <version>${jsch.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.glassfish.jersey.media</groupId>
       <artifactId>jersey-media-json-jackson</artifactId>

core/server/master/src/main/java/alluxio/master/MasterUtils.java

@@ -22,6 +22,8 @@
 import alluxio.master.metastore.heap.HeapInodeStore;
 import alluxio.master.metastore.rocks.RocksBlockMetaStore;
 import alluxio.master.metastore.rocks.RocksInodeStore;
+import alluxio.master.metastore.tikv.TiKVInodeStore;
+import alluxio.master.metastore.tikv.TiKVBlockMetaStore;
 import alluxio.util.CommonUtils;
 
 import java.util.ArrayList;
@@ -70,6 +72,8 @@ public static BlockMetaStore.Factory getBlockStoreFactory(String baseDir) {
         return HeapBlockMetaStore::new;
       case ROCKS:
         return () -> new RocksBlockMetaStore(baseDir);
+      case TIKV:
+        return () -> new TiKVBlockMetaStore(baseDir);
       default:
         throw new IllegalStateException("Unknown metastore type: " + type);
     }
@@ -92,6 +96,12 @@ public static InodeStore.Factory getInodeStoreFactory(String baseDir) {
         } else {
           return lockManager -> new CachingInodeStore(new RocksInodeStore(baseDir), lockManager);
         }
+      case TIKV:
+        if (Configuration.getInt(PropertyKey.MASTER_METASTORE_INODE_CACHE_MAX_SIZE) == 0) {
+          return lockManager -> new TiKVInodeStore(baseDir);
+        } else {
+          return lockManager -> new CachingInodeStore(new TiKVInodeStore(baseDir), lockManager);
+        }
       default:
         throw new IllegalStateException("Unknown metastore type: " + type);
     }