Fix Immutable Customizable Serialization (#8457)

Summary:
If a Customizable option was not mutable, it would still appear in the list of mutable options when serialized. This meant that when the immutable options were used to configure another immutable object, an "option not changeable" status would be returned.

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

Reviewed By: zhichao-cao

Differential Revision: D29428298

Pulled By: mrambacher

fbshipit-source-id: 3945b0b822f8e5955a7c5590fe64dfd5bc1fe6a0

options/cf_options.cc

@@ -539,8 +539,8 @@ static std::unordered_map<std::string, OptionTypeInfo>
              offset_of(&ImmutableCFOptions::user_comparator),
              OptionVerificationType::kByName, OptionTypeFlags::kCompareLoose,
              // Serializes a Comparator
-             [](const ConfigOptions& /*opts*/, const std::string&,
-                const void* addr, std::string* value) {
+             [](const ConfigOptions& opts, const std::string&, const void* addr,
+                std::string* value) {
                // it's a const pointer of const Comparator*
                const auto* ptr = static_cast<const Comparator* const*>(addr);
 
@@ -549,12 +549,14 @@ static std::unordered_map<std::string, OptionTypeInfo>
                // one instead of InternalKeyComparator.
                if (*ptr == nullptr) {
                  *value = kNullptrString;
+               } else if (opts.mutable_options_only) {
+                 *value = "";
                } else {
                  const Comparator* root_comp = (*ptr)->GetRootComparator();
                  if (root_comp == nullptr) {
                    root_comp = (*ptr);
                  }
-                 *value = root_comp->Name();
+                 *value = root_comp->ToString(opts);
                }
                return Status::OK();
              },

options/customizable_test.cc

@@ -132,7 +132,7 @@ static std::unordered_map<std::string, OptionTypeInfo> b_option_info = {
 #ifndef ROCKSDB_LITE
     {"string",
      {offsetof(struct BOptions, s), OptionType::kString,
-      OptionVerificationType::kNormal, OptionTypeFlags::kMutable}},
+      OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
     {"bool",
      {offsetof(struct BOptions, b), OptionType::kBoolean,
       OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
@@ -752,10 +752,25 @@ TEST_F(CustomizableTest, MutableOptionsTest) {
   ASSERT_OK(mc.ConfigureOption(options, "mutable", "{bool=true}"));
   auto* mm_a = mm->get()->GetOptions<AOptions>("A");
   ASSERT_EQ(mm_a->b, true);
-  ASSERT_OK(mc.ConfigureOption(options, "mutable", "{int=11;bool=false}"));
+  ASSERT_OK(mc.ConfigureOption(options, "mutable", "{int=22;bool=false}"));
   mm_a = mm->get()->GetOptions<AOptions>("A");
-  ASSERT_EQ(mm_a->i, 11);
+  ASSERT_EQ(mm_a->i, 22);
   ASSERT_EQ(mm_a->b, false);
+
+  // Only the mutable options should get serialized
+  options.mutable_options_only = false;
+  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));
+  ASSERT_EQ(mismatch, "immutable");
 }
 #endif  // !ROCKSDB_LITE
 

options/options_helper.cc

@@ -1096,8 +1096,6 @@ Status OptionTypeInfo::Serialize(const ConfigOptions& config_options,
     return Status::NotSupported("Cannot serialize option: ", opt_name);
   } else if (serialize_func_ != nullptr) {
     return serialize_func_(config_options, opt_name, opt_addr, opt_value);
-  } else if (SerializeSingleOptionHelper(opt_addr, type_, opt_value)) {
-    return Status::OK();
   } else if (IsCustomizable()) {
     const Customizable* custom = AsRawPointer<Customizable>(opt_ptr);
     if (custom == nullptr) {
@@ -1108,7 +1106,18 @@ Status OptionTypeInfo::Serialize(const ConfigOptions& config_options,
     } else {
       ConfigOptions embedded = config_options;
       embedded.delimiter = ";";
-      *opt_value = custom->ToString(embedded);
+      // If this option is mutable, everything inside it should be considered
+      // mutable
+      if (IsMutable()) {
+        embedded.mutable_options_only = false;
+      }
+      std::string value = custom->ToString(embedded);
+      if (!embedded.mutable_options_only ||
+          value.find("=") != std::string::npos) {
+        *opt_value = value;
+      } else {
+        *opt_value = "";
+      }
     }
     return Status::OK();
   } else if (IsConfigurable()) {
@@ -1119,6 +1128,10 @@ Status OptionTypeInfo::Serialize(const ConfigOptions& config_options,
       *opt_value = config->ToString(embedded);
     }
     return Status::OK();
+  } else if (config_options.mutable_options_only && !IsMutable()) {
+    return Status::OK();
+  } else if (SerializeSingleOptionHelper(opt_addr, type_, opt_value)) {
+    return Status::OK();
   } else {
     return Status::InvalidArgument("Cannot serialize option: ", opt_name);
   }

options/options_test.cc

@@ -1782,6 +1782,9 @@ TEST_F(OptionsTest, OnlyMutableDBOptions) {
   // Comparing all of the options, the two are not equivalent
   ASSERT_FALSE(mdb_config->AreEquivalent(cfg_opts, db_config.get(), &mismatch));
   ASSERT_FALSE(db_config->AreEquivalent(cfg_opts, mdb_config.get(), &mismatch));
+
+  // Make sure there are only mutable options being configured
+  ASSERT_OK(GetDBOptionsFromString(cfg_opts, DBOptions(), opt_str, &db_opts));
 }
 
 TEST_F(OptionsTest, OnlyMutableCFOptions) {
@@ -1795,6 +1798,7 @@ TEST_F(OptionsTest, OnlyMutableCFOptions) {
   std::unordered_set<std::string> a_names;
 
   test::RandomInitCFOptions(&cf_opts, db_opts, &rnd);
+  cf_opts.comparator = ReverseBytewiseComparator();
   auto cf_config = CFOptionsAsConfigurable(cf_opts);
 
   // Get all of the CF Option names (mutable or not)
@@ -1826,7 +1830,11 @@ TEST_F(OptionsTest, OnlyMutableCFOptions) {
   // Comparing all of the options, the two are not equivalent
   ASSERT_FALSE(mcf_config->AreEquivalent(cfg_opts, cf_config.get(), &mismatch));
   ASSERT_FALSE(cf_config->AreEquivalent(cfg_opts, mcf_config.get(), &mismatch));
+  delete cf_opts.compaction_filter;
 
+  // Make sure the options string contains only mutable options
+  ASSERT_OK(GetColumnFamilyOptionsFromString(cfg_opts, ColumnFamilyOptions(),
+                                             opt_str, &cf_opts));
   delete cf_opts.compaction_filter;
 }
 #endif  // !ROCKSDB_LITE