Add Functions to OptionTypeInfo (#6422)

Summary:
Added functions for parsing, serializing, and comparing elements to OptionTypeInfo.  These functions allow all of the special cases that could not be handled directly in the map of OptionTypeInfo to be moved into the map.  Using these functions, every type can be handled via the map rather than special cased.

By adding these functions, the code for handling options can become more standardized (fewer special cases) and (eventually) handled completely by common classes.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6422

Test Plan: pass make check

Reviewed By: siying

Differential Revision: D21269005

Pulled By: zhichao-cao

fbshipit-source-id: 9ba71c721a38ebf9ee88259d60bd81b3282b9077

CMakeLists.txt

@@ -518,6 +518,7 @@ find_package(Threads REQUIRED)
 # Main library source code
 
 set(SOURCES
+        cache/cache.cc
         cache/clock_cache.cc
         cache/lru_cache.cc
         cache/sharded_cache.cc
@@ -632,7 +633,6 @@ set(SOURCES
         options/options.cc
         options/options_helper.cc
         options/options_parser.cc
-        options/options_sanity_check.cc
         port/stack_trace.cc
         table/adaptive/adaptive_table_factory.cc
         table/block_based/binary_search_index_reader.cc

TARGETS

@@ -112,6 +112,7 @@ ROCKSDB_OS_DEPS += ([(
 cpp_library(
     name = "rocksdb_lib",
     srcs = [
+        "cache/cache.cc",
         "cache/clock_cache.cc",
         "cache/lru_cache.cc",
         "cache/sharded_cache.cc",
@@ -230,7 +231,6 @@ cpp_library(
         "options/options.cc",
         "options/options_helper.cc",
         "options/options_parser.cc",
-        "options/options_sanity_check.cc",
         "port/port_posix.cc",
         "port/stack_trace.cc",
         "table/adaptive/adaptive_table_factory.cc",

cache/cache.cc

@@ -0,0 +1,41 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+//
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#include "rocksdb/cache.h"
+
+#include "cache/lru_cache.h"
+#include "options/options_helper.h"
+#include "util/string_util.h"
+
+namespace ROCKSDB_NAMESPACE {
+Status Cache::CreateFromString(const ConfigOptions& /*opts*/,
+                               const std::string& value,
+                               std::shared_ptr<Cache>* result) {
+  Status status;
+  std::shared_ptr<Cache> cache;
+  if (value.find('=') == std::string::npos) {
+    cache = NewLRUCache(ParseSizeT(value));
+  } else {
+#ifndef ROCKSDB_LITE
+    LRUCacheOptions cache_opts;
+    if (!ParseOptionHelper(reinterpret_cast<char*>(&cache_opts),
+                           OptionType::kLRUCacheOptions, value)) {
+      status = Status::InvalidArgument("Invalid cache options");
+    }
+    cache = NewLRUCache(cache_opts);
+#else
+    status = Status::NotSupported("Cannot load cache in LITE mode ", value);
+#endif  //! ROCKSDB_LITE
+  }
+  if (status.ok()) {
+    result->swap(cache);
+  }
+  return status;
+}
+}  // namespace ROCKSDB_NAMESPACE

db/db_options_test.cc

@@ -40,7 +40,7 @@ class DBOptionsTest : public DBTestBase {
     StringToMap(options_str, &options_map);
     std::unordered_map<std::string, std::string> mutable_map;
     for (const auto opt : db_options_type_info) {
-      if (opt.second.IsMutable() && !opt.second.IsDeprecated()) {
+      if (opt.second.IsMutable() && opt.second.ShouldSerialize()) {
         mutable_map[opt.first] = options_map[opt.first];
       }
     }
@@ -58,7 +58,7 @@ class DBOptionsTest : public DBTestBase {
     StringToMap(options_str, &options_map);
     std::unordered_map<std::string, std::string> mutable_map;
     for (const auto opt : cf_options_type_info) {
-      if (opt.second.IsMutable() && !opt.second.IsDeprecated()) {
+      if (opt.second.IsMutable() && opt.second.ShouldSerialize()) {
         mutable_map[opt.first] = options_map[opt.first];
       }
     }

include/rocksdb/cache.h

@@ -33,6 +33,7 @@
 namespace ROCKSDB_NAMESPACE {
 
 class Cache;
+struct ConfigOptions;
 
 extern const bool kDefaultToAdaptiveMutex;
 
@@ -142,6 +143,21 @@ class Cache {
   Cache(const Cache&) = delete;
   Cache& operator=(const Cache&) = delete;
 
+  // Creates a new Cache based on the input value string and returns the result.
+  // Currently, this method can be used to create LRUCaches only
+  // @param config_options
+  // @param value  The value might be:
+  //   - an old-style cache ("1M") -- equivalent to NewLRUCache(1024*102(
+  //   - Name-value option pairs -- "capacity=1M; num_shard_bits=4;
+  //     For the LRUCache, the values are defined in LRUCacheOptions.
+  // @param result The new Cache object
+  // @return OK if the cache was sucessfully created
+  // @return NotFound if an invalid name was specified in the value
+  // @return InvalidArgument if either the options were not valid
+  static Status CreateFromString(const ConfigOptions& config_options,
+                                 const std::string& value,
+                                 std::shared_ptr<Cache>* result);
+
   // Destroys all existing entries by calling the "deleter"
   // function that was passed via the Insert() function.
   //

include/rocksdb/convenience.h

@@ -226,6 +226,12 @@ struct ConfigOptions {
 //     instead of resulting in an unknown-option error.
 // @return Status::OK() on success.  Otherwise, a non-ok status indicating
 //     error will be returned, and "new_options" will be set to "base_options".
+// @return Status::NotFound means the one (or more) of the option name in
+//     the opts_map is not valid for this option
+// @return Status::NotSupported means we do not know how to parse one of the
+//     value for this option
+// @return Status::InvalidArgument means the one of the option values is not
+//     valid for this option.
 Status GetColumnFamilyOptionsFromMap(
     const ConfigOptions& config_options,
     const ColumnFamilyOptions& base_options,
@@ -268,6 +274,12 @@ Status GetColumnFamilyOptionsFromMap(
 //     instead of resulting in an unknown-option error.
 // @return Status::OK() on success.  Otherwise, a non-ok status indicating
 //     error will be returned, and "new_options" will be set to "base_options".
+// @return Status::NotFound means the one (or more) of the option name in
+//     the opts_map is not valid for this option
+// @return Status::NotSupported means we do not know how to parse one of the
+//     value for this option
+// @return Status::InvalidArgument means the one of the option values is not
+//     valid for this option.
 Status GetDBOptionsFromMap(
     const ConfigOptions& cfg_options, const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,

include/rocksdb/filter_policy.h

@@ -20,17 +20,20 @@
 #pragma once
 
 #include <stdlib.h>
+
 #include <memory>
 #include <stdexcept>
 #include <string>
 #include <vector>
 
 #include "rocksdb/advanced_options.h"
+#include "rocksdb/status.h"
 
 namespace ROCKSDB_NAMESPACE {
 
 class Slice;
 struct BlockBasedTableOptions;
+struct ConfigOptions;
 
 // A class that takes a bunch of keys, then generates filter
 class FilterBitsBuilder {
@@ -125,6 +128,18 @@ class FilterPolicy {
  public:
   virtual ~FilterPolicy();
 
+  // Creates a new FilterPolicy based on the input value string and returns the
+  // result The value might be an ID, and ID with properties, or an old-style
+  // policy string.
+  // The value describes the FilterPolicy being created.
+  // For BloomFilters, value may be a ":"-delimited value of the form:
+  //   "bloomfilter:[bits_per_key]:[use_block_based_builder]",
+  //   e.g. ""bloomfilter:4:true"
+  //   The above string is equivalent to calling NewBloomFilterPolicy(4, true).
+  static Status CreateFromString(const ConfigOptions& config_options,
+                                 const std::string& value,
+                                 std::shared_ptr<const FilterPolicy>* result);
+
   // Return the name of this policy.  Note that if the filter encoding
   // changes in an incompatible way, the name returned by this method
   // must be changed.  Otherwise, old incompatible filters may be

options/cf_options.cc

@@ -47,8 +47,79 @@ int offset_of(T1 AdvancedColumnFamilyOptions::*member) {
              size_t(&OptionsHelper::dummy_cf_options));
 }
 
-const std::string kNameComparator = "comparator";
+static Status ParseCompressionOptions(const std::string& value,
-const std::string kNameMergeOperator = "merge_operator";
+                                      const std::string& name,
+                                      CompressionOptions& compression_opts) {
+  size_t start = 0;
+  size_t end = value.find(':');
+  if (end == std::string::npos) {
+    return Status::InvalidArgument("unable to parse the specified CF option " +
+                                   name);
+  }
+  compression_opts.window_bits = ParseInt(value.substr(start, end - start));
+  start = end + 1;
+  end = value.find(':', start);
+  if (end == std::string::npos) {
+    return Status::InvalidArgument("unable to parse the specified CF option " +
+                                   name);
+  }
+  compression_opts.level = ParseInt(value.substr(start, end - start));
+  start = end + 1;
+  if (start >= value.size()) {
+    return Status::InvalidArgument("unable to parse the specified CF option " +
+                                   name);
+  }
+  end = value.find(':', start);
+  compression_opts.strategy =
+      ParseInt(value.substr(start, value.size() - start));
+  // max_dict_bytes is optional for backwards compatibility
+  if (end != std::string::npos) {
+    start = end + 1;
+    if (start >= value.size()) {
+      return Status::InvalidArgument(
+          "unable to parse the specified CF option " + name);
+    }
+    compression_opts.max_dict_bytes =
+        ParseInt(value.substr(start, value.size() - start));
+    end = value.find(':', start);
+  }
+  // zstd_max_train_bytes is optional for backwards compatibility
+  if (end != std::string::npos) {
+    start = end + 1;
+    if (start >= value.size()) {
+      return Status::InvalidArgument(
+          "unable to parse the specified CF option " + name);
+    }
+    compression_opts.zstd_max_train_bytes =
+        ParseInt(value.substr(start, value.size() - start));
+    end = value.find(':', start);
+  }
+  // parallel_threads is optional for backwards compatibility
+  if (end != std::string::npos) {
+    start = end + 1;
+    if (start >= value.size()) {
+      return Status::InvalidArgument(
+          "unable to parse the specified CF option " + name);
+    }
+    compression_opts.parallel_threads =
+        ParseInt(value.substr(start, value.size() - start));
+    end = value.find(':', start);
+  }
+  // enabled is optional for backwards compatibility
+  if (end != std::string::npos) {
+    start = end + 1;
+    if (start >= value.size()) {
+      return Status::InvalidArgument(
+          "unable to parse the specified CF option " + name);
+    }
+    compression_opts.enabled =
+        ParseBoolean("", value.substr(start, value.size() - start));
+  }
+  return Status::OK();
+}
+
+const std::string kOptNameBMCompOpts = "bottommost_compression_opts";
+const std::string kOptNameCompOpts = "compression_opts";
 
 std::unordered_map<std::string, OptionTypeInfo>
     OptionsHelper::cf_options_type_info = {
@@ -280,9 +351,22 @@ std::unordered_map<std::string, OptionTypeInfo>
           OptionType::kCompressionType, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable,
           offsetof(struct MutableCFOptions, bottommost_compression)}},
-        {kNameComparator,
+        {"comparator",
          {offset_of(&ColumnFamilyOptions::comparator), OptionType::kComparator,
-          OptionVerificationType::kByName, OptionTypeFlags::kNone, 0}},
+          OptionVerificationType::kByName, OptionTypeFlags::kCompareLoose, 0,
+          // Parses the string and sets the corresponding comparator
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            auto old_comparator = reinterpret_cast<const Comparator**>(addr);
+            const Comparator* new_comparator = *old_comparator;
+            Status status = ObjectRegistry::NewInstance()->NewStaticObject(
+                value, &new_comparator);
+            if (status.ok()) {
+              *old_comparator = new_comparator;
+              return status;
+            }
+            return Status::OK();
+          }}},
         {"prefix_extractor",
          {offset_of(&ColumnFamilyOptions::prefix_extractor),
           OptionType::kSliceTransform, OptionVerificationType::kByNameAllowNull,
@@ -297,10 +381,74 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offset_of(&ColumnFamilyOptions::memtable_factory),
           OptionType::kMemTableRepFactory, OptionVerificationType::kByName,
           OptionTypeFlags::kNone, 0}},
+        {"memtable",
+         {offset_of(&ColumnFamilyOptions::memtable_factory),
+          OptionType::kMemTableRepFactory, OptionVerificationType::kAlias,
+          OptionTypeFlags::kNone, 0,
+          // Parses the value string and updates the memtable_factory
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            std::unique_ptr<MemTableRepFactory> new_mem_factory;
+            Status s = GetMemTableRepFactoryFromString(value, &new_mem_factory);
+            if (s.ok()) {
+              auto memtable_factory =
+                  reinterpret_cast<std::shared_ptr<MemTableRepFactory>*>(addr);
+              memtable_factory->reset(new_mem_factory.release());
+            }
+            return s;
+          }}},
         {"table_factory",
          {offset_of(&ColumnFamilyOptions::table_factory),
           OptionType::kTableFactory, OptionVerificationType::kByName,
-          OptionTypeFlags::kNone, 0}},
+          OptionTypeFlags::kCompareLoose, 0}},
+        {"block_based_table_factory",
+         {offset_of(&ColumnFamilyOptions::table_factory),
+          OptionType::kTableFactory, OptionVerificationType::kAlias,
+          OptionTypeFlags::kCompareLoose, 0,
+          // Parses the input value and creates a BlockBasedTableFactory
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            // Nested options
+            auto old_table_factory =
+                reinterpret_cast<std::shared_ptr<TableFactory>*>(addr);
+            BlockBasedTableOptions table_opts, base_opts;
+            BlockBasedTableFactory* block_based_table_factory =
+                static_cast_with_check<BlockBasedTableFactory, TableFactory>(
+                    old_table_factory->get());
+            if (block_based_table_factory != nullptr) {
+              base_opts = block_based_table_factory->table_options();
+            }
+            Status s = GetBlockBasedTableOptionsFromString(base_opts, value,
+                                                           &table_opts);
+            if (s.ok()) {
+              old_table_factory->reset(NewBlockBasedTableFactory(table_opts));
+            }
+            return s;
+          }}},
+        {"plain_table_factory",
+         {offset_of(&ColumnFamilyOptions::table_factory),
+          OptionType::kTableFactory, OptionVerificationType::kAlias,
+          OptionTypeFlags::kCompareLoose, 0,
+          // Parses the input value and creates a PlainTableFactory
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            // Nested options
+            auto old_table_factory =
+                reinterpret_cast<std::shared_ptr<TableFactory>*>(addr);
+            PlainTableOptions table_opts, base_opts;
+            PlainTableFactory* plain_table_factory =
+                static_cast_with_check<PlainTableFactory, TableFactory>(
+                    old_table_factory->get());
+            if (plain_table_factory != nullptr) {
+              base_opts = plain_table_factory->table_options();
+            }
+            Status s =
+                GetPlainTableOptionsFromString(base_opts, value, &table_opts);
+            if (s.ok()) {
+              old_table_factory->reset(NewPlainTableFactory(table_opts));
+            }
+            return s;
+          }}},
         {"compaction_filter",
          {offset_of(&ColumnFamilyOptions::compaction_filter),
           OptionType::kCompactionFilter, OptionVerificationType::kByName,
@@ -309,11 +457,19 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offset_of(&ColumnFamilyOptions::compaction_filter_factory),
           OptionType::kCompactionFilterFactory, OptionVerificationType::kByName,
           OptionTypeFlags::kNone, 0}},
-        {kNameMergeOperator,
+        {"merge_operator",
          {offset_of(&ColumnFamilyOptions::merge_operator),
           OptionType::kMergeOperator,
-          OptionVerificationType::kByNameAllowFromNull, OptionTypeFlags::kNone,
+          OptionVerificationType::kByNameAllowFromNull,
-          0}},
+          OptionTypeFlags::kCompareLoose, 0,
+          // Parses the input value as a MergeOperator, updating the value
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            auto mop = reinterpret_cast<std::shared_ptr<MergeOperator>*>(addr);
+            ObjectRegistry::NewInstance()->NewSharedObject<MergeOperator>(value,
+                                                                          mop);
+            return Status::OK();
+          }}},
         {"compaction_style",
          {offset_of(&ColumnFamilyOptions::compaction_style),
           OptionType::kCompactionStyle, OptionVerificationType::kNormal,
@@ -345,7 +501,36 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offset_of(&ColumnFamilyOptions::sample_for_compression),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable,
-          offsetof(struct MutableCFOptions, sample_for_compression)}}};
+          offsetof(struct MutableCFOptions, sample_for_compression)}},
+        // The following properties were handled as special cases in ParseOption
+        // This means that the properties could be read from the options file
+        // but never written to the file or compared to each other.
+        {kOptNameCompOpts,
+         {offset_of(&ColumnFamilyOptions::compression_opts),
+          OptionType::kUnknown, OptionVerificationType::kNormal,
+          (OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever |
+           OptionTypeFlags::kMutable),
+          offsetof(struct MutableCFOptions, compression_opts),
+          // Parses the value as a CompressionOptions
+          [](const ConfigOptions& /*opts*/, const std::string& name,
+             const std::string& value, char* addr) {
+            auto* compression = reinterpret_cast<CompressionOptions*>(addr);
+            return ParseCompressionOptions(value, name, *compression);
+          }}},
+        {kOptNameBMCompOpts,
+         {offset_of(&ColumnFamilyOptions::bottommost_compression_opts),
+          OptionType::kUnknown, OptionVerificationType::kNormal,
+          (OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever |
+           OptionTypeFlags::kMutable),
+          offsetof(struct MutableCFOptions, bottommost_compression_opts),
+          // Parses the value as a CompressionOptions
+          [](const ConfigOptions& /*opts*/, const std::string& name,
+             const std::string& value, char* addr) {
+            auto* compression = reinterpret_cast<CompressionOptions*>(addr);
+            return ParseCompressionOptions(value, name, *compression);
+          }}},
+        // End special case properties
+};
 
 Status ParseColumnFamilyOption(const ConfigOptions& config_options,
                                const std::string& name,
@@ -355,104 +540,19 @@ Status ParseColumnFamilyOption(const ConfigOptions& config_options,
                                  ? UnescapeOptionString(org_value)
                                  : org_value;
   try {
-    if (name == "block_based_table_factory") {
-      // Nested options
-      BlockBasedTableOptions table_opt, base_table_options;
-      BlockBasedTableFactory* block_based_table_factory =
-          static_cast_with_check<BlockBasedTableFactory, TableFactory>(
-              new_options->table_factory.get());
-      if (block_based_table_factory != nullptr) {
-        base_table_options = block_based_table_factory->table_options();
-      }
-      Status table_opt_s = GetBlockBasedTableOptionsFromString(
-          base_table_options, value, &table_opt);
-      if (!table_opt_s.ok()) {
-        return Status::InvalidArgument(
-            "unable to parse the specified CF option " + name);
-      }
-      new_options->table_factory.reset(NewBlockBasedTableFactory(table_opt));
-    } else if (name == "plain_table_factory") {
-      // Nested options
-      PlainTableOptions table_opt, base_table_options;
-      PlainTableFactory* plain_table_factory =
-          static_cast_with_check<PlainTableFactory, TableFactory>(
-              new_options->table_factory.get());
-      if (plain_table_factory != nullptr) {
-        base_table_options = plain_table_factory->table_options();
-      }
-      Status table_opt_s =
-          GetPlainTableOptionsFromString(base_table_options, value, &table_opt);
-      if (!table_opt_s.ok()) {
-        return Status::InvalidArgument(
-            "unable to parse the specified CF option " + name);
-      }
-      new_options->table_factory.reset(NewPlainTableFactory(table_opt));
-    } else if (name == "memtable") {
-      std::unique_ptr<MemTableRepFactory> new_mem_factory;
-      Status mem_factory_s =
-          GetMemTableRepFactoryFromString(value, &new_mem_factory);
-      if (!mem_factory_s.ok()) {
-        return Status::InvalidArgument(
-            "unable to parse the specified CF option " + name);
-      }
-      new_options->memtable_factory.reset(new_mem_factory.release());
-    } else if (name == "bottommost_compression_opts") {
-      Status s = ParseCompressionOptions(
-          value, name, new_options->bottommost_compression_opts);
-      if (!s.ok()) {
-        return s;
-      }
-    } else if (name == "compression_opts") {
-      Status s =
-          ParseCompressionOptions(value, name, new_options->compression_opts);
-      if (!s.ok()) {
-        return s;
-      }
-    } else {
-      if (name == kNameComparator) {
-        // Try to get comparator from object registry first.
-        // Only support static comparator for now.
-        Status status = ObjectRegistry::NewInstance()->NewStaticObject(
-            value, &new_options->comparator);
-        if (status.ok()) {
-          return status;
-        }
-      } else if (name == kNameMergeOperator) {
-        // Try to get merge operator from object registry first.
-        std::shared_ptr<MergeOperator> mo;
-        Status status =
-            ObjectRegistry::NewInstance()->NewSharedObject<MergeOperator>(
-                value, &new_options->merge_operator);
-        // Only support static comparator for now.
-        if (status.ok()) {
-          return status;
-        }
-      }
-
     auto iter = cf_options_type_info.find(name);
     if (iter == cf_options_type_info.end()) {
       return Status::InvalidArgument(
           "Unable to parse the specified CF option " + name);
-      }
-      const auto& opt_info = iter->second;
-      if (opt_info.IsDeprecated() ||
-          ParseOptionHelper(
-              reinterpret_cast<char*>(new_options) + opt_info.offset,
-              opt_info.type, value)) {
-        return Status::OK();
-      } else if (opt_info.IsByName()) {
-        return Status::NotSupported("Deserializing the specified CF option " +
-                                    name + " is not supported");
     } else {
-        return Status::InvalidArgument(
+      return iter->second.ParseOption(
-            "Unable to parse the specified CF option " + name);
+          config_options, name, value,
-      }
+          reinterpret_cast<char*>(new_options) + iter->second.offset);
     }
   } catch (const std::exception&) {
     return Status::InvalidArgument("unable to parse the specified option " +
                                    name);
   }
-  return Status::OK();
 }
 #endif  // ROCKSDB_LITE
 

options/db_options.cc

@@ -14,6 +14,7 @@
 #include "rocksdb/cache.h"
 #include "rocksdb/env.h"
 #include "rocksdb/file_system.h"
+#include "rocksdb/rate_limiter.h"
 #include "rocksdb/sst_file_manager.h"
 #include "rocksdb/wal_filter.h"
 
@@ -332,6 +333,37 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct DBOptions, best_efforts_recovery),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone, 0}},
+        // The following properties were handled as special cases in ParseOption
+        // This means that the properties could be read from the options file
+        // but never written to the file or compared to each other.
+        {"rate_limiter_bytes_per_sec",
+         {offsetof(struct DBOptions, rate_limiter), OptionType::kUnknown,
+          OptionVerificationType::kNormal,
+          (OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever), 0,
+          // Parse the input value as a RateLimiter
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            auto limiter =
+                reinterpret_cast<std::shared_ptr<RateLimiter>*>(addr);
+            limiter->reset(NewGenericRateLimiter(
+                static_cast<int64_t>(ParseUint64(value))));
+            return Status::OK();
+          }}},
+        {"env",
+         {offsetof(struct DBOptions, env), OptionType::kUnknown,
+          OptionVerificationType::kNormal,
+          (OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever), 0,
+          // Parse the input value as an Env
+          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+             const std::string& value, char* addr) {
+            auto old_env = reinterpret_cast<Env**>(addr);  // Get the old value
+            Env* new_env = *old_env;                       // Set new to old
+            Status s = Env::LoadEnv(value, &new_env);      // Update new value
+            if (s.ok()) {                                  // It worked
+              *old_env = new_env;                          // Update the old one
+            }
+            return s;
+          }}},
 };
 #endif  // ROCKSDB_LITE
 

options/options_helper.cc

@@ -272,10 +272,6 @@ std::vector<CompressionType> GetSupportedCompressions() {
 
 #ifndef ROCKSDB_LITE
 
-const std::string kNameEnv = "env";
-const std::string kOptNameBMCompOpts = "bottommost_compression_opts";
-const std::string kOptNameCompOpts = "compression_opts";
-
 namespace {
 template <typename T>
 bool ParseEnum(const std::unordered_map<std::string, T>& type_map,
@@ -382,7 +378,8 @@ static bool SerializeStruct(
 }
 
 static bool ParseSingleStructOption(
-    const std::string& opt_val_str, void* options,
+    const ConfigOptions& config_options, const std::string& opt_val_str,
+    void* options,
     const std::unordered_map<std::string, OptionTypeInfo>& type_info_map) {
   size_t end = opt_val_str.find('=');
   std::string key = opt_val_str.substr(0, end);
@@ -392,20 +389,17 @@ static bool ParseSingleStructOption(
     return false;
   }
   const auto& opt_info = iter->second;
-  if (opt_info.IsDeprecated()) {
+  Status s = opt_info.ParseOption(
-    // Should also skip deprecated sub-options such as
+      config_options, key, value,
-    // fifo_compaction_options_type_info.ttl
+      reinterpret_cast<char*>(options) + opt_info.mutable_offset);
-    return true;
+  return s.ok();
-  }
-  return ParseOptionHelper(
-      reinterpret_cast<char*>(options) + opt_info.mutable_offset, opt_info.type,
-      value);
 }
 
 static bool ParseStructOptions(
     const std::string& opt_str, void* options,
     const std::unordered_map<std::string, OptionTypeInfo>& type_info_map) {
   assert(!opt_str.empty());
+  ConfigOptions config_options;
 
   size_t start = 0;
   if (opt_str[0] == '{') {
@@ -417,8 +411,8 @@ static bool ParseStructOptions(
     }
     size_t end = opt_str.find(';', start);
     size_t len = (end == std::string::npos) ? end : end - start;
-    if (!ParseSingleStructOption(opt_str.substr(start, len), options,
+    if (!ParseSingleStructOption(config_options, opt_str.substr(start, len),
-                                 type_info_map)) {
+                                 options, type_info_map)) {
       return false;
     }
     start = (end == std::string::npos) ? end : end + 1;
@@ -792,129 +786,34 @@ bool SerializeSingleOptionHelper(const char* opt_address,
   return true;
 }
 
-Status ParseCompressionOptions(const std::string& value,
-                               const std::string& name,
-                               CompressionOptions& compression_opts) {
-  size_t start = 0;
-  size_t end = value.find(':');
-  if (end == std::string::npos) {
-    return Status::InvalidArgument("unable to parse the specified CF option " +
-                                   name);
-  }
-  compression_opts.window_bits = ParseInt(value.substr(start, end - start));
-  start = end + 1;
-  end = value.find(':', start);
-  if (end == std::string::npos) {
-    return Status::InvalidArgument("unable to parse the specified CF option " +
-                                   name);
-  }
-  compression_opts.level = ParseInt(value.substr(start, end - start));
-  start = end + 1;
-  if (start >= value.size()) {
-    return Status::InvalidArgument("unable to parse the specified CF option " +
-                                   name);
-  }
-  end = value.find(':', start);
-  compression_opts.strategy =
-      ParseInt(value.substr(start, value.size() - start));
-  // max_dict_bytes is optional for backwards compatibility
-  if (end != std::string::npos) {
-    start = end + 1;
-    if (start >= value.size()) {
-      return Status::InvalidArgument(
-          "unable to parse the specified CF option " + name);
-    }
-    compression_opts.max_dict_bytes =
-        ParseInt(value.substr(start, value.size() - start));
-    end = value.find(':', start);
-  }
-  // zstd_max_train_bytes is optional for backwards compatibility
-  if (end != std::string::npos) {
-    start = end + 1;
-    if (start >= value.size()) {
-      return Status::InvalidArgument(
-          "unable to parse the specified CF option " + name);
-    }
-    compression_opts.zstd_max_train_bytes =
-        ParseInt(value.substr(start, value.size() - start));
-    end = value.find(':', start);
-  }
-  // parallel_threads is optional for backwards compatibility
-  if (end != std::string::npos) {
-    start = end + 1;
-    if (start >= value.size()) {
-      return Status::InvalidArgument(
-          "unable to parse the specified CF option " + name);
-    }
-    compression_opts.parallel_threads =
-        ParseInt(value.substr(start, value.size() - start));
-    end = value.find(':', start);
-  }
-  // enabled is optional for backwards compatibility
-  if (end != std::string::npos) {
-    start = end + 1;
-    if (start >= value.size()) {
-      return Status::InvalidArgument(
-          "unable to parse the specified CF option " + name);
-    }
-    compression_opts.enabled =
-        ParseBoolean("", value.substr(start, value.size() - start));
-  }
-  return Status::OK();
-}
-
 Status GetMutableOptionsFromStrings(
     const MutableCFOptions& base_options,
     const std::unordered_map<std::string, std::string>& options_map,
     Logger* info_log, MutableCFOptions* new_options) {
   assert(new_options);
   *new_options = base_options;
+  ConfigOptions config_options;
   for (const auto& o : options_map) {
-    auto& option_name = o.first;
+    auto iter = cf_options_type_info.find(o.first);
-    auto& option_value = o.second;
-
-    try {
-      if (option_name == kOptNameBMCompOpts) {
-        Status s =
-            ParseCompressionOptions(option_value, option_name,
-                                    new_options->bottommost_compression_opts);
-        if (!s.ok()) {
-          return s;
-        }
-      } else if (option_name == kOptNameCompOpts) {
-        Status s = ParseCompressionOptions(option_value, option_name,
-                                           new_options->compression_opts);
-        if (!s.ok()) {
-          return s;
-        }
-      } else {
-        auto iter = cf_options_type_info.find(option_name);
     if (iter == cf_options_type_info.end()) {
-          return Status::InvalidArgument("Unrecognized option: " + option_name);
+      return Status::InvalidArgument("Unrecognized option: " + o.first);
     }
     const auto& opt_info = iter->second;
     if (!opt_info.IsMutable()) {
-          return Status::InvalidArgument("Option not changeable: " +
+      return Status::InvalidArgument("Option not changeable: " + o.first);
-                                         option_name);
     }
     if (opt_info.IsDeprecated()) {
-          // log warning when user tries to set a deprecated option but don't
+      // log warning when user tries to set a deprecated option but don't fail
-          // fail the call for compatibility.
+      // the call for compatibility.
-          ROCKS_LOG_WARN(info_log,
+      ROCKS_LOG_WARN(info_log, "%s is a deprecated option and cannot be set",
-                         "%s is a deprecated option and cannot be set",
+                     o.first.c_str());
-                         option_name.c_str());
       continue;
     }
-        bool is_ok = ParseOptionHelper(
+    Status s = opt_info.ParseOption(
-            reinterpret_cast<char*>(new_options) + opt_info.mutable_offset,
+        config_options, o.first, o.second,
-            opt_info.type, option_value);
+        reinterpret_cast<char*>(new_options) + opt_info.mutable_offset);
-        if (!is_ok) {
+    if (!s.ok()) {
-          return Status::InvalidArgument("Error parsing " + option_name);
+      return s;
-        }
-      }
-    } catch (std::exception& e) {
-      return Status::InvalidArgument("Error parsing " + option_name + ":" +
-                                     std::string(e.what()));
     }
   }
   return Status::OK();
@@ -926,6 +825,8 @@ Status GetMutableDBOptionsFromStrings(
     MutableDBOptions* new_options) {
   assert(new_options);
   *new_options = base_options;
+  ConfigOptions config_options;
+
   for (const auto& o : options_map) {
     try {
       auto iter = db_options_type_info.find(o.first);
@@ -936,11 +837,11 @@ Status GetMutableDBOptionsFromStrings(
       if (!opt_info.IsMutable()) {
         return Status::InvalidArgument("Option not changeable: " + o.first);
       }
-      bool is_ok = ParseOptionHelper(
+      Status s = opt_info.ParseOption(
-          reinterpret_cast<char*>(new_options) + opt_info.mutable_offset,
+          config_options, o.first, o.second,
-          opt_info.type, o.second);
+          reinterpret_cast<char*>(new_options) + opt_info.mutable_offset);
-      if (!is_ok) {
+      if (!s.ok()) {
-        return Status::InvalidArgument("Error parsing " + o.first);
+        return s;
       }
     } catch (std::exception& e) {
       return Status::InvalidArgument("Error parsing " + o.first + ":" +
@@ -1027,28 +928,26 @@ Status StringToMap(const std::string& opts_str,
 }
 
 Status GetStringFromStruct(
-    const ConfigOptions& cfg_options, const void* const opt_ptr,
+    const ConfigOptions& config_options, const void* const opt_ptr,
     const std::unordered_map<std::string, OptionTypeInfo>& type_info,
     std::string* opt_string) {
   assert(opt_string);
   opt_string->clear();
   for (const auto iter : type_info) {
     const auto& opt_info = iter.second;
-    if (opt_info.IsDeprecated()) {
     // If the option is no longer used in rocksdb and marked as deprecated,
     // we skip it in the serialization.
-      continue;
+    if (opt_info.ShouldSerialize()) {
-    }
+      const char* opt_addr =
-    const char* opt_address =
           reinterpret_cast<const char*>(opt_ptr) + opt_info.offset;
       std::string value;
-    bool result =
+      Status s = opt_info.SerializeOption(config_options, iter.first, opt_addr,
-        SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
+                                          &value);
-    if (result) {
+      if (s.ok()) {
-      opt_string->append(iter.first + "=" + value + cfg_options.delimiter);
+        opt_string->append(iter.first + "=" + value + config_options.delimiter);
       } else {
-      return Status::InvalidArgument("failed to serialize %s\n",
+        return s;
-                                     iter.first.c_str());
+      }
     }
   }
   return Status::OK();
@@ -1103,42 +1002,15 @@ static Status ParseDBOption(const ConfigOptions& config_options,
   const std::string& value = config_options.input_strings_escaped
                                  ? UnescapeOptionString(org_value)
                                  : org_value;
-  try {
-    if (name == "rate_limiter_bytes_per_sec") {
-      new_options->rate_limiter.reset(
-          NewGenericRateLimiter(static_cast<int64_t>(ParseUint64(value))));
-    } else if (name == kNameEnv) {
-      // Currently `Env` can be deserialized from object registry only.
-      Env* env = new_options->env;
-      Status status = Env::LoadEnv(value, &env);
-      // Only support static env for now.
-      if (status.ok()) {
-        new_options->env = env;
-      }
-    } else {
   auto iter = db_options_type_info.find(name);
   if (iter == db_options_type_info.end()) {
     return Status::InvalidArgument("Unrecognized option DBOptions:", name);
-      }
-      const auto& opt_info = iter->second;
-      if (opt_info.IsDeprecated() ||
-          ParseOptionHelper(
-              reinterpret_cast<char*>(new_options) + opt_info.offset,
-              opt_info.type, value)) {
-        return Status::OK();
-      } else if (opt_info.IsByName()) {
-        return Status::NotSupported("Deserializing the specified DB option " +
-                                    name + " is not supported");
   } else {
-        return Status::InvalidArgument(
+    return iter->second.ParseOption(
-            "Unable to parse the specified DB option " + name);
+        config_options, name, value,
+        reinterpret_cast<char*>(new_options) + iter->second.offset);
   }
 }
-  } catch (const std::exception&) {
-    return Status::InvalidArgument("Unable to parse DBOptions:", name);
-  }
-  return Status::OK();
-}
 
 Status GetColumnFamilyOptionsFromMap(
     const ColumnFamilyOptions& base_options,
@@ -1530,6 +1402,222 @@ std::unordered_map<std::string, OptionTypeInfo>
           OptionVerificationType::kNormal, OptionTypeFlags::kMutable,
           offsetof(struct LRUCacheOptions, high_pri_pool_ratio)}}};
 
+Status OptionTypeInfo::ParseOption(const ConfigOptions& config_options,
+                                   const std::string& opt_name,
+                                   const std::string& opt_value,
+                                   char* opt_addr) const {
+  if (IsDeprecated()) {
+    return Status::OK();
+  }
+  try {
+    if (opt_addr == nullptr) {
+      return Status::NotFound("Could not find option: ", opt_name);
+    } else if (parser_func != nullptr) {
+      return parser_func(config_options, opt_name, opt_value, opt_addr);
+    } else if (ParseOptionHelper(opt_addr, type, opt_value)) {
+      return Status::OK();
+    } else if (IsByName()) {
+      return Status::NotSupported("Deserializing the option " + opt_name +
+                                  " is not supported");
+    } else {
+      return Status::InvalidArgument("Error parsing:", opt_name);
+    }
+  } catch (std::exception& e) {
+    return Status::InvalidArgument("Error parsing " + opt_name + ":" +
+                                   std::string(e.what()));
+  }
+}
+
+Status OptionTypeInfo::SerializeOption(const ConfigOptions& config_options,
+                                       const std::string& opt_name,
+                                       const char* opt_addr,
+                                       std::string* opt_value) const {
+  // If the option is no longer used in rocksdb and marked as deprecated,
+  // we skip it in the serialization.
+  Status s;
+  if (opt_addr == nullptr || IsDeprecated()) {
+    return Status::OK();
+  } else if (string_func != nullptr) {
+    return string_func(config_options, opt_name, opt_addr, opt_value);
+  } else if (SerializeSingleOptionHelper(opt_addr, type, opt_value)) {
+    s = Status::OK();
+  } else {
+    s = Status::InvalidArgument("Cannot serialize option: ", opt_name);
+  }
+  return s;
+}
+
+template <typename T>
+bool IsOptionEqual(const char* offset1, const char* offset2) {
+  return (*reinterpret_cast<const T*>(offset1) ==
+          *reinterpret_cast<const T*>(offset2));
+}
+
+static bool AreEqualDoubles(const double a, const double b) {
+  return (fabs(a - b) < 0.00001);
+}
+
+static bool AreOptionsEqual(OptionType type, const char* this_offset,
+                            const char* that_offset) {
+  switch (type) {
+    case OptionType::kBoolean:
+      return IsOptionEqual<bool>(this_offset, that_offset);
+    case OptionType::kInt:
+      return IsOptionEqual<int>(this_offset, that_offset);
+    case OptionType::kUInt:
+      return IsOptionEqual<unsigned int>(this_offset, that_offset);
+    case OptionType::kInt32T:
+      return IsOptionEqual<int32_t>(this_offset, that_offset);
+    case OptionType::kInt64T: {
+      int64_t v1, v2;
+      GetUnaligned(reinterpret_cast<const int64_t*>(this_offset), &v1);
+      GetUnaligned(reinterpret_cast<const int64_t*>(that_offset), &v2);
+      return (v1 == v2);
+    }
+    case OptionType::kVectorInt:
+      return IsOptionEqual<std::vector<int> >(this_offset, that_offset);
+    case OptionType::kUInt32T:
+      return IsOptionEqual<uint32_t>(this_offset, that_offset);
+    case OptionType::kUInt64T: {
+      uint64_t v1, v2;
+      GetUnaligned(reinterpret_cast<const uint64_t*>(this_offset), &v1);
+      GetUnaligned(reinterpret_cast<const uint64_t*>(that_offset), &v2);
+      return (v1 == v2);
+    }
+    case OptionType::kSizeT: {
+      size_t v1, v2;
+      GetUnaligned(reinterpret_cast<const size_t*>(this_offset), &v1);
+      GetUnaligned(reinterpret_cast<const size_t*>(that_offset), &v2);
+      return (v1 == v2);
+    }
+    case OptionType::kString:
+      return IsOptionEqual<std::string>(this_offset, that_offset);
+    case OptionType::kDouble:
+      return AreEqualDoubles(*reinterpret_cast<const double*>(this_offset),
+                             *reinterpret_cast<const double*>(that_offset));
+    case OptionType::kVectorCompressionType:
+      return IsOptionEqual<std::vector<CompressionType> >(this_offset,
+                                                          that_offset);
+    case OptionType::kCompactionStyle:
+      return IsOptionEqual<CompactionStyle>(this_offset, that_offset);
+    case OptionType::kCompactionStopStyle:
+      return IsOptionEqual<CompactionStopStyle>(this_offset, that_offset);
+    case OptionType::kCompactionPri:
+      return IsOptionEqual<CompactionPri>(this_offset, that_offset);
+    case OptionType::kCompressionType:
+      return IsOptionEqual<CompressionType>(this_offset, that_offset);
+    case OptionType::kChecksumType:
+      return IsOptionEqual<ChecksumType>(this_offset, that_offset);
+    case OptionType::kEncodingType:
+      return IsOptionEqual<EncodingType>(this_offset, that_offset);
+    case OptionType::kBlockBasedTableIndexType:
+      return IsOptionEqual<BlockBasedTableOptions::IndexType>(this_offset,
+                                                              that_offset);
+    case OptionType::kBlockBasedTableDataBlockIndexType:
+      return IsOptionEqual<BlockBasedTableOptions::DataBlockIndexType>(
+          this_offset, that_offset);
+    case OptionType::kBlockBasedTableIndexShorteningMode:
+      return IsOptionEqual<BlockBasedTableOptions::IndexShorteningMode>(
+          this_offset, that_offset);
+    case OptionType::kWALRecoveryMode:
+      return IsOptionEqual<WALRecoveryMode>(this_offset, that_offset);
+    case OptionType::kAccessHint:
+      return IsOptionEqual<DBOptions::AccessHint>(this_offset, that_offset);
+    case OptionType::kInfoLogLevel:
+      return IsOptionEqual<InfoLogLevel>(this_offset, that_offset);
+    case OptionType::kCompactionOptionsFIFO: {
+      CompactionOptionsFIFO lhs =
+          *reinterpret_cast<const CompactionOptionsFIFO*>(this_offset);
+      CompactionOptionsFIFO rhs =
+          *reinterpret_cast<const CompactionOptionsFIFO*>(that_offset);
+      if (lhs.max_table_files_size == rhs.max_table_files_size &&
+          lhs.allow_compaction == rhs.allow_compaction) {
+        return true;
+      }
+      return false;
+    }
+    case OptionType::kCompactionOptionsUniversal: {
+      CompactionOptionsUniversal lhs =
+          *reinterpret_cast<const CompactionOptionsUniversal*>(this_offset);
+      CompactionOptionsUniversal rhs =
+          *reinterpret_cast<const CompactionOptionsUniversal*>(that_offset);
+      if (lhs.size_ratio == rhs.size_ratio &&
+          lhs.min_merge_width == rhs.min_merge_width &&
+          lhs.max_merge_width == rhs.max_merge_width &&
+          lhs.max_size_amplification_percent ==
+              rhs.max_size_amplification_percent &&
+          lhs.compression_size_percent == rhs.compression_size_percent &&
+          lhs.stop_style == rhs.stop_style &&
+          lhs.allow_trivial_move == rhs.allow_trivial_move) {
+        return true;
+      }
+      return false;
+    }
+    default:
+      return false;
+  }  // End switch
+}
+
+bool OptionTypeInfo::MatchesOption(const ConfigOptions& config_options,
+                                   const std::string& opt_name,
+                                   const char* this_addr, const char* that_addr,
+
+                                   std::string* mismatch) const {
+  if (!config_options.IsCheckEnabled(GetSanityLevel())) {
+    return true;  // If the sanity level is not being checked, skip it
+  }
+  if (this_addr == nullptr || that_addr == nullptr) {
+    if (this_addr == that_addr) {
+      return true;
+    }
+  } else if (equals_func != nullptr) {
+    if (equals_func(config_options, opt_name, this_addr, that_addr, mismatch)) {
+      return true;
+    }
+  } else if (AreOptionsEqual(type, this_addr, that_addr)) {
+    return true;
+  }
+  if (mismatch->empty()) {
+    *mismatch = opt_name;
+  }
+  return false;
+}
+
+bool OptionTypeInfo::MatchesByName(const ConfigOptions& config_options,
+                                   const std::string& opt_name,
+                                   const char* this_addr,
+                                   const char* that_addr) const {
+  if (IsByName()) {
+    std::string that_value;
+    if (SerializeOption(config_options, opt_name, that_addr, &that_value)
+            .ok()) {
+      return MatchesByName(config_options, opt_name, this_addr, that_value);
+    }
+  }
+  return false;
+}
+
+bool OptionTypeInfo::MatchesByName(const ConfigOptions& config_options,
+                                   const std::string& opt_name,
+                                   const char* opt_addr,
+                                   const std::string& that_value) const {
+  std::string this_value;
+  if (!IsByName()) {
+    return false;
+  } else if (!SerializeOption(config_options, opt_name, opt_addr, &this_value)
+                  .ok()) {
+    return false;
+  } else if (IsEnabled(OptionVerificationType::kByNameAllowFromNull)) {
+    if (that_value == kNullptrString) {
+      return true;
+    }
+  } else if (IsEnabled(OptionVerificationType::kByNameAllowNull)) {
+    if (that_value == kNullptrString) {
+      return true;
+    }
+  }
+  return (this_value == that_value);
+}
 #endif  // !ROCKSDB_LITE
 
 }  // namespace ROCKSDB_NAMESPACE

options/options_helper.h

@@ -57,10 +57,6 @@ Status GetTableFactoryFromMap(
     std::shared_ptr<TableFactory>* table_factory,
     bool ignore_unknown_options = false);
 
-Status ParseCompressionOptions(const std::string& value,
-                               const std::string& name,
-                               CompressionOptions& compression_opts);
-
 Status GetTableFactoryFromMap(
     const ConfigOptions& config_options, const std::string& factory_name,
     const std::unordered_map<std::string, std::string>& opt_map,

options/options_parser.cc

@@ -16,7 +16,6 @@
 #include "file/read_write_util.h"
 #include "file/writable_file_writer.h"
 #include "options/options_helper.h"
-#include "options/options_sanity_check.h"
 #include "port/port.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
@@ -518,170 +517,6 @@ std::string RocksDBOptionsParser::TrimAndRemoveComment(const std::string& line,
   return "";
 }
 
-namespace {
-bool AreEqualDoubles(const double a, const double b) {
-  return (fabs(a - b) < 0.00001);
-}
-}  // namespace
-
-bool AreEqualOptions(
-    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) {
-    case OptionType::kBoolean:
-      return (*reinterpret_cast<const bool*>(offset1) ==
-              *reinterpret_cast<const bool*>(offset2));
-    case OptionType::kInt:
-      return (*reinterpret_cast<const int*>(offset1) ==
-              *reinterpret_cast<const int*>(offset2));
-    case OptionType::kInt32T:
-      return (*reinterpret_cast<const int32_t*>(offset1) ==
-              *reinterpret_cast<const int32_t*>(offset2));
-    case OptionType::kInt64T:
-      {
-        int64_t v1, v2;
-        GetUnaligned(reinterpret_cast<const int64_t*>(offset1), &v1);
-        GetUnaligned(reinterpret_cast<const int64_t*>(offset2), &v2);
-        return (v1 == v2);
-      }
-    case OptionType::kVectorInt:
-      return (*reinterpret_cast<const std::vector<int>*>(offset1) ==
-              *reinterpret_cast<const std::vector<int>*>(offset2));
-    case OptionType::kUInt:
-      return (*reinterpret_cast<const unsigned int*>(offset1) ==
-              *reinterpret_cast<const unsigned int*>(offset2));
-    case OptionType::kUInt32T:
-      return (*reinterpret_cast<const uint32_t*>(offset1) ==
-              *reinterpret_cast<const uint32_t*>(offset2));
-    case OptionType::kUInt64T:
-      {
-        uint64_t v1, v2;
-        GetUnaligned(reinterpret_cast<const uint64_t*>(offset1), &v1);
-        GetUnaligned(reinterpret_cast<const uint64_t*>(offset2), &v2);
-        return (v1 == v2);
-      }
-    case OptionType::kSizeT:
-      {
-        size_t v1, v2;
-        GetUnaligned(reinterpret_cast<const size_t*>(offset1), &v1);
-        GetUnaligned(reinterpret_cast<const size_t*>(offset2), &v2);
-        return (v1 == v2);
-      }
-    case OptionType::kString:
-      return (*reinterpret_cast<const std::string*>(offset1) ==
-              *reinterpret_cast<const std::string*>(offset2));
-    case OptionType::kDouble:
-      return AreEqualDoubles(*reinterpret_cast<const double*>(offset1),
-                             *reinterpret_cast<const double*>(offset2));
-    case OptionType::kCompactionStyle:
-      return (*reinterpret_cast<const CompactionStyle*>(offset1) ==
-              *reinterpret_cast<const CompactionStyle*>(offset2));
-    case OptionType::kCompactionPri:
-      return (*reinterpret_cast<const CompactionPri*>(offset1) ==
-              *reinterpret_cast<const CompactionPri*>(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);
-    }
-    case OptionType::kChecksumType:
-      return (*reinterpret_cast<const ChecksumType*>(offset1) ==
-              *reinterpret_cast<const ChecksumType*>(offset2));
-    case OptionType::kBlockBasedTableIndexType:
-      return (
-          *reinterpret_cast<const BlockBasedTableOptions::IndexType*>(
-              offset1) ==
-          *reinterpret_cast<const BlockBasedTableOptions::IndexType*>(offset2));
-    case OptionType::kBlockBasedTableDataBlockIndexType:
-      return (
-          *reinterpret_cast<const BlockBasedTableOptions::DataBlockIndexType*>(
-              offset1) ==
-          *reinterpret_cast<const BlockBasedTableOptions::DataBlockIndexType*>(
-              offset2));
-    case OptionType::kBlockBasedTableIndexShorteningMode:
-      return (
-          *reinterpret_cast<const BlockBasedTableOptions::IndexShorteningMode*>(
-              offset1) ==
-          *reinterpret_cast<const BlockBasedTableOptions::IndexShorteningMode*>(
-              offset2));
-    case OptionType::kWALRecoveryMode:
-      return (*reinterpret_cast<const WALRecoveryMode*>(offset1) ==
-              *reinterpret_cast<const WALRecoveryMode*>(offset2));
-    case OptionType::kAccessHint:
-      return (*reinterpret_cast<const DBOptions::AccessHint*>(offset1) ==
-              *reinterpret_cast<const DBOptions::AccessHint*>(offset2));
-    case OptionType::kInfoLogLevel:
-      return (*reinterpret_cast<const InfoLogLevel*>(offset1) ==
-              *reinterpret_cast<const InfoLogLevel*>(offset2));
-    case OptionType::kCompactionOptionsFIFO: {
-      CompactionOptionsFIFO lhs =
-          *reinterpret_cast<const CompactionOptionsFIFO*>(offset1);
-      CompactionOptionsFIFO rhs =
-          *reinterpret_cast<const CompactionOptionsFIFO*>(offset2);
-      if (lhs.max_table_files_size == rhs.max_table_files_size &&
-          lhs.allow_compaction == rhs.allow_compaction) {
-        return true;
-      }
-      return false;
-    }
-    case OptionType::kCompactionOptionsUniversal: {
-      CompactionOptionsUniversal lhs =
-          *reinterpret_cast<const CompactionOptionsUniversal*>(offset1);
-      CompactionOptionsUniversal rhs =
-          *reinterpret_cast<const CompactionOptionsUniversal*>(offset2);
-      if (lhs.size_ratio == rhs.size_ratio &&
-          lhs.min_merge_width == rhs.min_merge_width &&
-          lhs.max_merge_width == rhs.max_merge_width &&
-          lhs.max_size_amplification_percent ==
-              rhs.max_size_amplification_percent &&
-          lhs.compression_size_percent == rhs.compression_size_percent &&
-          lhs.stop_style == rhs.stop_style &&
-          lhs.allow_trivial_move == rhs.allow_trivial_move) {
-        return true;
-      }
-      return false;
-    }
-    default:
-      if (type_info.IsByName()) {
-        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 {
-          if (type_info.IsEnabled(OptionVerificationType::kByNameAllowNull)) {
-            if (iter->second == kNullptrString || value1 == kNullptrString) {
-              return true;
-            }
-          } else if (type_info.IsEnabled(
-                         OptionVerificationType::kByNameAllowFromNull)) {
-            if (iter->second == kNullptrString) {
-              return true;
-            }
-          }
-          return (value1 == iter->second);
-        }
-      }
-      return false;
-  }
-}
-
 Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
     const ConfigOptions& config_options, const DBOptions& db_opt,
     const std::vector<std::string>& cf_names,
@@ -756,34 +591,33 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
 
 Status RocksDBOptionsParser::VerifyDBOptions(
     const ConfigOptions& config_options, const DBOptions& base_opt,
-    const DBOptions& persisted_opt,
+    const DBOptions& file_opt,
     const std::unordered_map<std::string, std::string>* /*opt_map*/) {
   for (const auto& pair : db_options_type_info) {
-    if (pair.second.IsDeprecated()) {
+    const auto& opt_info = pair.second;
-      // We skip checking deprecated variables as they might
+    if (config_options.IsCheckEnabled(opt_info.GetSanityLevel())) {
-      // contain random values since they might not be initialized
+      const char* base_addr =
-      continue;
+          reinterpret_cast<const char*>(&base_opt) + opt_info.offset;
-    }
+      const char* file_addr =
-    if (DBOptionSanityCheckLevel(pair.first) <= config_options.sanity_level) {
+          reinterpret_cast<const char*>(&file_opt) + opt_info.offset;
-      if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
+      std::string mismatch;
-                           reinterpret_cast<const char*>(&persisted_opt),
+      if (!opt_info.MatchesOption(config_options, pair.first, base_addr,
-                           pair.second, pair.first, nullptr)) {
+                                  file_addr, &mismatch) &&
-        constexpr size_t kBufferSize = 2048;
+          !opt_info.MatchesByName(config_options, pair.first, base_addr,
+                                  file_addr)) {
+        const size_t kBufferSize = 2048;
         char buffer[kBufferSize];
         std::string base_value;
-        std::string persisted_value;
+        std::string file_value;
-        SerializeSingleOptionHelper(
+        opt_info.SerializeOption(config_options, pair.first, base_addr,
-            reinterpret_cast<const char*>(&base_opt) + pair.second.offset,
+                                 &base_value);
-            pair.second.type, &base_value);
+        opt_info.SerializeOption(config_options, pair.first, file_addr,
-        SerializeSingleOptionHelper(
+                                 &file_value);
-            reinterpret_cast<const char*>(&persisted_opt) + pair.second.offset,
-            pair.second.type, &persisted_value);
         snprintf(buffer, sizeof(buffer),
                  "[RocksDBOptionsParser]: "
                  "failed the verification on DBOptions::%s --- "
                  "The specified one is %s while the persisted one is %s.\n",
-                 pair.first.c_str(), base_value.c_str(),
+                 pair.first.c_str(), base_value.c_str(), file_value.c_str());
-                 persisted_value.c_str());
         return Status::InvalidArgument(Slice(buffer, strlen(buffer)));
       }
     }
@@ -793,38 +627,51 @@ Status RocksDBOptionsParser::VerifyDBOptions(
 
 Status RocksDBOptionsParser::VerifyCFOptions(
     const ConfigOptions& config_options, const ColumnFamilyOptions& base_opt,
-    const ColumnFamilyOptions& persisted_opt,
+    const ColumnFamilyOptions& file_opt,
-    const std::unordered_map<std::string, std::string>* persisted_opt_map) {
+    const std::unordered_map<std::string, std::string>* opt_map) {
   for (const auto& pair : cf_options_type_info) {
-    if (pair.second.IsDeprecated()) {
+    const auto& opt_info = pair.second;
-      // We skip checking deprecated variables as they might
+
-      // contain random values since they might not be initialized
+    if (config_options.IsCheckEnabled(opt_info.GetSanityLevel())) {
-      continue;
+      std::string mismatch;
+      const char* base_addr =
+          reinterpret_cast<const char*>(&base_opt) + opt_info.offset;
+      const char* file_addr =
+          reinterpret_cast<const char*>(&file_opt) + opt_info.offset;
+      bool matches = opt_info.MatchesOption(config_options, pair.first,
+                                            base_addr, file_addr, &mismatch);
+      if (!matches && opt_info.IsByName()) {
+        if (opt_map == nullptr) {
+          matches = true;
+        } else {
+          auto iter = opt_map->find(pair.first);
+          if (iter == opt_map->end()) {
+            matches = true;
+          } else {
+            matches = opt_info.MatchesByName(config_options, pair.first,
+                                             base_addr, iter->second);
           }
-    if (CFOptionSanityCheckLevel(pair.first) <= config_options.sanity_level) {
+        }
-      if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
+      }
-                           reinterpret_cast<const char*>(&persisted_opt),
+      if (!matches) {
-                           pair.second, pair.first, persisted_opt_map)) {
+        // The options do not match
-        constexpr size_t kBufferSize = 2048;
+        const size_t kBufferSize = 2048;
         char buffer[kBufferSize];
         std::string base_value;
-        std::string persisted_value;
+        std::string file_value;
-        SerializeSingleOptionHelper(
+        opt_info.SerializeOption(config_options, pair.first, base_addr,
-            reinterpret_cast<const char*>(&base_opt) + pair.second.offset,
+                                 &base_value);
-            pair.second.type, &base_value);
+        opt_info.SerializeOption(config_options, pair.first, file_addr,
-        SerializeSingleOptionHelper(
+                                 &file_value);
-            reinterpret_cast<const char*>(&persisted_opt) + pair.second.offset,
-            pair.second.type, &persisted_value);
         snprintf(buffer, sizeof(buffer),
                  "[RocksDBOptionsParser]: "
                  "failed the verification on ColumnFamilyOptions::%s --- "
                  "The specified one is %s while the persisted one is %s.\n",
-                 pair.first.c_str(), base_value.c_str(),
+                 pair.first.c_str(), base_value.c_str(), file_value.c_str());
-                 persisted_value.c_str());
         return Status::InvalidArgument(Slice(buffer, sizeof(buffer)));
-      }
+      }  // if (! matches)
-    }
+    }    // CheckSanityLevel
-  }
+  }      // For each option
   return Status::OK();
 }
 

options/options_parser.h

@@ -43,11 +43,6 @@ Status PersistRocksDBOptions(const ConfigOptions& config_options,
                              const std::vector<ColumnFamilyOptions>& cf_opts,
                              const std::string& file_name, FileSystem* fs);
 
-extern bool AreEqualOptions(
-    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);
-
 class RocksDBOptionsParser {
  public:
   explicit RocksDBOptionsParser();

options/options_sanity_check.cc

@@ -1,39 +0,0 @@
-// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#ifndef ROCKSDB_LITE
-
-#include "options/options_sanity_check.h"
-
-namespace ROCKSDB_NAMESPACE {
-
-namespace {
-ConfigOptions::SanityLevel SanityCheckLevelHelper(
-    const std::unordered_map<std::string, ConfigOptions::SanityLevel>& smap,
-    const std::string& name) {
-  auto iter = smap.find(name);
-  return iter != smap.end() ? iter->second
-                            : ConfigOptions::kSanityLevelExactMatch;
-}
-}
-
-ConfigOptions::SanityLevel DBOptionSanityCheckLevel(
-    const std::string& option_name) {
-  return SanityCheckLevelHelper(sanity_level_db_options, option_name);
-}
-
-ConfigOptions::SanityLevel CFOptionSanityCheckLevel(
-    const std::string& option_name) {
-  return SanityCheckLevelHelper(sanity_level_cf_options, option_name);
-}
-
-ConfigOptions::SanityLevel BBTOptionSanityCheckLevel(
-    const std::string& option_name) {
-  return SanityCheckLevelHelper(sanity_level_bbt_options, option_name);
-}
-
-}  // namespace ROCKSDB_NAMESPACE
-
-#endif  // !ROCKSDB_LITE

options/options_sanity_check.h

@@ -1,45 +0,0 @@
-// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#pragma once
-
-#include <string>
-#include <unordered_map>
-
-#include "rocksdb/convenience.h"
-#include "rocksdb/rocksdb_namespace.h"
-
-#ifndef ROCKSDB_LITE
-namespace ROCKSDB_NAMESPACE {
-// This enum defines the RocksDB options sanity level.
-
-// The sanity check level for DB options
-static const std::unordered_map<std::string, ConfigOptions::SanityLevel>
-    sanity_level_db_options{};
-
-// The sanity check level for column-family options
-static const std::unordered_map<std::string, ConfigOptions::SanityLevel>
-    sanity_level_cf_options = {
-        {"comparator",
-         ConfigOptions::SanityLevel::kSanityLevelLooselyCompatible},
-        {"table_factory",
-         ConfigOptions::SanityLevel::kSanityLevelLooselyCompatible},
-        {"merge_operator",
-         ConfigOptions::SanityLevel::kSanityLevelLooselyCompatible}};
-
-// The sanity check level for block-based table options
-static const std::unordered_map<std::string, ConfigOptions::SanityLevel>
-    sanity_level_bbt_options{};
-
-ConfigOptions::SanityLevel DBOptionSanityCheckLevel(
-    const std::string& options_name);
-ConfigOptions::SanityLevel CFOptionSanityCheckLevel(
-    const std::string& options_name);
-ConfigOptions::SanityLevel BBTOptionSanityCheckLevel(
-    const std::string& options_name);
-
-}  // namespace ROCKSDB_NAMESPACE
-
-#endif  // !ROCKSDB_LITE

options/options_test.cc

@@ -16,7 +16,6 @@
 #include "cache/sharded_cache.h"
 #include "options/options_helper.h"
 #include "options/options_parser.h"
-#include "options/options_sanity_check.h"
 #include "port/port.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/convenience.h"
@@ -2986,6 +2985,281 @@ TEST_F(OptionsParserTest, EscapeOptionString) {
                 "Escape \\# and # comment together   ."),
             "Escape \\# and");
 }
+
+static void TestAndCompareOption(const ConfigOptions& config_options,
+                                 const OptionTypeInfo& opt_info,
+                                 const std::string& opt_name, void* base_ptr,
+                                 void* comp_ptr) {
+  std::string result, mismatch;
+  char* base_addr = reinterpret_cast<char*>(base_ptr) + opt_info.offset;
+  char* comp_addr = reinterpret_cast<char*>(comp_ptr) + opt_info.offset;
+  ASSERT_OK(
+      opt_info.SerializeOption(config_options, opt_name, base_addr, &result));
+  ASSERT_OK(opt_info.ParseOption(config_options, opt_name, result, comp_addr));
+  ASSERT_TRUE(opt_info.MatchesOption(config_options, opt_name, base_addr,
+                                     comp_addr, &mismatch));
+}
+
+template <typename T>
+void TestOptInfo(const ConfigOptions& config_options, OptionType opt_type,
+                 T* base, T* comp) {
+  std::string result;
+  OptionTypeInfo opt_info(0, opt_type);
+  char* base_addr = reinterpret_cast<char*>(base);
+  char* comp_addr = reinterpret_cast<char*>(comp);
+  ASSERT_FALSE(opt_info.MatchesOption(config_options, "base", base_addr,
+                                      comp_addr, &result));
+  ASSERT_EQ(result, "base");
+  ASSERT_NE(*base, *comp);
+  TestAndCompareOption(config_options, opt_info, "base", base_addr, comp_addr);
+  ASSERT_EQ(*base, *comp);
+}
+
+class OptionTypeInfoTest : public testing::Test {};
+
+TEST_F(OptionTypeInfoTest, BasicTypes) {
+  ConfigOptions config_options;
+  {
+    bool a = true, b = false;
+    TestOptInfo(config_options, OptionType::kBoolean, &a, &b);
+  }
+  {
+    int a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kInt, &a, &b);
+  }
+  {
+    int32_t a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kInt32T, &a, &b);
+  }
+  {
+    int64_t a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kInt64T, &a, &b);
+  }
+  {
+    unsigned int a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kUInt, &a, &b);
+  }
+  {
+    uint32_t a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kUInt32T, &a, &b);
+  }
+  {
+    uint64_t a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kUInt64T, &a, &b);
+  }
+  {
+    size_t a = 100, b = 200;
+    TestOptInfo(config_options, OptionType::kSizeT, &a, &b);
+  }
+  {
+    std::string a = "100", b = "200";
+    TestOptInfo(config_options, OptionType::kString, &a, &b);
+  }
+  {
+    double a = 1.0, b = 2.0;
+    TestOptInfo(config_options, OptionType::kDouble, &a, &b);
+  }
+}
+
+TEST_F(OptionTypeInfoTest, TestInvalidArgs) {
+  ConfigOptions config_options;
+  bool b;
+  int i;
+  int32_t i32;
+  int64_t i64;
+  unsigned int u;
+  int32_t u32;
+  int64_t u64;
+  size_t sz;
+  double d;
+
+  ASSERT_NOK(
+      OptionTypeInfo(0, OptionType::kBoolean)
+          .ParseOption(config_options, "b", "x", reinterpret_cast<char*>(&b)));
+  ASSERT_NOK(
+      OptionTypeInfo(0, OptionType::kInt)
+          .ParseOption(config_options, "b", "x", reinterpret_cast<char*>(&i)));
+  ASSERT_NOK(OptionTypeInfo(0, OptionType::kInt32T)
+                 .ParseOption(config_options, "b", "x",
+                              reinterpret_cast<char*>(&i32)));
+  ASSERT_NOK(OptionTypeInfo(0, OptionType::kInt64T)
+                 .ParseOption(config_options, "b", "x",
+                              reinterpret_cast<char*>(&i64)));
+  ASSERT_NOK(
+      OptionTypeInfo(0, OptionType::kUInt)
+          .ParseOption(config_options, "b", "x", reinterpret_cast<char*>(&u)));
+  ASSERT_NOK(OptionTypeInfo(0, OptionType::kUInt32T)
+                 .ParseOption(config_options, "b", "x",
+                              reinterpret_cast<char*>(&u32)));
+  ASSERT_NOK(OptionTypeInfo(0, OptionType::kUInt64T)
+                 .ParseOption(config_options, "b", "x",
+                              reinterpret_cast<char*>(&u64)));
+  ASSERT_NOK(
+      OptionTypeInfo(0, OptionType::kSizeT)
+          .ParseOption(config_options, "b", "x", reinterpret_cast<char*>(&sz)));
+  ASSERT_NOK(
+      OptionTypeInfo(0, OptionType::kDouble)
+          .ParseOption(config_options, "b", "x", reinterpret_cast<char*>(&d)));
+
+  // Don't know how to convert Unknowns to anything else
+  ASSERT_NOK(
+      OptionTypeInfo(0, OptionType::kUnknown)
+          .ParseOption(config_options, "b", "x", reinterpret_cast<char*>(&d)));
+
+  // Verify that if the parse function throws an exception, it is also trapped
+  OptionTypeInfo func_info(0, OptionType::kUnknown,
+                           OptionVerificationType::kNormal,
+                           OptionTypeFlags::kNone, 0,
+                           [](const ConfigOptions&, const std::string&,
+                              const std::string& value, char* addr) {
+                             auto ptr = reinterpret_cast<int*>(addr);
+                             *ptr = ParseInt(value);
+                             return Status::OK();
+                           });
+  ASSERT_OK(func_info.ParseOption(config_options, "b", "1",
+                                  reinterpret_cast<char*>(&i)));
+  ASSERT_NOK(func_info.ParseOption(config_options, "b", "x",
+                                   reinterpret_cast<char*>(&i)));
+}
+
+TEST_F(OptionTypeInfoTest, TestParseFunc) {
+  OptionTypeInfo opt_info(
+      0, OptionType::kUnknown, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone, 0,
+      [](const ConfigOptions& /*opts*/, const std::string& name,
+         const std::string& value, char* addr) {
+        auto ptr = reinterpret_cast<std::string*>(addr);
+        if (name == "Oops") {
+          return Status::InvalidArgument(value);
+        } else {
+          *ptr = value + " " + name;
+          return Status::OK();
+        }
+      });
+  ConfigOptions config_options;
+  std::string base;
+  ASSERT_OK(opt_info.ParseOption(config_options, "World", "Hello",
+                                 reinterpret_cast<char*>(&base)));
+  ASSERT_EQ(base, "Hello World");
+  ASSERT_NOK(opt_info.ParseOption(config_options, "Oops", "Hello",
+                                  reinterpret_cast<char*>(&base)));
+}
+
+TEST_F(OptionTypeInfoTest, TestSerializeFunc) {
+  OptionTypeInfo opt_info(
+      0, OptionType::kString, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone, 0, nullptr,
+      [](const ConfigOptions& /*opts*/, const std::string& name,
+         const char* /*addr*/, std::string* value) {
+        if (name == "Oops") {
+          return Status::InvalidArgument(name);
+        } else {
+          *value = name;
+          return Status::OK();
+        }
+      },
+      nullptr);
+  ConfigOptions config_options;
+  std::string base;
+  std::string value;
+  ASSERT_OK(opt_info.SerializeOption(config_options, "Hello",
+                                     reinterpret_cast<char*>(&base), &value));
+  ASSERT_EQ(value, "Hello");
+  ASSERT_NOK(opt_info.SerializeOption(config_options, "Oops",
+                                      reinterpret_cast<char*>(&base), &value));
+}
+
+TEST_F(OptionTypeInfoTest, TestEqualsFunc) {
+  OptionTypeInfo opt_info(
+      0, OptionType::kInt, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone, 0, nullptr, nullptr,
+      [](const ConfigOptions& /*opts*/, const std::string& name,
+         const char* addr1, const char* addr2, std::string* mismatch) {
+        auto i1 = *(reinterpret_cast<const int*>(addr1));
+        auto i2 = *(reinterpret_cast<const int*>(addr2));
+        if (name == "LT") {
+          return i1 < i2;
+        } else if (name == "GT") {
+          return i1 > i2;
+        } else if (name == "EQ") {
+          return i1 == i2;
+        } else {
+          *mismatch = name + "???";
+          return false;
+        }
+      });
+
+  ConfigOptions config_options;
+  int int1 = 100;
+  int int2 = 200;
+  std::string mismatch;
+  ASSERT_TRUE(opt_info.MatchesOption(
+      config_options, "LT", reinterpret_cast<const char*>(&int1),
+      reinterpret_cast<const char*>(&int2), &mismatch));
+  ASSERT_EQ(mismatch, "");
+  ASSERT_FALSE(opt_info.MatchesOption(
+      config_options, "GT", reinterpret_cast<char*>(&int1),
+      reinterpret_cast<char*>(&int2), &mismatch));
+  ASSERT_EQ(mismatch, "GT");
+  ASSERT_FALSE(opt_info.MatchesOption(
+      config_options, "NO", reinterpret_cast<char*>(&int1),
+      reinterpret_cast<char*>(&int2), &mismatch));
+  ASSERT_EQ(mismatch, "NO???");
+}
+
+TEST_F(OptionTypeInfoTest, TestOptionFlags) {
+  OptionTypeInfo opt_none(0, OptionType::kString,
+                          OptionVerificationType::kNormal,
+                          OptionTypeFlags::kDontSerialize, 0);
+  OptionTypeInfo opt_never(0, OptionType::kString,
+                           OptionVerificationType::kNormal,
+                           OptionTypeFlags::kCompareNever, 0);
+  OptionTypeInfo opt_alias(0, OptionType::kString,
+                           OptionVerificationType::kAlias,
+                           OptionTypeFlags::kNone, 0);
+  OptionTypeInfo opt_deprecated(0, OptionType::kString,
+                                OptionVerificationType::kDeprecated,
+                                OptionTypeFlags::kNone, 0);
+  ConfigOptions config_options;
+  std::string base = "base";
+  std::string comp = "comp";
+
+  // If marked string none, the serialization returns okay but does nothing
+  ASSERT_OK(opt_none.SerializeOption(config_options, "None",
+                                     reinterpret_cast<char*>(&base), &base));
+  // If marked never compare, they match even when they do not
+  ASSERT_TRUE(opt_never.MatchesOption(config_options, "Never",
+                                      reinterpret_cast<char*>(&base),
+                                      reinterpret_cast<char*>(&comp), &base));
+  ASSERT_FALSE(opt_none.MatchesOption(config_options, "Never",
+                                      reinterpret_cast<char*>(&base),
+                                      reinterpret_cast<char*>(&comp), &base));
+
+  // An alias can change the value via parse, but does nothing on serialize on
+  // match
+  std::string result;
+  ASSERT_OK(opt_alias.ParseOption(config_options, "Alias", "Alias",
+                                  reinterpret_cast<char*>(&base)));
+  ASSERT_OK(opt_alias.SerializeOption(config_options, "Alias",
+                                      reinterpret_cast<char*>(&base), &result));
+  ASSERT_TRUE(opt_alias.MatchesOption(config_options, "Alias",
+                                      reinterpret_cast<char*>(&base),
+                                      reinterpret_cast<char*>(&comp), &result));
+  ASSERT_EQ(base, "Alias");
+  ASSERT_NE(base, comp);
+
+  // Deprecated options do nothing on any of the commands
+  ASSERT_OK(opt_deprecated.ParseOption(config_options, "Alias", "Deprecated",
+                                       reinterpret_cast<char*>(&base)));
+  ASSERT_OK(opt_deprecated.SerializeOption(
+      config_options, "Alias", reinterpret_cast<char*>(&base), &result));
+  ASSERT_TRUE(opt_deprecated.MatchesOption(
+      config_options, "Alias", reinterpret_cast<char*>(&base),
+      reinterpret_cast<char*>(&comp), &result));
+  ASSERT_EQ(base, "Alias");
+  ASSERT_NE(base, comp);
+}
+
 #endif  // !ROCKSDB_LITE
 }  // namespace ROCKSDB_NAMESPACE
 

options/options_type.h

@@ -5,7 +5,13 @@
 
 #pragma once
 
+#include <functional>
+#include <memory>
+#include <unordered_map>
+
+#include "rocksdb/convenience.h"
 #include "rocksdb/rocksdb_namespace.h"
+#include "rocksdb/status.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -58,16 +64,25 @@ enum class OptionVerificationType {
                          // 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
+  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.
+  kAlias,                // This option represents is a name/shortcut for
+                         // another option and should not be written or verified
+                         // independently
 };
 
 enum class OptionTypeFlags : uint32_t {
   kNone = 0x00,  // No flags
-  kMutable = 0x01,  // Option is mutable
+  kCompareDefault = 0x0,
+  kCompareNever = ConfigOptions::kSanityLevelNone,
+  kCompareLoose = ConfigOptions::kSanityLevelLooselyCompatible,
+  kCompareExact = ConfigOptions::kSanityLevelExactMatch,
+
+  kMutable = 0x0100,        // Option is mutable
+  kDontSerialize = 0x2000,  // Don't serialize the option
 };
 
 inline OptionTypeFlags operator|(const OptionTypeFlags &a,
@@ -82,18 +97,55 @@ inline OptionTypeFlags operator&(const OptionTypeFlags &a,
                                       static_cast<uint32_t>(b));
 }
 
+// Function for converting a option string value into its underlying
+// representation in "addr"
+// On success, Status::OK is returned and addr is set to the parsed form
+// On failure, a non-OK status is returned
+// @param opts  The ConfigOptions controlling how the value is parsed
+// @param name  The name of the options being parsed
+// @param value The string representation of the option
+// @param addr  Pointer to the object
+using ParserFunc = std::function<Status(
+    const ConfigOptions& /*opts*/, const std::string& /*name*/,
+    const std::string& /*value*/, char* /*addr*/)>;
+
+// Function for converting an option "addr" into its string representation.
+// On success, Status::OK is returned and value is the serialized form.
+// On failure, a non-OK status is returned
+// @param opts  The ConfigOptions controlling how the values are serialized
+// @param name  The name of the options being serialized
+// @param addr  Pointer to the value being serialized
+// @param value The result of the serialization.
+using StringFunc = std::function<Status(
+    const ConfigOptions& /*opts*/, const std::string& /*name*/,
+    const char* /*addr*/, std::string* /*value*/)>;
+
+// Function for comparing two option values
+// If they are not equal, updates "mismatch" with the name of the bad option
+// @param opts  The ConfigOptions controlling how the values are compared
+// @param name  The name of the options being compared
+// @param addr1 The first address to compare
+// @param addr2 The address to compare to
+// @param mismatch If the values are not equal, the name of the option that
+// first differs
+using EqualsFunc = std::function<bool(
+    const ConfigOptions& /*opts*/, const std::string& /*name*/,
+    const char* /*addr1*/, const char* /*addr2*/, std::string* mismatch)>;
+
 // A struct for storing constant option information such as option name,
 // option type, and offset.
 class OptionTypeInfo {
  public:
   int offset;
   int mutable_offset;
-  OptionType type;
 
   // A simple "normal", non-mutable Type "_type" at _offset
   OptionTypeInfo(int _offset, OptionType _type)
       : offset(_offset),
         mutable_offset(0),
+        parser_func(nullptr),
+        string_func(nullptr),
+        equals_func(nullptr),
         type(_type),
         verification(OptionVerificationType::kNormal),
         flags(OptionTypeFlags::kNone) {}
@@ -102,6 +154,9 @@ class OptionTypeInfo {
   OptionTypeInfo(int _offset, OptionType _type, int _mutable_offset)
       : offset(_offset),
         mutable_offset(_mutable_offset),
+        parser_func(nullptr),
+        string_func(nullptr),
+        equals_func(nullptr),
         type(_type),
         verification(OptionVerificationType::kNormal),
         flags(OptionTypeFlags::kMutable) {}
@@ -111,6 +166,34 @@ class OptionTypeInfo {
                  int _mutable_offset)
       : offset(_offset),
         mutable_offset(_mutable_offset),
+        parser_func(nullptr),
+        string_func(nullptr),
+        equals_func(nullptr),
+        type(_type),
+        verification(_verification),
+        flags(_flags) {}
+
+  OptionTypeInfo(int _offset, OptionType _type,
+                 OptionVerificationType _verification, OptionTypeFlags _flags,
+                 int _mutable_offset, const ParserFunc& _pfunc)
+      : offset(_offset),
+        mutable_offset(_mutable_offset),
+        parser_func(_pfunc),
+        string_func(nullptr),
+        equals_func(nullptr),
+        type(_type),
+        verification(_verification),
+        flags(_flags) {}
+
+  OptionTypeInfo(int _offset, OptionType _type,
+                 OptionVerificationType _verification, OptionTypeFlags _flags,
+                 int _mutable_offset, const ParserFunc& _pfunc,
+                 const StringFunc& _sfunc, const EqualsFunc& _efunc)
+      : offset(_offset),
+        mutable_offset(_mutable_offset),
+        parser_func(_pfunc),
+        string_func(_sfunc),
+        equals_func(_efunc),
         type(_type),
         verification(_verification),
         flags(_flags) {}
@@ -123,18 +206,93 @@ class OptionTypeInfo {
     return IsEnabled(OptionVerificationType::kDeprecated);
   }
 
+  // Returns true if the option is marked as an Alias.
+  // Aliases are valid options that are parsed but are not converted to strings
+  // or compared.
+  bool IsAlias() const { return IsEnabled(OptionVerificationType::kAlias); }
+
   bool IsEnabled(OptionVerificationType ovf) const {
     return verification == ovf;
   }
 
+  // Returns the sanity level for comparing the option.
+  // If the options should not be compared, returns None
+  // If the option has a compare flag, returns it.
+  // Otherwise, returns "exact"
+  ConfigOptions::SanityLevel GetSanityLevel() const {
+    if (IsDeprecated() || IsAlias()) {
+      return ConfigOptions::SanityLevel::kSanityLevelNone;
+    } else {
+      auto match = (flags & OptionTypeFlags::kCompareExact);
+      if (match == OptionTypeFlags::kCompareDefault) {
+        return ConfigOptions::SanityLevel::kSanityLevelExactMatch;
+      } else {
+        return (ConfigOptions::SanityLevel)match;
+      }
+    }
+  }
+
+  // Returns true if the option should be serialized.
+  // Options should be serialized if the are not deprecated, aliases,
+  // or marked as "Don't Serialize".
+  bool ShouldSerialize() const {
+    if (IsDeprecated() || IsAlias()) {
+      return false;
+    } else if (IsEnabled(OptionTypeFlags::kDontSerialize)) {
+      return false;
+    } else {
+      return true;
+    }
+  }
+
   bool IsByName() const {
     return (verification == OptionVerificationType::kByName ||
             verification == OptionVerificationType::kByNameAllowNull ||
             verification == OptionVerificationType::kByNameAllowFromNull);
   }
 
- protected:
+  // Parses the option in "opt_value" according to the rules of this class
+  // and updates the value at "opt_addr".
+  // On success, Status::OK() is returned.  On failure:
+  // NotFound means the opt_name is not valid for this option
+  // NotSupported means we do not know how to parse the value for this option
+  // InvalidArgument means the opt_value is not valid for this option.
+  Status ParseOption(const ConfigOptions& config_options,
+                     const std::string& opt_name, const std::string& opt_value,
+                     char* opt_addr) const;
+
+  // Serializes the option in "opt_addr" according to the rules of this class
+  // into the value at "opt_value".
+  Status SerializeOption(const ConfigOptions& config_options,
+                         const std::string& opt_name, const char* opt_addr,
+                         std::string* opt_value) const;
+
+  // Compares the "addr1" and "addr2" values according to the rules of this
+  // class and returns true if they match.  On a failed match, mismatch is the
+  // name of the option that failed to match.
+  bool MatchesOption(const ConfigOptions& config_options,
+                     const std::string& opt_name, const char* addr1,
+                     const char* addr2, std::string* mismatch) const;
+
+  // Used to override the match rules for "ByName" options.
+  bool MatchesByName(const ConfigOptions& config_options,
+                     const std::string& opt_name, const char* this_offset,
+                     const char* that_offset) const;
+  bool MatchesByName(const ConfigOptions& config_options,
+                     const std::string& opt_name, const char* this_ptr,
+                     const std::string& that_value) const;
+
  private:
+  // The optional function to convert a string to its representation
+  ParserFunc parser_func;
+
+  // The optional function to convert a value to its string representation
+  StringFunc string_func;
+
+  // The optional function to convert a match to option values
+  EqualsFunc equals_func;
+
+  OptionType type;
   OptionVerificationType verification;
   OptionTypeFlags flags;
 };

src.mk

@@ -1,5 +1,6 @@
 # These are the sources from which librocksdb.a is built:
 LIB_SOURCES =                                                   \
+  cache/cache.cc                                                \
   cache/clock_cache.cc                                          \
   cache/lru_cache.cc                                            \
   cache/sharded_cache.cc                                        \
@@ -118,7 +119,6 @@ LIB_SOURCES =                                                   \
   options/options.cc                                            \
   options/options_helper.cc                                     \
   options/options_parser.cc                                     \
-  options/options_sanity_check.cc                               \
   port/port_posix.cc                                            \
   port/stack_trace.cc                                           \
   table/adaptive/adaptive_table_factory.cc                      \

table/block_based/block_based_table_factory.cc

@@ -17,7 +17,6 @@
 
 #include "options/options_helper.h"
 #include "options/options_parser.h"
-#include "options/options_sanity_check.h"
 #include "port/port.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/convenience.h"
@@ -170,7 +169,7 @@ static std::unordered_map<std::string, OptionTypeInfo>
         {"flush_block_policy_factory",
          {offsetof(struct BlockBasedTableOptions, flush_block_policy_factory),
           OptionType::kFlushBlockPolicyFactory, OptionVerificationType::kByName,
-          OptionTypeFlags::kNone, 0}},
+          OptionTypeFlags::kCompareNever, 0}},
         {"cache_index_and_filter_blocks",
          {offsetof(struct BlockBasedTableOptions,
                    cache_index_and_filter_blocks),
@@ -244,8 +243,46 @@ static std::unordered_map<std::string, OptionTypeInfo>
           OptionTypeFlags::kNone, 0}},
         {"filter_policy",
          {offsetof(struct BlockBasedTableOptions, filter_policy),
-          OptionType::kFilterPolicy, OptionVerificationType::kByName,
+          OptionType::kUnknown, OptionVerificationType::kByNameAllowFromNull,
-          OptionTypeFlags::kNone, 0}},
+          OptionTypeFlags::kNone, 0,
+          // Parses the Filter policy
+          [](const ConfigOptions& opts, const std::string&,
+             const std::string& value, char* addr) {
+            auto* policy =
+                reinterpret_cast<std::shared_ptr<const FilterPolicy>*>(addr);
+            return FilterPolicy::CreateFromString(opts, value, policy);
+          },
+          // Converts the FilterPolicy to its string representation
+          [](const ConfigOptions&, const std::string&, const char* addr,
+             std::string* value) {
+            const auto* policy =
+                reinterpret_cast<const std::shared_ptr<const FilterPolicy>*>(
+                    addr);
+            if (policy->get()) {
+              *value = (*policy)->Name();
+            } else {
+              *value = kNullptrString;
+            }
+            return Status::OK();
+          },
+          // Compares two FilterPolicy objects for equality
+          [](const ConfigOptions&, const std::string&, const char* addr1,
+             const char* addr2, std::string*) {
+            const auto* policy1 =
+                reinterpret_cast<const std::shared_ptr<const FilterPolicy>*>(
+                    addr1)
+                    ->get();
+            const auto* policy2 =
+                reinterpret_cast<const std::shared_ptr<FilterPolicy>*>(addr2)
+                    ->get();
+            if (policy1 == policy2) {
+              return true;
+            } else if (policy1 != nullptr && policy2 != nullptr) {
+              return (strcmp(policy1->Name(), policy2->Name()) == 0);
+            } else {
+              return false;
+            }
+          }}},
         {"whole_key_filtering",
          {offsetof(struct BlockBasedTableOptions, whole_key_filtering),
           OptionType::kBoolean, OptionVerificationType::kNormal,
@@ -277,7 +314,28 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct BlockBasedTableOptions,
                    pin_top_level_index_and_filter),
           OptionType::kBoolean, OptionVerificationType::kNormal,
-          OptionTypeFlags::kNone, 0}}};
+          OptionTypeFlags::kNone, 0}},
+        {"block_cache",
+         {offsetof(struct BlockBasedTableOptions, block_cache),
+          OptionType::kUnknown, OptionVerificationType::kNormal,
+          (OptionTypeFlags::kCompareNever | OptionTypeFlags::kDontSerialize), 0,
+          // Parses the input vsalue as a Cache
+          [](const ConfigOptions& opts, const std::string&,
+             const std::string& value, char* addr) {
+            auto* cache = reinterpret_cast<std::shared_ptr<Cache>*>(addr);
+            return Cache::CreateFromString(opts, value, cache);
+          }}},
+        {"block_cache_compressed",
+         {offsetof(struct BlockBasedTableOptions, block_cache_compressed),
+          OptionType::kUnknown, OptionVerificationType::kNormal,
+          (OptionTypeFlags::kCompareNever | OptionTypeFlags::kDontSerialize), 0,
+          // Parses the input vsalue as a Cache
+          [](const ConfigOptions& opts, const std::string&,
+             const std::string& value, char* addr) {
+            auto* cache = reinterpret_cast<std::shared_ptr<Cache>*>(addr);
+            return Cache::CreateFromString(opts, value, cache);
+          }}},
+};
 #endif  // ROCKSDB_LITE
 
 // TODO(myabandeh): We should return an error instead of silently changing the
