HDDS-1499. OzoneManager Cache. (#798)

Author: bharatviswa504

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/DBStore.java

@@ -44,6 +44,7 @@ public interface DBStore extends AutoCloseable {
    */
   Table<byte[], byte[]> getTable(String name) throws IOException;
 
+
   /**
    * Gets an existing TableStore with implicit key/value conversion.
    *

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBTable.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 
+import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdfs.DFSUtil;
 
 import org.rocksdb.ColumnFamilyHandle;
@@ -33,9 +34,12 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * RocksDB implementation of ozone metadata store.
+ * RocksDB implementation of ozone metadata store. This class should be only
+ * used as part of TypedTable as it's underlying implementation to access the
+ * metadata store content. All other user's using Table should use TypedTable.
  */
-public class RDBTable implements Table<byte[], byte[]> {
+@InterfaceAudience.Private
+class RDBTable implements Table<byte[], byte[]> {
 
 
   private static final Logger LOG =
@@ -52,7 +56,7 @@ public class RDBTable implements Table<byte[], byte[]> {
    * @param handle - ColumnFamily Handle.
    * @param writeOptions - RocksDB write Options.
    */
-  public RDBTable(RocksDB db, ColumnFamilyHandle handle,
+  RDBTable(RocksDB db, ColumnFamilyHandle handle,
       WriteOptions writeOptions) {
     this.db = db;
     this.handle = handle;

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/Table.java

@@ -21,8 +21,10 @@
 
 import java.io.IOException;
 
+import org.apache.commons.lang3.NotImplementedException;
 import org.apache.hadoop.classification.InterfaceStability;
-
+import org.apache.hadoop.utils.db.cache.CacheKey;
+import org.apache.hadoop.utils.db.cache.CacheValue;
 /**
  * Interface for key-value store that stores ozone metadata. Ozone metadata is
  * stored as key value pairs, both key and value are arbitrary byte arrays. Each
@@ -97,6 +99,28 @@ void putWithBatch(BatchOperation batch, KEY key, VALUE value)
    */
   String getName() throws IOException;
 
+  /**
+   * Add entry to the table cache.
+   *
+   * If the cacheKey already exists, it will override the entry.
+   * @param cacheKey
+   * @param cacheValue
+   */
+  default void addCacheEntry(CacheKey<KEY> cacheKey,
+      CacheValue<VALUE> cacheValue) {
+    throw new NotImplementedException("addCacheEntry is not implemented");
+  }
+
+  /**
+   * Removes all the entries from the table cache which are having epoch value
+   * less
+   * than or equal to specified epoch value.
+   * @param epoch
+   */
+  default void cleanupCache(long epoch) {
+    throw new NotImplementedException("cleanupCache is not implemented");
+  }
+
   /**
    * Class used to represent the key and value pair of a db entry.
    */

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/TypedTable.java

@@ -20,6 +20,12 @@
 
 import java.io.IOException;
 
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.utils.db.cache.CacheKey;
+import org.apache.hadoop.utils.db.cache.CacheValue;
+import org.apache.hadoop.utils.db.cache.PartialTableCache;
+import org.apache.hadoop.utils.db.cache.TableCache;
+
 /**
  * Strongly typed table implementation.
  * <p>
@@ -31,13 +37,16 @@
  */
 public class TypedTable<KEY, VALUE> implements Table<KEY, VALUE> {
 
-  private Table<byte[], byte[]> rawTable;
+  private final Table<byte[], byte[]> rawTable;
+
+  private final CodecRegistry codecRegistry;
 
-  private CodecRegistry codecRegistry;
+  private final Class<KEY> keyType;
 
-  private Class<KEY> keyType;
+  private final Class<VALUE> valueType;
+
+  private final TableCache<CacheKey<KEY>, CacheValue<VALUE>> cache;
 
-  private Class<VALUE> valueType;
 
   public TypedTable(
       Table<byte[], byte[]> rawTable,
@@ -47,6 +56,7 @@ public TypedTable(
     this.codecRegistry = codecRegistry;
     this.keyType = keyType;
     this.valueType = valueType;
+    cache = new PartialTableCache<>();
   }
 
   @Override
@@ -69,8 +79,34 @@ public boolean isEmpty() throws IOException {
     return rawTable.isEmpty();
   }
 
+  /**
+   * Returns the value mapped to the given key in byte array or returns null
+   * if the key is not found.
+   *
+   * Caller's of this method should use synchronization mechanism, when
+   * accessing. First it will check from cache, if it has entry return the
+   * value, otherwise get from the RocksDB table.
+   *
+   * @param key metadata key
+   * @return VALUE
+   * @throws IOException
+   */
   @Override
   public VALUE get(KEY key) throws IOException {
+    // Here the metadata lock will guarantee that cache is not updated for same
+    // key during get key.
+    CacheValue< VALUE > cacheValue = cache.get(new CacheKey<>(key));
+    if (cacheValue == null) {
+      // If no cache for the table or if it does not exist in cache get from
+      // RocksDB table.
+      return getFromTable(key);
+    } else {
+      // We have a value in cache, return the value.
+      return cacheValue.getValue();
+    }
+  }
+
+  private VALUE getFromTable(KEY key) throws IOException {
     byte[] keyBytes = codecRegistry.asRawData(key);
     byte[] valueBytes = rawTable.get(keyBytes);
     return codecRegistry.asObject(valueBytes, valueType);
@@ -106,6 +142,40 @@ public void close() throws Exception {
 
   }
 
+  @Override
+  public void addCacheEntry(CacheKey<KEY> cacheKey,
+      CacheValue<VALUE> cacheValue) {
+    // This will override the entry if there is already entry for this key.
+    cache.put(cacheKey, cacheValue);
+  }
+
+
+  @Override
+  public void cleanupCache(long epoch) {
+    cache.cleanup(epoch);
+  }
+
+  @VisibleForTesting
+  TableCache<CacheKey<KEY>, CacheValue<VALUE>> getCache() {
+    return cache;
+  }
+
+  public Table<byte[], byte[]> getRawTable() {
+    return rawTable;
+  }
+
+  public CodecRegistry getCodecRegistry() {
+    return codecRegistry;
+  }
+
+  public Class<KEY> getKeyType() {
+    return keyType;
+  }
+
+  public Class<VALUE> getValueType() {
+    return valueType;
+  }
+
   /**
    * Key value implementation for strongly typed tables.
    */

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/cache/CacheKey.java

@@ -0,0 +1,56 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.utils.db.cache;
+
+import java.util.Objects;
+
+/**
+ * CacheKey for the RocksDB table.
+ * @param <KEY>
+ */
+public class CacheKey<KEY> {
+
+  private final KEY key;
+
+  public CacheKey(KEY key) {
+    Objects.requireNonNull(key, "Key Should not be null in CacheKey");
+    this.key = key;
+  }
+
+  public KEY getKey() {
+    return key;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    CacheKey<?> cacheKey = (CacheKey<?>) o;
+    return Objects.equals(key, cacheKey.key);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(key);
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/cache/CacheValue.java

@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.utils.db.cache;
+
+import com.google.common.base.Optional;
+
+/**
+ * CacheValue for the RocksDB Table.
+ * @param <VALUE>
+ */
+public class CacheValue<VALUE> {
+
+  private Optional<VALUE> value;
+  // This value is used for evict entries from cache.
+  // This value is set with ratis transaction context log entry index.
+  private long epoch;
+
+  public CacheValue(Optional<VALUE> value, long epoch) {
+    this.value = value;
+    this.epoch = epoch;
+  }
+
+  public VALUE getValue() {
+    return value.orNull();
+  }
+
+  public long getEpoch() {
+    return epoch;
+  }
+
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/cache/EpochEntry.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.hadoop.utils.db.cache;
+
+import java.util.Objects;
+
+/**
+ * Class used which describes epoch entry. This will be used during deletion
+ * entries from cache for partial table cache.
+ * @param <CACHEKEY>
+ */
+public class EpochEntry<CACHEKEY> implements Comparable<CACHEKEY> {
+
+  private long epoch;
+  private CACHEKEY cachekey;
+
+  EpochEntry(long epoch, CACHEKEY cachekey) {
+    this.epoch = epoch;
+    this.cachekey = cachekey;
+  }
+
+  public long getEpoch() {
+    return epoch;
+  }
+
+  public CACHEKEY getCachekey() {
+    return cachekey;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    EpochEntry<?> that = (EpochEntry<?>) o;
+    return epoch == that.epoch && cachekey == that.cachekey;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(epoch, cachekey);
+  }
+
+  public int compareTo(Object o) {
+    if(this.epoch == ((EpochEntry<?>)o).epoch) {
+      return 0;
+    } else if (this.epoch < ((EpochEntry<?>)o).epoch) {
+      return -1;
+    } else {
+      return 1;
+    }
+  }
+
+}