Option string/map can set merge operator from object registry (#5123)

Summary:
Allow customized merge operator to be loaded from option file/map/string
by allowing users to pre-regiester merge operators to object registry.

Also update HISTORY.md and header files for the same feature for comparator.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5123

Differential Revision: D14658488

Pulled By: siying

fbshipit-source-id: 86ea2fbd2a0a04632d8ea9fceaffefd041f6ae61

HISTORY.md

 # Rocksdb Change Log
 
+### Unreleased
+### New Features
+* When reading from option file/string/map, customized comparators and/or merge operators can be filled according to object registry.
+
 ## 6.1.0 (3/27/2019)
 ### New Features
 * Introduce two more stats levels, kExceptHistogramOrTimers and kExceptTimers.

include/rocksdb/utilities/options_util.h

@@ -33,6 +33,12 @@ namespace rocksdb {
 // * merge_operator
 // * compaction_filter
 //
+// User can also choose to load customized comparator and/or merge_operator
+// through object registry:
+// * comparator needs to be registered through Registrar<const Comparator>
+// * merge operator needs to be registered through
+//     Registrar<std::shared_ptr<MergeOperator>>.
+//
 // For table_factory, this function further supports deserializing
 // BlockBasedTableFactory and its BlockBasedTableOptions except the
 // pointer options of BlockBasedTableOptions (flush_block_policy_factory,

options/options_helper.cc

@@ -242,6 +242,7 @@ std::unordered_map<std::string, CompressionType>
 #ifndef ROCKSDB_LITE
 
 const std::string kNameComparator = "comparator";
+const std::string kNameMergeOperator = "merge_operator";
 
 template <typename T>
 Status GetStringFromStruct(
@@ -1025,7 +1026,17 @@ Status ParseColumnFamilyOption(const std::string& name,
         if (comp != nullptr && !comp_guard) {
           new_options->comparator = comp;
         }
+      } else if (name == kNameMergeOperator) {
+        // Try to get merge operator from object registry first.
+        std::unique_ptr<std::shared_ptr<MergeOperator>> mo_guard;
+        std::shared_ptr<MergeOperator>* mo =
+            NewCustomObject<std::shared_ptr<MergeOperator>>(value, &mo_guard);
+        // Only support static comparator for now.
+        if (mo != nullptr) {
+          new_options->merge_operator = *mo;
+        }
       }
+
       auto iter = cf_options_type_info.find(name);
       if (iter == cf_options_type_info.end()) {
         return Status::InvalidArgument(
@@ -1917,7 +1928,7 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offset_of(&ColumnFamilyOptions::compaction_filter_factory),
           OptionType::kCompactionFilterFactory, OptionVerificationType::kByName,
           false, 0}},
-        {"merge_operator",
+        {kNameMergeOperator,
          {offset_of(&ColumnFamilyOptions::merge_operator),
           OptionType::kMergeOperator,
           OptionVerificationType::kByNameAllowFromNull, false, 0}},

options/options_test.cc

@@ -32,6 +32,7 @@
 #include "util/string_util.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
+#include "utilities/merge_operators/bytesxor.h"
 
 #ifndef GFLAGS
 bool FLAGS_enable_print = false;
@@ -348,6 +349,22 @@ TEST_F(OptionsTest, GetColumnFamilyOptionsFromStringTest) {
       base_cf_opt, "comparator=" + kCompName + ";", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.comparator, ReverseBytewiseComparator());
 
+  // MergeOperator from object registry
+  std::unique_ptr<BytesXOROperator> bxo(new BytesXOROperator());
+  std::string kMoName = bxo->Name();
+  static Registrar<std::shared_ptr<MergeOperator>> test_reg_b(
+      kMoName, [](const std::string& /*name*/,
+                  std::unique_ptr<std::shared_ptr<MergeOperator>>*
+                      merge_operator_guard) {
+        merge_operator_guard->reset(
+            new std::shared_ptr<MergeOperator>(new BytesXOROperator()));
+        return merge_operator_guard->get();
+      });
+
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      base_cf_opt, "merge_operator=" + kMoName + ";", &new_cf_opt));
+  ASSERT_EQ(kMoName, std::string(new_cf_opt.merge_operator->Name()));
+
   // Wrong key/value pair
   ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
              "write_buffer_size=13;max_write_buffer_number;", &new_cf_opt));