Fix an issue with MemTableRepFactory::CreateFromString (#9273)

Summary:
If ignore_unsupported_options=true, then it is possible for MemTableRepFactory::CreateFromString to succeed without setting a result (result=nullptr).  This would cause the original value to be overwritten with null and an error would be raised later when PrepareOptions is invoked.

Added unit test for this condition.  Will add (in another PR unless required by reviewers) comparable tests for all of the other Customizable classes.

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

Reviewed By: ltamasi

Differential Revision: D32990365

Pulled By: mrambacher

fbshipit-source-id: b150724c3f5ae7346357b3866244fd93466875c7

db/db_options_test.cc

@@ -266,6 +266,50 @@ TEST_F(DBOptionsTest, SetMutableTableOptions) {
   ASSERT_EQ(c_bbto->block_restart_interval, 13);
 }
 
+TEST_F(DBOptionsTest, SetWithCustomMemTableFactory) {
+  class DummySkipListFactory : public SkipListFactory {
+   public:
+    static const char* kClassName() { return "DummySkipListFactory"; }
+    const char* Name() const override { return kClassName(); }
+    explicit DummySkipListFactory() : SkipListFactory(2) {}
+  };
+  {
+    // Verify the DummySkipList cannot be created
+    ConfigOptions config_options;
+    config_options.ignore_unsupported_options = false;
+    std::unique_ptr<MemTableRepFactory> factory;
+    ASSERT_NOK(MemTableRepFactory::CreateFromString(
+        config_options, DummySkipListFactory::kClassName(), &factory));
+  }
+  Options options;
+  options.create_if_missing = true;
+  // Try with fail_if_options_file_error=false/true to update the options
+  for (bool on_error : {false, true}) {
+    options.fail_if_options_file_error = on_error;
+    options.env = env_;
+    options.disable_auto_compactions = false;
+
+    options.memtable_factory.reset(new DummySkipListFactory());
+    Reopen(options);
+
+    ColumnFamilyHandle* cfh = dbfull()->DefaultColumnFamily();
+    ASSERT_OK(
+        dbfull()->SetOptions(cfh, {{"disable_auto_compactions", "true"}}));
+    ColumnFamilyDescriptor cfd;
+    ASSERT_OK(cfh->GetDescriptor(&cfd));
+    ASSERT_STREQ(cfd.options.memtable_factory->Name(),
+                 DummySkipListFactory::kClassName());
+    ColumnFamilyHandle* test = nullptr;
+    ASSERT_OK(dbfull()->CreateColumnFamily(options, "test", &test));
+    ASSERT_OK(test->GetDescriptor(&cfd));
+    ASSERT_STREQ(cfd.options.memtable_factory->Name(),
+                 DummySkipListFactory::kClassName());
+
+    ASSERT_OK(dbfull()->DropColumnFamily(test));
+    delete test;
+  }
+}
+
 TEST_F(DBOptionsTest, SetBytesPerSync) {
   const size_t kValueSize = 1024 * 1024;  // 1MB
   Options options;

options/cf_options.cc

@@ -597,7 +597,7 @@ static std::unordered_map<std::string, OptionTypeInfo>
                 static_cast<std::shared_ptr<MemTableRepFactory>*>(addr);
             Status s =
                 MemTableRepFactory::CreateFromString(opts, value, &factory);
-            if (s.ok()) {
+            if (factory && s.ok()) {
               shared->reset(factory.release());
             }
             return s;
@@ -613,7 +613,7 @@ static std::unordered_map<std::string, OptionTypeInfo>
                 static_cast<std::shared_ptr<MemTableRepFactory>*>(addr);
             Status s =
                 MemTableRepFactory::CreateFromString(opts, value, &factory);
-            if (s.ok()) {
+            if (factory && s.ok()) {
               shared->reset(factory.release());
             }
             return s;

options/configurable.cc

@@ -53,13 +53,13 @@ Status Configurable::PrepareOptions(const ConfigOptions& opts) {
                 opt_info.AsRawPointer<Configurable>(opt_iter.opt_ptr);
             if (config != nullptr) {
               status = config->PrepareOptions(opts);
-              if (!status.ok()) {
-                return status;
-              }
             } else if (!opt_info.CanBeNull()) {
               status = Status::NotFound("Missing configurable object",
                                         map_iter.first);
             }
+            if (!status.ok()) {
+              return status;
+            }
           }
         }
       }