Added GetFactoryCount/Names/Types to ObjectRegistry (#9358)

Summary:
These methods allow for more thorough testing of the ObjectRegistry and Customizable infrastructure in a simpler manner.  With this change, the Customizable tests can now check what factories are registered and attempt to create each of them in a systematic fashion.

With this change, I think all of the factories registered with the ObjectRegistry/CreateFromString are now tested via the customizable_test classes.

Note that there were a few other minor changes.  There was a "posix://*" register with the ObjectRegistry which was missed during the PatternEntry conversion -- these changes found that.  The nickname and default names for the FileSystem classes was also inverted.

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

Reviewed By: pdillinger

Differential Revision: D33433542

Pulled By: mrambacher

fbshipit-source-id: 9a32da74e6620745b4eeffb2712be70eeeadfa7e

env/env_encryption.cc

@@ -1139,6 +1139,15 @@ CTREncryptionProvider::CTREncryptionProvider(
   RegisterOptions("Cipher", &cipher_, &ctr_encryption_provider_type_info);
 }
 
+bool CTREncryptionProvider::IsInstanceOf(const std::string& name) const {
+  // Special case for test purposes.
+  if (name == "1://test" && cipher_ != nullptr) {
+    return cipher_->IsInstanceOf(ROT13BlockCipher::kClassName());
+  } else {
+    return EncryptionProvider::IsInstanceOf(name);
+  }
+}
+
 // GetPrefixLength returns the length of the prefix that is added to every file
 // and used for storing encryption options.
 // For optimal performance, the prefix length should be a multiple of

env/env_encryption_ctr.h

@@ -66,7 +66,7 @@ class CTREncryptionProvider : public EncryptionProvider {
 
   static const char* kClassName() { return "CTR"; }
   const char* Name() const override { return kClassName(); }
-
+  bool IsInstanceOf(const std::string& name) const override;
   // GetPrefixLength returns the length of the prefix that is added to every
   // file
   // and used for storing encryption options.

env/env_posix.cc

@@ -130,8 +130,8 @@ class PosixDynamicLibrary : public DynamicLibrary {
 class PosixClock : public SystemClock {
  public:
   static const char* kClassName() { return "PosixClock"; }
-  const char* Name() const override { return kClassName(); }
-  const char* NickName() const override { return kDefaultName(); }
+  const char* Name() const override { return kDefaultName(); }
+  const char* NickName() const override { return kClassName(); }
 
   uint64_t NowMicros() override {
     struct timeval tv;

env/fs_posix.cc

@@ -146,6 +146,13 @@ class PosixFileSystem : public FileSystem {
   const char* NickName() const override { return kDefaultName(); }
 
   ~PosixFileSystem() override {}
+  bool IsInstanceOf(const std::string& name) const override {
+    if (name == "posix") {
+      return true;
+    } else {
+      return FileSystem::IsInstanceOf(name);
+    }
+  }
 
   void SetFD_CLOEXEC(int fd, const EnvOptions* options) {
     if ((options == nullptr || options->set_fd_cloexec) && fd > 0) {

include/rocksdb/memtablerep.h

@@ -300,6 +300,9 @@ class MemTableRepFactory : public Customizable {
   static Status CreateFromString(const ConfigOptions& config_options,
                                  const std::string& id,
                                  std::unique_ptr<MemTableRepFactory>* factory);
+  static Status CreateFromString(const ConfigOptions& config_options,
+                                 const std::string& id,
+                                 std::shared_ptr<MemTableRepFactory>* factory);
 
   virtual MemTableRep* CreateMemTableRep(const MemTableRep::KeyComparator&,
                                          Allocator*, const SliceTransform*,

include/rocksdb/utilities/object_registry.h

@@ -13,6 +13,7 @@
 #include <mutex>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 #include "rocksdb/status.h"
@@ -217,6 +218,18 @@ class ObjectLibrary {
   // @param num_types returns how many unique types are registered.
   size_t GetFactoryCount(size_t* num_types) const;
 
+  // Returns the number of factories registered for this library
+  // for the input type.
+  // @param num_types returns how many unique types are registered.
+  size_t GetFactoryCount(const std::string& type) const;
+
+  // Returns the registered factory names for the input type
+  // names is updated to include the names for the type
+  void GetFactoryNames(const std::string& type,
+                       std::vector<std::string>* names) const;
+
+  void GetFactoryTypes(std::unordered_set<std::string>* types) const;
+
   void Dump(Logger* logger) const;
 
   // Registers the factory with the library for the name.
@@ -497,6 +510,18 @@ class ObjectRegistry {
     }
   }
 
+  // Returns the number of factories registered for this library
+  // for the input type.
+  // @param num_types returns how many unique types are registered.
+  size_t GetFactoryCount(const std::string& type) const;
+
+  // Returns the names of registered factories for the input type.
+  // names is updated to include the names for the type
+  void GetFactoryNames(const std::string& type,
+                       std::vector<std::string>* names) const;
+
+  void GetFactoryTypes(std::unordered_set<std::string>* types) const;
+
   // Dump the contents of the registry to the logger
   void Dump(Logger* logger) const;
 

options/cf_options.cc

@@ -596,10 +596,7 @@ static std::unordered_map<std::string, OptionTypeInfo>
             auto* shared =
                 static_cast<std::shared_ptr<MemTableRepFactory>*>(addr);
             Status s =
-                MemTableRepFactory::CreateFromString(opts, value, &factory);
-            if (factory && s.ok()) {
-              shared->reset(factory.release());
-            }
+                MemTableRepFactory::CreateFromString(opts, value, shared);
             return s;
           }}},
         {"memtable",
@@ -612,10 +609,7 @@ static std::unordered_map<std::string, OptionTypeInfo>
             auto* shared =
                 static_cast<std::shared_ptr<MemTableRepFactory>*>(addr);
             Status s =
-                MemTableRepFactory::CreateFromString(opts, value, &factory);
-            if (factory && s.ok()) {
-              shared->reset(factory.release());
-            }
+                MemTableRepFactory::CreateFromString(opts, value, shared);
             return s;
           }}},
         {"table_factory",

options/customizable_test.cc

@@ -13,8 +13,11 @@
 #include <cinttypes>
 #include <cstring>
 #include <unordered_map>
+#include <unordered_set>
 
 #include "db/db_test_util.h"
+#include "memory/jemalloc_nodump_allocator.h"
+#include "memory/memkind_kmem_allocator.h"
 #include "options/options_helper.h"
 #include "options/options_parser.h"
 #include "port/stack_trace.h"
@@ -1655,6 +1658,190 @@ class LoadCustomizableTest : public testing::Test {
 #endif  // !ROCKSDB_LITE
   }
 
+  template <typename T, typename U>
+  Status TestCreateStatic(const std::string& name, U** result,
+                          bool delete_result = false) {
+    Status s = T::CreateFromString(config_options_, name, result);
+    if (s.ok()) {
+      EXPECT_NE(*result, nullptr);
+      EXPECT_TRUE(*result != nullptr && (*result)->IsInstanceOf(name));
+    }
+    if (delete_result) {
+      delete *result;
+      *result = nullptr;
+    }
+    return s;
+  }
+
+  template <typename T, typename U>
+  std::shared_ptr<U> ExpectCreateShared(const std::string& name,
+                                        std::shared_ptr<U>* object) {
+    EXPECT_OK(T::CreateFromString(config_options_, name, object));
+    EXPECT_NE(object->get(), nullptr);
+    EXPECT_TRUE(object->get()->IsInstanceOf(name));
+    return *object;
+  }
+
+  template <typename T>
+  std::shared_ptr<T> ExpectCreateShared(const std::string& name) {
+    std::shared_ptr<T> result;
+    return ExpectCreateShared<T>(name, &result);
+  }
+
+  template <typename T, typename U>
+  Status TestExpectedBuiltins(
+      const std::string& mock, const std::unordered_set<std::string>& expected,
+      std::shared_ptr<U>* object, std::vector<std::string>* failed,
+      const std::function<std::vector<std::string>(const std::string&)>& alt =
+          nullptr) {
+    std::unordered_set<std::string> factories = expected;
+    Status s = T::CreateFromString(config_options_, mock, object);
+    EXPECT_NOK(s);
+#ifndef ROCKSDB_LITE
+    std::vector<std::string> builtins;
+    ObjectLibrary::Default()->GetFactoryNames(T::Type(), &builtins);
+    factories.insert(builtins.begin(), builtins.end());
+#endif  // ROCKSDB_LITE
+    Status result;
+    int created = 0;
+    for (const auto& name : factories) {
+      created++;
+      s = T::CreateFromString(config_options_, name, object);
+      if (!s.ok() && alt != nullptr) {
+        for (const auto& alt_name : alt(name)) {
+          s = T::CreateFromString(config_options_, alt_name, object);
+          if (s.ok()) {
+            break;
+          }
+        }
+      }
+      if (!s.ok()) {
+        result = s;
+        failed->push_back(name);
+      } else {
+        EXPECT_NE(object->get(), nullptr);
+        EXPECT_TRUE(object->get()->IsInstanceOf(name));
+      }
+    }
+#ifndef ROCKSDB_LITE
+    std::vector<std::string> plugins;
+    ObjectRegistry::Default()->GetFactoryNames(T::Type(), &plugins);
+    if (plugins.size() > builtins.size()) {
+      for (const auto& name : plugins) {
+        if (factories.find(name) == factories.end()) {
+          created++;
+          s = T::CreateFromString(config_options_, name, object);
+          if (!s.ok() && alt != nullptr) {
+            for (const auto& alt_name : alt(name)) {
+              s = T::CreateFromString(config_options_, alt_name, object);
+              if (s.ok()) {
+                break;
+              }
+            }
+          }
+          if (!s.ok()) {
+            failed->push_back(name);
+            if (result.ok()) {
+              result = s;
+            }
+            printf("%s: Failed creating plugin[%s]: %s\n", T::Type(),
+                   name.c_str(), s.ToString().c_str());
+          } else if (object->get() == nullptr ||
+                     !object->get()->IsInstanceOf(name)) {
+            failed->push_back(name);
+            printf("%s: Invalid plugin[%s]\n", T::Type(), name.c_str());
+          }
+        }
+      }
+    }
+    printf("%s: Created %d (expected+builtins+plugins %d+%d+%d) %d Failed\n",
+           T::Type(), created, (int)expected.size(),
+           (int)(factories.size() - expected.size()),
+           (int)(plugins.size() - builtins.size()), (int)failed->size());
+#else
+    printf("%s: Created %d (expected %d) %d Failed\n", T::Type(), created,
+           (int)expected.size(), (int)failed->size());
+#endif  // ROCKSDB_LITE
+    return result;
+  }
+
+  template <typename T>
+  Status TestSharedBuiltins(const std::string& mock,
+                            const std::string& expected,
+                            std::vector<std::string>* failed = nullptr) {
+    std::unordered_set<std::string> values;
+    if (!expected.empty()) {
+      values.insert(expected);
+    }
+    std::shared_ptr<T> object;
+    if (failed != nullptr) {
+      return TestExpectedBuiltins<T>(mock, values, &object, failed);
+    } else {
+      std::vector<std::string> failures;
+      Status s = TestExpectedBuiltins<T>(mock, values, &object, &failures);
+      EXPECT_EQ(0U, failures.size());
+      return s;
+    }
+  }
+
+  template <typename T, typename U>
+  Status TestStaticBuiltins(const std::string& mock, U** object,
+                            const std::unordered_set<std::string>& expected,
+                            std::vector<std::string>* failed,
+                            bool delete_objects = false) {
+    std::unordered_set<std::string> factories = expected;
+    Status s = TestCreateStatic<T>(mock, object, delete_objects);
+    EXPECT_NOK(s);
+#ifndef ROCKSDB_LITE
+    std::vector<std::string> builtins;
+    ObjectLibrary::Default()->GetFactoryNames(T::Type(), &builtins);
+    factories.insert(builtins.begin(), builtins.end());
+#endif  // ROCKSDB_LITE
+    int created = 0;
+    Status result;
+    for (const auto& name : factories) {
+      created++;
+      s = TestCreateStatic<T>(name, object, delete_objects);
+      if (!s.ok()) {
+        result = s;
+        failed->push_back(name);
+      }
+    }
+#ifndef ROCKSDB_LITE
+    std::vector<std::string> plugins;
+    ObjectRegistry::Default()->GetFactoryNames(T::Type(), &plugins);
+    if (plugins.size() > builtins.size()) {
+      for (const auto& name : plugins) {
+        if (factories.find(name) == factories.end()) {
+          created++;
+          s = T::CreateFromString(config_options_, name, object);
+          if (!s.ok() || *object == nullptr ||
+              !((*object)->IsInstanceOf(name))) {
+            failed->push_back(name);
+            if (result.ok() && !s.ok()) {
+              result = s;
+            }
+            printf("%s: Failed creating plugin[%s]: %s\n", T::Type(),
+                   name.c_str(), s.ToString().c_str());
+          }
+          if (delete_objects) {
+            delete *object;
+            *object = nullptr;
+          }
+        }
+      }
+    }
+    printf("%s: Created %d (expected+builtins+plugins %d+%d+%d) %d Failed\n",
+           T::Type(), created, (int)expected.size(),
+           (int)(factories.size() - expected.size()),
+           (int)(plugins.size() - builtins.size()), (int)failed->size());
+#else
+    printf("%s: Created %d (expected %d) %d Failed\n", T::Type(), created,
+           (int)expected.size(), (int)failed->size());
+#endif  // ROCKSDB_LITE
+    return result;
+  }
+
  protected:
   DBOptions db_opts_;
   ColumnFamilyOptions cf_opts_;
@@ -1662,13 +1849,9 @@ class LoadCustomizableTest : public testing::Test {
 };
 
 TEST_F(LoadCustomizableTest, LoadTableFactoryTest) {
-  std::shared_ptr<TableFactory> factory;
-  ASSERT_NOK(TableFactory::CreateFromString(
-      config_options_, mock::MockTableFactory::kClassName(), &factory));
-  ASSERT_OK(TableFactory::CreateFromString(
-      config_options_, TableFactory::kBlockBasedTableName(), &factory));
-  ASSERT_NE(factory, nullptr);
-  ASSERT_STREQ(factory->Name(), TableFactory::kBlockBasedTableName());
+  ASSERT_OK(
+      TestSharedBuiltins<TableFactory>(mock::MockTableFactory::kClassName(),
+                                       TableFactory::kBlockBasedTableName()));
 #ifndef ROCKSDB_LITE
   std::string opts_str = "table_factory=";
   ASSERT_OK(GetColumnFamilyOptionsFromString(
@@ -1679,10 +1862,7 @@ TEST_F(LoadCustomizableTest, LoadTableFactoryTest) {
                TableFactory::kBlockBasedTableName());
 #endif  // ROCKSDB_LITE
   if (RegisterTests("Test")) {
-    ASSERT_OK(TableFactory::CreateFromString(
-        config_options_, mock::MockTableFactory::kClassName(), &factory));
-    ASSERT_NE(factory, nullptr);
-    ASSERT_STREQ(factory->Name(), mock::MockTableFactory::kClassName());
+    ExpectCreateShared<TableFactory>(mock::MockTableFactory::kClassName());
 #ifndef ROCKSDB_LITE
     ASSERT_OK(GetColumnFamilyOptionsFromString(
         config_options_, cf_opts_,
@@ -1695,151 +1875,97 @@ TEST_F(LoadCustomizableTest, LoadTableFactoryTest) {
 }
 
 TEST_F(LoadCustomizableTest, LoadFileSystemTest) {
-  ColumnFamilyOptions cf_opts;
-  std::shared_ptr<FileSystem> result;
-  ASSERT_NOK(FileSystem::CreateFromString(
-      config_options_, DummyFileSystem::kClassName(), &result));
-  ASSERT_OK(FileSystem::CreateFromString(config_options_,
-                                         FileSystem::kDefaultName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_TRUE(result->IsInstanceOf(FileSystem::kDefaultName()));
+  ASSERT_OK(TestSharedBuiltins<FileSystem>(DummyFileSystem::kClassName(),
+                                           FileSystem::kDefaultName()));
   if (RegisterTests("Test")) {
-    ASSERT_OK(FileSystem::CreateFromString(
-        config_options_, DummyFileSystem::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), DummyFileSystem::kClassName());
-    ASSERT_FALSE(result->IsInstanceOf(FileSystem::kDefaultName()));
+    auto fs = ExpectCreateShared<FileSystem>(DummyFileSystem::kClassName());
+    ASSERT_FALSE(fs->IsInstanceOf(FileSystem::kDefaultName()));
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadSecondaryCacheTest) {
-  std::shared_ptr<SecondaryCache> result;
-  ASSERT_NOK(SecondaryCache::CreateFromString(
-      config_options_, TestSecondaryCache::kClassName(), &result));
+  ASSERT_OK(
+      TestSharedBuiltins<SecondaryCache>(TestSecondaryCache::kClassName(), ""));
   if (RegisterTests("Test")) {
-    ASSERT_OK(SecondaryCache::CreateFromString(
-        config_options_, TestSecondaryCache::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), TestSecondaryCache::kClassName());
+    ExpectCreateShared<SecondaryCache>(TestSecondaryCache::kClassName());
   }
 }
 
 #ifndef ROCKSDB_LITE
 TEST_F(LoadCustomizableTest, LoadSstPartitionerFactoryTest) {
-  std::shared_ptr<SstPartitionerFactory> factory;
-  ASSERT_NOK(SstPartitionerFactory::CreateFromString(config_options_, "Mock",
-                                                     &factory));
-  ASSERT_OK(SstPartitionerFactory::CreateFromString(
-      config_options_, SstPartitionerFixedPrefixFactory::kClassName(),
-      &factory));
-  ASSERT_NE(factory, nullptr);
-  ASSERT_STREQ(factory->Name(), SstPartitionerFixedPrefixFactory::kClassName());
-
+  ASSERT_OK(TestSharedBuiltins<SstPartitionerFactory>(
+      "Mock", SstPartitionerFixedPrefixFactory::kClassName()));
   if (RegisterTests("Test")) {
-    ASSERT_OK(SstPartitionerFactory::CreateFromString(config_options_, "Mock",
-                                                      &factory));
-    ASSERT_NE(factory, nullptr);
-    ASSERT_STREQ(factory->Name(), "Mock");
+    ExpectCreateShared<SstPartitionerFactory>("Mock");
   }
 }
 #endif  // ROCKSDB_LITE
 
 TEST_F(LoadCustomizableTest, LoadChecksumGenFactoryTest) {
-  std::shared_ptr<FileChecksumGenFactory> factory;
-  ASSERT_NOK(FileChecksumGenFactory::CreateFromString(config_options_, "Mock",
-                                                      &factory));
-  ASSERT_OK(FileChecksumGenFactory::CreateFromString(
-      config_options_, FileChecksumGenCrc32cFactory::kClassName(), &factory));
-  ASSERT_NE(factory, nullptr);
-  ASSERT_STREQ(factory->Name(), FileChecksumGenCrc32cFactory::kClassName());
-
+  ASSERT_OK(TestSharedBuiltins<FileChecksumGenFactory>("Mock", ""));
   if (RegisterTests("Test")) {
-    ASSERT_OK(FileChecksumGenFactory::CreateFromString(config_options_, "Mock",
-                                                       &factory));
-    ASSERT_NE(factory, nullptr);
-    ASSERT_STREQ(factory->Name(), "Mock");
+    ExpectCreateShared<FileChecksumGenFactory>("Mock");
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadTablePropertiesCollectorFactoryTest) {
-  std::shared_ptr<TablePropertiesCollectorFactory> factory;
-  ASSERT_NOK(TablePropertiesCollectorFactory::CreateFromString(
-      config_options_, MockTablePropertiesCollectorFactory::kClassName(),
-      &factory));
+  ASSERT_OK(TestSharedBuiltins<TablePropertiesCollectorFactory>(
+      MockTablePropertiesCollectorFactory::kClassName(), ""));
   if (RegisterTests("Test")) {
-    ASSERT_OK(TablePropertiesCollectorFactory::CreateFromString(
-        config_options_, MockTablePropertiesCollectorFactory::kClassName(),
-        &factory));
-    ASSERT_NE(factory, nullptr);
-    ASSERT_STREQ(factory->Name(),
-                 MockTablePropertiesCollectorFactory::kClassName());
+    ExpectCreateShared<TablePropertiesCollectorFactory>(
+        MockTablePropertiesCollectorFactory::kClassName());
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadComparatorTest) {
   const Comparator* bytewise = BytewiseComparator();
   const Comparator* reverse = ReverseBytewiseComparator();
-
   const Comparator* result = nullptr;
-  ASSERT_NOK(Comparator::CreateFromString(
-      config_options_, test::SimpleSuffixReverseComparator::kClassName(),
-      &result));
-  ASSERT_OK(
-      Comparator::CreateFromString(config_options_, bytewise->Name(), &result));
-  ASSERT_EQ(result, bytewise);
-  ASSERT_OK(
-      Comparator::CreateFromString(config_options_, reverse->Name(), &result));
-  ASSERT_EQ(result, reverse);
-
+  std::unordered_set<std::string> expected = {bytewise->Name(),
+                                              reverse->Name()};
+  std::vector<std::string> failures;
+  ASSERT_OK(TestStaticBuiltins<Comparator>(
+      test::SimpleSuffixReverseComparator::kClassName(), &result, expected,
+      &failures));
   if (RegisterTests("Test")) {
-    ASSERT_OK(Comparator::CreateFromString(
-        config_options_, test::SimpleSuffixReverseComparator::kClassName(),
-        &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(),
-                 test::SimpleSuffixReverseComparator::kClassName());
+    ASSERT_OK(TestCreateStatic<Comparator>(
+        test::SimpleSuffixReverseComparator::kClassName(), &result));
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadSliceTransformFactoryTest) {
   std::shared_ptr<const SliceTransform> result;
-  ASSERT_NOK(
-      SliceTransform::CreateFromString(config_options_, "Mock", &result));
-  ASSERT_OK(
-      SliceTransform::CreateFromString(config_options_, "fixed:16", &result));
-  ASSERT_NE(result.get(), nullptr);
-  ASSERT_TRUE(result->IsInstanceOf("fixed"));
+  std::vector<std::string> failures;
+  std::unordered_set<std::string> expected = {"rocksdb.Noop", "fixed",
+                                              "rocksdb.FixedPrefix", "capped",
+                                              "rocksdb.CappedPrefix"};
+  ASSERT_OK(TestExpectedBuiltins<SliceTransform>(
+      "Mock", expected, &result, &failures, [](const std::string& name) {
+        std::vector<std::string> names = {name + ":22", name + ".22"};
+        return names;
+      }));
   ASSERT_OK(SliceTransform::CreateFromString(
       config_options_, "rocksdb.FixedPrefix.22", &result));
   ASSERT_NE(result.get(), nullptr);
   ASSERT_TRUE(result->IsInstanceOf("fixed"));
-
-  ASSERT_OK(
-      SliceTransform::CreateFromString(config_options_, "capped:16", &result));
-  ASSERT_NE(result.get(), nullptr);
-  ASSERT_TRUE(result->IsInstanceOf("capped"));
-
   ASSERT_OK(SliceTransform::CreateFromString(
-      config_options_, "rocksdb.CappedPrefix.11", &result));
+      config_options_, "rocksdb.CappedPrefix.22", &result));
   ASSERT_NE(result.get(), nullptr);
   ASSERT_TRUE(result->IsInstanceOf("capped"));
-
   if (RegisterTests("Test")) {
-    ASSERT_OK(
-        SliceTransform::CreateFromString(config_options_, "Mock", &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "Mock");
+    ExpectCreateShared<SliceTransform>("Mock", &result);
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadStatisticsTest) {
-  std::shared_ptr<Statistics> stats;
-  ASSERT_NOK(Statistics::CreateFromString(
-      config_options_, TestStatistics::kClassName(), &stats));
+  ASSERT_OK(TestSharedBuiltins<Statistics>(TestStatistics::kClassName(),
+                                           "BasicStatistics"));
+  // Empty will create a default BasicStatistics
   ASSERT_OK(
-      Statistics::CreateFromString(config_options_, "BasicStatistics", &stats));
-  ASSERT_NE(stats, nullptr);
-  ASSERT_EQ(stats->Name(), std::string("BasicStatistics"));
+      Statistics::CreateFromString(config_options_, "", &db_opts_.statistics));
+  ASSERT_NE(db_opts_.statistics, nullptr);
+  ASSERT_STREQ(db_opts_.statistics->Name(), "BasicStatistics");
+
 #ifndef ROCKSDB_LITE
   ASSERT_NOK(GetDBOptionsFromString(config_options_, db_opts_,
                                     "statistics=Test", &db_opts_));
@@ -1849,10 +1975,7 @@ TEST_F(LoadCustomizableTest, LoadStatisticsTest) {
   ASSERT_STREQ(db_opts_.statistics->Name(), "BasicStatistics");
 
   if (RegisterTests("test")) {
-    ASSERT_OK(Statistics::CreateFromString(
-        config_options_, TestStatistics::kClassName(), &stats));
-    ASSERT_NE(stats, nullptr);
-    ASSERT_STREQ(stats->Name(), TestStatistics::kClassName());
+    auto stats = ExpectCreateShared<Statistics>(TestStatistics::kClassName());
 
     ASSERT_OK(GetDBOptionsFromString(config_options_, db_opts_,
                                      "statistics=Test", &db_opts_));
@@ -1883,167 +2006,90 @@ TEST_F(LoadCustomizableTest, LoadStatisticsTest) {
 }
 
 TEST_F(LoadCustomizableTest, LoadMemTableRepFactoryTest) {
-  std::unique_ptr<MemTableRepFactory> result;
-  ASSERT_NOK(MemTableRepFactory::CreateFromString(
-      config_options_, "SpecialSkipListFactory", &result));
-  ASSERT_OK(MemTableRepFactory::CreateFromString(
-      config_options_, SkipListFactory::kClassName(), &result));
-  ASSERT_NE(result.get(), nullptr);
-  ASSERT_TRUE(result->IsInstanceOf(SkipListFactory::kClassName()));
+  std::unordered_set<std::string> expected = {
+      SkipListFactory::kClassName(),
+      SkipListFactory::kNickName(),
+  };
 
+  std::vector<std::string> failures;
+  std::shared_ptr<MemTableRepFactory> factory;
+  Status s = TestExpectedBuiltins<MemTableRepFactory>(
+      "SpecialSkipListFactory", expected, &factory, &failures);
+  // There is a "cuckoo" factory registered that we expect to fail.  Ignore the
+  // error if this is the one
+  if (s.ok() || failures.size() > 1 || failures[0] != "cuckoo") {
+    ASSERT_OK(s);
+  }
   if (RegisterTests("Test")) {
-    ASSERT_OK(MemTableRepFactory::CreateFromString(
-        config_options_, "SpecialSkipListFactory", &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "SpecialSkipListFactory");
+    ExpectCreateShared<MemTableRepFactory>("SpecialSkipListFactory");
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadMergeOperatorTest) {
   std::shared_ptr<MergeOperator> result;
-
-  ASSERT_NOK(
-      MergeOperator::CreateFromString(config_options_, "Changling", &result));
-  //**TODO: MJR: Use the constants when these names are in public classes
-  ASSERT_OK(MergeOperator::CreateFromString(config_options_, "put", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "PutOperator");
-  ASSERT_OK(
-      MergeOperator::CreateFromString(config_options_, "PutOperator", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "PutOperator");
-  ASSERT_OK(
-      MergeOperator::CreateFromString(config_options_, "put_v1", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "PutOperator");
-
-  ASSERT_OK(
-      MergeOperator::CreateFromString(config_options_, "uint64add", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "UInt64AddOperator");
-  ASSERT_OK(MergeOperator::CreateFromString(config_options_,
-                                            "UInt64AddOperator", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "UInt64AddOperator");
-
-  ASSERT_OK(MergeOperator::CreateFromString(config_options_, "max", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "MaxOperator");
-  ASSERT_OK(
-      MergeOperator::CreateFromString(config_options_, "MaxOperator", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "MaxOperator");
+  std::vector<std::string> failed;
+  std::unordered_set<std::string> expected = {
+      "put", "put_v1",      "PutOperator", "uint64add", "UInt64AddOperator",
+      "max", "MaxOperator",
+  };
 #ifndef ROCKSDB_LITE
-  ASSERT_OK(MergeOperator::CreateFromString(
-      config_options_, StringAppendOperator::kNickName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), StringAppendOperator::kClassName());
-  ASSERT_OK(MergeOperator::CreateFromString(
-      config_options_, StringAppendOperator::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), StringAppendOperator::kClassName());
-
-  ASSERT_OK(MergeOperator::CreateFromString(
-      config_options_, StringAppendTESTOperator::kNickName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), StringAppendTESTOperator::kClassName());
-  ASSERT_OK(MergeOperator::CreateFromString(
-      config_options_, StringAppendTESTOperator::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), StringAppendTESTOperator::kClassName());
-
-  ASSERT_OK(MergeOperator::CreateFromString(config_options_,
-                                            SortList::kNickName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), SortList::kClassName());
-  ASSERT_OK(MergeOperator::CreateFromString(config_options_,
-                                            SortList::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), SortList::kClassName());
-
-  ASSERT_OK(MergeOperator::CreateFromString(
-      config_options_, BytesXOROperator::kNickName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), BytesXOROperator::kClassName());
-  ASSERT_OK(MergeOperator::CreateFromString(
-      config_options_, BytesXOROperator::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), BytesXOROperator::kClassName());
+  expected.insert({
+      StringAppendOperator::kClassName(),
+      StringAppendOperator::kNickName(),
+      StringAppendTESTOperator::kClassName(),
+      StringAppendTESTOperator::kNickName(),
+      SortList::kClassName(),
+      SortList::kNickName(),
+      BytesXOROperator::kClassName(),
+      BytesXOROperator::kNickName(),
+  });
 #endif  // ROCKSDB_LITE
-  ASSERT_NOK(
-      MergeOperator::CreateFromString(config_options_, "Changling", &result));
+
+  ASSERT_OK(TestExpectedBuiltins<MergeOperator>("Changling", expected, &result,
+                                                &failed));
   if (RegisterTests("Test")) {
-    ASSERT_OK(
-        MergeOperator::CreateFromString(config_options_, "Changling", &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "ChanglingMergeOperator");
+    ExpectCreateShared<MergeOperator>("Changling");
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadCompactionFilterFactoryTest) {
-  std::shared_ptr<CompactionFilterFactory> result;
-
-  ASSERT_NOK(CompactionFilterFactory::CreateFromString(config_options_,
-                                                       "Changling", &result));
+  ASSERT_OK(TestSharedBuiltins<CompactionFilterFactory>("Changling", ""));
   if (RegisterTests("Test")) {
-    ASSERT_OK(CompactionFilterFactory::CreateFromString(config_options_,
-                                                        "Changling", &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "ChanglingCompactionFilterFactory");
+    ExpectCreateShared<CompactionFilterFactory>("Changling");
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadCompactionFilterTest) {
   const CompactionFilter* result = nullptr;
-
-  ASSERT_NOK(CompactionFilter::CreateFromString(config_options_, "Changling",
-                                                &result));
-#ifndef ROCKSDB_LITE
-  ASSERT_OK(CompactionFilter::CreateFromString(
-      config_options_, RemoveEmptyValueCompactionFilter::kClassName(),
-      &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), RemoveEmptyValueCompactionFilter::kClassName());
-  delete result;
-  result = nullptr;
+  std::vector<std::string> failures;
+  ASSERT_OK(TestStaticBuiltins<CompactionFilter>("Changling", &result, {},
+                                                 &failures, true));
   if (RegisterTests("Test")) {
-    ASSERT_OK(CompactionFilter::CreateFromString(config_options_, "Changling",
-                                                 &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "ChanglingCompactionFilter");
-    delete result;
+    ASSERT_OK(TestCreateStatic<CompactionFilter>("Changling", &result, true));
   }
-#endif  // ROCKSDB_LITE
 }
 
 #ifndef ROCKSDB_LITE
 TEST_F(LoadCustomizableTest, LoadEventListenerTest) {
-  std::shared_ptr<EventListener> result;
-
-  ASSERT_NOK(EventListener::CreateFromString(
-      config_options_, OnFileDeletionListener::kClassName(), &result));
-  ASSERT_NOK(EventListener::CreateFromString(
-      config_options_, FlushCounterListener::kClassName(), &result));
+  ASSERT_OK(TestSharedBuiltins<EventListener>(
+      OnFileDeletionListener::kClassName(), ""));
   if (RegisterTests("Test")) {
-    ASSERT_OK(EventListener::CreateFromString(
-        config_options_, OnFileDeletionListener::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), OnFileDeletionListener::kClassName());
-    ASSERT_OK(EventListener::CreateFromString(
-        config_options_, FlushCounterListener::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), FlushCounterListener::kClassName());
+    ExpectCreateShared<EventListener>(OnFileDeletionListener::kClassName());
+    ExpectCreateShared<EventListener>(FlushCounterListener::kClassName());
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadEncryptionProviderTest) {
+  std::vector<std::string> failures;
   std::shared_ptr<EncryptionProvider> result;
-  ASSERT_NOK(
-      EncryptionProvider::CreateFromString(config_options_, "Mock", &result));
   ASSERT_OK(
-      EncryptionProvider::CreateFromString(config_options_, "CTR", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "CTR");
+      TestExpectedBuiltins<EncryptionProvider>("Mock", {}, &result, &failures));
+  if (!failures.empty()) {
+    ASSERT_EQ(failures[0], "1://test");
+    ASSERT_EQ(failures.size(), 1U);
+  }
+
+  result = ExpectCreateShared<EncryptionProvider>("CTR");
   ASSERT_NOK(result->ValidateOptions(db_opts_, cf_opts_));
   ASSERT_OK(EncryptionProvider::CreateFromString(config_options_, "CTR://test",
                                                  &result));
@@ -2052,10 +2098,7 @@ TEST_F(LoadCustomizableTest, LoadEncryptionProviderTest) {
   ASSERT_OK(result->ValidateOptions(db_opts_, cf_opts_));
 
   if (RegisterTests("Test")) {
-    ASSERT_OK(
-        EncryptionProvider::CreateFromString(config_options_, "Mock", &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "Mock");
+    ExpectCreateShared<EncryptionProvider>("Mock");
     ASSERT_OK(EncryptionProvider::CreateFromString(config_options_,
                                                    "Mock://test", &result));
     ASSERT_NE(result, nullptr);
@@ -2065,72 +2108,69 @@ TEST_F(LoadCustomizableTest, LoadEncryptionProviderTest) {
 }
 
 TEST_F(LoadCustomizableTest, LoadEncryptionCipherTest) {
-  std::shared_ptr<BlockCipher> result;
-  ASSERT_NOK(BlockCipher::CreateFromString(config_options_, "Mock", &result));
-  ASSERT_OK(BlockCipher::CreateFromString(config_options_, "ROT13", &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), "ROT13");
+  ASSERT_OK(TestSharedBuiltins<BlockCipher>("Mock", "ROT13"));
   if (RegisterTests("Test")) {
-    ASSERT_OK(BlockCipher::CreateFromString(config_options_, "Mock", &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), "Mock");
+    ExpectCreateShared<BlockCipher>("Mock");
   }
 }
 #endif  // !ROCKSDB_LITE
 
 TEST_F(LoadCustomizableTest, LoadSystemClockTest) {
-  std::shared_ptr<SystemClock> result;
-  ASSERT_NOK(SystemClock::CreateFromString(
-      config_options_, MockSystemClock::kClassName(), &result));
-  ASSERT_OK(SystemClock::CreateFromString(
-      config_options_, SystemClock::kDefaultName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_TRUE(result->IsInstanceOf(SystemClock::kDefaultName()));
+  ASSERT_OK(TestSharedBuiltins<SystemClock>(MockSystemClock::kClassName(),
+                                            SystemClock::kDefaultName()));
   if (RegisterTests("Test")) {
-    ASSERT_OK(SystemClock::CreateFromString(
-        config_options_, MockSystemClock::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), MockSystemClock::kClassName());
+    auto result =
+        ExpectCreateShared<SystemClock>(MockSystemClock::kClassName());
+    ASSERT_FALSE(result->IsInstanceOf(SystemClock::kDefaultName()));
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadMemoryAllocatorTest) {
-  std::shared_ptr<MemoryAllocator> result;
-  ASSERT_NOK(MemoryAllocator::CreateFromString(
-      config_options_, MockMemoryAllocator::kClassName(), &result));
-  ASSERT_OK(MemoryAllocator::CreateFromString(
-      config_options_, DefaultMemoryAllocator::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), DefaultMemoryAllocator::kClassName());
+  std::vector<std::string> failures;
+  Status s = TestSharedBuiltins<MemoryAllocator>(
+      MockMemoryAllocator::kClassName(), DefaultMemoryAllocator::kClassName(),
+      &failures);
+  if (failures.empty()) {
+    ASSERT_OK(s);
+  } else {
+    ASSERT_NOK(s);
+    for (const auto& failure : failures) {
+      if (failure == JemallocNodumpAllocator::kClassName()) {
+        ASSERT_FALSE(JemallocNodumpAllocator::IsSupported());
+      } else if (failure == MemkindKmemAllocator::kClassName()) {
+        ASSERT_FALSE(MemkindKmemAllocator::IsSupported());
+      } else {
+        printf("BYPASSED: %s -- %s\n", failure.c_str(), s.ToString().c_str());
+      }
+    }
+  }
   if (RegisterTests("Test")) {
-    ASSERT_OK(MemoryAllocator::CreateFromString(
-        config_options_, MockMemoryAllocator::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), MockMemoryAllocator::kClassName());
+    ExpectCreateShared<MemoryAllocator>(MockMemoryAllocator::kClassName());
   }
 }
 
 TEST_F(LoadCustomizableTest, LoadRateLimiterTest) {
+#ifndef ROCKSDB_LITE
+  ASSERT_OK(TestSharedBuiltins<RateLimiter>(MockRateLimiter::kClassName(),
+                                            GenericRateLimiter::kClassName()));
+#else
+  ASSERT_OK(TestSharedBuiltins<RateLimiter>(MockRateLimiter::kClassName(), ""));
+#endif  // ROCKSDB_LITE
+
   std::shared_ptr<RateLimiter> result;
-  ASSERT_NOK(RateLimiter::CreateFromString(
-      config_options_, MockRateLimiter::kClassName(), &result));
   ASSERT_OK(RateLimiter::CreateFromString(
       config_options_, std::string(GenericRateLimiter::kClassName()) + ":1234",
       &result));
   ASSERT_NE(result, nullptr);
+  ASSERT_TRUE(result->IsInstanceOf(GenericRateLimiter::kClassName()));
 #ifndef ROCKSDB_LITE
-  ASSERT_OK(RateLimiter::CreateFromString(
-      config_options_, GenericRateLimiter::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
   ASSERT_OK(GetDBOptionsFromString(
       config_options_, db_opts_,
       std::string("rate_limiter=") + GenericRateLimiter::kClassName(),
       &db_opts_));
   ASSERT_NE(db_opts_.rate_limiter, nullptr);
   if (RegisterTests("Test")) {
-    ASSERT_OK(RateLimiter::CreateFromString(
-        config_options_, MockRateLimiter::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
+    ExpectCreateShared<RateLimiter>(MockRateLimiter::kClassName());
     ASSERT_OK(GetDBOptionsFromString(
         config_options_, db_opts_,
         std::string("rate_limiter=") + MockRateLimiter::kClassName(),
@@ -2141,17 +2181,52 @@ TEST_F(LoadCustomizableTest, LoadRateLimiterTest) {
 }
 
 TEST_F(LoadCustomizableTest, LoadFilterPolicyTest) {
-  std::shared_ptr<TableFactory> table;
+  const std::string kAutoBloom = BloomFilterPolicy::kClassName();
+  const std::string kAutoRibbon = RibbonFilterPolicy::kClassName();
+
   std::shared_ptr<const FilterPolicy> result;
-  ASSERT_NOK(FilterPolicy::CreateFromString(
-      config_options_, MockFilterPolicy::kClassName(), &result));
+  std::vector<std::string> failures;
+  std::unordered_set<std::string> expected = {
+      ReadOnlyBuiltinFilterPolicy::kClassName(),
+  };
 
-  ASSERT_OK(FilterPolicy::CreateFromString(config_options_, "", &result));
-  ASSERT_EQ(result, nullptr);
+#ifndef ROCKSDB_LITE
+  expected.insert({
+      kAutoBloom,
+      BloomFilterPolicy::kNickName(),
+      kAutoRibbon,
+      RibbonFilterPolicy::kNickName(),
+  });
+#endif  // ROCKSDB_LITE
+  ASSERT_OK(TestExpectedBuiltins<const FilterPolicy>(
+      "Mock", expected, &result, &failures, [](const std::string& name) {
+        std::vector<std::string> names = {name + ":1.234"};
+        return names;
+      }));
+#ifndef ROCKSDB_LITE
   ASSERT_OK(FilterPolicy::CreateFromString(
-      config_options_, ReadOnlyBuiltinFilterPolicy::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), ReadOnlyBuiltinFilterPolicy::kClassName());
+      config_options_, kAutoBloom + ":1.234:false", &result));
+  ASSERT_NE(result.get(), nullptr);
+  ASSERT_TRUE(result->IsInstanceOf(kAutoBloom));
+  ASSERT_OK(FilterPolicy::CreateFromString(
+      config_options_, kAutoBloom + ":1.234:false", &result));
+  ASSERT_NE(result.get(), nullptr);
+  ASSERT_TRUE(result->IsInstanceOf(kAutoBloom));
+  ASSERT_OK(FilterPolicy::CreateFromString(config_options_,
+                                           kAutoRibbon + ":1.234:-1", &result));
+  ASSERT_NE(result.get(), nullptr);
+  ASSERT_TRUE(result->IsInstanceOf(kAutoRibbon));
+  ASSERT_OK(FilterPolicy::CreateFromString(config_options_,
+                                           kAutoRibbon + ":1.234:56", &result));
+  ASSERT_NE(result.get(), nullptr);
+  ASSERT_TRUE(result->IsInstanceOf(kAutoRibbon));
+#endif  // ROCKSDB_LITE
+
+  if (RegisterTests("Test")) {
+    ExpectCreateShared<FilterPolicy>(MockFilterPolicy::kClassName(), &result);
+  }
+
+  std::shared_ptr<TableFactory> table;
 
 #ifndef ROCKSDB_LITE
   std::string table_opts = "id=BlockBasedTable; filter_policy=";
@@ -2173,42 +2248,30 @@ TEST_F(LoadCustomizableTest, LoadFilterPolicyTest) {
       config_options_, table_opts + MockFilterPolicy::kClassName(), &table));
   bbto = table->GetOptions<BlockBasedTableOptions>();
   ASSERT_NE(bbto, nullptr);
-  ASSERT_EQ(bbto->filter_policy.get(), nullptr);
-  if (RegisterTests("Test")) {
-    ASSERT_OK(FilterPolicy::CreateFromString(
-        config_options_, MockFilterPolicy::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), MockFilterPolicy::kClassName());
-    ASSERT_OK(TableFactory::CreateFromString(
-        config_options_, table_opts + MockFilterPolicy::kClassName(), &table));
-    bbto = table->GetOptions<BlockBasedTableOptions>();
-    ASSERT_NE(bbto, nullptr);
-    ASSERT_NE(bbto->filter_policy.get(), nullptr);
-    ASSERT_STREQ(bbto->filter_policy->Name(), MockFilterPolicy::kClassName());
-  }
+  ASSERT_NE(bbto->filter_policy.get(), nullptr);
+  ASSERT_TRUE(
+      bbto->filter_policy->IsInstanceOf(MockFilterPolicy::kClassName()));
 #endif  // ROCKSDB_LITE
 }
 
 TEST_F(LoadCustomizableTest, LoadFlushBlockPolicyFactoryTest) {
-  std::shared_ptr<TableFactory> table;
   std::shared_ptr<FlushBlockPolicyFactory> result;
-  ASSERT_NOK(FlushBlockPolicyFactory::CreateFromString(
-      config_options_, TestFlushBlockPolicyFactory::kClassName(), &result));
+  std::shared_ptr<TableFactory> table;
+  std::vector<std::string> failed;
+  std::unordered_set<std::string> expected = {
+      FlushBlockBySizePolicyFactory::kClassName(),
+      FlushBlockEveryKeyPolicyFactory::kClassName(),
+  };
+
+  ASSERT_OK(TestExpectedBuiltins<FlushBlockPolicyFactory>(
+      TestFlushBlockPolicyFactory::kClassName(), expected, &result, &failed));
 
+  // An empty policy name creates a BySize policy
   ASSERT_OK(
       FlushBlockPolicyFactory::CreateFromString(config_options_, "", &result));
   ASSERT_NE(result, nullptr);
   ASSERT_STREQ(result->Name(), FlushBlockBySizePolicyFactory::kClassName());
 
-  ASSERT_OK(FlushBlockPolicyFactory::CreateFromString(
-      config_options_, FlushBlockEveryKeyPolicyFactory::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), FlushBlockEveryKeyPolicyFactory::kClassName());
-
-  ASSERT_OK(FlushBlockPolicyFactory::CreateFromString(
-      config_options_, FlushBlockBySizePolicyFactory::kClassName(), &result));
-  ASSERT_NE(result, nullptr);
-  ASSERT_STREQ(result->Name(), FlushBlockBySizePolicyFactory::kClassName());
 #ifndef ROCKSDB_LITE
   std::string table_opts = "id=BlockBasedTable; flush_block_policy_factory=";
   ASSERT_OK(TableFactory::CreateFromString(
@@ -2220,10 +2283,8 @@ TEST_F(LoadCustomizableTest, LoadFlushBlockPolicyFactoryTest) {
   ASSERT_STREQ(bbto->flush_block_policy_factory->Name(),
                FlushBlockEveryKeyPolicyFactory::kClassName());
   if (RegisterTests("Test")) {
-    ASSERT_OK(FlushBlockPolicyFactory::CreateFromString(
-        config_options_, TestFlushBlockPolicyFactory::kClassName(), &result));
-    ASSERT_NE(result, nullptr);
-    ASSERT_STREQ(result->Name(), TestFlushBlockPolicyFactory::kClassName());
+    ExpectCreateShared<FlushBlockPolicyFactory>(
+        TestFlushBlockPolicyFactory::kClassName());
     ASSERT_OK(TableFactory::CreateFromString(
         config_options_, table_opts + TestFlushBlockPolicyFactory::kClassName(),
         &table));

port/win/env_win.h

@@ -80,8 +80,8 @@ class WinClock : public SystemClock {
   virtual ~WinClock() {}
 
   static const char* kClassName() { return "WindowsClock"; }
-  const char* Name() const override { return kClassName(); }
-  const char* NickName() const override { return kDefaultName(); }
+  const char* Name() const override { return kDefaultName(); }
+  const char* NickName() const override { return kClassName(); }
 
   uint64_t NowMicros() override;
 

table/plain/plain_table_factory.cc

@@ -275,7 +275,7 @@ Status MemTableRepFactory::CreateFromString(
     if (opts_list.empty() || opts_list.size() > 2 || !opt_map.empty()) {
       status = Status::InvalidArgument("Can't parse memtable_factory option ",
                                        value);
-    } else if (opts_list[0] == "skip_list" ||
+    } else if (opts_list[0] == SkipListFactory::kNickName() ||
                opts_list[0] == SkipListFactory::kClassName()) {
       // Expecting format
       // skip_list:<lookahead>
@@ -293,6 +293,17 @@ Status MemTableRepFactory::CreateFromString(
   return status;
 }
 
+Status MemTableRepFactory::CreateFromString(
+    const ConfigOptions& config_options, const std::string& value,
+    std::shared_ptr<MemTableRepFactory>* result) {
+  std::unique_ptr<MemTableRepFactory> factory;
+  Status s = CreateFromString(config_options, value, &factory);
+  if (factory && s.ok()) {
+    result->reset(factory.release());
+  }
+  return s;
+}
+
 #ifndef ROCKSDB_LITE
 Status GetPlainTableOptionsFromMap(
     const PlainTableOptions& table_options,

test_util/testutil.h

@@ -714,7 +714,9 @@ class ChanglingMergeOperator : public MergeOperator {
     return false;
   }
   static const char* kClassName() { return "ChanglingMergeOperator"; }
-  virtual bool IsInstanceOf(const std::string& id) const override {
+  const char* NickName() const override { return kNickName(); }
+  static const char* kNickName() { return "Changling"; }
+  bool IsInstanceOf(const std::string& id) const override {
     if (id == kClassName()) {
       return true;
     } else {
@@ -747,7 +749,10 @@ class ChanglingCompactionFilter : public CompactionFilter {
   }
 
   static const char* kClassName() { return "ChanglingCompactionFilter"; }
-  virtual bool IsInstanceOf(const std::string& id) const override {
+  const char* NickName() const override { return kNickName(); }
+  static const char* kNickName() { return "Changling"; }
+
+  bool IsInstanceOf(const std::string& id) const override {
     if (id == kClassName()) {
       return true;
     } else {
@@ -781,7 +786,10 @@ class ChanglingCompactionFilterFactory : public CompactionFilterFactory {
   // Returns a name that identifies this compaction filter factory.
   const char* Name() const override { return name_.c_str(); }
   static const char* kClassName() { return "ChanglingCompactionFilterFactory"; }
-  virtual bool IsInstanceOf(const std::string& id) const override {
+  const char* NickName() const override { return kNickName(); }
+  static const char* kNickName() { return "Changling"; }
+
+  bool IsInstanceOf(const std::string& id) const override {
     if (id == kClassName()) {
       return true;
     } else {

utilities/object_registry.cc

@@ -157,6 +157,37 @@ size_t ObjectLibrary::GetFactoryCount(size_t *types) const {
   return factories;
 }
 
+size_t ObjectLibrary::GetFactoryCount(const std::string &type) const {
+  std::unique_lock<std::mutex> lock(mu_);
+  auto iter = factories_.find(type);
+  if (iter != factories_.end()) {
+    return iter->second.size();
+  } else {
+    return 0;
+  }
+}
+
+void ObjectLibrary::GetFactoryNames(const std::string &type,
+                                    std::vector<std::string> *names) const {
+  assert(names);
+  std::unique_lock<std::mutex> lock(mu_);
+  auto iter = factories_.find(type);
+  if (iter != factories_.end()) {
+    for (const auto &f : iter->second) {
+      names->push_back(f->Name());
+    }
+  }
+}
+
+void ObjectLibrary::GetFactoryTypes(
+    std::unordered_set<std::string> *types) const {
+  assert(types);
+  std::unique_lock<std::mutex> lock(mu_);
+  for (const auto &iter : factories_) {
+    types->insert(iter.first);
+  }
+}
+
 void ObjectLibrary::Dump(Logger *logger) const {
   std::unique_lock<std::mutex> lock(mu_);
   if (logger != nullptr && !factories_.empty()) {
@@ -276,6 +307,46 @@ Status ObjectRegistry::ListManagedObjects(
   }
 }
 
+// Returns the number of registered types for this registry.
+// If specified (not-null), types is updated to include the names of the
+// registered types.
+size_t ObjectRegistry::GetFactoryCount(const std::string &type) const {
+  size_t count = 0;
+  if (parent_ != nullptr) {
+    count = parent_->GetFactoryCount(type);
+  }
+  std::unique_lock<std::mutex> lock(library_mutex_);
+  for (const auto &library : libraries_) {
+    count += library->GetFactoryCount(type);
+  }
+  return count;
+}
+
+void ObjectRegistry::GetFactoryNames(const std::string &type,
+                                     std::vector<std::string> *names) const {
+  assert(names);
+  names->clear();
+  if (parent_ != nullptr) {
+    parent_->GetFactoryNames(type, names);
+  }
+  std::unique_lock<std::mutex> lock(library_mutex_);
+  for (const auto &library : libraries_) {
+    library->GetFactoryNames(type, names);
+  }
+}
+
+void ObjectRegistry::GetFactoryTypes(
+    std::unordered_set<std::string> *types) const {
+  assert(types);
+  if (parent_ != nullptr) {
+    parent_->GetFactoryTypes(types);
+  }
+  std::unique_lock<std::mutex> lock(library_mutex_);
+  for (const auto &library : libraries_) {
+    library->GetFactoryTypes(types);
+  }
+}
+
 void ObjectRegistry::Dump(Logger *logger) const {
   if (logger != nullptr) {
     std::unique_lock<std::mutex> lock(library_mutex_);

utilities/object_registry_test.cc

@@ -260,6 +260,77 @@ class MyCustomizable : public Customizable {
   std::string name_;
 };
 
+TEST_F(ObjRegistryTest, TestFactoryCount) {
+  std::string msg;
+  auto grand = ObjectRegistry::Default();
+  auto local = ObjectRegistry::NewInstance();
+  std::unordered_set<std::string> grand_types, local_types;
+  std::vector<std::string> grand_names, local_names;
+
+  // Check how many types we have on startup.
+  // Grand should equal local
+  grand->GetFactoryTypes(&grand_types);
+  local->GetFactoryTypes(&local_types);
+  ASSERT_EQ(grand_types, local_types);
+  size_t grand_count = grand->GetFactoryCount(Env::Type());
+  size_t local_count = local->GetFactoryCount(Env::Type());
+
+  ASSERT_EQ(grand_count, local_count);
+  grand->GetFactoryNames(Env::Type(), &grand_names);
+  local->GetFactoryNames(Env::Type(), &local_names);
+  ASSERT_EQ(grand_names.size(), grand_count);
+  ASSERT_EQ(local_names.size(), local_count);
+  ASSERT_EQ(grand_names, local_names);
+
+  // Add an Env to the local registry.
+  // This will add one factory.
+  auto library = local->AddLibrary("local");
+  library->AddFactory<Env>(
+      "A", [](const std::string& /*uri*/, std::unique_ptr<Env>* /*guard */,
+              std::string* /* errmsg */) { return nullptr; });
+  ASSERT_EQ(local_count + 1, local->GetFactoryCount(Env::Type()));
+  ASSERT_EQ(grand_count, grand->GetFactoryCount(Env::Type()));
+  local->GetFactoryTypes(&local_types);
+  local->GetFactoryNames(Env::Type(), &local_names);
+  ASSERT_EQ(grand_names.size() + 1, local_names.size());
+  ASSERT_EQ(local_names.size(), local->GetFactoryCount(Env::Type()));
+
+  if (grand_count == 0) {
+    // There were no Env when we started.  Should have one more type
+    // than previously
+    ASSERT_NE(grand_types, local_types);
+    ASSERT_EQ(grand_types.size() + 1, local_types.size());
+  } else {
+    // There was an Env type when we started.  The types should match
+    ASSERT_EQ(grand_types, local_types);
+  }
+
+  // Add a MyCustomizable to the registry.  This should be a new type
+  library->AddFactory<MyCustomizable>(
+      "MY", [](const std::string& /*uri*/,
+               std::unique_ptr<MyCustomizable>* /*guard */,
+               std::string* /* errmsg */) { return nullptr; });
+  ASSERT_EQ(local_count + 1, local->GetFactoryCount(Env::Type()));
+  ASSERT_EQ(grand_count, grand->GetFactoryCount(Env::Type()));
+  ASSERT_EQ(0U, grand->GetFactoryCount(MyCustomizable::Type()));
+  ASSERT_EQ(1U, local->GetFactoryCount(MyCustomizable::Type()));
+
+  local->GetFactoryNames(MyCustomizable::Type(), &local_names);
+  ASSERT_EQ(1U, local_names.size());
+  ASSERT_EQ(local_names[0], "MY");
+
+  local->GetFactoryTypes(&local_types);
+  ASSERT_EQ(grand_count == 0 ? 2 : grand_types.size() + 1, local_types.size());
+
+  // Add the same name again.  We should now have 2 factories.
+  library->AddFactory<MyCustomizable>(
+      "MY", [](const std::string& /*uri*/,
+               std::unique_ptr<MyCustomizable>* /*guard */,
+               std::string* /* errmsg */) { return nullptr; });
+  local->GetFactoryNames(MyCustomizable::Type(), &local_names);
+  ASSERT_EQ(2U, local_names.size());
+}
+
 TEST_F(ObjRegistryTest, TestManagedObjects) {
   auto registry = ObjectRegistry::NewInstance();
   auto m_a1 = std::make_shared<MyCustomizable>("", "A");