@@ -566,55 +624,6 @@ std::string ParseBlockBasedTableOption(const ConfigOptions& config_options,
   const std::string& value = config_options.input_strings_escaped
                                  ? UnescapeOptionString(org_value)
                                  : org_value;
-  if (!config_options.input_strings_escaped) {
-    // if the input string is not escaped, it means this function is
-    // invoked from SetOptions, which takes the old format.
-    if (name == "block_cache" || name == "block_cache_compressed") {
-      // cache options can be specified in the following format
-      //   "block_cache={capacity=1M;num_shard_bits=4;
-      //    strict_capacity_limit=true;high_pri_pool_ratio=0.5;}"
-      // To support backward compatibility, the following format
-      // is also supported.
-      //   "block_cache=1M"
-      std::shared_ptr<Cache> cache;
-      // block_cache is specified in format block_cache=<cache_size>.
-      if (value.find('=') == std::string::npos) {
-        cache = NewLRUCache(ParseSizeT(value));
-      } else {
-        LRUCacheOptions cache_opts;
-        if (!ParseOptionHelper(reinterpret_cast<char*>(&cache_opts),
-                               OptionType::kLRUCacheOptions, value)) {
-          return "Invalid cache options";
-        }
-        cache = NewLRUCache(cache_opts);
-      }
-
-      if (name == "block_cache") {
-        new_options->block_cache = cache;
-      } else {
-        new_options->block_cache_compressed = cache;
-      }
-      return "";
-    } else if (name == "filter_policy") {
-      // Expect the following format
-      // bloomfilter:int:bool
-      const std::string kName = "bloomfilter:";
-      if (value.compare(0, kName.size(), kName) != 0) {
-        return "Invalid filter policy name";
-      }
-      size_t pos = value.find(':', kName.size());
-      if (pos == std::string::npos) {
-        return "Invalid filter policy config, missing bits_per_key";
-      }
-      double bits_per_key =
-          ParseDouble(trim(value.substr(kName.size(), pos - kName.size())));
-      bool use_block_based_builder =
-          ParseBoolean("use_block_based_builder", trim(value.substr(pos + 1)));
-      new_options->filter_policy.reset(
-          NewBloomFilterPolicy(bits_per_key, use_block_based_builder));
-      return "";
-    }
-  }
   const auto iter = block_based_table_type_info.find(name);
   if (iter == block_based_table_type_info.end()) {
     if (config_options.ignore_unknown_options) {
@@ -624,12 +633,14 @@ std::string ParseBlockBasedTableOption(const ConfigOptions& config_options,
     }
   }
   const auto& opt_info = iter->second;
-  if (!opt_info.IsDeprecated() &&
+  Status s = opt_info.ParseOption(
-      !ParseOptionHelper(reinterpret_cast<char*>(new_options) + opt_info.offset,
+      config_options, iter->first, value,
-                         opt_info.type, value)) {
+      reinterpret_cast<char*>(new_options) + opt_info.offset);
-    return "Invalid value";
+  if (s.ok()) {
-  }
     return "";
+  } else {
+    return s.ToString();
+  }
 }
 }  // namespace
 
