Make MemTableRepFactory into a Customizable class (#8419)

Summary:
This PR does the following:
-> Makes the MemTableRepFactory into a Customizable class and creatable/configurable via CreateFromString
-> Makes the existing implementations compatible with configurations
-> Moves the "SpecialRepFactory" test class into testutil, accessible via the ObjectRegistry or a NewSpecial API

New tests were added to validate the functionality and all existing tests pass.  db_bench and memtablerep_bench were hand-tested to verify the functionality in those tools.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/8419

Reviewed By: zhichao-cao

Differential Revision: D29558961

Pulled By: mrambacher

fbshipit-source-id: 81b7229636e4e649a0c914e73ac7b0f8454c931c

7 files changed, 173 insertions, 113 deletions

diff --git a/memtable/hash_linklist_rep.cc b/memtable/hash_linklist_rep.cc
index 15bc648a2..cf1f1f85f 100644
--- a/memtable/hash_linklist_rep.cc
+++ b/memtable/hash_linklist_rep.cc
@@ -5,10 +5,10 @@
 //
 
 #ifndef ROCKSDB_LITE
-#include "memtable/hash_linklist_rep.h"
 
 #include <algorithm>
 #include <atomic>
+
 #include "db/memtable.h"
 #include "memory/arena.h"
 #include "memtable/skiplist.h"
@@ -17,6 +17,7 @@
 #include "rocksdb/memtablerep.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/slice_transform.h"
+#include "rocksdb/utilities/options_type.h"
 #include "util/hash.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -820,15 +821,77 @@ Node* HashLinkListRep::FindGreaterOrEqualInBucket(Node* head,
   return x;
 }
 
