[RocksDB Options] Support more options in RocksDBOptionParser for sanity check.

Summary:
RocksDBOptionsParser now supports CompressionType and the following
pointer-typed options in RocksDBOptionParser
for sanity check:
  prefix_extractor
  table_factory
  comparator
  compaction_filter
  compaction_filter_factory
  merge_operator
  memtable_factory

In the RocksDB Options file, only high level information about pointer-typed
options are serialized, and those information is only used for verification
/ sanity check purpose.

Test Plan: added more tests in options_test

Reviewers: igor, IslamAbdelRahman, sdong, anthony

Reviewed By: sdong

Subscribers: dhruba, leveldb

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

util/options_helper.cc

@@ -8,9 +8,13 @@
 #include <cctype>
 #include <cstdlib>
 #include <unordered_set>
+#include <vector>
 #include "rocksdb/cache.h"
+#include "rocksdb/compaction_filter.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/filter_policy.h"
+#include "rocksdb/memtablerep.h"
+#include "rocksdb/merge_operator.h"
 #include "rocksdb/options.h"
 #include "rocksdb/rate_limiter.h"
 #include "rocksdb/slice_transform.h"
@@ -85,25 +89,89 @@ std::string UnescapeOptionString(const std::string& escaped_string) {
 }
 
 namespace {
-CompressionType ParseCompressionType(const std::string& type) {
+std::string trim(const std::string& str) {
-  if (type == "kNoCompression") {
+  if (str.empty()) return std::string();
-    return kNoCompression;
+  size_t start = 0;
-  } else if (type == "kSnappyCompression") {
+  size_t end = str.size() - 1;
-    return kSnappyCompression;
+  while (isspace(str[start]) != 0 && start <= end) {
-  } else if (type == "kZlibCompression") {
+    ++start;
-    return kZlibCompression;
+  }
-  } else if (type == "kBZip2Compression") {
+  while (isspace(str[end]) != 0 && start <= end) {
-    return kBZip2Compression;
+    --end;
-  } else if (type == "kLZ4Compression") {
+  }
-    return kLZ4Compression;
+  if (start <= end) {
-  } else if (type == "kLZ4HCCompression") {
+    return str.substr(start, end - start + 1);
-    return kLZ4HCCompression;
+  }
-  } else if (type == "kZSTDNotFinalCompression") {
+  return std::string();
-    return kZSTDNotFinalCompression;
+}
+
+bool SerializeCompressionType(const CompressionType& type, std::string* value) {
+  switch (type) {
+    case kNoCompression:
+      *value = "kNoCompression";
+      return true;
+    case kSnappyCompression:
+      *value = "kSnappyCompression";
+      return true;
+    case kZlibCompression:
+      *value = "kZlibCompression";
+      return true;
+    case kBZip2Compression:
+      *value = "kBZip2Compression";
+      return true;
+    case kLZ4Compression:
+      *value = "kLZ4Compression";
+      return true;
+    case kLZ4HCCompression:
+      *value = "kLZ4HCCompression";
+      return true;
+    case kZSTDNotFinalCompression:
+      *value = "kZSTDNotFinalCompression";
+      return true;
+    default:
+      return false;
+  }
+}
+
+bool SerializeVectorCompressionType(const std::vector<CompressionType>& types,
+                                    std::string* value) {
+  std::stringstream ss;
+  bool result;
+  for (size_t i = 0; i < types.size(); ++i) {
+    if (i > 0) {
+      ss << ':';
+    }
+    std::string string_type;
+    result = SerializeCompressionType(types[i], &string_type);
+    if (result == false) {
+      return result;
+    }
+    ss << string_type;
+  }
+  *value = ss.str();
+  return true;
+}
+
+bool ParseCompressionType(const std::string& string_value,
+                          CompressionType* type) {
+  if (string_value == "kNoCompression") {
+    *type = kNoCompression;
+  } else if (string_value == "kSnappyCompression") {
+    *type = kSnappyCompression;
+  } else if (string_value == "kZlibCompression") {
+    *type = kZlibCompression;
+  } else if (string_value == "kBZip2Compression") {
+    *type = kBZip2Compression;
+  } else if (string_value == "kLZ4Compression") {
+    *type = kLZ4Compression;
+  } else if (string_value == "kLZ4HCCompression") {
+    *type = kLZ4HCCompression;
+  } else if (string_value == "kZSTDNotFinalCompression") {
+    *type = kZSTDNotFinalCompression;
   } else {
-    throw std::invalid_argument("Unknown compression type: " + type);
+    return false;
   }
-  return kNoCompression;
+  return true;
 }
 
 BlockBasedTableOptions::IndexType ParseBlockBasedTableIndexType(
@@ -205,7 +273,6 @@ double ParseDouble(const std::string& value) {
   return std::strtod(value.c_str(), 0);
 #endif
 }
-
 static const std::unordered_map<char, std::string>
     compaction_style_to_string_map = {
         {kCompactionStyleLevel, "kCompactionStyleLevel"},
@@ -229,6 +296,83 @@ std::string CompactionStyleToString(const CompactionStyle style) {
   return iter->second;
 }
 
+bool ParseVectorCompressionType(
+    const std::string& value,
+    std::vector<CompressionType>* compression_per_level) {
+  compression_per_level->clear();
+  size_t start = 0;
+  while (start < value.size()) {
+    size_t end = value.find(':', start);
+    bool is_ok;
+    CompressionType type;
+    if (end == std::string::npos) {
+      is_ok = ParseCompressionType(value.substr(start), &type);
+      if (!is_ok) {
+        return false;
+      }
+      compression_per_level->emplace_back(type);
+      break;
+    } else {
+      is_ok = ParseCompressionType(value.substr(start, end - start), &type);
+      if (!is_ok) {
+        return false;
+      }
+      compression_per_level->emplace_back(type);
+      start = end + 1;
+    }
+  }
+  return true;
+}
+
+bool ParseSliceTransformHelper(
+    const std::string& kFixedPrefixName, const std::string& kCappedPrefixName,
+    const std::string& value,
+    std::shared_ptr<const SliceTransform>* slice_transform) {
+  auto& pe_value = value;
+  if (pe_value.size() > kFixedPrefixName.size() &&
+      pe_value.compare(0, kFixedPrefixName.size(), kFixedPrefixName) == 0) {
+    int prefix_length = ParseInt(trim(value.substr(kFixedPrefixName.size())));
+    slice_transform->reset(NewFixedPrefixTransform(prefix_length));
+  } else if (pe_value.size() > kCappedPrefixName.size() &&
+             pe_value.compare(0, kCappedPrefixName.size(), kCappedPrefixName) ==
+                 0) {
+    int prefix_length =
+        ParseInt(trim(pe_value.substr(kCappedPrefixName.size())));
+    slice_transform->reset(NewCappedPrefixTransform(prefix_length));
+  } else if (value == "nullptr") {
+    slice_transform->reset();
+  } else {
+    return false;
+  }
+
+  return true;
+}
+
+bool ParseSliceTransform(
+    const std::string& value,
+    std::shared_ptr<const SliceTransform>* slice_transform) {
+  // While we normally don't convert the string representation of a
+  // pointer-typed option into its instance, here we do so for backward
+  // compatibility as we allow this action in SetOption().
+
+  // TODO(yhchiang): A possible better place for these serialization /
+  // deserialization is inside the class definition of pointer-typed
+  // option itself, but this requires a bigger change of public API.
+  bool result =
+      ParseSliceTransformHelper("fixed:", "capped:", value, slice_transform);
+  if (result) {
+    return result;
+  }
+  result = ParseSliceTransformHelper(
+      "rocksdb.FixedPrefix.", "rocksdb.CappedPrefix.", value, slice_transform);
+  if (result) {
+    return result;
+  }
+  // TODO(yhchiang): we can further support other default
+  //                 SliceTransforms here.
+  return false;
+}
+
 bool ParseOptionHelper(char* opt_address, const OptionType& opt_type,
                        const std::string& value) {
   switch (opt_type) {
@@ -260,12 +404,24 @@ bool ParseOptionHelper(char* opt_address, const OptionType& opt_type,
       *reinterpret_cast<CompactionStyle*>(opt_address) =
           ParseCompactionStyle(value);
       break;
+    case OptionType::kCompressionType:
+      return ParseCompressionType(
+          value, reinterpret_cast<CompressionType*>(opt_address));
+    case OptionType::kVectorCompressionType:
+      return ParseVectorCompressionType(
+          value, reinterpret_cast<std::vector<CompressionType>*>(opt_address));
+    case OptionType::kSliceTransform:
+      return ParseSliceTransform(
+          value, reinterpret_cast<std::shared_ptr<const SliceTransform>*>(
+                     opt_address));
     default:
       return false;
   }
   return true;
 }
 
+}  // anonymouse namespace
+
 bool SerializeSingleOptionHelper(const char* opt_address,
                                  const OptionType opt_type,
                                  std::string* value) {
@@ -300,13 +456,69 @@ bool SerializeSingleOptionHelper(const char* opt_address,
       *value = CompactionStyleToString(
           *(reinterpret_cast<const CompactionStyle*>(opt_address)));
       break;
+    case OptionType::kCompressionType:
+      return SerializeCompressionType(
+          *(reinterpret_cast<const CompressionType*>(opt_address)), value);
+    case OptionType::kVectorCompressionType:
+      return SerializeVectorCompressionType(
+          *(reinterpret_cast<const std::vector<CompressionType>*>(opt_address)),
+          value);
+      break;
+    case OptionType::kSliceTransform: {
+      const auto* slice_transform_ptr =
+          reinterpret_cast<const std::shared_ptr<const SliceTransform>*>(
+              opt_address);
+      *value = slice_transform_ptr->get() ? slice_transform_ptr->get()->Name()
+                                          : "nullptr";
+      break;
+    }
+    case OptionType::kTableFactory: {
+      const auto* table_factory_ptr =
+          reinterpret_cast<const std::shared_ptr<const TableFactory>*>(
+              opt_address);
+      *value = table_factory_ptr->get() ? table_factory_ptr->get()->Name()
+                                        : "nullptr";
+      break;
+    }
+    case OptionType::kComparator: {
+      // it's a const pointer of const Comparator*
+      const auto* ptr = reinterpret_cast<const Comparator* const*>(opt_address);
+      *value = *ptr ? (*ptr)->Name() : "nullptr";
+      break;
+    }
+    case OptionType::kCompactionFilter: {
+      // it's a const pointer of const CompactionFilter*
+      const auto* ptr =
+          reinterpret_cast<const CompactionFilter* const*>(opt_address);
+      *value = *ptr ? (*ptr)->Name() : "nullptr";
+      break;
+    }
+    case OptionType::kCompactionFilterFactory: {
+      const auto* ptr =
+          reinterpret_cast<const std::shared_ptr<CompactionFilterFactory>*>(
+              opt_address);
+      *value = ptr->get() ? ptr->get()->Name() : "nullptr";
+      break;
+    }
+    case OptionType::kMemTableRepFactory: {
+      const auto* ptr =
+          reinterpret_cast<const std::shared_ptr<MemTableRepFactory>*>(
+              opt_address);
+      *value = ptr->get() ? ptr->get()->Name() : "nullptr";
+      break;
+    }
+    case OptionType::kMergeOperator: {
+      const auto* ptr =
+          reinterpret_cast<const std::shared_ptr<MergeOperator>*>(opt_address);
+      *value = ptr->get() ? ptr->get()->Name() : "nullptr";
+      break;
+    }
     default:
       return false;
   }
   return true;
 }
 
-}  // anonymouse namespace
 
 template<typename OptionsType>
 bool ParseMemtableOptions(const std::string& name, const std::string& value,
@@ -427,26 +639,6 @@ Status GetMutableOptionsFromStrings(
   return Status::OK();
 }
 
-namespace {
-
-std::string trim(const std::string& str) {
-  if (str.empty()) return std::string();
-  size_t start = 0;
-  size_t end = str.size() - 1;
-  while (isspace(str[start]) != 0 && start <= end) {
-    ++start;
-  }
-  while (isspace(str[end]) != 0 && start <= end) {
-    --end;
-  }
-  if (start <= end) {
-    return str.substr(start, end - start + 1);
-  }
-  return std::string();
-}
-
-}  // anonymous namespace
-
 Status StringToMap(const std::string& opts_str,
                    std::unordered_map<std::string, std::string>* opts_map) {
   assert(opts_map);
@@ -559,23 +751,6 @@ bool ParseColumnFamilyOption(const std::string& name,
         return false;
       }
       new_options->table_factory.reset(NewBlockBasedTableFactory(table_opt));
-    } else if (name == "compression") {
-      new_options->compression = ParseCompressionType(value);
-    } else if (name == "compression_per_level") {
-      new_options->compression_per_level.clear();
-      size_t start = 0;
-      while (true) {
-        size_t end = value.find(':', start);
-        if (end == std::string::npos) {
-          new_options->compression_per_level.push_back(
-              ParseCompressionType(value.substr(start)));
-          break;
-        } else {
-          new_options->compression_per_level.push_back(
-              ParseCompressionType(value.substr(start, end - start)));
-          start = end + 1;
-        }
-      }
     } else if (name == "compression_opts") {
       size_t start = 0;
       size_t end = value.find(':');
@@ -603,26 +778,6 @@ bool ParseColumnFamilyOption(const std::string& name,
     } else if (name == "compaction_options_fifo") {
       new_options->compaction_options_fifo.max_table_files_size =
           ParseUint64(value);
-    } else if (name == "prefix_extractor") {
-      const std::string kFixedPrefixName = "fixed:";
-      const std::string kCappedPrefixName = "capped:";
-      auto& pe_value = value;
-      if (pe_value.size() > kFixedPrefixName.size() &&
-          pe_value.compare(0, kFixedPrefixName.size(), kFixedPrefixName) == 0) {
-        int prefix_length =
-            ParseInt(trim(value.substr(kFixedPrefixName.size())));
-        new_options->prefix_extractor.reset(
-            NewFixedPrefixTransform(prefix_length));
-      } else if (pe_value.size() > kCappedPrefixName.size() &&
-                 pe_value.compare(0, kCappedPrefixName.size(),
-                                  kCappedPrefixName) == 0) {
-        int prefix_length =
-            ParseInt(trim(pe_value.substr(kCappedPrefixName.size())));
-        new_options->prefix_extractor.reset(
-            NewCappedPrefixTransform(prefix_length));
-      } else {
-        return false;
-      }
     } else {
       auto iter = cf_options_type_info.find(name);
       if (iter == cf_options_type_info.end()) {
@@ -740,9 +895,12 @@ bool ParseDBOption(const std::string& name, const std::string& org_value,
         return false;
       }
       const auto& opt_info = iter->second;
-      return ParseOptionHelper(
+      if (opt_info.verification != OptionVerificationType::kByName &&
-          reinterpret_cast<char*>(new_options) + opt_info.offset, opt_info.type,
+          opt_info.verification != OptionVerificationType::kDeprecated) {
-          value);
+        return ParseOptionHelper(
+            reinterpret_cast<char*>(new_options) + opt_info.offset,
+            opt_info.type, value);
+      }
     }
   } catch (const std::exception& e) {
     return false;
@@ -881,7 +1039,12 @@ Status GetColumnFamilyOptionsFromMap(
   for (const auto& o : opts_map) {
     if (!ParseColumnFamilyOption(o.first, o.second, new_options,
                                  input_strings_escaped)) {
-      return Status::InvalidArgument("Can't parse option " + o.first);
+      auto iter = cf_options_type_info.find(o.first);
+      if (iter == cf_options_type_info.end() ||
+          (iter->second.verification != OptionVerificationType::kByName &&
+           iter->second.verification != OptionVerificationType::kDeprecated)) {
+        return Status::InvalidArgument("Can't parse option " + o.first);
+      }
     }
   }
   return Status::OK();
@@ -907,6 +1070,8 @@ Status GetDBOptionsFromMap(
   *new_options = base_options;
   for (const auto& o : opts_map) {
     if (!ParseDBOption(o.first, o.second, new_options, input_strings_escaped)) {
+      // Note that options with kDeprecated validation will pass ParseDBOption
+      // and will not hit the below statement.
       return Status::InvalidArgument("Can't parse option " + o.first);
     }
   }

util/options_helper.h

@@ -65,11 +65,22 @@ enum class OptionType {
   kString,
   kDouble,
   kCompactionStyle,
+  kSliceTransform,
+  kCompressionType,
+  kVectorCompressionType,
+  kTableFactory,
+  kComparator,
+  kCompactionFilter,
+  kCompactionFilterFactory,
+  kMergeOperator,
+  kMemTableRepFactory,
   kUnknown
 };
 
 enum class OptionVerificationType {
   kNormal,
+  kByName,     // The option is pointer typed so we can only verify
+               // based on it's name.
   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
@@ -85,6 +96,11 @@ struct OptionTypeInfo {
   OptionVerificationType verification;
 };
 
+// A helper function that converts "opt_address" to a std::string
+// based on the specified OptionType.
+bool SerializeSingleOptionHelper(const char* opt_address,
+                                 const OptionType opt_type, std::string* value);
+
 static std::unordered_map<std::string, OptionTypeInfo> db_options_type_info = {
     /*
      // not yet supported
@@ -226,7 +242,6 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
     CompactionOptionsFIFO compaction_options_fifo;
     CompactionOptionsUniversal compaction_options_universal;
     CompressionOptions compression_opts;
-    CompressionType compression;
     TablePropertiesCollectorFactories table_properties_collector_factories;
     typedef std::vector<std::shared_ptr<TablePropertiesCollectorFactory>>
         TablePropertiesCollectorFactories;
@@ -234,14 +249,6 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
                                      uint34_t* existing_value_size,
                                      Slice delta_value,
                                      std::string* merged_value);
-    const CompactionFilter* compaction_filter;
-    const Comparator* comparator;
-    std::shared_ptr<CompactionFilterFactory> compaction_filter_factory;
-    std::shared_ptr<MemTableRepFactory> memtable_factory;
-    std::shared_ptr<MergeOperator> merge_operator;
-    std::shared_ptr<TableFactory> table_factory;
-    std::shared_ptr<const SliceTransform> prefix_extractor;
-    std::vector<CompressionType> compression_per_level;
     std::vector<int> max_bytes_for_level_multiplier_additional;
      */
     {"compaction_measure_io_stats",
@@ -360,6 +367,33 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
     {"rate_limit_delay_max_milliseconds",
      {offsetof(struct ColumnFamilyOptions, rate_limit_delay_max_milliseconds),
       OptionType::kUInt, OptionVerificationType::kDeprecated}},
+    {"compression",
+     {offsetof(struct ColumnFamilyOptions, compression),
+      OptionType::kCompressionType, OptionVerificationType::kNormal}},
+    {"compression_per_level",
+     {offsetof(struct ColumnFamilyOptions, compression_per_level),
+      OptionType::kVectorCompressionType, OptionVerificationType::kNormal}},
+    {"comparator",
+     {offsetof(struct ColumnFamilyOptions, comparator), OptionType::kComparator,
+      OptionVerificationType::kByName}},
+    {"prefix_extractor",
+     {offsetof(struct ColumnFamilyOptions, prefix_extractor),
+      OptionType::kSliceTransform, OptionVerificationType::kByName}},
+    {"memtable_factory",
+     {offsetof(struct ColumnFamilyOptions, memtable_factory),
+      OptionType::kMemTableRepFactory, OptionVerificationType::kByName}},
+    {"table_factory",
+     {offsetof(struct ColumnFamilyOptions, table_factory),
+      OptionType::kTableFactory, OptionVerificationType::kByName}},
+    {"compaction_filter",
+     {offsetof(struct ColumnFamilyOptions, compaction_filter),
+      OptionType::kCompactionFilter, OptionVerificationType::kByName}},
+    {"compaction_filter_factory",
+     {offsetof(struct ColumnFamilyOptions, compaction_filter_factory),
+      OptionType::kCompactionFilterFactory, OptionVerificationType::kByName}},
+    {"merge_operator",
+     {offsetof(struct ColumnFamilyOptions, merge_operator),
+      OptionType::kMergeOperator, OptionVerificationType::kByName}},
     {"compaction_style",
      {offsetof(struct ColumnFamilyOptions, compaction_style),
       OptionType::kCompactionStyle, OptionVerificationType::kNormal}}};

util/options_parser.cc

@@ -88,8 +88,10 @@ RocksDBOptionsParser::RocksDBOptionsParser() { Reset(); }
 
 void RocksDBOptionsParser::Reset() {
   db_opt_ = DBOptions();
+  db_opt_map_.clear();
   cf_names_.clear();
   cf_opts_.clear();
+  cf_opt_maps_.clear();
   has_version_section_ = false;
   has_db_options_ = false;
   has_default_cf_options_ = false;
@@ -356,6 +358,7 @@ Status RocksDBOptionsParser::EndSection(
     if (!s.ok()) {
       return s;
     }
+    db_opt_map_ = opt_map;
   } else if (section == kOptionSectionCFOptions) {
     // This condition should be ensured earlier in ParseSection
     // so we make an assertion here.
@@ -367,6 +370,8 @@ Status RocksDBOptionsParser::EndSection(
     if (!s.ok()) {
       return s;
     }
+    // keep the parsed string.
+    cf_opt_maps_.emplace_back(opt_map);
   } else if (section == kOptionSectionVersion) {
     for (const auto pair : opt_map) {
       if (pair.first == "rocksdb_version") {
@@ -444,8 +449,10 @@ bool AreEqualDoubles(const double a, const double b) {
   return (fabs(a - b) < 0.00001);
 }
 
-bool AreEqualOptions(const char* opt1, const char* opt2,
+bool AreEqualOptions(
-                     const OptionTypeInfo& type_info) {
+    const char* opt1, const char* opt2, const OptionTypeInfo& type_info,
+    const std::string& opt_name,
+    const std::unordered_map<std::string, std::string>* opt_map) {
   const char* offset1 = opt1 + type_info.offset;
   const char* offset2 = opt2 + type_info.offset;
   switch (type_info.type) {
@@ -476,7 +483,34 @@ bool AreEqualOptions(const char* opt1, const char* opt2,
     case OptionType::kCompactionStyle:
       return (*reinterpret_cast<const CompactionStyle*>(offset1) ==
               *reinterpret_cast<const CompactionStyle*>(offset2));
+    case OptionType::kCompressionType:
+      return (*reinterpret_cast<const CompressionType*>(offset1) ==
+              *reinterpret_cast<const CompressionType*>(offset2));
+    case OptionType::kVectorCompressionType: {
+      const auto* vec1 =
+          reinterpret_cast<const std::vector<CompressionType>*>(offset1);
+      const auto* vec2 =
+          reinterpret_cast<const std::vector<CompressionType>*>(offset2);
+      return (*vec1 == *vec2);
+    }
     default:
+      if (type_info.verification == OptionVerificationType::kByName) {
+        std::string value1;
+        bool result =
+            SerializeSingleOptionHelper(offset1, type_info.type, &value1);
+        if (result == false) {
+          return false;
+        }
+        if (opt_map == nullptr) {
+          return true;
+        }
+        auto iter = opt_map->find(opt_name);
+        if (iter == opt_map->end()) {
+          return true;
+        } else {
+          return (value1 == iter->second);
+        }
+      }
       return false;
   }
 }
@@ -495,7 +529,7 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
   }
 
   // Verify DBOptions
-  s = VerifyDBOptions(db_opt, *parser.db_opt());
+  s = VerifyDBOptions(db_opt, *parser.db_opt(), parser.db_opt_map());
   if (!s.ok()) {
     return s;
   }
@@ -522,7 +556,8 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
         "the same number of column families as the db instance.");
   }
   for (size_t i = 0; i < cf_opts.size(); ++i) {
-    s = VerifyCFOptions(cf_opts[i], parser.cf_opts()->at(i));
+    s = VerifyCFOptions(cf_opts[i], parser.cf_opts()->at(i),
+                        &(parser.cf_opt_maps()->at(i)));
     if (!s.ok()) {
       return s;
     }
@@ -531,8 +566,9 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
   return Status::OK();
 }
 
-Status RocksDBOptionsParser::VerifyDBOptions(const DBOptions& base_opt,
+Status RocksDBOptionsParser::VerifyDBOptions(
-                                             const DBOptions& new_opt) {
+    const DBOptions& base_opt, const DBOptions& new_opt,
+    const std::unordered_map<std::string, std::string>* opt_map) {
   for (auto pair : db_options_type_info) {
     if (pair.second.verification == OptionVerificationType::kDeprecated) {
       // We skip checking deprecated variables as they might
@@ -540,8 +576,8 @@ Status RocksDBOptionsParser::VerifyDBOptions(const DBOptions& base_opt,
       continue;
     }
     if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
-                         reinterpret_cast<const char*>(&new_opt),
+                         reinterpret_cast<const char*>(&new_opt), pair.second,
-                         pair.second)) {
+                         pair.first, nullptr)) {
       return Status::Corruption(
           "[RocksDBOptionsParser]: "
           "failed the verification on DBOptions::",
@@ -552,7 +588,8 @@ Status RocksDBOptionsParser::VerifyDBOptions(const DBOptions& base_opt,
 }
 
 Status RocksDBOptionsParser::VerifyCFOptions(
-    const ColumnFamilyOptions& base_opt, const ColumnFamilyOptions& new_opt) {
+    const ColumnFamilyOptions& base_opt, const ColumnFamilyOptions& new_opt,
+    const std::unordered_map<std::string, std::string>* new_opt_map) {
   for (auto& pair : cf_options_type_info) {
     if (pair.second.verification == OptionVerificationType::kDeprecated) {
       // We skip checking deprecated variables as they might
@@ -560,8 +597,8 @@ Status RocksDBOptionsParser::VerifyCFOptions(
       continue;
     }
     if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
-                         reinterpret_cast<const char*>(&new_opt),
+                         reinterpret_cast<const char*>(&new_opt), pair.second,
-                         pair.second)) {
+                         pair.first, new_opt_map)) {
       return Status::Corruption(
           "[RocksDBOptionsParser]: "
           "failed the verification on ColumnFamilyOptions::",

util/options_parser.h

@@ -45,8 +45,15 @@ class RocksDBOptionsParser {
                                           const bool trim_only = false);
 
   const DBOptions* db_opt() const { return &db_opt_; }
+  const std::unordered_map<std::string, std::string>* db_opt_map() const {
+    return &db_opt_map_;
+  }
   const std::vector<ColumnFamilyOptions>* cf_opts() const { return &cf_opts_; }
   const std::vector<std::string>* cf_names() const { return &cf_names_; }
+  const std::vector<std::unordered_map<std::string, std::string>>* cf_opt_maps()
+      const {
+    return &cf_opt_maps_;
+  }
 
   const ColumnFamilyOptions* GetCFOptions(const std::string& name) const {
     assert(cf_names_.size() == cf_opts_.size());
@@ -64,11 +71,15 @@ class RocksDBOptionsParser {
       const std::vector<ColumnFamilyOptions>& cf_opts,
       const std::string& file_name, Env* env);
 
-  static Status VerifyDBOptions(const DBOptions& base_opt,
+  static Status VerifyDBOptions(
-                                const DBOptions& new_opt);
+      const DBOptions& base_opt, const DBOptions& new_opt,
+      const std::unordered_map<std::string, std::string>* new_opt_map =
+          nullptr);
 
-  static Status VerifyCFOptions(const ColumnFamilyOptions& base_opt,
+  static Status VerifyCFOptions(
-                                const ColumnFamilyOptions& new_opt);
+      const ColumnFamilyOptions& base_opt, const ColumnFamilyOptions& new_opt,
+      const std::unordered_map<std::string, std::string>* new_opt_map =
+          nullptr);
 
   static Status ExtraParserCheck(const RocksDBOptionsParser& input_parser);
 
@@ -97,8 +108,10 @@ class RocksDBOptionsParser {
 
  private:
   DBOptions db_opt_;
+  std::unordered_map<std::string, std::string> db_opt_map_;
   std::vector<std::string> cf_names_;
   std::vector<ColumnFamilyOptions> cf_opts_;
+  std::vector<std::unordered_map<std::string, std::string>> cf_opt_maps_;
   bool has_version_section_;
   bool has_db_options_;
   bool has_default_cf_options_;

util/options_test.cc

@@ -16,7 +16,9 @@
 #include <inttypes.h>
 
 #include "rocksdb/cache.h"
+#include "rocksdb/compaction_filter.h"
 #include "rocksdb/convenience.h"
+#include "rocksdb/merge_operator.h"
 #include "rocksdb/options.h"
 #include "rocksdb/table.h"
 #include "rocksdb/utilities/leveldb_options.h"
@@ -748,7 +750,132 @@ TEST_F(OptionsTest, DBOptionsSerialization) {
 }
 
 namespace {
+CompressionType RandomCompressionType(Random* rnd) {
+  return static_cast<CompressionType>(rnd->Uniform(6));
+}
+
+void RandomCompressionTypeVector(const size_t count,
+                                 std::vector<CompressionType>* types,
+                                 Random* rnd) {
+  types->clear();
+  for (size_t i = 0; i < count; ++i) {
+    types->emplace_back(RandomCompressionType(rnd));
+  }
+}
+
+const SliceTransform* RandomSliceTransform(Random* rnd, int pre_defined = -1) {
+  int random_num = pre_defined >= 0 ? pre_defined : rnd->Uniform(4);
+  switch (random_num) {
+    case 0:
+      return NewFixedPrefixTransform(rnd->Uniform(20) + 1);
+    case 1:
+      return NewCappedPrefixTransform(rnd->Uniform(20) + 1);
+    case 2:
+      return NewNoopTransform();
+    default:
+      return nullptr;
+  }
+}
+
+TableFactory* RandomTableFactory(Random* rnd, int pre_defined = -1) {
+  int random_num = pre_defined >= 0 ? pre_defined : rnd->Uniform(3);
+  switch (random_num) {
+    case 0:
+      return NewPlainTableFactory();
+    case 1:
+      return NewCuckooTableFactory();
+    default:
+      return NewBlockBasedTableFactory();
+  }
+}
+
+std::string RandomString(Random* rnd, const size_t len) {
+  std::stringstream ss;
+  for (size_t i = 0; i < len; ++i) {
+    ss << static_cast<char>(rnd->Uniform(26) + 'a');
+  }
+  return ss.str();
+}
+
+class ChanglingMergeOperator : public MergeOperator {
+ public:
+  explicit ChanglingMergeOperator(const std::string& name)
+      : name_(name + "MergeOperator") {}
+  ~ChanglingMergeOperator() {}
+
+  void SetName(const std::string& name) { name_ = name; }
+
+  virtual bool FullMerge(const Slice& key, const Slice* existing_value,
+                         const std::deque<std::string>& operand_list,
+                         std::string* new_value,
+                         Logger* logger) const override {
+    return false;
+  }
+  virtual bool PartialMergeMulti(const Slice& key,
+                                 const std::deque<Slice>& operand_list,
+                                 std::string* new_value,
+                                 Logger* logger) const override {
+    return false;
+  }
+  virtual const char* Name() const override { return name_.c_str(); }
+
+ protected:
+  std::string name_;
+};
+
+MergeOperator* RandomMergeOperator(Random* rnd) {
+  return new ChanglingMergeOperator(RandomString(rnd, 10));
+}
+
+class ChanglingCompactionFilter : public CompactionFilter {
+ public:
+  explicit ChanglingCompactionFilter(const std::string& name)
+      : name_(name + "CompactionFilter") {}
+  ~ChanglingCompactionFilter() {}
+
+  void SetName(const std::string& name) { name_ = name; }
+
+  bool Filter(int level, const Slice& key, const Slice& existing_value,
+              std::string* new_value, bool* value_changed) const override {
+    return false;
+  }
+
+  const char* Name() const override { return name_.c_str(); }
+
+ private:
+  std::string name_;
+};
+
+CompactionFilter* RandomCompactionFilter(Random* rnd) {
+  return new ChanglingCompactionFilter(RandomString(rnd, 10));
+}
+
+class ChanglingCompactionFilterFactory : public CompactionFilterFactory {
+ public:
+  explicit ChanglingCompactionFilterFactory(const std::string& name)
+      : name_(name + "CompactionFilterFactory") {}
+  ~ChanglingCompactionFilterFactory() {}
+
+  void SetName(const std::string& name) { name_ = name; }
+
+  std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+      const CompactionFilter::Context& context) override {
+    return std::unique_ptr<CompactionFilter>();
+  }
 
+  // Returns a name that identifies this compaction filter factory.
+  const char* Name() const override { return name_.c_str(); }
+
+ protected:
+  std::string name_;
+};
+
+CompactionFilterFactory* RandomCompactionFilterFactory(Random* rnd) {
+  return new ChanglingCompactionFilterFactory(RandomString(rnd, 10));
+}
+
+// Note that the caller is responsible for releasing non-null
+// cf_opt->compaction_filter.
 void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, Random* rnd) {
   cf_opt->compaction_style = (CompactionStyle)(rnd->Uniform(4));
 
@@ -803,6 +930,21 @@ void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, Random* rnd) {
 
   // unsigned int options
   cf_opt->rate_limit_delay_max_milliseconds = rnd->Uniform(10000);
+
+  // pointer typed options
+  cf_opt->prefix_extractor.reset(RandomSliceTransform(rnd));
+  cf_opt->table_factory.reset(RandomTableFactory(rnd));
+  cf_opt->merge_operator.reset(RandomMergeOperator(rnd));
+  if (cf_opt->compaction_filter) {
+    delete cf_opt->compaction_filter;
+  }
+  cf_opt->compaction_filter = RandomCompactionFilter(rnd);
+  cf_opt->compaction_filter_factory.reset(RandomCompactionFilterFactory(rnd));
+
+  // custom typed options
+  cf_opt->compression = RandomCompressionType(rnd);
+  RandomCompressionTypeVector(cf_opt->num_levels,
+                              &cf_opt->compression_per_level, rnd);
 }
 
 }  // namespace
@@ -824,6 +966,9 @@ TEST_F(OptionsTest, ColumnFamilyOptionsSerialization) {
   ASSERT_OK(GetColumnFamilyOptionsFromString(
       ColumnFamilyOptions(), base_options_file_content, &new_opt));
   ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(base_opt, new_opt));
+  if (base_opt.compaction_filter) {
+    delete base_opt.compaction_filter;
+  }
 }
 
 #endif  // !ROCKSDB_LITE
@@ -1268,12 +1413,101 @@ TEST_F(OptionsParserTest, ParseVersion) {
   }
 }
 
+void VerifyCFPointerTypedOptions(
+    ColumnFamilyOptions* base_cf_opt, const ColumnFamilyOptions* new_cf_opt,
+    const std::unordered_map<std::string, std::string>* new_cf_opt_map) {
+  std::string name_buffer;
+  ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*base_cf_opt, *new_cf_opt,
+                                                  new_cf_opt_map));
+
+  // change the name of merge operator back-and-forth
+  {
+    auto* merge_operator = dynamic_cast<ChanglingMergeOperator*>(
+        base_cf_opt->merge_operator.get());
+    if (merge_operator != nullptr) {
+      name_buffer = merge_operator->Name();
+      // change the name  and expect non-ok status
+      merge_operator->SetName("some-other-name");
+      ASSERT_NOK(RocksDBOptionsParser::VerifyCFOptions(
+          *base_cf_opt, *new_cf_opt, new_cf_opt_map));
+      // change the name back and expect ok status
+      merge_operator->SetName(name_buffer);
+      ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*base_cf_opt, *new_cf_opt,
+                                                      new_cf_opt_map));
+    }
+  }
+
+  // change the name of the compaction filter factory back-and-forth
+  {
+    auto* compaction_filter_factory =
+        dynamic_cast<ChanglingCompactionFilterFactory*>(
+            base_cf_opt->compaction_filter_factory.get());
+    if (compaction_filter_factory != nullptr) {
+      name_buffer = compaction_filter_factory->Name();
+      // change the name and expect non-ok status
+      compaction_filter_factory->SetName("some-other-name");
+      ASSERT_NOK(RocksDBOptionsParser::VerifyCFOptions(
+          *base_cf_opt, *new_cf_opt, new_cf_opt_map));
+      // change the name back and expect ok status
+      compaction_filter_factory->SetName(name_buffer);
+      ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*base_cf_opt, *new_cf_opt,
+                                                      new_cf_opt_map));
+    }
+  }
+
+  // test by setting compaction_filter to nullptr
+  {
+    auto* tmp_compaction_filter = base_cf_opt->compaction_filter;
+    if (tmp_compaction_filter != nullptr) {
+      base_cf_opt->compaction_filter = nullptr;
+      // set compaction_filter to nullptr and expect non-ok status
+      ASSERT_NOK(RocksDBOptionsParser::VerifyCFOptions(
+          *base_cf_opt, *new_cf_opt, new_cf_opt_map));
+      // set the value back and expect ok status
+      base_cf_opt->compaction_filter = tmp_compaction_filter;
+      ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*base_cf_opt, *new_cf_opt,
+                                                      new_cf_opt_map));
+    }
+  }
+
+  // test by setting table_factory to nullptr
+  {
+    auto tmp_table_factory = base_cf_opt->table_factory;
+    if (tmp_table_factory != nullptr) {
+      base_cf_opt->table_factory.reset();
+      // set table_factory to nullptr and expect non-ok status
+      ASSERT_NOK(RocksDBOptionsParser::VerifyCFOptions(
+          *base_cf_opt, *new_cf_opt, new_cf_opt_map));
+      // set the value back and expect ok status
+      base_cf_opt->table_factory = tmp_table_factory;
+      ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*base_cf_opt, *new_cf_opt,
+                                                      new_cf_opt_map));
+    }
+  }
+
+  // test by setting memtable_factory to nullptr
+  {
+    auto tmp_memtable_factory = base_cf_opt->memtable_factory;
+    if (tmp_memtable_factory != nullptr) {
+      base_cf_opt->memtable_factory.reset();
+      // set memtable_factory to nullptr and expect non-ok status
+      ASSERT_NOK(RocksDBOptionsParser::VerifyCFOptions(
+          *base_cf_opt, *new_cf_opt, new_cf_opt_map));
+      // set the value back and expect ok status
+      base_cf_opt->memtable_factory = tmp_memtable_factory;
+      ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*base_cf_opt, *new_cf_opt,
+                                                      new_cf_opt_map));
+    }
+  }
+}
+
 TEST_F(OptionsParserTest, DumpAndParse) {
   DBOptions base_db_opt;
   std::vector<ColumnFamilyOptions> base_cf_opts;
-  std::vector<std::string> cf_names = {
+  std::vector<std::string> cf_names = {"default", "cf1", "cf2", "cf3",
-      // special characters are also included.
+                                       "c:f:4:4:4"
-      "default", "p\\i\\k\\a\\chu\\\\\\", "###rocksdb#1-testcf#2###"};
+                                       "p\\i\\k\\a\\chu\\\\\\",
+                                       "###rocksdb#1-testcf#2###"};
   const int num_cf = static_cast<int>(cf_names.size());
   Random rnd(302);
   RandomInitDBOptions(&base_db_opt, &rnd);
@@ -1282,6 +1516,12 @@ TEST_F(OptionsParserTest, DumpAndParse) {
     ColumnFamilyOptions cf_opt;
     Random cf_rnd(0xFB + c);
     RandomInitCFOptions(&cf_opt, &cf_rnd);
+    if (c < 4) {
+      cf_opt.prefix_extractor.reset(RandomSliceTransform(&rnd, c));
+    }
+    if (c < 3) {
+      cf_opt.table_factory.reset(RandomTableFactory(&rnd, c));
+    }
     base_cf_opts.emplace_back(cf_opt);
   }
 
@@ -1300,13 +1540,29 @@ TEST_F(OptionsParserTest, DumpAndParse) {
   for (int c = 0; c < num_cf; ++c) {
     const auto* cf_opt = parser.GetCFOptions(cf_names[c]);
     ASSERT_NE(cf_opt, nullptr);
-    ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*cf_opt, base_cf_opts[c]));
+    ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(
+        base_cf_opts[c], *cf_opt, &(parser.cf_opt_maps()->at(c))));
   }
+
+  // Further verify pointer-typed options
+  for (int c = 0; c < num_cf; ++c) {
+    const auto* cf_opt = parser.GetCFOptions(cf_names[c]);
+    ASSERT_NE(cf_opt, nullptr);
+    VerifyCFPointerTypedOptions(&base_cf_opts[c], cf_opt,
+                                &(parser.cf_opt_maps()->at(c)));
+  }
+
   ASSERT_EQ(parser.GetCFOptions("does not exist"), nullptr);
 
   base_db_opt.max_open_files++;
   ASSERT_NOK(RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
       base_db_opt, cf_names, base_cf_opts, kOptionsFileName, env_.get()));
+
+  for (int c = 0; c < num_cf; ++c) {
+    if (base_cf_opts[c].compaction_filter) {
+      delete base_cf_opts[c].compaction_filter;
+    }
+  }
 }
 
 namespace {

util/slice.cc

@@ -25,6 +25,11 @@ class FixedPrefixTransform : public SliceTransform {
  public:
   explicit FixedPrefixTransform(size_t prefix_len)
       : prefix_len_(prefix_len),
+        // Note that if any part of the name format changes, it will require
+        // changes on options_helper in order to make RocksDBOptionsParser work
+        // for the new change.
+        // TODO(yhchiang): move serialization / deserializaion code inside
+        // the class implementation itself.
         name_("rocksdb.FixedPrefix." + ToString(prefix_len_)) {}
 
   virtual const char* Name() const override { return name_.c_str(); }
@@ -55,6 +60,11 @@ class CappedPrefixTransform : public SliceTransform {
  public:
   explicit CappedPrefixTransform(size_t cap_len)
       : cap_len_(cap_len),
+        // Note that if any part of the name format changes, it will require
+        // changes on options_helper in order to make RocksDBOptionsParser work
+        // for the new change.
+        // TODO(yhchiang): move serialization / deserializaion code inside
+        // the class implementation itself.
         name_("rocksdb.CappedPrefix." + ToString(cap_len_)) {}
 
   virtual const char* Name() const override { return name_.c_str(); }