@@ -712,16 +723,20 @@ Status VerifyBlockBasedTableFactory(const ConfigOptions& config_options,
   const auto& base_opt = base_tf->table_options();
   const auto& file_opt = file_tf->table_options();
 
+  std::string mismatch;
   for (auto& pair : block_based_table_type_info) {
-    if (pair.second.IsDeprecated()) {
     // We skip checking deprecated variables as they might
     // contain random values since they might not be initialized
-      continue;
+    if (config_options.IsCheckEnabled(pair.second.GetSanityLevel())) {
-    }
+      const char* base_addr =
-    if (BBTOptionSanityCheckLevel(pair.first) <= config_options.sanity_level) {
+          reinterpret_cast<const char*>(&base_opt) + pair.second.offset;
-      if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
+      const char* file_addr =
-                           reinterpret_cast<const char*>(&file_opt),
+          reinterpret_cast<const char*>(&file_opt) + pair.second.offset;
-                           pair.second, pair.first, nullptr)) {
+
+      if (!pair.second.MatchesOption(config_options, pair.first, base_addr,
+                                     file_addr, &mismatch) &&
+          !pair.second.MatchesByName(config_options, pair.first, base_addr,
+                                     file_addr)) {
         return Status::Corruption(
             "[RocksDBOptionsParser]: "
             "failed the verification on BlockBasedTableOptions::",

table/block_based/filter_policy.cc

@@ -756,4 +756,34 @@ FilterBuildingContext::FilterBuildingContext(
 
 FilterPolicy::~FilterPolicy() { }
 
+Status FilterPolicy::CreateFromString(
+    const ConfigOptions& /*options*/, const std::string& value,
+    std::shared_ptr<const FilterPolicy>* policy) {
+  const std::string kBloomName = "bloomfilter:";
+  if (value == kNullptrString || value == "rocksdb.BuiltinBloomFilter") {
+    policy->reset();
+#ifndef ROCKSDB_LITE
+  } else if (value.compare(0, kBloomName.size(), kBloomName) == 0) {
+    size_t pos = value.find(':', kBloomName.size());
+    if (pos == std::string::npos) {
+      return Status::InvalidArgument(
+          "Invalid filter policy config, missing bits_per_key");
+    } else {
+      double bits_per_key = ParseDouble(
+          trim(value.substr(kBloomName.size(), pos - kBloomName.size())));
+      bool use_block_based_builder =
+          ParseBoolean("use_block_based_builder", trim(value.substr(pos + 1)));
+      policy->reset(
+          NewBloomFilterPolicy(bits_per_key, use_block_based_builder));
+    }
+  } else {
+    return Status::InvalidArgument("Invalid filter policy name ", value);
+#else
+  } else {
+    return Status::NotSupported("Cannot load filter policy in LITE mode ",
+                                value);
+#endif  // ROCKSDB_LITE
+  }
+  return Status::OK();
+}
 }  // namespace ROCKSDB_NAMESPACE

table/plain/plain_table_factory.cc

@@ -220,12 +220,14 @@ std::string ParsePlainTableOptions(const ConfigOptions& config_options,
     }
   }
   const auto& opt_info = iter->second;
-  if (!opt_info.IsDeprecated() &&
+  Status s = opt_info.ParseOption(
-      !ParseOptionHelper(reinterpret_cast<char*>(new_options) + opt_info.offset,
+      config_options, name, value,
-                         opt_info.type, value)) {
+      reinterpret_cast<char*>(new_options) + opt_info.offset);
-    return "Invalid value";
+  if (s.ok()) {
-  }
     return "";
+  } else {
+    return s.ToString();
+  }
 }
 
 Status GetPlainTableOptionsFromMap(