From 4774a9409b150a40e5cb48df65099b0111b6391a Mon Sep 17 00:00:00 2001
From: Siying Dong <siying.d@fb.com>
Date: Tue, 26 Mar 2019 14:15:26 -0700
Subject: [PATCH] Allow option string to get comparator from object registry
 (#5106)

Summary:
Even customized ldb may not be able to read data from some databases if
comparator is not standard. We modify option helper to get comparator from
object registry so that we can use customized ldb to read non-standard
comparator.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5106

Differential Revision: D14622107

Pulled By: siying

fbshipit-source-id: 151dcb295a35a4c7d54f919cd4e322a89dc601c9
---
 options/options_helper.cc | 15 ++++++++++++++-
 options/options_test.cc   | 13 +++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/options/options_helper.cc b/options/options_helper.cc
index d8b094021..ed0efa3e4 100644
--- a/options/options_helper.cc
+++ b/options/options_helper.cc
@@ -19,6 +19,7 @@
 #include "rocksdb/rate_limiter.h"
 #include "rocksdb/slice_transform.h"
 #include "rocksdb/table.h"
+#include "rocksdb/utilities/object_registry.h"
 #include "table/block_based_table_factory.h"
 #include "table/plain_table_factory.h"
 #include "util/cast_util.h"
@@ -240,6 +241,8 @@ std::unordered_map<std::string, CompressionType>
         {"kDisableCompressionOption", kDisableCompressionOption}};
 #ifndef ROCKSDB_LITE
 
+const std::string kNameComparator = "comparator";
+
 template <typename T>
 Status GetStringFromStruct(
     std::string* opt_string, const T& options,
@@ -1013,6 +1016,16 @@ Status ParseColumnFamilyOption(const std::string& name,
         return s;
       }
     } else {
+      if (name == kNameComparator) {
+        // Try to get comparator from object registry first.
+        std::unique_ptr<const Comparator> comp_guard;
+        const Comparator* comp =
+            NewCustomObject<const Comparator>(value, &comp_guard);
+        // Only support static comparator for now.
+        if (comp != nullptr && !comp_guard) {
+          new_options->comparator = comp;
+        }
+      }
       auto iter = cf_options_type_info.find(name);
       if (iter == cf_options_type_info.end()) {
         return Status::InvalidArgument(
@@ -1876,7 +1889,7 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offset_of(&ColumnFamilyOptions::bottommost_compression),
           OptionType::kCompressionType, OptionVerificationType::kNormal, false,
           0}},
-        {"comparator",
+        {kNameComparator,
          {offset_of(&ColumnFamilyOptions::comparator), OptionType::kComparator,
           OptionVerificationType::kByName, false, 0}},
         {"prefix_extractor",
diff --git a/options/options_test.cc b/options/options_test.cc
index d6ccf531a..32f178df9 100644
--- a/options/options_test.cc
+++ b/options/options_test.cc
@@ -26,6 +26,7 @@
 #include "rocksdb/convenience.h"
 #include "rocksdb/memtablerep.h"
 #include "rocksdb/utilities/leveldb_options.h"
+#include "rocksdb/utilities/object_registry.h"
 #include "util/random.h"
 #include "util/stderr_logger.h"
 #include "util/string_util.h"
@@ -335,6 +336,18 @@ TEST_F(OptionsTest, GetColumnFamilyOptionsFromStringTest) {
               &new_cf_opt));
   ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(base_cf_opt, new_cf_opt));
 
+  // Comparator from object registry
+  std::string kCompName = "reverse_comp";
+  static Registrar<const Comparator> test_reg_a(
+      kCompName, [](const std::string& /*name*/,
+                    std::unique_ptr<const Comparator>* /*comparator_guard*/) {
+        return ReverseBytewiseComparator();
+      });
+
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      base_cf_opt, "comparator=" + kCompName + ";", &new_cf_opt));
+  ASSERT_EQ(new_cf_opt.comparator, ReverseBytewiseComparator());
+
   // Wrong key/value pair
   ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
              "write_buffer_size=13;max_write_buffer_number;", &new_cf_opt));