Allow upgrades from nullptr to some merge operator

Summary:
Currently, RocksDB does not allow reopening a preexisting DB with no merge operator defined, with a merge operator defined. This means that if a DB ever want to add a merge operator, there's no way to do so currently.

Fix this by adding a new verification type `kByNameAllowFromNull` which will allow old values to be nullptr, and new values to be non-nullptr.
Closes https://github.com/facebook/rocksdb/pull/2958

Differential Revision: D5961131

Pulled By: lth

fbshipit-source-id: 06179bebd0d90db3d43690b5eb7345e2d5bab1eb

options/options_helper.cc

@@ -787,6 +787,7 @@ Status ParseColumnFamilyOption(const std::string& name,
       switch (opt_info.verification) {
         case OptionVerificationType::kByName:
         case OptionVerificationType::kByNameAllowNull:
+        case OptionVerificationType::kByNameAllowFromNull:
           return Status::NotSupported(
               "Deserializing the specified CF option " + name +
                   " is not supported");

options/options_helper.h

@@ -94,15 +94,17 @@ enum class OptionType {
 
 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
-                     // parser will not include it in serialization and
-                     // verification processes.
+  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.
+  kByNameAllowFromNull,  // Same as kByName, but it also allows the case
+                         // where the old option is 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 parser will not include it in serialization
+                         // and verification processes.
 };
 
 // A struct for storing constant option information such as option name,
@@ -575,7 +577,8 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
       false, 0}},
     {"merge_operator",
      {offset_of(&ColumnFamilyOptions::merge_operator),
-      OptionType::kMergeOperator, OptionVerificationType::kByName, false, 0}},
+      OptionType::kMergeOperator, OptionVerificationType::kByNameAllowFromNull,
+      false, 0}},
     {"compaction_style",
      {offset_of(&ColumnFamilyOptions::compaction_style),
       OptionType::kCompactionStyle, OptionVerificationType::kNormal, false, 0}},

options/options_parser.cc

@@ -589,6 +589,8 @@ bool AreEqualOptions(
               *reinterpret_cast<const InfoLogLevel*>(offset2));
     default:
       if (type_info.verification == OptionVerificationType::kByName ||
+          type_info.verification ==
+              OptionVerificationType::kByNameAllowFromNull ||
           type_info.verification == OptionVerificationType::kByNameAllowNull) {
         std::string value1;
         bool result =
@@ -608,6 +610,11 @@ bool AreEqualOptions(
             if (iter->second == kNullptrString || value1 == kNullptrString) {
               return true;
             }
+          } else if (type_info.verification ==
+                     OptionVerificationType::kByNameAllowFromNull) {
+            if (iter->second == kNullptrString) {
+              return true;
+            }
           }
           return (value1 == iter->second);
         }

options/options_test.cc

@@ -1544,6 +1544,15 @@ TEST_F(OptionsSanityCheckTest, SanityCheck) {
 
   // merge_operator
   {
+    // Test when going from nullptr -> merge operator
+    opts.merge_operator.reset(test::RandomMergeOperator(&rnd));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelNone));
+
+    // persist the change
+    ASSERT_OK(PersistCFOptions(opts));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelExactMatch));
+
     for (int test = 0; test < 5; ++test) {
       // change the merge operator
       opts.merge_operator.reset(test::RandomMergeOperator(&rnd));
@@ -1554,6 +1563,15 @@ TEST_F(OptionsSanityCheckTest, SanityCheck) {
       ASSERT_OK(PersistCFOptions(opts));
       ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelExactMatch));
     }
+
+    // Test when going from merge operator -> nullptr
+    opts.merge_operator = nullptr;
+    ASSERT_NOK(SanityCheckCFOptions(opts, kSanityLevelLooselyCompatible));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelNone));
+
+    // persist the change
+    ASSERT_OK(PersistCFOptions(opts));
+    ASSERT_OK(SanityCheckCFOptions(opts, kSanityLevelExactMatch));
   }
 
   // compaction_filter

table/block_based_table_factory.cc

@@ -346,6 +346,8 @@ Status GetBlockBasedTableOptionsFromMap(
           (iter->second.verification != OptionVerificationType::kByName &&
            iter->second.verification !=
                OptionVerificationType::kByNameAllowNull &&
+           iter->second.verification !=
+               OptionVerificationType::kByNameAllowFromNull &&
            iter->second.verification != OptionVerificationType::kDeprecated)) {
         // Restore "new_options" to the default "base_options".
         *new_table_options = table_options;

table/plain_table_factory.cc

@@ -210,6 +210,8 @@ Status GetPlainTableOptionsFromMap(
           (iter->second.verification != OptionVerificationType::kByName &&
            iter->second.verification !=
                OptionVerificationType::kByNameAllowNull &&
+           iter->second.verification !=
+               OptionVerificationType::kByNameAllowFromNull &&
            iter->second.verification != OptionVerificationType::kDeprecated)) {
         // Restore "new_options" to the default "base_options".
         *new_table_options = table_options;