-} // anon namespace
+struct HashLinkListRepOptions {
+  static const char* kName() { return "HashLinkListRepFactoryOptions"; }
+  size_t bucket_count;
+  uint32_t threshold_use_skiplist;
+  size_t huge_page_tlb_size;
+  int bucket_entries_logging_threshold;
+  bool if_log_bucket_dist_when_flash;
+};
+
+static std::unordered_map<std::string, OptionTypeInfo> hash_linklist_info = {
+    {"bucket_count",
+     {offsetof(struct HashLinkListRepOptions, bucket_count), OptionType::kSizeT,
+      OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
+    {"threshold",
+     {offsetof(struct HashLinkListRepOptions, threshold_use_skiplist),
+      OptionType::kUInt32T, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+    {"huge_page_size",
+     {offsetof(struct HashLinkListRepOptions, huge_page_tlb_size),
+      OptionType::kSizeT, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+    {"logging_threshold",
+     {offsetof(struct HashLinkListRepOptions, bucket_entries_logging_threshold),
+      OptionType::kInt, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+    {"log_when_flash",
+     {offsetof(struct HashLinkListRepOptions, if_log_bucket_dist_when_flash),
+      OptionType::kBoolean, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+};
+
+class HashLinkListRepFactory : public MemTableRepFactory {
+ public:
+  explicit HashLinkListRepFactory(size_t bucket_count,
+                                  uint32_t threshold_use_skiplist,
+                                  size_t huge_page_tlb_size,
+                                  int bucket_entries_logging_threshold,
+                                  bool if_log_bucket_dist_when_flash) {
+    options_.bucket_count = bucket_count;
+    options_.threshold_use_skiplist = threshold_use_skiplist;
+    options_.huge_page_tlb_size = huge_page_tlb_size;
+    options_.bucket_entries_logging_threshold =
+        bucket_entries_logging_threshold;
+    options_.if_log_bucket_dist_when_flash = if_log_bucket_dist_when_flash;
+    RegisterOptions(&options_, &hash_linklist_info);
+  }
+
+  using MemTableRepFactory::CreateMemTableRep;
+  virtual MemTableRep* CreateMemTableRep(
+      const MemTableRep::KeyComparator& compare, Allocator* allocator,
+      const SliceTransform* transform, Logger* logger) override;
+
+  static const char* kClassName() { return "HashLinkListRepFactory"; }
+  static const char* kNickName() { return "hash_linkedlist"; }
+  virtual const char* Name() const override { return kClassName(); }
+  virtual const char* NickName() const override { return kNickName(); }
+
+ private:
+  HashLinkListRepOptions options_;
+};
+
+}  // namespace
 
 MemTableRep* HashLinkListRepFactory::CreateMemTableRep(
     const MemTableRep::KeyComparator& compare, Allocator* allocator,
     const SliceTransform* transform, Logger* logger) {
-  return new HashLinkListRep(compare, allocator, transform, bucket_count_,
-                             threshold_use_skiplist_, huge_page_tlb_size_,
-                             logger, bucket_entries_logging_threshold_,
-                             if_log_bucket_dist_when_flash_);
+  return new HashLinkListRep(
+      compare, allocator, transform, options_.bucket_count,
+      options_.threshold_use_skiplist, options_.huge_page_tlb_size, logger,
+      options_.bucket_entries_logging_threshold,
+      options_.if_log_bucket_dist_when_flash);
 }
 
 MemTableRepFactory* NewHashLinkListRepFactory(
diff --git a/memtable/hash_linklist_rep.h b/memtable/hash_linklist_rep.h
deleted file mode 100644
index 7e7195526..000000000
--- a/memtable/hash_linklist_rep.h
+++ /dev/null
@@ -1,49 +0,0 @@
-// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file. See the AUTHORS file for names of contributors.
-
-#pragma once
-#ifndef ROCKSDB_LITE
-#include "rocksdb/slice_transform.h"
-#include "rocksdb/memtablerep.h"
-
-namespace ROCKSDB_NAMESPACE {
-
-class HashLinkListRepFactory : public MemTableRepFactory {
- public:
-  explicit HashLinkListRepFactory(size_t bucket_count,
-                                  uint32_t threshold_use_skiplist,
-                                  size_t huge_page_tlb_size,
-                                  int bucket_entries_logging_threshold,
-                                  bool if_log_bucket_dist_when_flash)
-      : bucket_count_(bucket_count),
-        threshold_use_skiplist_(threshold_use_skiplist),
-        huge_page_tlb_size_(huge_page_tlb_size),
-        bucket_entries_logging_threshold_(bucket_entries_logging_threshold),
-        if_log_bucket_dist_when_flash_(if_log_bucket_dist_when_flash) {}
-
-  virtual ~HashLinkListRepFactory() {}
-
-  using MemTableRepFactory::CreateMemTableRep;
-  virtual MemTableRep* CreateMemTableRep(
-      const MemTableRep::KeyComparator& compare, Allocator* allocator,
-      const SliceTransform* transform, Logger* logger) override;
-
-  virtual const char* Name() const override {
-    return "HashLinkListRepFactory";
-  }
-
- private:
-  const size_t bucket_count_;
-  const uint32_t threshold_use_skiplist_;
-  const size_t huge_page_tlb_size_;
-  int bucket_entries_logging_threshold_;
-  bool if_log_bucket_dist_when_flash_;
-};
-
-}  // namespace ROCKSDB_NAMESPACE
-#endif  // ROCKSDB_LITE
diff --git a/memtable/hash_skiplist_rep.cc b/memtable/hash_skiplist_rep.cc
index 72fc391e1..dc58046a4 100644
--- a/memtable/hash_skiplist_rep.cc
+++ b/memtable/hash_skiplist_rep.cc
@@ -5,8 +5,6 @@
 //
 
 #ifndef ROCKSDB_LITE
-#include "memtable/hash_skiplist_rep.h"
-
 #include <atomic>
 
 #include "db/memtable.h"
@@ -16,6 +14,7 @@
 #include "rocksdb/memtablerep.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/slice_transform.h"
+#include "rocksdb/utilities/options_type.h"
 #include "util/murmurhash.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -329,13 +328,60 @@ MemTableRep::Iterator* HashSkipListRep::GetDynamicPrefixIterator(Arena* arena) {
   }
 }
 
-} // anon namespace
+struct HashSkipListRepOptions {
+  static const char* kName() { return "HashSkipListRepFactoryOptions"; }
+  size_t bucket_count;
+  int32_t skiplist_height;
+  int32_t skiplist_branching_factor;
+};
+
+static std::unordered_map<std::string, OptionTypeInfo> hash_skiplist_info = {
+    {"bucket_count",
+     {offsetof(struct HashSkipListRepOptions, bucket_count), OptionType::kSizeT,
+      OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
+    {"skiplist_height",
+     {offsetof(struct HashSkipListRepOptions, skiplist_height),
+      OptionType::kInt32T, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+    {"branching_factor",
+     {offsetof(struct HashSkipListRepOptions, skiplist_branching_factor),
+      OptionType::kInt32T, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+};
+
+class HashSkipListRepFactory : public MemTableRepFactory {
+ public:
+  explicit HashSkipListRepFactory(size_t bucket_count, int32_t skiplist_height,
+                                  int32_t skiplist_branching_factor) {
+    options_.bucket_count = bucket_count;
+    options_.skiplist_height = skiplist_height;
+    options_.skiplist_branching_factor = skiplist_branching_factor;
+    RegisterOptions(&options_, &hash_skiplist_info);
+  }
+
+  using MemTableRepFactory::CreateMemTableRep;
+  virtual MemTableRep* CreateMemTableRep(
+      const MemTableRep::KeyComparator& compare, Allocator* allocator,
+      const SliceTransform* transform, Logger* logger) override;
+
+  static const char* kClassName() { return "HashSkipListRepFactory"; }
+  static const char* kNickName() { return "prefix_hash"; }
+
+  virtual const char* Name() const override { return kClassName(); }
+  virtual const char* NickName() const override { return kNickName(); }
+
+ private:
+  HashSkipListRepOptions options_;
+};
+
+}  // namespace
 
 MemTableRep* HashSkipListRepFactory::CreateMemTableRep(
     const MemTableRep::KeyComparator& compare, Allocator* allocator,
     const SliceTransform* transform, Logger* /*logger*/) {
-  return new HashSkipListRep(compare, allocator, transform, bucket_count_,
-                             skiplist_height_, skiplist_branching_factor_);
+  return new HashSkipListRep(compare, allocator, transform,
+                             options_.bucket_count, options_.skiplist_height,
+                             options_.skiplist_branching_factor);
 }
 
 MemTableRepFactory* NewHashSkipListRepFactory(
diff --git a/memtable/hash_skiplist_rep.h b/memtable/hash_skiplist_rep.h
deleted file mode 100644
index 6da5a4e94..000000000
--- a/memtable/hash_skiplist_rep.h
+++ /dev/null
@@ -1,44 +0,0 @@
-// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file. See the AUTHORS file for names of contributors.
-
-#pragma once
-#ifndef ROCKSDB_LITE
-#include "rocksdb/slice_transform.h"
-#include "rocksdb/memtablerep.h"
-
-namespace ROCKSDB_NAMESPACE {
-
-class HashSkipListRepFactory : public MemTableRepFactory {
- public:
-  explicit HashSkipListRepFactory(
-    size_t bucket_count,
-    int32_t skiplist_height,
-    int32_t skiplist_branching_factor)
-      : bucket_count_(bucket_count),
-        skiplist_height_(skiplist_height),
-        skiplist_branching_factor_(skiplist_branching_factor) { }
-
-  virtual ~HashSkipListRepFactory() {}
-
-  using MemTableRepFactory::CreateMemTableRep;
-  virtual MemTableRep* CreateMemTableRep(
-      const MemTableRep::KeyComparator& compare, Allocator* allocator,
-      const SliceTransform* transform, Logger* logger) override;
-
-  virtual const char* Name() const override {
-    return "HashSkipListRepFactory";
-  }
-
- private:
-  const size_t bucket_count_;
-  const int32_t skiplist_height_;
-  const int32_t skiplist_branching_factor_;
-};
-
-}  // namespace ROCKSDB_NAMESPACE
-#endif  // ROCKSDB_LITE
diff --git a/memtable/memtablerep_bench.cc b/memtable/memtablerep_bench.cc
index d476d03fb..1eaa7658f 100644
--- a/memtable/memtablerep_bench.cc
+++ b/memtable/memtablerep_bench.cc
@@ -28,6 +28,7 @@ int main() {
 #include "port/port.h"
 #include "port/stack_trace.h"
 #include "rocksdb/comparator.h"
+#include "rocksdb/convenience.h"
 #include "rocksdb/memtablerep.h"
 #include "rocksdb/options.h"
 #include "rocksdb/slice_transform.h"
@@ -581,13 +582,15 @@ int main(int argc, char** argv) {
 #ifndef ROCKSDB_LITE
   } else if (FLAGS_memtablerep == "vector") {
     factory.reset(new ROCKSDB_NAMESPACE::VectorRepFactory);
-  } else if (FLAGS_memtablerep == "hashskiplist") {
+  } else if (FLAGS_memtablerep == "hashskiplist" ||
+             FLAGS_memtablerep == "prefix_hash") {
     factory.reset(ROCKSDB_NAMESPACE::NewHashSkipListRepFactory(
         FLAGS_bucket_count, FLAGS_hashskiplist_height,
         FLAGS_hashskiplist_branching_factor));
     options.prefix_extractor.reset(
         ROCKSDB_NAMESPACE::NewFixedPrefixTransform(FLAGS_prefix_length));
-  } else if (FLAGS_memtablerep == "hashlinklist") {
+  } else if (FLAGS_memtablerep == "hashlinklist" ||
+             FLAGS_memtablerep == "hash_linkedlist") {
     factory.reset(ROCKSDB_NAMESPACE::NewHashLinkListRepFactory(
         FLAGS_bucket_count, FLAGS_huge_page_tlb_size,
         FLAGS_bucket_entries_logging_threshold,
@@ -596,8 +599,16 @@ int main(int argc, char** argv) {
         ROCKSDB_NAMESPACE::NewFixedPrefixTransform(FLAGS_prefix_length));
 #endif  // ROCKSDB_LITE
   } else {
-    fprintf(stdout, "Unknown memtablerep: %s\n", FLAGS_memtablerep.c_str());
-    exit(1);
+    ROCKSDB_NAMESPACE::ConfigOptions config_options;
+    config_options.ignore_unsupported_options = false;
+
+    ROCKSDB_NAMESPACE::Status s =
+        ROCKSDB_NAMESPACE::MemTableRepFactory::CreateFromString(
+            config_options, FLAGS_memtablerep, &factory);
+    if (!s.ok()) {
+      fprintf(stdout, "Unknown memtablerep: %s\n", s.ToString().c_str());
+      exit(1);
+    }
   }
 
   ROCKSDB_NAMESPACE::InternalKeyComparator internal_key_comp(
diff --git a/memtable/skiplistrep.cc b/memtable/skiplistrep.cc
index d7f78672f..016515c44 100644
--- a/memtable/skiplistrep.cc
+++ b/memtable/skiplistrep.cc
@@ -9,6 +9,8 @@
 #include "memory/arena.h"
 #include "memtable/inlineskiplist.h"
 #include "rocksdb/memtablerep.h"
+#include "rocksdb/utilities/options_type.h"
+#include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
 namespace {
@@ -335,6 +337,27 @@ public:
 };
 }
 
+static std::unordered_map<std::string, OptionTypeInfo> skiplist_factory_info = {
+#ifndef ROCKSDB_LITE
+    {"lookahead",
+     {0, OptionType::kSizeT, OptionVerificationType::kNormal,
+      OptionTypeFlags::kDontSerialize /*Since it is part of the ID*/}},
+#endif
+};
+
+SkipListFactory::SkipListFactory(size_t lookahead) : lookahead_(lookahead) {
+  RegisterOptions("SkipListFactoryOptions", &lookahead_,
+                  &skiplist_factory_info);
+}
+
+std::string SkipListFactory::GetId() const {
+  std::string id = Name();
+  if (lookahead_ > 0) {
+    id.append(":").append(ROCKSDB_NAMESPACE::ToString(lookahead_));
+  }
+  return id;
+}
+
 MemTableRep* SkipListFactory::CreateMemTableRep(
     const MemTableRep::KeyComparator& compare, Allocator* allocator,
     const SliceTransform* transform, Logger* /*logger*/) {
diff --git a/memtable/vectorrep.cc b/memtable/vectorrep.cc
index 7dde80b22..2386e46ab 100644
--- a/memtable/vectorrep.cc
+++ b/memtable/vectorrep.cc
@@ -4,18 +4,18 @@
 //  (found in the LICENSE.Apache file in the root directory).
 //
 #ifndef ROCKSDB_LITE
-#include "rocksdb/memtablerep.h"
-
-#include <unordered_set>
-#include <set>
-#include <memory>
 #include <algorithm>
+#include <memory>
+#include <set>
 #include <type_traits>
+#include <unordered_set>
 
 #include "db/memtable.h"
 #include "memory/arena.h"
 #include "memtable/stl_wrappers.h"
 #include "port/port.h"
+#include "rocksdb/memtablerep.h"
+#include "rocksdb/utilities/options_type.h"
 #include "util/mutexlock.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -292,6 +292,16 @@ MemTableRep::Iterator* VectorRep::GetIterator(Arena* arena) {
 }
 } // anon namespace
 
+static std::unordered_map<std::string, OptionTypeInfo> vector_rep_table_info = {
+    {"count",
+     {0, OptionType::kSizeT, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
+};
+
+VectorRepFactory::VectorRepFactory(size_t count) : count_(count) {
+  RegisterOptions("VectorRepFactoryOptions", &count_, &vector_rep_table_info);
+}
+
 MemTableRep* VectorRepFactory::CreateMemTableRep(
     const MemTableRep::KeyComparator& compare, Allocator* allocator,
     const SliceTransform*, Logger* /*logger*/) {