Fix some minor issues in the Customizable infrastructure (#8566)

Summary:
- Fix issue with OptionType::Vector when the nested item is a Customizable with no names
- Fix issue with OptionType::Vector to appropriately wrap the elements in a Vector;
- Fix an issue with nested Customizable object with a null immutable object still appearing in the mutable options;
- Fix/Add tests for null/empty customizable objects
- Move the RegisterTestObjects from customizable_test into testutil.

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

Reviewed By: zhichao-cao

Differential Revision: D30303724

Pulled By: mrambacher

fbshipit-source-id: 33fa8ea2a3b663210cb356da05e64aab7585b1b5

HISTORY.md

@@ -23,6 +23,7 @@
 
 ## Public API change
 * Added APIs to decode and replay trace file via Replayer class. Added `DB::NewDefaultReplayer()` to create a default Replayer instance. Added `TraceReader::Reset()` to restart reading a trace file. Created trace_record.h, trace_record_result.h and utilities/replayer.h files to access the decoded Trace records, replay them, and query the actual operation results.
+* Added Configurable::GetOptionsMap to the public API for use in creating new Customizable classes.
 
 ### Performance Improvements
 * Try to avoid updating DBOptions if `SetDBOptions()` does not change any option value.

include/rocksdb/configurable.h

@@ -267,6 +267,24 @@ class Configurable {
   // changed.
   virtual bool IsPrepared() const { return is_prepared_; }
 
+  // Splits the input opt_value into the ID field and the remaining options.
+  // The input opt_value can be in the form of "name" or "name=value
+  // [;name=value]". The first form uses the "name" as an id with no options The
+  // latter form converts the input into a map of name=value pairs and sets "id"
+  // to the "id" value from the map.
+  // @param opt_value The value to split into id and options
+  // @param id The id field from the opt_value
+  // @param options The remaining name/value pairs from the opt_value
+  // @param default_id If specified and there is no id field in the map, this
+  // value is returned as the ID
+  // @return OK if the value was converted to a map successfully and an ID was
+  // found.
+  // @return InvalidArgument if the value could not be converted to a map or
+  // there was or there is no id property in the map.
+  static Status GetOptionsMap(
+      const std::string& opt_value, const std::string& default_id,
+      std::string* id, std::unordered_map<std::string, std::string>* options);
+
  protected:
   // True once the object is prepared.  Once the object is prepared, only
   // mutable options can be configured.

include/rocksdb/utilities/customizable_util.h

@@ -58,24 +58,26 @@ static Status NewSharedObject(
     const ConfigOptions& config_options, const std::string& id,
     const std::unordered_map<std::string, std::string>& opt_map,
     std::shared_ptr<T>* result) {
-  Status status;
   if (!id.empty()) {
+    Status status;
 #ifndef ROCKSDB_LITE
     status = config_options.registry->NewSharedObject(id, result);
 #else
     status = Status::NotSupported("Cannot load object in LITE mode ", id);
 #endif  // ROCKSDB_LITE
     if (config_options.ignore_unsupported_options && status.IsNotSupported()) {
-      return Status::OK();
+      status = Status::OK();
+    } else if (status.ok()) {
+      status = Customizable::ConfigureNewObject(config_options, result->get(),
+                                                opt_map);
     }
-  } else {
-    status = Status::NotSupported("Cannot reset object ");
-  }
-  if (!status.ok()) {
     return status;
+  } else if (opt_map.empty()) {
+    // There was no ID and no map (everything empty), so reset/clear the result
+    result->reset();
+    return Status::OK();
   } else {
-    return Customizable::ConfigureNewObject(config_options, result->get(),
+    return Status::NotSupported("Cannot reset object ");
-                                            opt_map);
   }
 }
 
@@ -119,12 +121,7 @@ static Status LoadSharedObject(const ConfigOptions& config_options,
     return status;
   } else if (func == nullptr ||
              !func(id, result)) {  // No factory, or it failed
-    if (value.empty()) {           // No Id and no options.  Clear the object
+    return NewSharedObject(config_options, id, opt_map, result);
-      *result = nullptr;
-      return Status::OK();
-    } else {
-      return NewSharedObject(config_options, id, opt_map, result);
-    }
   } else {
     return Customizable::ConfigureNewObject(config_options, result->get(),
                                             opt_map);
@@ -149,24 +146,26 @@ static Status NewUniqueObject(
     const ConfigOptions& config_options, const std::string& id,
     const std::unordered_map<std::string, std::string>& opt_map,
     std::unique_ptr<T>* result) {
-  Status status;
+  if (!id.empty()) {
-  if (id.empty()) {
+    Status status;
-    status = Status::NotSupported("Cannot reset object ");
-  } else {
 #ifndef ROCKSDB_LITE
     status = config_options.registry->NewUniqueObject(id, result);
 #else
     status = Status::NotSupported("Cannot load object in LITE mode ", id);
 #endif  // ROCKSDB_LITE
     if (config_options.ignore_unsupported_options && status.IsNotSupported()) {
-      return Status::OK();
+      status = Status::OK();
+    } else if (status.ok()) {
+      status = Customizable::ConfigureNewObject(config_options, result->get(),
+                                                opt_map);
     }
-  }
-  if (!status.ok()) {
     return status;
+  } else if (opt_map.empty()) {
+    // There was no ID and no map (everything empty), so reset/clear the result
+    result->reset();
+    return Status::OK();
   } else {
-    return Customizable::ConfigureNewObject(config_options, result->get(),
+    return Status::NotSupported("Cannot reset object ");
-                                            opt_map);
   }
 }
 
@@ -194,12 +193,7 @@ static Status LoadUniqueObject(const ConfigOptions& config_options,
     return status;
   } else if (func == nullptr ||
              !func(id, result)) {  // No factory, or it failed
-    if (value.empty()) {           // No Id and no options.  Clear the object
+    return NewUniqueObject(config_options, id, opt_map, result);
-      *result = nullptr;
-      return Status::OK();
-    } else {
-      return NewUniqueObject(config_options, id, opt_map, result);
-    }
   } else {
     return Customizable::ConfigureNewObject(config_options, result->get(),
                                             opt_map);
@@ -223,23 +217,26 @@ template <typename T>
 static Status NewStaticObject(
     const ConfigOptions& config_options, const std::string& id,
     const std::unordered_map<std::string, std::string>& opt_map, T** result) {
-  Status status;
+  if (!id.empty()) {
-  if (id.empty()) {
+    Status status;
-    status = Status::NotSupported("Cannot reset object ");
-  } else {
 #ifndef ROCKSDB_LITE
     status = config_options.registry->NewStaticObject(id, result);
 #else
     status = Status::NotSupported("Cannot load object in LITE mode ", id);
 #endif  // ROCKSDB_LITE
     if (config_options.ignore_unsupported_options && status.IsNotSupported()) {
-      return Status::OK();
+      status = Status::OK();
+    } else if (status.ok()) {
+      status =
+          Customizable::ConfigureNewObject(config_options, *result, opt_map);
     }
-  }
-  if (!status.ok()) {
     return status;
+  } else if (opt_map.empty()) {
+    // There was no ID and no map (everything empty), so reset/clear the result
+    *result = nullptr;
+    return Status::OK();
   } else {
-    return Customizable::ConfigureNewObject(config_options, *result, opt_map);
+    return Status::NotSupported("Cannot reset object ");
   }
 }
 
@@ -266,12 +263,7 @@ static Status LoadStaticObject(const ConfigOptions& config_options,
     return status;
   } else if (func == nullptr ||
              !func(id, result)) {  // No factory, or it failed
-    if (value.empty()) {           // No Id and no options.  Clear the object
+    return NewStaticObject(config_options, id, opt_map, result);
-      *result = nullptr;
-      return Status::OK();
-    } else {
-      return NewStaticObject(config_options, id, opt_map, result);
-    }
   } else {
     return Customizable::ConfigureNewObject(config_options, *result, opt_map);
   }

include/rocksdb/utilities/options_type.h

@@ -430,10 +430,15 @@ class OptionTypeInfo {
     return OptionTypeInfo(
         offset, OptionType::kCustomizable, ovt,
         flags | OptionTypeFlags::kShared,
-        [](const ConfigOptions& opts, const std::string&,
+        [](const ConfigOptions& opts, const std::string& name,
            const std::string& value, void* addr) {
           auto* shared = static_cast<std::shared_ptr<T>*>(addr);
-          return T::CreateFromString(opts, value, shared);
+          if (name == kIdPropName() && value.empty()) {
+            shared->reset();
+            return Status::OK();
+          } else {
+            return T::CreateFromString(opts, value, shared);
+          }
         },
         serialize_func, equals_func);
   }
@@ -463,10 +468,15 @@ class OptionTypeInfo {
     return OptionTypeInfo(
         offset, OptionType::kCustomizable, ovt,
         flags | OptionTypeFlags::kUnique,
-        [](const ConfigOptions& opts, const std::string&,
+        [](const ConfigOptions& opts, const std::string& name,
            const std::string& value, void* addr) {
           auto* unique = static_cast<std::unique_ptr<T>*>(addr);
-          return T::CreateFromString(opts, value, unique);
+          if (name == kIdPropName() && value.empty()) {
+            unique->reset();
+            return Status::OK();
+          } else {
+            return T::CreateFromString(opts, value, unique);
+          }
         },
         serialize_func, equals_func);
   }
@@ -494,10 +504,15 @@ class OptionTypeInfo {
     return OptionTypeInfo(
         offset, OptionType::kCustomizable, ovt,
         flags | OptionTypeFlags::kRawPointer,
-        [](const ConfigOptions& opts, const std::string&,
+        [](const ConfigOptions& opts, const std::string& name,
            const std::string& value, void* addr) {
           auto** pointer = static_cast<T**>(addr);
-          return T::CreateFromString(opts, value, pointer);
+          if (name == kIdPropName() && value.empty()) {
+            *pointer = nullptr;
+            return Status::OK();
+          } else {
+            return T::CreateFromString(opts, value, pointer);
+          }
         },
         serialize_func, equals_func);
   }
@@ -773,6 +788,9 @@ class OptionTypeInfo {
   static Status NextToken(const std::string& opts, char delimiter, size_t start,
                           size_t* end, std::string* token);
 
+  constexpr static const char* kIdPropName() { return "id"; }
+  constexpr static const char* kIdPropSuffix() { return ".id"; }
+
  private:
   int offset_;
 
@@ -867,18 +885,18 @@ Status SerializeVector(const ConfigOptions& config_options,
   std::string result;
   ConfigOptions embedded = config_options;
   embedded.delimiter = ";";
-  for (size_t i = 0; i < vec.size(); ++i) {
+  int printed = 0;
+  for (const auto& elem : vec) {
     std::string elem_str;
-    Status s = elem_info.Serialize(
+    Status s = elem_info.Serialize(embedded, name, &elem, &elem_str);
-        embedded, name, reinterpret_cast<const char*>(&vec[i]), &elem_str);
     if (!s.ok()) {
       return s;
-    } else {
+    } else if (!elem_str.empty()) {
-      if (i > 0) {
+      if (printed++ > 0) {
         result += separator;
       }
       // If the element contains embedded separators, put it inside of brackets
-      if (result.find(separator) != std::string::npos) {
+      if (elem_str.find(separator) != std::string::npos) {
         result += "{" + elem_str + "}";
       } else {
         result += elem_str;
@@ -887,6 +905,8 @@ Status SerializeVector(const ConfigOptions& config_options,
   }
   if (result.find("=") != std::string::npos) {
     *value = "{" + result + "}";
+  } else if (printed > 1 && result.at(0) == '{') {
+    *value = "{" + result + "}";
   } else {
     *value = result;
   }

options/configurable.cc

@@ -415,8 +415,8 @@ Status ConfigurableHelper::ConfigureCustomizableOption(
 
   if (opt_info.IsMutable() || !config_options.mutable_options_only) {
     // Either the option is mutable, or we are processing all of the options
-    if (opt_name == name || name == ConfigurableHelper::kIdPropName ||
+    if (opt_name == name || name == OptionTypeInfo::kIdPropName() ||
-        EndsWith(opt_name, ConfigurableHelper::kIdPropSuffix)) {
+        EndsWith(opt_name, OptionTypeInfo::kIdPropSuffix())) {
       return configurable.ParseOption(copy, opt_info, name, value, opt_ptr);
     } else if (value.empty()) {
       return Status::OK();
@@ -439,8 +439,8 @@ Status ConfigurableHelper::ConfigureCustomizableOption(
       } else {
         return Status::InvalidArgument("Option not changeable: " + opt_name);
       }
-    } else if (EndsWith(opt_name, ConfigurableHelper::kIdPropSuffix) ||
+    } else if (EndsWith(opt_name, OptionTypeInfo::kIdPropSuffix()) ||
-               name == ConfigurableHelper::kIdPropName) {
+               name == OptionTypeInfo::kIdPropName()) {
       // We have a property of the form "id=value" or "table.id=value"
       // This is OK if we ID/value matches the current customizable object
       if (custom->GetId() == value) {
@@ -459,7 +459,8 @@ Status ConfigurableHelper::ConfigureCustomizableOption(
       // map
       std::unordered_map<std::string, std::string> props;
       std::string id;
-      Status s = GetOptionsMap(value, custom->GetId(), &id, &props);
+      Status s =
+          Configurable::GetOptionsMap(value, custom->GetId(), &id, &props);
       if (!s.ok()) {
         return s;
       } else if (custom->GetId() != id) {
@@ -734,7 +735,7 @@ bool ConfigurableHelper::AreEquivalent(const ConfigOptions& config_options,
 }
 #endif  // ROCKSDB_LITE
 
-Status ConfigurableHelper::GetOptionsMap(
+Status Configurable::GetOptionsMap(
     const std::string& value, const std::string& default_id, std::string* id,
     std::unordered_map<std::string, std::string>* props) {
   assert(id);
@@ -752,10 +753,13 @@ Status ConfigurableHelper::GetOptionsMap(
       props->clear();         // Clear the properties
       status = Status::OK();  // and ignore the error
     } else {
-      auto iter = props->find(ConfigurableHelper::kIdPropName);
+      auto iter = props->find(OptionTypeInfo::kIdPropName());
       if (iter != props->end()) {
         *id = iter->second;
         props->erase(iter);
+        if (*id == kNullptrString) {
+          id->clear();
+        }
       } else if (!default_id.empty()) {
         *id = default_id;
       } else {           // No id property and no default

options/configurable_helper.h

@@ -20,8 +20,6 @@ namespace ROCKSDB_NAMESPACE {
 // of configuring the objects.
 class ConfigurableHelper {
  public:
-  constexpr static const char* kIdPropName = "id";
-  constexpr static const char* kIdPropSuffix = ".id";
   // Configures the input Configurable object based on the parameters.
   // On successful completion, the Configurable is updated with the settings
   // from the opt_map.
@@ -48,24 +46,6 @@ class ConfigurableHelper {
       const std::unordered_map<std::string, std::string>& options,
       std::unordered_map<std::string, std::string>* unused);
 
-  // Splits the input opt_value into the ID field and the remaining options.
-  // The input opt_value can be in the form of "name" or "name=value
-  // [;name=value]". The first form uses the "name" as an id with no options The
-  // latter form converts the input into a map of name=value pairs and sets "id"
-  // to the "id" value from the map.
-  // @param opt_value The value to split into id and options
-  // @param id The id field from the opt_value
-  // @param options The remaining name/value pairs from the opt_value
-  // @param default_id If specified and there is no id field in the map, this
-  // value is returned as the ID
-  // @return OK if the value was converted to a map succesfully and an ID was
-  // found.
-  // @return InvalidArgument if the value could not be converted to a map or
-  // there was or there is no id property in the map.
-  static Status GetOptionsMap(
-      const std::string& opt_value, const std::string& default_id,
-      std::string* id, std::unordered_map<std::string, std::string>* options);
-
 #ifndef ROCKSDB_LITE
   // Internal method to configure a set of options for this object.
   // Classes may override this value to change its behavior.

options/customizable.cc

@@ -5,10 +5,10 @@
 
 #include "rocksdb/customizable.h"
 
-#include "options/configurable_helper.h"
 #include "options/options_helper.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/status.h"
+#include "rocksdb/utilities/options_type.h"
 #include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -29,7 +29,7 @@ std::string Customizable::GetOptionName(const std::string& long_name) const {
 Status Customizable::GetOption(const ConfigOptions& config_options,
                                const std::string& opt_name,
                                std::string* value) const {
-  if (opt_name == ConfigurableHelper::kIdPropName) {
+  if (opt_name == OptionTypeInfo::kIdPropName()) {
     *value = GetId();
     return Status::OK();
   } else {
@@ -49,8 +49,10 @@ std::string Customizable::SerializeOptions(const ConfigOptions& config_options,
     result = id;
   } else {
     result.append(prefix);
-    result.append(ConfigurableHelper::kIdPropName).append("=");
+    result.append(OptionTypeInfo::kIdPropName());
-    result.append(id).append(config_options.delimiter);
+    result.append("=");
+    result.append(id);
+    result.append(config_options.delimiter);
     result.append(parent);
   }
   return result;
@@ -65,7 +67,7 @@ bool Customizable::AreEquivalent(const ConfigOptions& config_options,
       this != other) {
     const Customizable* custom = reinterpret_cast<const Customizable*>(other);
     if (GetId() != custom->GetId()) {
-      *mismatch = ConfigurableHelper::kIdPropName;
+      *mismatch = OptionTypeInfo::kIdPropName();
       return false;
     } else if (config_options.sanity_level >
                ConfigOptions::kSanityLevelLooselyCompatible) {
@@ -81,9 +83,13 @@ Status Customizable::GetOptionsMap(
     const ConfigOptions& config_options, const Customizable* customizable,
     const std::string& value, std::string* id,
     std::unordered_map<std::string, std::string>* props) {
-  if (customizable != nullptr) {
+  Status status;
-    Status status = ConfigurableHelper::GetOptionsMap(
+  if (value.empty() || value == kNullptrString) {
-        value, customizable->GetId(), id, props);
+    *id = "";
+    props->clear();
+  } else if (customizable != nullptr) {
+    status =
+        Configurable::GetOptionsMap(value, customizable->GetId(), id, props);
 #ifdef ROCKSDB_LITE
     (void)config_options;
 #else
@@ -101,10 +107,10 @@ Status Customizable::GetOptionsMap(
       }
     }
 #endif  // ROCKSDB_LITE
-    return status;
   } else {
-    return ConfigurableHelper::GetOptionsMap(value, "", id, props);
+    status = Configurable::GetOptionsMap(value, "", id, props);
   }
+  return status;
 }
 
 Status Customizable::ConfigureNewObject(

options/customizable_test.cc

@@ -40,6 +40,7 @@ DEFINE_bool(enable_print, false, "Print options generated to console.");
 #endif  // GFLAGS
 
 namespace ROCKSDB_NAMESPACE {
+namespace {
 class StringLogger : public Logger {
  public:
   using Logger::Logv;
@@ -87,6 +88,7 @@ class TestCustomizable : public Customizable {
 };
 
 struct AOptions {
+  static const char* kName() { return "A"; }
   int i = 0;
   bool b = false;
 };
@@ -101,11 +103,12 @@ static std::unordered_map<std::string, OptionTypeInfo> a_option_info = {
       OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
 #endif  // ROCKSDB_LITE
 };
+
 class ACustomizable : public TestCustomizable {
  public:
   explicit ACustomizable(const std::string& id)
       : TestCustomizable("A"), id_(id) {
-    RegisterOptions("A", &opts_, &a_option_info);
+    RegisterOptions(&opts_, &a_option_info);
   }
   std::string GetId() const override { return id_; }
   static const char* kClassName() { return "A"; }
@@ -115,19 +118,6 @@ class ACustomizable : public TestCustomizable {
   const std::string id_;
 };
 
-#ifndef ROCKSDB_LITE
-static int A_count = 0;
-const FactoryFunc<TestCustomizable>& a_func =
-    ObjectLibrary::Default()->Register<TestCustomizable>(
-        "A.*",
-        [](const std::string& name, std::unique_ptr<TestCustomizable>* guard,
-           std::string* /* msg */) {
-          guard->reset(new ACustomizable(name));
-          A_count++;
-          return guard->get();
-        });
-#endif  // ROCKSDB_LITE
-
 struct BOptions {
   std::string s;
   bool b = false;
@@ -168,6 +158,89 @@ static bool LoadSharedB(const std::string& id,
     return false;
   }
 }
+
+#ifndef ROCKSDB_LITE
+static int A_count = 0;
+static int RegisterCustomTestObjects(ObjectLibrary& library,
+                                     const std::string& /*arg*/) {
+  library.Register<TestCustomizable>(
+      "A.*",
+      [](const std::string& name, std::unique_ptr<TestCustomizable>* guard,
+         std::string* /* msg */) {
+        guard->reset(new ACustomizable(name));
+        A_count++;
+        return guard->get();
+      });
+
+  library.Register<TestCustomizable>(
+      "S", [](const std::string& name,
+              std::unique_ptr<TestCustomizable>* /* guard */,
+              std::string* /* msg */) { return new BCustomizable(name); });
+  size_t num_types;
+  return static_cast<int>(library.GetFactoryCount(&num_types));
+}
+#endif  // ROCKSDB_LITE
+
+struct SimpleOptions {
+  static const char* kName() { return "simple"; }
+  bool b = true;
+  std::unique_ptr<TestCustomizable> cu;
+  std::shared_ptr<TestCustomizable> cs;
+  TestCustomizable* cp = nullptr;
+};
+
+static std::unordered_map<std::string, OptionTypeInfo> simple_option_info = {
+#ifndef ROCKSDB_LITE
+    {"bool",
+     {offsetof(struct SimpleOptions, b), OptionType::kBoolean,
+      OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
+    {"unique",
+     OptionTypeInfo::AsCustomUniquePtr<TestCustomizable>(
+         offsetof(struct SimpleOptions, cu), OptionVerificationType::kNormal,
+         OptionTypeFlags::kAllowNull)},
+    {"shared",
+     OptionTypeInfo::AsCustomSharedPtr<TestCustomizable>(
+         offsetof(struct SimpleOptions, cs), OptionVerificationType::kNormal,
+         OptionTypeFlags::kAllowNull)},
+    {"pointer",
+     OptionTypeInfo::AsCustomRawPtr<TestCustomizable>(
+         offsetof(struct SimpleOptions, cp), OptionVerificationType::kNormal,
+         OptionTypeFlags::kAllowNull)},
+#endif  // ROCKSDB_LITE
+};
+
+class SimpleConfigurable : public Configurable {
+ private:
+  SimpleOptions simple_;
+
+ public:
+  SimpleConfigurable() { RegisterOptions(&simple_, &simple_option_info); }
+
+  explicit SimpleConfigurable(
+      const std::unordered_map<std::string, OptionTypeInfo>* map) {
+    RegisterOptions(&simple_, map);
+  }
+};
+
+#ifndef ROCKSDB_LITE
+static void GetMapFromProperties(
+    const std::string& props,
+    std::unordered_map<std::string, std::string>* map) {
+  std::istringstream iss(props);
+  std::unordered_map<std::string, std::string> copy_map;
+  std::string line;
+  map->clear();
+  for (int line_num = 0; std::getline(iss, line); line_num++) {
+    std::string name;
+    std::string value;
+    ASSERT_OK(
+        RocksDBOptionsParser::ParseStatement(&name, &value, line, line_num));
+    (*map)[name] = value;
+  }
+}
+#endif  // ROCKSDB_LITE
+}  // namespace
+
 Status TestCustomizable::CreateFromString(
     const ConfigOptions& config_options, const std::string& value,
     std::shared_ptr<TestCustomizable>* result) {
@@ -213,58 +286,16 @@ Status TestCustomizable::CreateFromString(const ConfigOptions& config_options,
       result);
 }
 
-#ifndef ROCKSDB_LITE
-const FactoryFunc<TestCustomizable>& s_func =
-    ObjectLibrary::Default()->Register<TestCustomizable>(
-        "S", [](const std::string& name,
-                std::unique_ptr<TestCustomizable>* /* guard */,
-                std::string* /* msg */) { return new BCustomizable(name); });
-#endif  // ROCKSDB_LITE
-
-struct SimpleOptions {
-  static const char* kName() { return "simple"; }
-  bool b = true;
-  std::unique_ptr<TestCustomizable> cu;
-  std::shared_ptr<TestCustomizable> cs;
-  TestCustomizable* cp = nullptr;
-};
-
-static std::unordered_map<std::string, OptionTypeInfo> simple_option_info = {
-#ifndef ROCKSDB_LITE
-    {"bool",
-     {offsetof(struct SimpleOptions, b), OptionType::kBoolean,
-      OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
-    {"unique",
-     OptionTypeInfo::AsCustomUniquePtr<TestCustomizable>(
-         offsetof(struct SimpleOptions, cu), OptionVerificationType::kNormal,
-         OptionTypeFlags::kAllowNull)},
-    {"shared",
-     OptionTypeInfo::AsCustomSharedPtr<TestCustomizable>(
-         offsetof(struct SimpleOptions, cs), OptionVerificationType::kNormal,
-         OptionTypeFlags::kAllowNull)},
-    {"pointer",
-     OptionTypeInfo::AsCustomRawPtr<TestCustomizable>(
-         offsetof(struct SimpleOptions, cp), OptionVerificationType::kNormal,
-         OptionTypeFlags::kAllowNull)},
-#endif  // ROCKSDB_LITE
-};
-
-class SimpleConfigurable : public Configurable {
- private:
-  SimpleOptions simple_;
-
- public:
-  SimpleConfigurable() { RegisterOptions(&simple_, &simple_option_info); }
-
-  explicit SimpleConfigurable(
-      const std::unordered_map<std::string, OptionTypeInfo>* map) {
-    RegisterOptions(&simple_, map);
-  }
-};
-
 class CustomizableTest : public testing::Test {
  public:
-  CustomizableTest() { config_options_.invoke_prepare_options = false; }
+  CustomizableTest() {
+    config_options_.invoke_prepare_options = false;
+#ifndef ROCKSDB_LITE
+    // GetOptionsFromMap is not supported in ROCKSDB_LITE
+    config_options_.registry->AddLibrary("CustomizableTest",
+                                         RegisterCustomTestObjects, "");
+#endif  // ROCKSDB_LITE
+  }
 
   ConfigOptions config_options_;
 };
@@ -324,22 +355,6 @@ TEST_F(CustomizableTest, SimpleConfigureTest) {
       configurable->AreEquivalent(config_options_, copy.get(), &mismatch));
 }
 
-static void GetMapFromProperties(
-    const std::string& props,
-    std::unordered_map<std::string, std::string>* map) {
-  std::istringstream iss(props);
-  std::unordered_map<std::string, std::string> copy_map;
-  std::string line;
-  map->clear();
-  for (int line_num = 0; std::getline(iss, line); line_num++) {
-    std::string name;
-    std::string value;
-    ASSERT_OK(
-        RocksDBOptionsParser::ParseStatement(&name, &value, line, line_num));
-    (*map)[name] = value;
-  }
-}
-
 TEST_F(CustomizableTest, ConfigureFromPropsTest) {
   std::unordered_map<std::string, std::string> opt_map = {
       {"unique.id", "A"}, {"unique.A.int", "1"},    {"unique.A.bool", "true"},
@@ -412,7 +427,7 @@ TEST_F(CustomizableTest, AreEquivalentOptionsTest) {
 // Tests that we can initialize a customizable from its options
 TEST_F(CustomizableTest, ConfigureStandaloneCustomTest) {
   std::unique_ptr<TestCustomizable> base, copy;
-  auto registry = ObjectRegistry::NewInstance();
+  const auto& registry = config_options_.registry;
   ASSERT_OK(registry->NewUniqueObject<TestCustomizable>("A", &base));
   ASSERT_OK(registry->NewUniqueObject<TestCustomizable>("A", &copy));
   ASSERT_OK(base->ConfigureFromString(config_options_, "int=33;bool=true"));
@@ -476,11 +491,13 @@ TEST_F(CustomizableTest, UniqueIdTest) {
 }
 
 TEST_F(CustomizableTest, IsInstanceOfTest) {
-  std::shared_ptr<TestCustomizable> tc = std::make_shared<ACustomizable>("A");
+  std::shared_ptr<TestCustomizable> tc = std::make_shared<ACustomizable>("A_1");
 
+  ASSERT_EQ(tc->GetId(), std::string("A_1"));
   ASSERT_TRUE(tc->IsInstanceOf("A"));
   ASSERT_TRUE(tc->IsInstanceOf("TestCustomizable"));
   ASSERT_FALSE(tc->IsInstanceOf("B"));
+  ASSERT_FALSE(tc->IsInstanceOf("A_1"));
   ASSERT_EQ(tc->CheckedCast<ACustomizable>(), tc.get());
   ASSERT_EQ(tc->CheckedCast<TestCustomizable>(), tc.get());
   ASSERT_EQ(tc->CheckedCast<BCustomizable>(), nullptr);
@@ -566,16 +583,16 @@ TEST_F(CustomizableTest, PrepareOptionsTest) {
   ASSERT_TRUE(simple->cp->IsPrepared());
   delete simple->cp;
   base.reset(new SimpleConfigurable());
+  simple = base->GetOptions<SimpleOptions>();
+  ASSERT_NE(simple, nullptr);
 
   ASSERT_NOK(
       base->ConfigureFromString(prepared, "unique={id=P; can_prepare=false}"));
-  simple = base->GetOptions<SimpleOptions>();
+  ASSERT_EQ(simple->cu, nullptr);
-  ASSERT_NE(simple, nullptr);
-  ASSERT_NE(simple->cu, nullptr);
-  ASSERT_FALSE(simple->cu->IsPrepared());
 
   ASSERT_OK(
       base->ConfigureFromString(prepared, "unique={id=P; can_prepare=true}"));
+  ASSERT_NE(simple->cu, nullptr);
   ASSERT_TRUE(simple->cu->IsPrepared());
 
   ASSERT_OK(base->ConfigureFromString(config_options_,
@@ -593,6 +610,7 @@ TEST_F(CustomizableTest, PrepareOptionsTest) {
   ASSERT_FALSE(simple->cu->IsPrepared());
 }
 
+namespace {
 static std::unordered_map<std::string, OptionTypeInfo> inner_option_info = {
 #ifndef ROCKSDB_LITE
     {"inner",
@@ -636,6 +654,7 @@ class WrappedCustomizable2 : public InnerCustomizable {
   const char* Name() const override { return kClassName(); }
   static const char* kClassName() { return "Wrapped2"; }
 };
+}  // namespace
 
 TEST_F(CustomizableTest, WrappedInnerTest) {
   std::shared_ptr<TestCustomizable> ac =
@@ -665,20 +684,19 @@ TEST_F(CustomizableTest, WrappedInnerTest) {
   ASSERT_EQ(wc2->CheckedCast<TestCustomizable>(), ac.get());
 }
 
-class ShallowCustomizable : public Customizable {
- public:
-  ShallowCustomizable() {
-    inner_ = std::make_shared<ACustomizable>("a");
-    RegisterOptions("inner", &inner_, &inner_option_info);
-  };
-  static const char* kClassName() { return "shallow"; }
-  const char* Name() const override { return kClassName(); }
-
- private:
-  std::shared_ptr<TestCustomizable> inner_;
-};
-
 TEST_F(CustomizableTest, TestStringDepth) {
+  class ShallowCustomizable : public Customizable {
+   public:
+    ShallowCustomizable() {
+      inner_ = std::make_shared<ACustomizable>("a");
+      RegisterOptions("inner", &inner_, &inner_option_info);
+    }
+    static const char* kClassName() { return "shallow"; }
+    const char* Name() const override { return kClassName(); }
+
+   private:
+    std::shared_ptr<TestCustomizable> inner_;
+  };
   ConfigOptions shallow = config_options_;
   std::unique_ptr<Configurable> c(new ShallowCustomizable());
   std::string opt_str;
@@ -691,7 +709,7 @@ TEST_F(CustomizableTest, TestStringDepth) {
 }
 
 // Tests that we only get a new customizable when it changes
-TEST_F(CustomizableTest, NewCustomizableTest) {
+TEST_F(CustomizableTest, NewUniqueCustomizableTest) {
   std::unique_ptr<Configurable> base(new SimpleConfigurable());
   A_count = 0;
   ASSERT_OK(base->ConfigureFromString(config_options_,
@@ -711,9 +729,140 @@ TEST_F(CustomizableTest, NewCustomizableTest) {
   ASSERT_EQ(A_count, 3);  // Created another A
   ASSERT_OK(base->ConfigureFromString(config_options_, "unique.id="));
   ASSERT_EQ(simple->cu, nullptr);
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique=nullptr"));
+  ASSERT_EQ(simple->cu, nullptr);
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique.id=nullptr"));
+  ASSERT_EQ(simple->cu, nullptr);
   ASSERT_EQ(A_count, 3);
 }
 
+TEST_F(CustomizableTest, NewEmptyUniqueTest) {
+  std::unique_ptr<Configurable> base(new SimpleConfigurable());
+  SimpleOptions* simple = base->GetOptions<SimpleOptions>();
+  ASSERT_EQ(simple->cu, nullptr);
+  simple->cu.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique={id=}"));
+  ASSERT_EQ(simple->cu, nullptr);
+  simple->cu.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique={id=nullptr}"));
+  ASSERT_EQ(simple->cu, nullptr);
+  simple->cu.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique.id="));
+  ASSERT_EQ(simple->cu, nullptr);
+  simple->cu.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique=nullptr"));
+  ASSERT_EQ(simple->cu, nullptr);
+  simple->cu.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "unique.id=nullptr"));
+  ASSERT_EQ(simple->cu, nullptr);
+}
+
+TEST_F(CustomizableTest, NewEmptySharedTest) {
+  std::unique_ptr<Configurable> base(new SimpleConfigurable());
+
+  SimpleOptions* simple = base->GetOptions<SimpleOptions>();
+  ASSERT_NE(simple, nullptr);
+  ASSERT_EQ(simple->cs, nullptr);
+  simple->cs.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "shared={id=}"));
+  ASSERT_NE(simple, nullptr);
+  ASSERT_EQ(simple->cs, nullptr);
+  simple->cs.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "shared={id=nullptr}"));
+  ASSERT_EQ(simple->cs, nullptr);
+  simple->cs.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "shared.id="));
+  ASSERT_EQ(simple->cs, nullptr);
+  simple->cs.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "shared.id=nullptr"));
+  ASSERT_EQ(simple->cs, nullptr);
+  simple->cs.reset(new BCustomizable("B"));
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "shared=nullptr"));
+  ASSERT_EQ(simple->cs, nullptr);
+}
+
+TEST_F(CustomizableTest, NewEmptyStaticTest) {
+  std::unique_ptr<Configurable> base(new SimpleConfigurable());
+  ASSERT_OK(base->ConfigureFromString(config_options_, "pointer={id=}"));
+  SimpleOptions* simple = base->GetOptions<SimpleOptions>();
+  ASSERT_NE(simple, nullptr);
+  ASSERT_EQ(simple->cp, nullptr);
+  ASSERT_OK(base->ConfigureFromString(config_options_, "pointer={id=nullptr}"));
+  ASSERT_EQ(simple->cp, nullptr);
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "pointer="));
+  ASSERT_EQ(simple->cp, nullptr);
+  ASSERT_OK(base->ConfigureFromString(config_options_, "pointer=nullptr"));
+  ASSERT_EQ(simple->cp, nullptr);
+
+  ASSERT_OK(base->ConfigureFromString(config_options_, "pointer.id="));
+  ASSERT_EQ(simple->cp, nullptr);
+  ASSERT_OK(base->ConfigureFromString(config_options_, "pointer.id=nullptr"));
+  ASSERT_EQ(simple->cp, nullptr);
+}
+
+namespace {
+#ifndef ROCKSDB_LITE
+static std::unordered_map<std::string, OptionTypeInfo> vector_option_info = {
+    {"vector",
+     OptionTypeInfo::Vector<std::shared_ptr<TestCustomizable>>(
+         0, OptionVerificationType::kNormal,
+
+         OptionTypeFlags::kNone,
+
+         OptionTypeInfo::AsCustomSharedPtr<TestCustomizable>(
+             0, OptionVerificationType::kNormal, OptionTypeFlags::kNone))},
+};
+class VectorConfigurable : public SimpleConfigurable {
+ public:
+  VectorConfigurable() { RegisterOptions("vector", &cv, &vector_option_info); }
+  std::vector<std::shared_ptr<TestCustomizable>> cv;
+};
+}  // namespace
+
+TEST_F(CustomizableTest, VectorConfigTest) {
+  VectorConfigurable orig, copy;
+  std::shared_ptr<TestCustomizable> c1, c2;
+  ASSERT_OK(TestCustomizable::CreateFromString(config_options_, "A", &c1));
+  ASSERT_OK(TestCustomizable::CreateFromString(config_options_, "B", &c2));
+  orig.cv.push_back(c1);
+  orig.cv.push_back(c2);
+  ASSERT_OK(orig.ConfigureFromString(config_options_, "unique=A2"));
+  std::string opt_str, mismatch;
+  ASSERT_OK(orig.GetOptionString(config_options_, &opt_str));
+  ASSERT_OK(copy.ConfigureFromString(config_options_, opt_str));
+  ASSERT_TRUE(orig.AreEquivalent(config_options_, &copy, &mismatch));
+}
+
+TEST_F(CustomizableTest, NoNameTest) {
+  // If Customizables are created without names, they are not
+  // part of the serialization (since they cannot be recreated)
+  VectorConfigurable orig, copy;
+  auto sopts = orig.GetOptions<SimpleOptions>();
+  auto copts = copy.GetOptions<SimpleOptions>();
+  sopts->cu.reset(new ACustomizable(""));
+  orig.cv.push_back(std::make_shared<ACustomizable>(""));
+  orig.cv.push_back(std::make_shared<ACustomizable>("A1"));
+  std::string opt_str, mismatch;
+  ASSERT_OK(orig.GetOptionString(config_options_, &opt_str));
+  ASSERT_OK(copy.ConfigureFromString(config_options_, opt_str));
+  ASSERT_EQ(copy.cv.size(), 1U);
+  ASSERT_EQ(copy.cv[0]->GetId(), "A1");
+  ASSERT_EQ(copts->cu, nullptr);
+}
+
+#endif  // ROCKSDB_LITE
+
 TEST_F(CustomizableTest, IgnoreUnknownObjects) {
   ConfigOptions ignore = config_options_;
   std::shared_ptr<TestCustomizable> shared;
@@ -825,11 +974,19 @@ TEST_F(CustomizableTest, MutableOptionsTest) {
     }
     const char* Name() const override { return "MutableCustomizable"; }
   };
-  MutableCustomizable mc;
+  MutableCustomizable mc, mc2;
+  std::string mismatch;
+  std::string opt_str;
 
   ConfigOptions options = config_options_;
   ASSERT_FALSE(mc.IsPrepared());
   ASSERT_OK(mc.ConfigureOption(options, "mutable", "{id=B;}"));
+  options.mutable_options_only = true;
+  ASSERT_OK(mc.GetOptionString(options, &opt_str));
+  ASSERT_OK(mc2.ConfigureFromString(options, opt_str));
+  ASSERT_TRUE(mc.AreEquivalent(options, &mc2, &mismatch));
+
+  options.mutable_options_only = false;
   ASSERT_OK(mc.ConfigureOption(options, "immutable", "{id=A; int=10}"));
   auto* mm = mc.GetOptions<std::shared_ptr<TestCustomizable>>("mutable");
   auto* im = mc.GetOptions<std::shared_ptr<TestCustomizable>>("immutable");
@@ -875,14 +1032,12 @@ TEST_F(CustomizableTest, MutableOptionsTest) {
 
   // Only the mutable options should get serialized
   options.mutable_options_only = false;
+  ASSERT_OK(mc.GetOptionString(options, &opt_str));
   ASSERT_OK(mc.ConfigureOption(options, "immutable", "{id=B;}"));
   options.mutable_options_only = true;
 
-  std::string opt_str;
   ASSERT_OK(mc.GetOptionString(options, &opt_str));
-  MutableCustomizable mc2;
   ASSERT_OK(mc2.ConfigureFromString(options, opt_str));
-  std::string mismatch;
   ASSERT_TRUE(mc.AreEquivalent(options, &mc2, &mismatch));
   options.mutable_options_only = false;
   ASSERT_FALSE(mc.AreEquivalent(options, &mc2, &mismatch));
@@ -890,6 +1045,7 @@ TEST_F(CustomizableTest, MutableOptionsTest) {
 }
 #endif  // !ROCKSDB_LITE
 
+namespace {
 class TestSecondaryCache : public SecondaryCache {
  public:
   static const char* kClassName() { return "Test"; }
@@ -912,63 +1068,6 @@ class TestSecondaryCache : public SecondaryCache {
 };
 
 #ifndef ROCKSDB_LITE
-// This method loads existing test classes into the ObjectRegistry
-static int RegisterTestObjects(ObjectLibrary& library,
-                               const std::string& /*arg*/) {
-  size_t num_types;
-  library.Register<TableFactory>(
-      "MockTable",
-      [](const std::string& /*uri*/, std::unique_ptr<TableFactory>* guard,
-         std::string* /* errmsg */) {
-        guard->reset(new mock::MockTableFactory());
-        return guard->get();
-      });
-  library.Register<EventListener>(
-      OnFileDeletionListener::kClassName(),
-      [](const std::string& /*uri*/, std::unique_ptr<EventListener>* guard,
-         std::string* /* errmsg */) {
-        guard->reset(new OnFileDeletionListener());
-        return guard->get();
-      });
-  library.Register<EventListener>(
-      FlushCounterListener::kClassName(),
-      [](const std::string& /*uri*/, std::unique_ptr<EventListener>* guard,
-         std::string* /* errmsg */) {
-        guard->reset(new FlushCounterListener());
-        return guard->get();
-      });
-  library.Register<const Comparator>(
-      test::SimpleSuffixReverseComparator::kClassName(),
-      [](const std::string& /*uri*/,
-         std::unique_ptr<const Comparator>* /*guard*/,
-         std::string* /* errmsg */) {
-        static test::SimpleSuffixReverseComparator ssrc;
-        return &ssrc;
-      });
-  library.Register<MergeOperator>(
-      "Changling",
-      [](const std::string& uri, std::unique_ptr<MergeOperator>* guard,
-         std::string* /* errmsg */) {
-        guard->reset(new test::ChanglingMergeOperator(uri));
-        return guard->get();
-      });
-  library.Register<CompactionFilter>(
-      "Changling",
-      [](const std::string& uri, std::unique_ptr<CompactionFilter>* /*guard*/,
-         std::string* /* errmsg */) {
-        return new test::ChanglingCompactionFilter(uri);
-      });
-  library.Register<CompactionFilterFactory>(
-      "Changling", [](const std::string& uri,
-                      std::unique_ptr<CompactionFilterFactory>* guard,
-                      std::string* /* errmsg */) {
-        guard->reset(new test::ChanglingCompactionFilterFactory(uri));
-        return guard->get();
-      });
-
-  return static_cast<int>(library.GetFactoryCount(&num_types));
-}
-
 class MockEncryptionProvider : public EncryptionProvider {
  public:
   explicit MockEncryptionProvider(const std::string& id) : id_(id) {}
@@ -1010,6 +1109,7 @@ class MockCipher : public BlockCipher {
   Status Encrypt(char* /*data*/) override { return Status::NotSupported(); }
   Status Decrypt(char* data) override { return Encrypt(data); }
 };
+#endif  // ROCKSDB_LITE
 
 class TestFlushBlockPolicyFactory : public FlushBlockPolicyFactory {
  public:
@@ -1025,9 +1125,31 @@ class TestFlushBlockPolicyFactory : public FlushBlockPolicyFactory {
   }
 };
 
+#ifndef ROCKSDB_LITE
 static int RegisterLocalObjects(ObjectLibrary& library,
                                 const std::string& /*arg*/) {
   size_t num_types;
+  library.Register<TableFactory>(
+      mock::MockTableFactory::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<TableFactory>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new mock::MockTableFactory());
+        return guard->get();
+      });
+  library.Register<EventListener>(
+      OnFileDeletionListener::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<EventListener>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new OnFileDeletionListener());
+        return guard->get();
+      });
+  library.Register<EventListener>(
+      FlushCounterListener::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<EventListener>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new FlushCounterListener());
+        return guard->get();
+      });
   // Load any locally defined objects here
   library.Register<EncryptionProvider>(
       "Mock(://test)?",
@@ -1061,6 +1183,7 @@ static int RegisterLocalObjects(ObjectLibrary& library,
   return static_cast<int>(library.GetFactoryCount(&num_types));
 }
 #endif  // !ROCKSDB_LITE
+}  // namespace
 
 class LoadCustomizableTest : public testing::Test {
  public:
@@ -1070,8 +1193,8 @@ class LoadCustomizableTest : public testing::Test {
   }
   bool RegisterTests(const std::string& arg) {
 #ifndef ROCKSDB_LITE
-    config_options_.registry->AddLibrary("custom-tests", RegisterTestObjects,
+    config_options_.registry->AddLibrary("custom-tests",
-                                         arg);
+                                         test::RegisterTestObjects, arg);
     config_options_.registry->AddLibrary("local-tests", RegisterLocalObjects,
                                          arg);
     return true;
@@ -1090,8 +1213,8 @@ class LoadCustomizableTest : public testing::Test {
 TEST_F(LoadCustomizableTest, LoadTableFactoryTest) {
   ColumnFamilyOptions cf_opts;
   std::shared_ptr<TableFactory> factory;
-  ASSERT_NOK(
+  ASSERT_NOK(TableFactory::CreateFromString(
-      TableFactory::CreateFromString(config_options_, "MockTable", &factory));
+      config_options_, mock::MockTableFactory::kClassName(), &factory));
   ASSERT_OK(TableFactory::CreateFromString(
       config_options_, TableFactory::kBlockBasedTableName(), &factory));
   ASSERT_NE(factory, nullptr);
@@ -1106,16 +1229,17 @@ TEST_F(LoadCustomizableTest, LoadTableFactoryTest) {
                TableFactory::kBlockBasedTableName());
 #endif  // ROCKSDB_LITE
   if (RegisterTests("Test")) {
-    ASSERT_OK(
+    ASSERT_OK(TableFactory::CreateFromString(
-        TableFactory::CreateFromString(config_options_, "MockTable", &factory));
+        config_options_, mock::MockTableFactory::kClassName(), &factory));
     ASSERT_NE(factory, nullptr);
-    ASSERT_STREQ(factory->Name(), "MockTable");
+    ASSERT_STREQ(factory->Name(), mock::MockTableFactory::kClassName());
 #ifndef ROCKSDB_LITE
-    ASSERT_OK(
+    ASSERT_OK(GetColumnFamilyOptionsFromString(
-        GetColumnFamilyOptionsFromString(config_options_, ColumnFamilyOptions(),
+        config_options_, ColumnFamilyOptions(),
-                                         opts_str + "MockTable", &cf_opts));
+        opts_str + mock::MockTableFactory::kClassName(), &cf_opts));
     ASSERT_NE(cf_opts.table_factory.get(), nullptr);
-    ASSERT_STREQ(cf_opts.table_factory->Name(), "MockTable");
+    ASSERT_STREQ(cf_opts.table_factory->Name(),
+                 mock::MockTableFactory::kClassName());
 #endif  // ROCKSDB_LITE
   }
 }

options/options_helper.cc

@@ -1075,7 +1075,16 @@ Status OptionTypeInfo::Serialize(const ConfigOptions& config_options,
   } else if (IsCustomizable()) {
     const Customizable* custom = AsRawPointer<Customizable>(opt_ptr);
     if (custom == nullptr) {
-      *opt_value = kNullptrString;
+      // We do not have a custom object to serialize.
+      // If the option is not mutable and we are doing only mutable options,
+      // we return an empty string (which will cause the option not to be
+      // printed). Otherwise, we return the "nullptr" string, which will result
+      // in "option=nullptr" being printed.
+      if (IsMutable() || !config_options.mutable_options_only) {
+        *opt_value = kNullptrString;
+      } else {
+        *opt_value = "";
+      }
     } else if (IsEnabled(OptionTypeFlags::kStringNameOnly) &&
                !config_options.IsDetailed()) {
       *opt_value = custom->GetId();

options/options_test.cc

@@ -3668,21 +3668,28 @@ TEST_F(OptionsParserTest, EscapeOptionString) {
 static void TestAndCompareOption(const ConfigOptions& config_options,
                                  const OptionTypeInfo& opt_info,
                                  const std::string& opt_name, void* base_ptr,
-                                 void* comp_ptr) {
+                                 void* comp_ptr, bool strip = false) {
   std::string result, mismatch;
   ASSERT_OK(opt_info.Serialize(config_options, opt_name, base_ptr, &result));
+  if (strip) {
+    ASSERT_EQ(result.at(0), '{');
+    ASSERT_EQ(result.at(result.size() - 1), '}');
+    result = result.substr(1, result.size() - 2);
+  }
   ASSERT_OK(opt_info.Parse(config_options, opt_name, result, comp_ptr));
   ASSERT_TRUE(opt_info.AreEqual(config_options, opt_name, base_ptr, comp_ptr,
                                 &mismatch));
 }
 
-static void TestAndCompareOption(const ConfigOptions& config_options,
+static void TestParseAndCompareOption(const ConfigOptions& config_options,
-                                 const OptionTypeInfo& opt_info,
+                                      const OptionTypeInfo& opt_info,
-                                 const std::string& opt_name,
+                                      const std::string& opt_name,
-                                 const std::string& opt_value, void* base_ptr,
+                                      const std::string& opt_value,
-                                 void* comp_ptr) {
+                                      void* base_ptr, void* comp_ptr,
+                                      bool strip = false) {
   ASSERT_OK(opt_info.Parse(config_options, opt_name, opt_value, base_ptr));
-  TestAndCompareOption(config_options, opt_info, opt_name, base_ptr, comp_ptr);
+  TestAndCompareOption(config_options, opt_info, opt_name, base_ptr, comp_ptr,
+                       strip);
 }
 
 template <typename T>
@@ -3934,7 +3941,7 @@ TEST_F(OptionTypeInfoTest, TestCustomEnum) {
   ASSERT_FALSE(opt_info.AreEqual(config_options, "Enum", &e1, &e2, &mismatch));
   ASSERT_EQ(mismatch, "Enum");
 
-  TestAndCompareOption(config_options, opt_info, "", "C", &e1, &e2);
+  TestParseAndCompareOption(config_options, opt_info, "", "C", &e1, &e2);
   ASSERT_EQ(e2, TestEnum::kC);
 
   ASSERT_NOK(opt_info.Parse(config_options, "", "D", &e1));
@@ -3945,44 +3952,44 @@ TEST_F(OptionTypeInfoTest, TestBuiltinEnum) {
   ConfigOptions config_options;
   for (auto iter : OptionsHelper::compaction_style_string_map) {
     CompactionStyle e1, e2;
-    TestAndCompareOption(config_options,
+    TestParseAndCompareOption(config_options,
-                         OptionTypeInfo(0, OptionType::kCompactionStyle),
+                              OptionTypeInfo(0, OptionType::kCompactionStyle),
-                         "CompactionStyle", iter.first, &e1, &e2);
+                              "CompactionStyle", iter.first, &e1, &e2);
     ASSERT_EQ(e1, iter.second);
   }
   for (auto iter : OptionsHelper::compaction_pri_string_map) {
     CompactionPri e1, e2;
-    TestAndCompareOption(config_options,
+    TestParseAndCompareOption(config_options,
-                         OptionTypeInfo(0, OptionType::kCompactionPri),
+                              OptionTypeInfo(0, OptionType::kCompactionPri),
-                         "CompactionPri", iter.first, &e1, &e2);
+                              "CompactionPri", iter.first, &e1, &e2);
     ASSERT_EQ(e1, iter.second);
   }
   for (auto iter : OptionsHelper::compression_type_string_map) {
     CompressionType e1, e2;
-    TestAndCompareOption(config_options,
+    TestParseAndCompareOption(config_options,
-                         OptionTypeInfo(0, OptionType::kCompressionType),
+                              OptionTypeInfo(0, OptionType::kCompressionType),
-                         "CompressionType", iter.first, &e1, &e2);
+                              "CompressionType", iter.first, &e1, &e2);
     ASSERT_EQ(e1, iter.second);
   }
   for (auto iter : OptionsHelper::compaction_stop_style_string_map) {
     CompactionStopStyle e1, e2;
-    TestAndCompareOption(config_options,
+    TestParseAndCompareOption(
-                         OptionTypeInfo(0, OptionType::kCompactionStopStyle),
+        config_options, OptionTypeInfo(0, OptionType::kCompactionStopStyle),
-                         "CompactionStopStyle", iter.first, &e1, &e2);
+        "CompactionStopStyle", iter.first, &e1, &e2);
     ASSERT_EQ(e1, iter.second);
   }
   for (auto iter : OptionsHelper::checksum_type_string_map) {
     ChecksumType e1, e2;
-    TestAndCompareOption(config_options,
+    TestParseAndCompareOption(config_options,
-                         OptionTypeInfo(0, OptionType::kChecksumType),
+                              OptionTypeInfo(0, OptionType::kChecksumType),
-                         "CheckSumType", iter.first, &e1, &e2);
+                              "CheckSumType", iter.first, &e1, &e2);
     ASSERT_EQ(e1, iter.second);
   }
   for (auto iter : OptionsHelper::encoding_type_string_map) {
     EncodingType e1, e2;
-    TestAndCompareOption(config_options,
+    TestParseAndCompareOption(config_options,
-                         OptionTypeInfo(0, OptionType::kEncodingType),
+                              OptionTypeInfo(0, OptionType::kEncodingType),
-                         "EncodingType", iter.first, &e1, &e2);
+                              "EncodingType", iter.first, &e1, &e2);
     ASSERT_EQ(e1, iter.second);
   }
 }
@@ -4021,15 +4028,17 @@ TEST_F(OptionTypeInfoTest, TestStruct) {
   Extended e1, e2;
   ConfigOptions config_options;
   std::string mismatch;
-  TestAndCompareOption(config_options, basic_info, "b", "{i=33;s=33}", &e1.b,
+  TestParseAndCompareOption(config_options, basic_info, "b", "{i=33;s=33}",
-                       &e2.b);
+                            &e1.b, &e2.b);
   ASSERT_EQ(e1.b.i, 33);
   ASSERT_EQ(e1.b.s, "33");
 
-  TestAndCompareOption(config_options, basic_info, "b.i", "44", &e1.b, &e2.b);
+  TestParseAndCompareOption(config_options, basic_info, "b.i", "44", &e1.b,
+                            &e2.b);
   ASSERT_EQ(e1.b.i, 44);
 
-  TestAndCompareOption(config_options, basic_info, "i", "55", &e1.b, &e2.b);
+  TestParseAndCompareOption(config_options, basic_info, "i", "55", &e1.b,
+                            &e2.b);
   ASSERT_EQ(e1.b.i, 55);
 
   e1.b.i = 0;
@@ -4052,17 +4061,18 @@ TEST_F(OptionTypeInfoTest, TestStruct) {
   ASSERT_NOK(basic_info.Parse(config_options, "b.j", "44", &e1.b));
   ASSERT_NOK(basic_info.Parse(config_options, "j", "44", &e1.b));
 
-  TestAndCompareOption(config_options, extended_info, "e",
+  TestParseAndCompareOption(config_options, extended_info, "e",
-                       "b={i=55;s=55}; j=22;", &e1, &e2);
+                            "b={i=55;s=55}; j=22;", &e1, &e2);
   ASSERT_EQ(e1.b.i, 55);
   ASSERT_EQ(e1.j, 22);
   ASSERT_EQ(e1.b.s, "55");
-  TestAndCompareOption(config_options, extended_info, "e.b", "{i=66;s=66;}",
+  TestParseAndCompareOption(config_options, extended_info, "e.b",
-                       &e1, &e2);
+                            "{i=66;s=66;}", &e1, &e2);
   ASSERT_EQ(e1.b.i, 66);
   ASSERT_EQ(e1.j, 22);
   ASSERT_EQ(e1.b.s, "66");
-  TestAndCompareOption(config_options, extended_info, "e.b.i", "77", &e1, &e2);
+  TestParseAndCompareOption(config_options, extended_info, "e.b.i", "77", &e1,
+                            &e2);
   ASSERT_EQ(e1.b.i, 77);
   ASSERT_EQ(e1.j, 22);
   ASSERT_EQ(e1.b.s, "66");
@@ -4076,7 +4086,8 @@ TEST_F(OptionTypeInfoTest, TestVectorType) {
   std::string mismatch;
 
   ConfigOptions config_options;
-  TestAndCompareOption(config_options, vec_info, "v", "a:b:c:d", &vec1, &vec2);
+  TestParseAndCompareOption(config_options, vec_info, "v", "a:b:c:d", &vec1,
+                            &vec2);
   ASSERT_EQ(vec1.size(), 4);
   ASSERT_EQ(vec1[0], "a");
   ASSERT_EQ(vec1[1], "b");
@@ -4087,8 +4098,8 @@ TEST_F(OptionTypeInfoTest, TestVectorType) {
   ASSERT_EQ(mismatch, "v");
 
   // Test vectors with inner brackets
-  TestAndCompareOption(config_options, vec_info, "v", "a:{b}:c:d", &vec1,
+  TestParseAndCompareOption(config_options, vec_info, "v", "a:{b}:c:d", &vec1,
-                       &vec2);
+                            &vec2);
   ASSERT_EQ(vec1.size(), 4);
   ASSERT_EQ(vec1[0], "a");
   ASSERT_EQ(vec1[1], "b");
@@ -4098,14 +4109,33 @@ TEST_F(OptionTypeInfoTest, TestVectorType) {
   OptionTypeInfo bar_info = OptionTypeInfo::Vector<std::string>(
       0, OptionVerificationType::kNormal, OptionTypeFlags::kNone,
       {0, OptionType::kString}, '|');
-  TestAndCompareOption(config_options, vec_info, "v", "x|y|z", &vec1, &vec2);
+  TestParseAndCompareOption(config_options, vec_info, "v", "x|y|z", &vec1,
+                            &vec2);
   // Test vectors with inner vector
-  TestAndCompareOption(config_options, bar_info, "v",
+  TestParseAndCompareOption(config_options, bar_info, "v",
-                       "a|{b1|b2}|{c1|c2|{d1|d2}}", &vec1, &vec2);
+                            "a|{b1|b2}|{c1|c2|{d1|d2}}", &vec1, &vec2, false);
   ASSERT_EQ(vec1.size(), 3);
   ASSERT_EQ(vec1[0], "a");
   ASSERT_EQ(vec1[1], "b1|b2");
   ASSERT_EQ(vec1[2], "c1|c2|{d1|d2}");
+
+  TestParseAndCompareOption(config_options, bar_info, "v",
+                            "{a1|a2}|{b1|{c1|c2}}|d1", &vec1, &vec2, true);
+  ASSERT_EQ(vec1.size(), 3);
+  ASSERT_EQ(vec1[0], "a1|a2");
+  ASSERT_EQ(vec1[1], "b1|{c1|c2}");
+  ASSERT_EQ(vec1[2], "d1");
+
+  TestParseAndCompareOption(config_options, bar_info, "v", "{a1}", &vec1, &vec2,
+                            false);
+  ASSERT_EQ(vec1.size(), 1);
+  ASSERT_EQ(vec1[0], "a1");
+
+  TestParseAndCompareOption(config_options, bar_info, "v", "{a1|a2}|{b1|b2}",
+                            &vec1, &vec2, true);
+  ASSERT_EQ(vec1.size(), 2);
+  ASSERT_EQ(vec1[0], "a1|a2");
+  ASSERT_EQ(vec1[1], "b1|b2");
 }
 
 TEST_F(OptionTypeInfoTest, TestStaticType) {

table/mock_table.h

@@ -49,7 +49,8 @@ class MockTableFactory : public TableFactory {
   };
 
   MockTableFactory();
-  const char* Name() const override { return "MockTable"; }
+  static const char* kClassName() { return "MockTable"; }
+  const char* Name() const override { return kClassName(); }
   using TableFactory::NewTableReader;
   Status NewTableReader(
       const ReadOptions& ro, const TableReaderOptions& table_reader_options,

test_util/testutil.cc

@@ -25,6 +25,7 @@
 #include "port/port.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/system_clock.h"
+#include "rocksdb/utilities/object_registry.h"
 #include "test_util/sync_point.h"
 #include "util/random.h"
 
@@ -601,5 +602,41 @@ Status CreateEnvFromSystem(const ConfigOptions& config_options, Env** result,
   }
 }
 
+#ifndef ROCKSDB_LITE
+// This method loads existing test classes into the ObjectRegistry
+int RegisterTestObjects(ObjectLibrary& library, const std::string& /*arg*/) {
+  size_t num_types;
+  library.Register<const Comparator>(
+      test::SimpleSuffixReverseComparator::kClassName(),
+      [](const std::string& /*uri*/,
+         std::unique_ptr<const Comparator>* /*guard*/,
+         std::string* /* errmsg */) {
+        static test::SimpleSuffixReverseComparator ssrc;
+        return &ssrc;
+      });
+  library.Register<MergeOperator>(
+      "Changling",
+      [](const std::string& uri, std::unique_ptr<MergeOperator>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new test::ChanglingMergeOperator(uri));
+        return guard->get();
+      });
+  library.Register<CompactionFilter>(
+      "Changling",
+      [](const std::string& uri, std::unique_ptr<CompactionFilter>* /*guard*/,
+         std::string* /* errmsg */) {
+        return new test::ChanglingCompactionFilter(uri);
+      });
+  library.Register<CompactionFilterFactory>(
+      "Changling", [](const std::string& uri,
+                      std::unique_ptr<CompactionFilterFactory>* guard,
+                      std::string* /* errmsg */) {
+        guard->reset(new test::ChanglingCompactionFilterFactory(uri));
+        return guard->get();
+      });
+
+  return static_cast<int>(library.GetFactoryCount(&num_types));
+}
+#endif  // ROCKSDB_LITE
 }  // namespace test
 }  // namespace ROCKSDB_NAMESPACE

test_util/testutil.h

@@ -27,6 +27,7 @@
 
 namespace ROCKSDB_NAMESPACE {
 class FileSystem;
+class ObjectLibrary;
 class Random;
 class SequentialFile;
 class SequentialFileReader;
@@ -895,5 +896,10 @@ void DeleteDir(Env* env, const std::string& dirname);
 // environment variables.
 Status CreateEnvFromSystem(const ConfigOptions& options, Env** result,
                            std::shared_ptr<Env>* guard);
+
+#ifndef ROCKSDB_LITE
+// Registers the testutil classes with the ObjectLibrary
+int RegisterTestObjects(ObjectLibrary& library, const std::string& /*arg*/);
+#endif  // ROCKSDB_LITE
 }  // namespace test
 }  // namespace ROCKSDB_NAMESPACE