Relax the check condition of prefix_extractor in CheckOptionsCompatibility

Summary:
Relax the check condition of prefix_extractor in CheckOptionsCompatibility
by allowing changing value from non-nullptr to nullptr or nullptr to
non-nullptr.

Test Plan:
options_test
options_util_test

Reviewers: sdong, anthony, IslamAbdelRahman, kradhakrishnan, gunnarku

Reviewed By: gunnarku

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D54477

util/options_helper.cc

@@ -805,6 +805,7 @@ Status ParseColumnFamilyOption(const std::string& name,
       }
       switch (opt_info.verification) {
         case OptionVerificationType::kByName:
+        case OptionVerificationType::kByNameAllowNull:
           return Status::NotSupported(
               "Deserializing the specified CF option " + name +
                   " is not supported");
@@ -985,6 +986,7 @@ Status ParseDBOption(const std::string& name,
       }
       switch (opt_info.verification) {
         case OptionVerificationType::kByName:
+        case OptionVerificationType::kByNameAllowNull:
           return Status::NotSupported(
               "Deserializing the specified DB option " + name +
                   " is not supported");
@@ -1082,6 +1084,8 @@ Status GetBlockBasedTableOptionsFromMap(
                                      // the old API, where everything is
                                      // parsable.
           (iter->second.verification != OptionVerificationType::kByName &&
+           iter->second.verification !=
+               OptionVerificationType::kByNameAllowNull &&
            iter->second.verification != OptionVerificationType::kDeprecated)) {
         return Status::InvalidArgument("Can't parse BlockBasedTableOptions:",
                                        o.first + " " + error_message);
@@ -1120,6 +1124,8 @@ Status GetPlainTableOptionsFromMap(
                                      // the old API, where everything is
                                      // parsable.
           (iter->second.verification != OptionVerificationType::kByName &&
+           iter->second.verification !=
+               OptionVerificationType::kByNameAllowNull &&
            iter->second.verification != OptionVerificationType::kDeprecated)) {
         return Status::InvalidArgument("Can't parse PlainTableOptions:",
                                         o.first + " " + error_message);

util/options_helper.h

@@ -100,6 +100,8 @@ enum class OptionVerificationType {
   kNormal,
   kByName,           // The option is pointer typed so we can only verify
                      // based on it's name.
+  kByNameAllowNull,  // Same as kByName, but it also allows the case
+                     // where one of them is a nullptr.
   kDeprecated        // The option is no longer used in rocksdb. The RocksDB
                      // OptionsParser will still accept this option if it
                      // happen to exists in some Options file.  However, the
@@ -433,7 +435,7 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
       OptionVerificationType::kByName}},
     {"prefix_extractor",
      {offsetof(struct ColumnFamilyOptions, prefix_extractor),
-      OptionType::kSliceTransform, OptionVerificationType::kByName}},
+      OptionType::kSliceTransform, OptionVerificationType::kByNameAllowNull}},
     {"memtable_factory",
      {offsetof(struct ColumnFamilyOptions, memtable_factory),
       OptionType::kMemTableRepFactory, OptionVerificationType::kByName}},

util/options_parser.cc

@@ -510,6 +510,7 @@ bool AreEqualOptions(
     const std::unordered_map<std::string, std::string>* opt_map) {
   const char* offset1 = opt1 + type_info.offset;
   const char* offset2 = opt2 + type_info.offset;
+  static const std::string kNullptrString = "nullptr";
   switch (type_info.type) {
     case OptionType::kBoolean:
       return (*reinterpret_cast<const bool*>(offset1) ==
@@ -557,7 +558,8 @@ bool AreEqualOptions(
               offset1) ==
           *reinterpret_cast<const BlockBasedTableOptions::IndexType*>(offset2));
     default:
-      if (type_info.verification == OptionVerificationType::kByName) {
+      if (type_info.verification == OptionVerificationType::kByName ||
+          type_info.verification == OptionVerificationType::kByNameAllowNull) {
         std::string value1;
         bool result =
             SerializeSingleOptionHelper(offset1, type_info.type, &value1);
@@ -571,6 +573,12 @@ bool AreEqualOptions(
         if (iter == opt_map->end()) {
           return true;
         } else {
+          if (type_info.verification ==
+              OptionVerificationType::kByNameAllowNull) {
+            if (iter->second == kNullptrString || value1 == kNullptrString) {
+              return true;
+            }
+          }
           return (value1 == iter->second);
         }
       }

util/options_test.cc

@@ -1308,10 +1308,10 @@ TEST_F(OptionsSanityCheckTest, SanityCheck) {
 
   // prefix_extractor
   {
-    // change the prefix extractor and expect only pass when
-    // sanity-level == kSanityLevelNone
+    // Okay to change prefix_extractor form nullptr to non-nullptr
+    ASSERT_EQ(opts.prefix_extractor.get(), nullptr);
     opts.prefix_extractor.reset(NewCappedPrefixTransform(10));
-    ASSERT_NOK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));
     ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelNone));
 
     // persist the change
@@ -1338,11 +1338,21 @@ TEST_F(OptionsSanityCheckTest, SanityCheck) {
     // expect pass only in kSanityLevelNone
     ASSERT_NOK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));
     ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelNone));
+
+    // Change prefix extractor from non-nullptr to nullptr
+    opts.prefix_extractor.reset();
+    // expect pass as it's safe to change prefix_extractor
+    // from non-null to null
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelNone));
   }
+  // persist the change
+  ASSERT_OK(PersistCFOptions(opts));
+  ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelExactMatch));
 
   // table_factory
   {
-    for (int tb = 2; tb >= 0; --tb) {
+    for (int tb = 0; tb <= 2; ++tb) {
       // change the table factory
       opts.table_factory.reset(test::RandomTableFactory(&rnd, tb));
       ASSERT_NOK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));

utilities/options/options_util_test.cc

@@ -173,8 +173,9 @@ TEST_F(OptionsUtilTest, SanityCheck) {
         (i == 0) ? kDefaultColumnFamilyName : test::RandomName(&rnd_, 10);
 
     cf_descs.back().options.table_factory.reset(NewBlockBasedTableFactory());
+    // Assign non-null values to prefix_extractors except the first cf.
     cf_descs.back().options.prefix_extractor.reset(
-        test::RandomSliceTransform(&rnd_));
+        i != 0 ? test::RandomSliceTransform(&rnd_) : nullptr);
     cf_descs.back().options.merge_operator.reset(
         test::RandomMergeOperator(&rnd_));
   }
@@ -223,9 +224,10 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     std::shared_ptr<const SliceTransform> prefix_extractor =
         cf_descs[1].options.prefix_extractor;
 
+    // It's okay to set prefix_extractor to nullptr.
     ASSERT_NE(prefix_extractor, nullptr);
     cf_descs[1].options.prefix_extractor.reset();
-    ASSERT_NOK(
+    ASSERT_OK(
         CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
 
     cf_descs[1].options.prefix_extractor.reset(new DummySliceTransform());
@@ -237,6 +239,27 @@ TEST_F(OptionsUtilTest, SanityCheck) {
         CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
   }
 
+  // prefix extractor nullptr case
+  {
+    std::shared_ptr<const SliceTransform> prefix_extractor =
+        cf_descs[0].options.prefix_extractor;
+
+    // It's okay to set prefix_extractor to nullptr.
+    ASSERT_EQ(prefix_extractor, nullptr);
+    cf_descs[0].options.prefix_extractor.reset();
+    ASSERT_OK(
+        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+
+    // It's okay to change prefix_extractor from nullptr to non-nullptr
+    cf_descs[0].options.prefix_extractor.reset(new DummySliceTransform());
+    ASSERT_OK(
+        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+
+    cf_descs[0].options.prefix_extractor = prefix_extractor;
+    ASSERT_OK(
+        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+  }
+
   // comparator
   {
     test::SimpleSuffixReverseComparator comparator;