Add a ConfigOptions for use in comparing objects and converting to/from strings (#6389)

Summary:
The methods in convenience.h are used to compare/convert objects to/from strings.  There is a mishmash of parameters in use here with more needed in the future.  This PR replaces those parameters with a single structure.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6389

Reviewed By: siying

Differential Revision: D21163707

Pulled By: zhichao-cao

fbshipit-source-id: f807b4cc7e2b0af3871536b69546b2604dfa81bd

HISTORY.md

@@ -5,6 +5,7 @@
 * Finish implementation of BlockBasedTableOptions::IndexType::kBinarySearchWithFirstKey. It's now ready for use. Significantly reduces read amplification in some setups, especially for iterator seeks.
 
 ### Public API Change
+* Add a ConfigOptions argument to the APIs dealing with converting options to and from strings and files.  The ConfigOptions is meant to replace some of the options (such as input_strings_escaped and ignore_unknown_options) and allow for more parameters to be passed in the future without changing the function signature.
 * Add NewFileChecksumGenCrc32cFactory to the file checksum public API, such that the builtin Crc32c based file checksum generator factory can be used by applications.
 * Add IsDirectory to Env and FS to indicate if a path is a directory.
 

db/column_family_test.cc

@@ -8,16 +8,16 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include <algorithm>
-#include <vector>
 #include <string>
 #include <thread>
+#include <vector>
 
 #include "db/db_impl/db_impl.h"
 #include "db/db_test_util.h"
-#include "memtable/hash_skiplist_rep.h"
 #include "options/options_parser.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
+#include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
 #include "rocksdb/iterator.h"
@@ -287,7 +287,9 @@ class ColumnFamilyTestBase : public testing::Test {
       // Verify the CF options of the returned CF handle.
       ColumnFamilyDescriptor desc;
       ASSERT_OK(handles_[cfi]->GetDescriptor(&desc));
-      RocksDBOptionsParser::VerifyCFOptions(desc.options, current_cf_opt);
+      RocksDBOptionsParser::VerifyCFOptions(ConfigOptions(), desc.options,
+                                            current_cf_opt);
+
 #endif  // !ROCKSDB_LITE
       cfi++;
     }

db/db_bloom_filter_test.cc

@@ -8,6 +8,7 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include "db/db_test_util.h"
+#include "options/options_helper.h"
 #include "port/stack_trace.h"
 #include "rocksdb/perf_context.h"
 #include "table/block_based/filter_policy_internal.h"

db/db_options_test.cc

@@ -33,7 +33,9 @@ class DBOptionsTest : public DBTestBase {
   std::unordered_map<std::string, std::string> GetMutableDBOptionsMap(
       const DBOptions& options) {
     std::string options_str;
-    GetStringFromDBOptions(&options_str, options);
+    ConfigOptions config_options;
+    config_options.delimiter = "; ";
+    GetStringFromDBOptions(config_options, options, &options_str);
     std::unordered_map<std::string, std::string> options_map;
     StringToMap(options_str, &options_map);
     std::unordered_map<std::string, std::string> mutable_map;
@@ -48,7 +50,10 @@ class DBOptionsTest : public DBTestBase {
   std::unordered_map<std::string, std::string> GetMutableCFOptionsMap(
       const ColumnFamilyOptions& options) {
     std::string options_str;
-    GetStringFromColumnFamilyOptions(&options_str, options);
+    ConfigOptions config_options;
+    config_options.delimiter = "; ";
+
+    GetStringFromColumnFamilyOptions(config_options, options, &options_str);
     std::unordered_map<std::string, std::string> options_map;
     StringToMap(options_str, &options_map);
     std::unordered_map<std::string, std::string> mutable_map;

examples/options_file_example.cc

@@ -79,7 +79,8 @@ int main() {
   // Load the options file.
   DBOptions loaded_db_opt;
   std::vector<ColumnFamilyDescriptor> loaded_cf_descs;
-  s = LoadLatestOptions(kDBPath, Env::Default(), &loaded_db_opt,
+  ConfigOptions config_options;
+  s = LoadLatestOptions(config_options, kDBPath, &loaded_db_opt,
                         &loaded_cf_descs);
   assert(s.ok());
   assert(loaded_db_opt.create_if_missing == db_opt.create_if_missing);

include/rocksdb/convenience.h

@@ -10,12 +10,74 @@
 #include <vector>
 
 #include "rocksdb/db.h"
-#include "rocksdb/options.h"
+#include "rocksdb/status.h"
 #include "rocksdb/table.h"
 
 namespace ROCKSDB_NAMESPACE {
+class Env;
+struct ColumnFamilyOptions;
+struct DBOptions;
+struct Options;
+
+// ConfigOptions containing the parameters/controls for
+// comparing objects and converting to/from strings.
+// These settings control how the methods
+// treat errors (e.g. ignore_unknown_objects), the format
+// of the serialization (e.g. delimiter), and how to compare
+// options (sanity_level).
+struct ConfigOptions {
+  // This enum defines the RocksDB options sanity level.
+  enum SanityLevel : unsigned char {
+    kSanityLevelNone = 0x01,  // Performs no sanity check at all.
+    // Performs minimum check to ensure the RocksDB instance can be
+    // opened without corrupting / mis-interpreting the data.
+    kSanityLevelLooselyCompatible = 0x02,
+    // Perform exact match sanity check.
+    kSanityLevelExactMatch = 0xFF,
+  };
+
+  enum Depth {
+    kDepthDefault,  // Traverse nested options that are not flagged as "shallow"
+    kDepthShallow,  // Do not traverse into any nested options
+    kDepthDetailed,  // Traverse nested options, overriding the options shallow
+                     // setting
+  };
+
+  // When true, any unused options will be ignored and OK will be returned
+  bool ignore_unknown_options = false;
+
+  // If the strings are escaped (old-style?)
+  bool input_strings_escaped = true;
+
+  // The separator between options when converting to a string
+  std::string delimiter = ";";
+
+  // Controls how to traverse options during print/match stages
+  Depth depth = Depth::kDepthDefault;
+
+  // Controls how options are serialized
+  // Controls how pedantic the comparison must be for equivalency
+  SanityLevel sanity_level = SanityLevel::kSanityLevelExactMatch;
+  // `file_readahead_size` is used for readahead for the option file.
+  size_t file_readahead_size = 512 * 1024;
+
+  // The environment to use for this option
+  Env* env = Env::Default();
+
+  bool IsShallow() const { return depth == Depth::kDepthShallow; }
+  bool IsDetailed() const { return depth == Depth::kDepthDetailed; }
+
+  bool IsCheckDisabled() const {
+    return sanity_level == SanityLevel::kSanityLevelNone;
+  }
+
+  bool IsCheckEnabled(SanityLevel level) const {
+    return (level > SanityLevel::kSanityLevelNone && level <= sanity_level);
+  }
+};
 
 #ifndef ROCKSDB_LITE
+
 // The following set of functions provide a way to construct RocksDB Options
 // from a string or a string-to-string map.  Here're the general rule of
 // setting option values from strings by type.  Some RocksDB types are also
@@ -134,13 +196,6 @@ namespace ROCKSDB_NAMESPACE {
 //     [Example]:
 //     * {"memtable", "vector:1024"} is equivalent to setting memtable
 //       to VectorRepFactory(1024).
-//   - HashCuckooRepFactory:
-//     Pass "cuckoo:<write_buffer_size>" to use HashCuckooRepFactory with the
-//     specified write buffer size, or simply "cuckoo" to use the default
-//     HashCuckooRepFactory.
-//     [Example]:
-//     * {"memtable", "cuckoo:1024"} is equivalent to setting memtable
-//       to NewHashCuckooRepFactory(1024).
 //
 //  * compression_opts:
 //    Use "compression_opts" to config compression_opts.  The value format
@@ -153,6 +208,12 @@ namespace ROCKSDB_NAMESPACE {
 //        cf_opt.compression_opts.strategy = 6;
 //        cf_opt.compression_opts.max_dict_bytes = 7;
 //
+// The GetColumnFamilyOptionsFromMap(ConfigOptions, ...) should be used; the
+// alternative signature may be deprecated in a future release.  The equivalent
+// functionality can be achieved by setting the corresponding options in
+// the ConfigOptions parameter.
+//
+// @param config_options controls how the map is processed.
 // @param base_options the default options of the output "new_options".
 // @param opts_map an option name to value map for specifying how "new_options"
 //     should be set.
@@ -165,6 +226,11 @@ namespace ROCKSDB_NAMESPACE {
 //     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".
+Status GetColumnFamilyOptionsFromMap(
+    const ConfigOptions& config_options,
+    const ColumnFamilyOptions& base_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    ColumnFamilyOptions* new_options);
 Status GetColumnFamilyOptionsFromMap(
     const ColumnFamilyOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
@@ -184,6 +250,12 @@ Status GetColumnFamilyOptionsFromMap(
 //   - Passing {"rate_limiter_bytes_per_sec", "1024"} is equivalent to
 //     passing NewGenericRateLimiter(1024) to rate_limiter_bytes_per_sec.
 //
+// The GetDBOptionsFromMap(ConfigOptions, ...) should be used; the
+// alternative signature may be deprecated in a future release. The equivalent
+// functionality can be achieved by setting the corresponding options in
+// the ConfigOptions parameter.
+//
+// @param config_options controls how the map is processed.
 // @param base_options the default options of the output "new_options".
 // @param opts_map an option name to value map for specifying how "new_options"
 //     should be set.
@@ -196,6 +268,10 @@ 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".
+Status GetDBOptionsFromMap(
+    const ConfigOptions& cfg_options, const DBOptions& base_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    DBOptions* new_options);
 Status GetDBOptionsFromMap(
     const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
@@ -227,6 +303,12 @@ Status GetDBOptionsFromMap(
 //   - Passing {"block_cache", "1M"} in GetBlockBasedTableOptionsFromMap is
 //     equivalent to setting block_cache using NewLRUCache(1024 * 1024).
 //
+// The GetBlockBasedTableOptionsFromMap(ConfigOptions, ...) should be used;
+// the alternative signature may be deprecated in a future release. The
+// equivalent functionality can be achieved by setting the corresponding
+// options in the ConfigOptions parameter.
+//
+// @param config_options controls how the map is processed.
 // @param table_options the default options of the output "new_table_options".
 // @param opts_map an option name to value map for specifying how
 //     "new_table_options" should be set.
@@ -240,6 +322,11 @@ Status GetDBOptionsFromMap(
 // @return Status::OK() on success.  Otherwise, a non-ok status indicating
 //     error will be returned, and "new_table_options" will be set to
 //     "table_options".
+Status GetBlockBasedTableOptionsFromMap(
+    const ConfigOptions& config_options,
+    const BlockBasedTableOptions& table_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    BlockBasedTableOptions* new_table_options);
 Status GetBlockBasedTableOptionsFromMap(
     const BlockBasedTableOptions& table_options,
     const std::unordered_map<std::string, std::string>& opts_map,
@@ -250,6 +337,12 @@ Status GetBlockBasedTableOptionsFromMap(
 // map "opts_map" of option name to option value to construct the new
 // PlainTableOptions "new_table_options".
 //
+// The GetPlainTableOptionsFromMap(ConfigOptions, ...) should be used; the
+// alternative signature may be deprecated in a future release. The equivalent
+// functionality can be achieved by setting the corresponding options in
+// the ConfigOptions parameter.
+//
+// @param config_options controls how the map is processed.
 // @param table_options the default options of the output "new_table_options".
 // @param opts_map an option name to value map for specifying how
 //     "new_table_options" should be set.
@@ -263,6 +356,10 @@ Status GetBlockBasedTableOptionsFromMap(
 // @return Status::OK() on success.  Otherwise, a non-ok status indicating
 //     error will be returned, and "new_table_options" will be set to
 //     "table_options".
+Status GetPlainTableOptionsFromMap(
+    const ConfigOptions& config_options, const PlainTableOptions& table_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    PlainTableOptions* new_table_options);
 Status GetPlainTableOptionsFromMap(
     const PlainTableOptions& table_options,
     const std::unordered_map<std::string, std::string>& opts_map,
@@ -277,22 +374,43 @@ Status GetPlainTableOptionsFromMap(
 // BlockBasedTableOptions as part of the string for block-based table factory:
 //   "write_buffer_size=1024;block_based_table_factory={block_size=4k};"
 //   "max_write_buffer_num=2"
+//
+//
+// The GetColumnFamilyOptionsFromString(ConfigOptions, ...) should be used; the
+// alternative signature may be deprecated in a future release. The equivalent
+// functionality can be achieved by setting the corresponding options in
+// the ConfigOptions parameter.
+Status GetColumnFamilyOptionsFromString(const ConfigOptions& config_options,
+                                        const ColumnFamilyOptions& base_options,
+                                        const std::string& opts_str,
+                                        ColumnFamilyOptions* new_options);
 Status GetColumnFamilyOptionsFromString(const ColumnFamilyOptions& base_options,
                                         const std::string& opts_str,
                                         ColumnFamilyOptions* new_options);
 
+Status GetDBOptionsFromString(const ConfigOptions& config_options,
+                              const DBOptions& base_options,
+                              const std::string& opts_str,
+                              DBOptions* new_options);
+
 Status GetDBOptionsFromString(const DBOptions& base_options,
                               const std::string& opts_str,
                               DBOptions* new_options);
 
+Status GetStringFromDBOptions(const ConfigOptions& config_options,
+                              const DBOptions& db_options,
+                              std::string* opts_str);
+
 Status GetStringFromDBOptions(std::string* opts_str,
                               const DBOptions& db_options,
                               const std::string& delimiter = ";  ");
 
+Status GetStringFromColumnFamilyOptions(const ConfigOptions& config_options,
+                                        const ColumnFamilyOptions& cf_options,
+                                        std::string* opts_str);
 Status GetStringFromColumnFamilyOptions(std::string* opts_str,
                                         const ColumnFamilyOptions& cf_options,
                                         const std::string& delimiter = ";  ");
-
 Status GetStringFromCompressionType(std::string* compression_str,
                                     CompressionType compression_type);
 
@@ -301,10 +419,18 @@ std::vector<CompressionType> GetSupportedCompressions();
 Status GetBlockBasedTableOptionsFromString(
     const BlockBasedTableOptions& table_options, const std::string& opts_str,
     BlockBasedTableOptions* new_table_options);
+Status GetBlockBasedTableOptionsFromString(
+    const ConfigOptions& config_options,
+    const BlockBasedTableOptions& table_options, const std::string& opts_str,
+    BlockBasedTableOptions* new_table_options);
 
 Status GetPlainTableOptionsFromString(const PlainTableOptions& table_options,
                                       const std::string& opts_str,
                                       PlainTableOptions* new_table_options);
+Status GetPlainTableOptionsFromString(const ConfigOptions& config_options,
+                                      const PlainTableOptions& table_options,
+                                      const std::string& opts_str,
+                                      PlainTableOptions* new_table_options);
 
 Status GetMemTableRepFactoryFromString(
     const std::string& opts_str,
@@ -312,6 +438,9 @@ Status GetMemTableRepFactoryFromString(
 
 Status GetOptionsFromString(const Options& base_options,
                             const std::string& opts_str, Options* new_options);
+Status GetOptionsFromString(const ConfigOptions& config_options,
+                            const Options& base_options,
+                            const std::string& opts_str, Options* new_options);
 
 Status StringToMap(const std::string& opts_str,
                    std::unordered_map<std::string, std::string>* opts_map);

include/rocksdb/table.h

@@ -25,20 +25,24 @@
 #include "rocksdb/cache.h"
 #include "rocksdb/env.h"
 #include "rocksdb/iterator.h"
-#include "rocksdb/options.h"
 #include "rocksdb/status.h"
 
 namespace ROCKSDB_NAMESPACE {
 
 // -- Block-based Table
+class FilterPolicy;
 class FlushBlockPolicyFactory;
 class PersistentCache;
 class RandomAccessFile;
 struct TableReaderOptions;
 struct TableBuilderOptions;
 class TableBuilder;
+class TableFactory;
 class TableReader;
 class WritableFileWriter;
+struct ColumnFamilyOptions;
+struct ConfigOptions;
+struct DBOptions;
 struct EnvOptions;
 struct Options;
 
@@ -555,8 +559,8 @@ class TableFactory {
   // RocksDB prints configurations at DB Open().
   virtual std::string GetPrintableTableOptions() const = 0;
 
-  virtual Status GetOptionString(std::string* /*opt_string*/,
+  virtual Status GetOptionString(const ConfigOptions& /*config_options*/,
-                                 const std::string& /*delimiter*/) const {
+                                 std::string* /*opt_string*/) const {
     return Status::NotSupported(
         "The table factory doesn't implement GetOptionString().");
   }

include/rocksdb/utilities/ldb_cmd.h

@@ -9,6 +9,7 @@
 
 #include <stdio.h>
 #include <stdlib.h>
+
 #include <algorithm>
 #include <functional>
 #include <map>
@@ -16,6 +17,7 @@
 #include <string>
 #include <vector>
 
+#include "rocksdb/convenience.h"
 #include "rocksdb/env.h"
 #include "rocksdb/iterator.h"
 #include "rocksdb/ldb_tool.h"
@@ -161,8 +163,6 @@ class LDBCommand {
   // If true, try to construct options from DB's option files.
   bool try_load_options_;
 
-  bool ignore_unknown_options_;
-
   bool create_if_missing_;
 
   /**
@@ -237,6 +237,7 @@ class LDBCommand {
 
   Options options_;
   std::vector<ColumnFamilyDescriptor> column_families_;
+  ConfigOptions config_options_;
   LDBOptions ldb_options_;
 
  private:

include/rocksdb/utilities/options_util.h

@@ -11,12 +11,14 @@
 #include <string>
 #include <vector>
 
+#include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
 #include "rocksdb/status.h"
 
 namespace ROCKSDB_NAMESPACE {
+struct ConfigOptions;
 // Constructs the DBOptions and ColumnFamilyDescriptors by loading the
 // latest RocksDB options file stored in the specified rocksdb database.
 //
@@ -45,13 +47,19 @@ namespace ROCKSDB_NAMESPACE {
 // pointer options of BlockBasedTableOptions (flush_block_policy_factory,
 // block_cache, and block_cache_compressed), which will be initialized with
 // default values.  Developers can further specify these three options by
-// casting the return value of TableFactoroy::GetOptions() to
+// casting the return value of TableFactory::GetOptions() to
 // BlockBasedTableOptions and making necessary changes.
 //
 // ignore_unknown_options can be set to true if you want to ignore options
 // that are from a newer version of the db, esentially for forward
 // compatibility.
 //
+// config_options contains a set of options that controls the processing
+// of the options.  The LoadLatestOptions(ConfigOptions...) should be preferred;
+// the alternative signature may be deprecated in a future release. The
+// equivalent functionality can be achieved by setting the corresponding options
+// in the ConfigOptions parameter.
+//
 // examples/options_file_example.cc demonstrates how to use this function
 // to open a RocksDB instance.
 //
@@ -67,16 +75,30 @@ Status LoadLatestOptions(const std::string& dbpath, Env* env,
                          std::vector<ColumnFamilyDescriptor>* cf_descs,
                          bool ignore_unknown_options = false,
                          std::shared_ptr<Cache>* cache = {});
+Status LoadLatestOptions(const ConfigOptions& config_options,
+                         const std::string& dbpath, DBOptions* db_options,
+                         std::vector<ColumnFamilyDescriptor>* cf_descs,
+                         std::shared_ptr<Cache>* cache = {});
 
 // Similar to LoadLatestOptions, this function constructs the DBOptions
 // and ColumnFamilyDescriptors based on the specified RocksDB Options file.
 //
+// The LoadOptionsFile(ConfigOptions...) should be preferred;
+// the alternative signature may be deprecated in a future release. The
+// equivalent functionality can be achieved by setting the corresponding
+// options in the ConfigOptions parameter.
+//
 // @see LoadLatestOptions
 Status LoadOptionsFromFile(const std::string& options_file_name, Env* env,
                            DBOptions* db_options,
                            std::vector<ColumnFamilyDescriptor>* cf_descs,
                            bool ignore_unknown_options = false,
                            std::shared_ptr<Cache>* cache = {});
+Status LoadOptionsFromFile(const ConfigOptions& config_options,
+                           const std::string& options_file_name,
+                           DBOptions* db_options,
+                           std::vector<ColumnFamilyDescriptor>* cf_descs,
+                           std::shared_ptr<Cache>* cache = {});
 
 // Returns the latest options file name under the specified db path.
 Status GetLatestOptionsFileName(const std::string& dbpath, Env* env,
@@ -97,6 +119,10 @@ Status CheckOptionsCompatibility(
     const std::string& dbpath, Env* env, const DBOptions& db_options,
     const std::vector<ColumnFamilyDescriptor>& cf_descs,
     bool ignore_unknown_options = false);
+Status CheckOptionsCompatibility(
+    const ConfigOptions& config_options, const std::string& dbpath,
+    const DBOptions& db_options,
+    const std::vector<ColumnFamilyDescriptor>& cf_descs);
 
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // !ROCKSDB_LITE

java/CMakeLists.txt

@@ -26,6 +26,7 @@ set(JNI_NATIVE_SOURCES
         rocksjni/comparator.cc
         rocksjni/comparatorjnicallback.cc
         rocksjni/compression_options.cc
+        rocksjni/config_options.cc
         rocksjni/env.cc
         rocksjni/env_options.cc
         rocksjni/filter.cc
@@ -128,6 +129,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/ComparatorType.java
   src/main/java/org/rocksdb/CompressionOptions.java
   src/main/java/org/rocksdb/CompressionType.java
+  src/main/java/org/rocksdb/ConfigOptions.java
   src/main/java/org/rocksdb/DataBlockIndexType.java
   src/main/java/org/rocksdb/DBOptionsInterface.java
   src/main/java/org/rocksdb/DBOptions.java
@@ -191,6 +193,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/RocksMemEnv.java
   src/main/java/org/rocksdb/RocksMutableObject.java
   src/main/java/org/rocksdb/RocksObject.java
+  src/main/java/org/rocksdb/SanityLevel.java
   src/main/java/org/rocksdb/SizeApproximationFlag.java
   src/main/java/org/rocksdb/SkipListMemTableConfig.java
   src/main/java/org/rocksdb/Slice.java

java/Makefile

@@ -25,6 +25,7 @@ NATIVE_JAVA_CLASSES = \
 	org.rocksdb.CompactRangeOptions\
 	org.rocksdb.ComparatorOptions\
 	org.rocksdb.CompressionOptions\
+	org.rocksdb.ConfigOptions\
 	org.rocksdb.DBOptions\
 	org.rocksdb.DirectSlice\
 	org.rocksdb.Env\

java/rocksjni/config_options.cc

@@ -0,0 +1,88 @@
+// 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).
+//
+// This file implements the "bridge" between Java and C++ and enables
+// calling C++ ROCKSDB_NAMESPACE::ConfigOptions methods
+// from Java side.
+
+#include <jni.h>
+
+#include "include/org_rocksdb_ConfigOptions.h"
+#include "rocksdb/convenience.h"
+#include "rocksjni/portal.h"
+
+/*
+ * Class:     org_rocksdb_ConfigOptions
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_ConfigOptions_disposeInternal(JNIEnv *, jobject,
+                                                    jlong jhandle) {
+  auto *co = reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions *>(jhandle);
+  assert(co != nullptr);
+  delete co;
+}
+
+/*
+ * Class:     org_rocksdb_ConfigOptions
+ * Method:    newConfigOptions
+ * Signature: ()J
+ */
+jlong Java_org_rocksdb_ConfigOptions_newConfigOptions(JNIEnv *, jclass) {
+  auto *cfg_opt = new ROCKSDB_NAMESPACE::ConfigOptions();
+  return reinterpret_cast<jlong>(cfg_opt);
+}
+
+/*
+ * Class:     org_rocksdb_ConfigOptions
+ * Method:    setDelimiter
+ * Signature: (JLjava/lang/String;)V
+ */
+void Java_org_rocksdb_ConfigOptions_setDelimiter(JNIEnv *env, jclass,
+                                                 jlong handle, jstring s) {
+  auto *cfg_opt = reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions *>(handle);
+  const char *delim = env->GetStringUTFChars(s, nullptr);
+  if (delim == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return;
+  }
+  cfg_opt->delimiter = delim;
+  env->ReleaseStringUTFChars(s, delim);
+}
+
+/*
+ * Class:     org_rocksdb_ConfigOptions
+ * Method:    setIgnoreUnknownOptions
+ * Signature: (JZ)V
+ */
+void Java_org_rocksdb_ConfigOptions_setIgnoreUnknownOptions(JNIEnv *, jclass,
+                                                            jlong handle,
+                                                            jboolean b) {
+  auto *cfg_opt = reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions *>(handle);
+  cfg_opt->ignore_unknown_options = static_cast<bool>(b);
+}
+
+/*
+ * Class:     org_rocksdb_ConfigOptions
+ * Method:    setInputStringsEscaped
+ * Signature: (JZ)V
+ */
+void Java_org_rocksdb_ConfigOptions_setInputStringsEscaped(JNIEnv *, jclass,
+                                                           jlong handle,
+                                                           jboolean b) {
+  auto *cfg_opt = reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions *>(handle);
+  cfg_opt->input_strings_escaped = static_cast<bool>(b);
+}
+
+/*
+ * Class:     org_rocksdb_ConfigOptions
+ * Method:    setSanityLevel
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_ConfigOptions_setSanityLevel(JNIEnv *, jclass,
+                                                   jlong handle, jbyte level) {
+  auto *cfg_opt = reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions *>(handle);
+  cfg_opt->sanity_level = ROCKSDB_NAMESPACE::SanityLevelJni::toCppSanityLevel(level);
+}

java/rocksjni/options.cc

@@ -3274,12 +3274,46 @@ jlong Java_org_rocksdb_ColumnFamilyOptions_newColumnFamilyOptionsFromOptions(
   return reinterpret_cast<jlong>(new_opt);
 }
 
+/*
+ * Class:     org_rocksdb_ColumnFamilyOptions
+ * Method:    getColumnFamilyOptionsFromProps
+ * Signature: (JLjava/lang/String;)J
+ */
+jlong Java_org_rocksdb_ColumnFamilyOptions_getColumnFamilyOptionsFromProps__JLjava_lang_String_2(
+    JNIEnv* env, jclass, jlong cfg_handle, jstring jopt_string) {
+  const char* opt_string = env->GetStringUTFChars(jopt_string, nullptr);
+  if (opt_string == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return 0;
+  }
+  auto* config_options =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions*>(cfg_handle);
+  auto* cf_options = new ROCKSDB_NAMESPACE::ColumnFamilyOptions();
+  ROCKSDB_NAMESPACE::Status status =
+      ROCKSDB_NAMESPACE::GetColumnFamilyOptionsFromString(
+          *config_options, ROCKSDB_NAMESPACE::ColumnFamilyOptions(), opt_string,
+          cf_options);
+
+  env->ReleaseStringUTFChars(jopt_string, opt_string);
+
+  // Check if ColumnFamilyOptions creation was possible.
+  jlong ret_value = 0;
+  if (status.ok()) {
+    ret_value = reinterpret_cast<jlong>(cf_options);
+  } else {
+    // if operation failed the ColumnFamilyOptions need to be deleted
+    // again to prevent a memory leak.
+    delete cf_options;
+  }
+  return ret_value;
+}
+
 /*
  * Class:     org_rocksdb_ColumnFamilyOptions
  * Method:    getColumnFamilyOptionsFromProps
  * Signature: (Ljava/util/String;)J
  */
-jlong Java_org_rocksdb_ColumnFamilyOptions_getColumnFamilyOptionsFromProps(
+jlong Java_org_rocksdb_ColumnFamilyOptions_getColumnFamilyOptionsFromProps__Ljava_lang_String_2(
     JNIEnv* env, jclass, jstring jopt_string) {
   const char* opt_string = env->GetStringUTFChars(jopt_string, nullptr);
   if (opt_string == nullptr) {
@@ -4667,12 +4701,45 @@ jlong Java_org_rocksdb_DBOptions_newDBOptionsFromOptions(
   return reinterpret_cast<jlong>(new_opt);
 }
 
+/*
+ * Class:     org_rocksdb_DBOptions
+ * Method:    getDBOptionsFromProps
+ * Signature: (JLjava/lang/String;)J
+ */
+jlong Java_org_rocksdb_DBOptions_getDBOptionsFromProps__JLjava_lang_String_2(
+    JNIEnv* env, jclass, jlong config_handle, jstring jopt_string) {
+  const char* opt_string = env->GetStringUTFChars(jopt_string, nullptr);
+  if (opt_string == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return 0;
+  }
+
+  auto* config_options =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions*>(config_handle);
+  auto* db_options = new ROCKSDB_NAMESPACE::DBOptions();
+  ROCKSDB_NAMESPACE::Status status = ROCKSDB_NAMESPACE::GetDBOptionsFromString(
+      *config_options, ROCKSDB_NAMESPACE::DBOptions(), opt_string, db_options);
+
+  env->ReleaseStringUTFChars(jopt_string, opt_string);
+
+  // Check if DBOptions creation was possible.
+  jlong ret_value = 0;
+  if (status.ok()) {
+    ret_value = reinterpret_cast<jlong>(db_options);
+  } else {
+    // if operation failed the DBOptions need to be deleted
+    // again to prevent a memory leak.
+    delete db_options;
+  }
+  return ret_value;
+}
+
 /*
  * Class:     org_rocksdb_DBOptions
  * Method:    getDBOptionsFromProps
  * Signature: (Ljava/util/String;)J
  */
-jlong Java_org_rocksdb_DBOptions_getDBOptionsFromProps(
+jlong Java_org_rocksdb_DBOptions_getDBOptionsFromProps__Ljava_lang_String_2(
     JNIEnv* env, jclass, jstring jopt_string) {
   const char* opt_string = env->GetStringUTFChars(jopt_string, nullptr);
   if (opt_string == nullptr) {

java/rocksjni/options_util.cc

@@ -55,7 +55,7 @@ void build_column_family_descriptor_list(
  * Method:    loadLatestOptions
  * Signature: (Ljava/lang/String;JLjava/util/List;Z)V
  */
-void Java_org_rocksdb_OptionsUtil_loadLatestOptions(
+void Java_org_rocksdb_OptionsUtil_loadLatestOptions__Ljava_lang_String_2JJLjava_util_List_2Z(
     JNIEnv* env, jclass /*jcls*/, jstring jdbpath, jlong jenv_handle,
     jlong jdb_opts_handle, jobject jcfds, jboolean ignore_unknown_options) {
   jboolean has_exception = JNI_FALSE;
@@ -78,12 +78,42 @@ void Java_org_rocksdb_OptionsUtil_loadLatestOptions(
   }
 }
 
+/*
+ * Class:     org_rocksdb_OptionsUtil
+ * Method:    loadLatestOptions_1
+ * Signature: (JLjava/lang/String;JLjava/util/List;)V
+ */
+void Java_org_rocksdb_OptionsUtil_loadLatestOptions__JLjava_lang_String_2JLjava_util_List_2(
+    JNIEnv* env, jclass /*jcls*/, jlong cfg_handle, jstring jdbpath,
+    jlong jdb_opts_handle, jobject jcfds) {
+  jboolean has_exception = JNI_FALSE;
+  auto db_path =
+      ROCKSDB_NAMESPACE::JniUtil::copyStdString(env, jdbpath, &has_exception);
+  if (has_exception == JNI_TRUE) {
+    // exception occurred
+    return;
+  }
+  std::vector<ROCKSDB_NAMESPACE::ColumnFamilyDescriptor> cf_descs;
+  auto* config_options =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions*>(cfg_handle);
+  auto* db_options =
+      reinterpret_cast<ROCKSDB_NAMESPACE::DBOptions*>(jdb_opts_handle);
+  ROCKSDB_NAMESPACE::Status s = ROCKSDB_NAMESPACE::LoadLatestOptions(
+      *config_options, db_path, db_options, &cf_descs);
+  if (!s.ok()) {
+    // error, raise an exception
+    ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(env, s);
+  } else {
+    build_column_family_descriptor_list(env, jcfds, cf_descs);
+  }
+}
+
 /*
  * Class:     org_rocksdb_OptionsUtil
  * Method:    loadOptionsFromFile
  * Signature: (Ljava/lang/String;JJLjava/util/List;Z)V
  */
-void Java_org_rocksdb_OptionsUtil_loadOptionsFromFile(
+void Java_org_rocksdb_OptionsUtil_loadOptionsFromFile__Ljava_lang_String_2JJLjava_util_List_2Z(
     JNIEnv* env, jclass /*jcls*/, jstring jopts_file_name, jlong jenv_handle,
     jlong jdb_opts_handle, jobject jcfds, jboolean ignore_unknown_options) {
   jboolean has_exception = JNI_FALSE;
@@ -106,6 +136,36 @@ void Java_org_rocksdb_OptionsUtil_loadOptionsFromFile(
   }
 }
 
+/*
+ * Class:     org_rocksdb_OptionsUtil
+ * Method:    loadOptionsFromFile
+ * Signature: (JLjava/lang/String;JLjava/util/List;)V
+ */
+void Java_org_rocksdb_OptionsUtil_loadOptionsFromFile__JLjava_lang_String_2JLjava_util_List_2(
+    JNIEnv* env, jclass /*jcls*/, jlong cfg_handle, jstring jopts_file_name,
+    jlong jdb_opts_handle, jobject jcfds) {
+  jboolean has_exception = JNI_FALSE;
+  auto opts_file_name = ROCKSDB_NAMESPACE::JniUtil::copyStdString(
+      env, jopts_file_name, &has_exception);
+  if (has_exception == JNI_TRUE) {
+    // exception occurred
+    return;
+  }
+  std::vector<ROCKSDB_NAMESPACE::ColumnFamilyDescriptor> cf_descs;
+  auto* config_options =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ConfigOptions*>(cfg_handle);
+  auto* db_options =
+      reinterpret_cast<ROCKSDB_NAMESPACE::DBOptions*>(jdb_opts_handle);
+  ROCKSDB_NAMESPACE::Status s = ROCKSDB_NAMESPACE::LoadOptionsFromFile(
+      *config_options, opts_file_name, db_options, &cf_descs);
+  if (!s.ok()) {
+    // error, raise an exception
+    ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(env, s);
+  } else {
+    build_column_family_descriptor_list(env, jcfds, cf_descs);
+  }
+}
+
 /*
  * Class:     org_rocksdb_OptionsUtil
  * Method:    getLatestOptionsFileName

java/rocksjni/portal.h

@@ -22,6 +22,7 @@
 #include <type_traits>
 #include <vector>
 
+#include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/rate_limiter.h"
@@ -7530,5 +7531,41 @@ class ReusedSynchronisationTypeJni {
     }
   }
 };
+// The portal class for org.rocksdb.SanityLevel
+class SanityLevelJni {
+ public:
+  // Returns the equivalent org.rocksdb.SanityLevel for the provided
+  // C++ ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel enum
+  static jbyte toJavaSanityLevel(
+      const ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel &sanity_level) {
+    switch (sanity_level) {
+      case ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel::kSanityLevelNone:
+        return 0x0;
+      case ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel::
+          kSanityLevelLooselyCompatible:
+        return 0x1;
+      case ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel::
+          kSanityLevelExactMatch:
+        return -0x01;
+      default:
+        return -0x01;  // undefined
+    }
+  }
+
+  // Returns the equivalent C++ ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel enum for
+  // the provided Java org.rocksdb.SanityLevel
+  static ROCKSDB_NAMESPACE::ConfigOptions::SanityLevel toCppSanityLevel(
+      jbyte sanity_level) {
+    switch (sanity_level) {
+      case 0x0:
+        return ROCKSDB_NAMESPACE::ConfigOptions::kSanityLevelNone;
+      case 0x1:
+        return ROCKSDB_NAMESPACE::ConfigOptions::kSanityLevelLooselyCompatible;
+      default:
+        // undefined/default
+        return ROCKSDB_NAMESPACE::ConfigOptions::kSanityLevelExactMatch;
+    }
+  }
+};
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // JAVA_ROCKSJNI_PORTAL_H_

java/src/main/java/org/rocksdb/ColumnFamilyOptions.java

@@ -96,20 +96,40 @@ public class ColumnFamilyOptions extends RocksObject
    */
   public static ColumnFamilyOptions getColumnFamilyOptionsFromProps(
       final Properties properties) {
-    if (properties == null || properties.size() == 0) {
-      throw new IllegalArgumentException(
-          "Properties value must contain at least one value.");
-    }
     ColumnFamilyOptions columnFamilyOptions = null;
-    StringBuilder stringBuilder = new StringBuilder();
+    final long handle =
-    for (final String name : properties.stringPropertyNames()){
+        getColumnFamilyOptionsFromProps(Options.getOptionStringFromProps(properties));
-      stringBuilder.append(name);
+    if (handle != 0) {
-      stringBuilder.append("=");
+      columnFamilyOptions = new ColumnFamilyOptions(handle);
-      stringBuilder.append(properties.getProperty(name));
-      stringBuilder.append(";");
     }
-    long handle = getColumnFamilyOptionsFromProps(
+    return columnFamilyOptions;
-        stringBuilder.toString());
+  }
+
+  /**
+   * <p>Method to get a options instance by using pre-configured
+   * property values. If one or many values are undefined in
+   * the context of RocksDB the method will return a null
+   * value.</p>
+   *
+   * <p><strong>Note</strong>: Property keys can be derived from
+   * getter methods within the options class. Example: the method
+   * {@code writeBufferSize()} has a property key:
+   * {@code write_buffer_size}.</p>
+   *
+   * @param cfgOpts  ConfigOptions controlling how the properties are parsed.
+   * @param properties {@link java.util.Properties} instance.
+   *
+   * @return {@link org.rocksdb.ColumnFamilyOptions instance}
+   *     or null.
+   *
+   * @throws java.lang.IllegalArgumentException if null or empty
+   *     {@link Properties} instance is passed to the method call.
+   */
+  public static ColumnFamilyOptions getColumnFamilyOptionsFromProps(
+      final ConfigOptions cfgOpts, final Properties properties) {
+    ColumnFamilyOptions columnFamilyOptions = null;
+    final long handle = getColumnFamilyOptionsFromProps(
+        cfgOpts.nativeHandle_, Options.getOptionStringFromProps(properties));
     if (handle != 0){
       columnFamilyOptions = new ColumnFamilyOptions(handle);
     }
@@ -825,7 +845,8 @@ public class ColumnFamilyOptions extends RocksObject
   }
 
   private static native long getColumnFamilyOptionsFromProps(
-      String optString);
+      final long cfgHandle, String optString);
+  private static native long getColumnFamilyOptionsFromProps(final String optString);
 
   private static native long newColumnFamilyOptions();
   private static native long copyColumnFamilyOptions(final long handle);

java/src/main/java/org/rocksdb/ConfigOptions.java

@@ -0,0 +1,47 @@
+package org.rocksdb;
+
+public class ConfigOptions extends RocksObject {
+  static {
+    RocksDB.loadLibrary();
+  }
+
+  /**
+   * Construct with default Options
+   */
+  public ConfigOptions() {
+    super(newConfigOptions());
+  }
+
+  public ConfigOptions setDelimiter(final String delimiter) {
+    setDelimiter(nativeHandle_, delimiter);
+    return this;
+  }
+  public ConfigOptions setIgnoreUnknownOptions(final boolean ignore) {
+    setIgnoreUnknownOptions(nativeHandle_, ignore);
+    return this;
+  }
+
+  public ConfigOptions setEnv(final Env env) {
+    setEnv(nativeHandle_, env.nativeHandle_);
+    return this;
+  }
+
+  public ConfigOptions setInputStringsEscaped(final boolean escaped) {
+    setInputStringsEscaped(nativeHandle_, escaped);
+    return this;
+  }
+
+  public ConfigOptions setSanityLevel(final SanityLevel level) {
+    setSanityLevel(nativeHandle_, level.getValue());
+    return this;
+  }
+
+  @Override protected final native void disposeInternal(final long handle);
+
+  private native static long newConfigOptions();
+  private native static void setEnv(final long handle, final long envHandle);
+  private native static void setDelimiter(final long handle, final String delimiter);
+  private native static void setIgnoreUnknownOptions(final long handle, final boolean ignore);
+  private native static void setInputStringsEscaped(final long handle, final boolean escaped);
+  private native static void setSanityLevel(final long handle, final byte level);
+}

java/src/main/java/org/rocksdb/DBOptions.java

@@ -71,6 +71,7 @@ public class DBOptions extends RocksObject
    * {@code allowMmapReads()} has a property key:
    * {@code allow_mmap_reads}.</p>
    *
+   * @param cfgOpts The ConfigOptions to control how the string is processed.
    * @param properties {@link java.util.Properties} instance.
    *
    * @return {@link org.rocksdb.DBOptions instance}
@@ -80,22 +81,40 @@ public class DBOptions extends RocksObject
    *     {@link java.util.Properties} instance is passed to the method call.
    */
   public static DBOptions getDBOptionsFromProps(
-      final Properties properties) {
+      final ConfigOptions cfgOpts, final Properties properties) {
-    if (properties == null || properties.size() == 0) {
-      throw new IllegalArgumentException(
-          "Properties value must contain at least one value.");
-    }
     DBOptions dbOptions = null;
-    StringBuilder stringBuilder = new StringBuilder();
+    final String optionsString = Options.getOptionStringFromProps(properties);
-    for (final String name : properties.stringPropertyNames()){
+    final long handle = getDBOptionsFromProps(cfgOpts.nativeHandle_, optionsString);
-      stringBuilder.append(name);
+    if (handle != 0) {
-      stringBuilder.append("=");
+      dbOptions = new DBOptions(handle);
-      stringBuilder.append(properties.getProperty(name));
-      stringBuilder.append(";");
     }
-    long handle = getDBOptionsFromProps(
+    return dbOptions;
-        stringBuilder.toString());
+  }
-    if (handle != 0){
+
+  /**
+   * <p>Method to get a options instance by using pre-configured
+   * property values. If one or many values are undefined in
+   * the context of RocksDB the method will return a null
+   * value.</p>
+   *
+   * <p><strong>Note</strong>: Property keys can be derived from
+   * getter methods within the options class. Example: the method
+   * {@code allowMmapReads()} has a property key:
+   * {@code allow_mmap_reads}.</p>
+   *
+   * @param properties {@link java.util.Properties} instance.
+   *
+   * @return {@link org.rocksdb.DBOptions instance}
+   *     or null.
+   *
+   * @throws java.lang.IllegalArgumentException if null or empty
+   *     {@link java.util.Properties} instance is passed to the method call.
+   */
+  public static DBOptions getDBOptionsFromProps(final Properties properties) {
+    DBOptions dbOptions = null;
+    final String optionsString = Options.getOptionStringFromProps(properties);
+    final long handle = getDBOptionsFromProps(optionsString);
+    if (handle != 0) {
       dbOptions = new DBOptions(handle);
     }
     return dbOptions;
@@ -1175,8 +1194,8 @@ public class DBOptions extends RocksObject
     super(nativeHandle);
   }
 
-  private static native long getDBOptionsFromProps(
+  private static native long getDBOptionsFromProps(long cfgHandle, String optString);
-      String optString);
+  private static native long getDBOptionsFromProps(String optString);
 
   private static native long newDBOptions();
   private static native long copyDBOptions(final long handle);

java/src/main/java/org/rocksdb/Options.java

@@ -6,10 +6,7 @@
 package org.rocksdb;
 
 import java.nio.file.Paths;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
 
 /**
  * Options to control the behavior of a database.  It will be used
@@ -27,6 +24,25 @@ public class Options extends RocksObject
     RocksDB.loadLibrary();
   }
 
+  /**
+   * Converts the input properties into a Options-style formatted string
+   * @param properties   The set of properties to convert
+   * @return The Options-style representation of those properties.
+   */
+  public static String getOptionStringFromProps(final Properties properties) {
+    if (properties == null || properties.size() == 0) {
+      throw new IllegalArgumentException("Properties value must contain at least one value.");
+    }
+    StringBuilder stringBuilder = new StringBuilder();
+    for (final String name : properties.stringPropertyNames()) {
+      stringBuilder.append(name);
+      stringBuilder.append("=");
+      stringBuilder.append(properties.getProperty(name));
+      stringBuilder.append(";");
+    }
+    return stringBuilder.toString();
+  }
+
   /**
    * Construct options for opening a RocksDB.
    *

java/src/main/java/org/rocksdb/OptionsUtil.java

@@ -59,7 +59,7 @@ public class OptionsUtil {
    * @param cfDescs A list of {@link org.rocksdb.ColumnFamilyDescriptor}'s be
    *     returned.
    * @param ignoreUnknownOptions this flag can be set to true if you want to
-   *     ignore options that are from a newer version of the db, esentially for
+   *     ignore options that are from a newer version of the db, essentially for
    *     forward compatibility.
    *
    * @throws RocksDBException thrown if error happens in underlying
@@ -71,6 +71,25 @@ public class OptionsUtil {
         dbPath, env.nativeHandle_, dbOptions.nativeHandle_, cfDescs, ignoreUnknownOptions);
   }
 
+  /**
+   * Similar to LoadLatestOptions, this function constructs the DBOptions
+   * and ColumnFamilyDescriptors based on the specified RocksDB Options file.
+   * See LoadLatestOptions above.
+   *
+   * @param dbPath the path to the RocksDB.
+   * @param configOptions {@link org.rocksdb.ConfigOptions} instance.
+   * @param dbOptions {@link org.rocksdb.DBOptions} instance. This will be
+   *     filled and returned.
+   * @param cfDescs A list of {@link org.rocksdb.ColumnFamilyDescriptor}'s be
+   *     returned.
+   * @throws RocksDBException thrown if error happens in underlying
+   *     native library.
+   */
+  public static void loadLatestOptions(ConfigOptions configOptions, String dbPath,
+      DBOptions dbOptions, List<ColumnFamilyDescriptor> cfDescs) throws RocksDBException {
+    loadLatestOptions(configOptions.nativeHandle_, dbPath, dbOptions.nativeHandle_, cfDescs);
+  }
+
   /**
    * Similar to LoadLatestOptions, this function constructs the DBOptions
    * and ColumnFamilyDescriptors based on the specified RocksDB Options file.
@@ -111,6 +130,26 @@ public class OptionsUtil {
         optionsFileName, env.nativeHandle_, dbOptions.nativeHandle_, cfDescs, ignoreUnknownOptions);
   }
 
+  /**
+   * Similar to LoadLatestOptions, this function constructs the DBOptions
+   * and ColumnFamilyDescriptors based on the specified RocksDB Options file.
+   * See LoadLatestOptions above.
+   *
+   * @param optionsFileName the RocksDB options file path.
+   * @param configOptions {@link org.rocksdb.ConfigOptions} instance.
+   * @param dbOptions {@link org.rocksdb.DBOptions} instance. This will be
+   *     filled and returned.
+   * @param cfDescs A list of {@link org.rocksdb.ColumnFamilyDescriptor}'s be
+   *     returned.
+   * @throws RocksDBException thrown if error happens in underlying
+   *     native library.
+   */
+  public static void loadOptionsFromFile(ConfigOptions configOptions, String optionsFileName,
+      DBOptions dbOptions, List<ColumnFamilyDescriptor> cfDescs) throws RocksDBException {
+    loadOptionsFromFile(
+        configOptions.nativeHandle_, optionsFileName, dbOptions.nativeHandle_, cfDescs);
+  }
+
   /**
    * Returns the latest options file name under the specified RocksDB path.
    *
@@ -134,9 +173,13 @@ public class OptionsUtil {
   // native methods
   private native static void loadLatestOptions(String dbPath, long envHandle, long dbOptionsHandle,
       List<ColumnFamilyDescriptor> cfDescs, boolean ignoreUnknownOptions) throws RocksDBException;
+  private native static void loadLatestOptions(long cfgHandle, String dbPath, long dbOptionsHandle,
+      List<ColumnFamilyDescriptor> cfDescs) throws RocksDBException;
   private native static void loadOptionsFromFile(String optionsFileName, long envHandle,
       long dbOptionsHandle, List<ColumnFamilyDescriptor> cfDescs, boolean ignoreUnknownOptions)
       throws RocksDBException;
+  private native static void loadOptionsFromFile(long cfgHandle, String optionsFileName,
+      long dbOptionsHandle, List<ColumnFamilyDescriptor> cfDescs) throws RocksDBException;
   private native static String getLatestOptionsFileName(String dbPath, long envHandle)
       throws RocksDBException;
 }

java/src/main/java/org/rocksdb/SanityLevel.java

@@ -0,0 +1,42 @@
+package org.rocksdb;
+
+public enum SanityLevel {
+  NONE((byte) 0x0),
+  LOOSELY_COMPATIBLE((byte) 0x1),
+  EXACT_MATCH((byte) 0xFF);
+
+  private final byte value;
+
+  SanityLevel(final byte value) {
+    this.value = value;
+  }
+
+  /**
+   * Get the internal representation value.
+   *
+   * @return the internal representation value.
+   */
+  // TODO(AR) should be made package-private
+  public byte getValue() {
+    return value;
+  }
+
+  /**
+   * Get the SanityLevel from the internal representation value.
+   *
+   * @param value the internal representation value.
+   *
+   * @return the SanityLevel
+   *
+   * @throws IllegalArgumentException if the value does not match a
+   *     SanityLevel
+   */
+  static SanityLevel fromValue(final byte value) throws IllegalArgumentException {
+    for (final SanityLevel level : SanityLevel.values()) {
+      if (level.value == value) {
+        return level;
+      }
+    }
+    throw new IllegalArgumentException("Unknown value for SanityLevel: " + value);
+  }
+}

java/src/test/java/org/rocksdb/ColumnFamilyOptionsTest.java

@@ -54,6 +54,27 @@ public class ColumnFamilyOptionsTest {
     }
   }
 
+  @Test
+  public void getColumnFamilyOptionsFromPropsWithIgnoreIllegalValue() {
+    // setup sample properties
+    final Properties properties = new Properties();
+    properties.put("tomato", "1024");
+    properties.put("burger", "2");
+    properties.put("write_buffer_size", "112");
+    properties.put("max_write_buffer_number", "13");
+
+    try (final ConfigOptions cfgOpts = new ConfigOptions().setIgnoreUnknownOptions(true);
+         final ColumnFamilyOptions opt =
+             ColumnFamilyOptions.getColumnFamilyOptionsFromProps(cfgOpts, properties)) {
+      // setup sample properties
+      assertThat(opt).isNotNull();
+      assertThat(String.valueOf(opt.writeBufferSize()))
+          .isEqualTo(properties.get("write_buffer_size"));
+      assertThat(String.valueOf(opt.maxWriteBufferNumber()))
+          .isEqualTo(properties.get("max_write_buffer_number"));
+    }
+  }
+
   @Test
   public void failColumnFamilyOptionsFromPropsWithIllegalValue() {
     // setup sample properties

options/cf_options.cc

@@ -347,12 +347,13 @@ std::unordered_map<std::string, OptionTypeInfo>
           OptionTypeFlags::kMutable,
           offsetof(struct MutableCFOptions, sample_for_compression)}}};
 
-Status ParseColumnFamilyOption(const std::string& name,
+Status ParseColumnFamilyOption(const ConfigOptions& config_options,
+                               const std::string& name,
                                const std::string& org_value,
-                               ColumnFamilyOptions* new_options,
+                               ColumnFamilyOptions* new_options) {
-                               bool input_strings_escaped) {
+  const std::string& value = config_options.input_strings_escaped
-  const std::string& value =
+                                 ? UnescapeOptionString(org_value)
-      input_strings_escaped ? UnescapeOptionString(org_value) : org_value;
+                                 : org_value;
   try {
     if (name == "block_based_table_factory") {
       // Nested options

options/options_helper.cc

@@ -10,6 +10,7 @@
 #include <unordered_set>
 #include <vector>
 
+#include "options/options_type.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/compaction_filter.h"
 #include "rocksdb/convenience.h"
@@ -353,10 +354,14 @@ bool FIFOCompactionOptionsSpecialCase(const std::string& opt_str,
 }
 
 static bool SerializeStruct(
-    const void* const options, std::string* value,
+    const void* const opt_ptr, std::string* value,
     const std::unordered_map<std::string, OptionTypeInfo>& type_info_map) {
+  ConfigOptions config_options;
+  config_options.delimiter = ";";
+
   std::string opt_str;
-  Status s = GetStringFromStruct(&opt_str, options, type_info_map, ";");
+  Status s =
+      GetStringFromStruct(config_options, opt_ptr, type_info_map, &opt_str);
   if (!s.ok()) {
     return false;
   }
@@ -1010,9 +1015,9 @@ Status StringToMap(const std::string& opts_str,
 }
 
 Status GetStringFromStruct(
-    std::string* opt_string, const void* const options,
+    const ConfigOptions& cfg_options, const void* const opt_ptr,
     const std::unordered_map<std::string, OptionTypeInfo>& type_info,
-    const std::string& delimiter) {
+    std::string* opt_string) {
   assert(opt_string);
   opt_string->clear();
   for (const auto iter : type_info) {
@@ -1023,12 +1028,12 @@ Status GetStringFromStruct(
       continue;
     }
     const char* opt_address =
-        reinterpret_cast<const char*>(options) + opt_info.offset;
+        reinterpret_cast<const char*>(opt_ptr) + opt_info.offset;
     std::string value;
     bool result =
         SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
     if (result) {
-      opt_string->append(iter.first + "=" + value + delimiter);
+      opt_string->append(iter.first + "=" + value + cfg_options.delimiter);
     } else {
       return Status::InvalidArgument("failed to serialize %s\n",
                                      iter.first.c_str());
@@ -1040,15 +1045,32 @@ Status GetStringFromStruct(
 Status GetStringFromDBOptions(std::string* opt_string,
                               const DBOptions& db_options,
                               const std::string& delimiter) {
-  return GetStringFromStruct(opt_string, &db_options, db_options_type_info,
+  ConfigOptions config_options;
-                             delimiter);
+  config_options.delimiter = delimiter;
+  return GetStringFromDBOptions(config_options, db_options, opt_string);
+}
+
+Status GetStringFromDBOptions(const ConfigOptions& cfg_options,
+                              const DBOptions& db_options,
+                              std::string* opt_string) {
+  return GetStringFromStruct(cfg_options, &db_options, db_options_type_info,
+                             opt_string);
 }
 
 Status GetStringFromColumnFamilyOptions(std::string* opt_string,
                                         const ColumnFamilyOptions& cf_options,
                                         const std::string& delimiter) {
-  return GetStringFromStruct(opt_string, &cf_options, cf_options_type_info,
+  ConfigOptions config_options;
-                             delimiter);
+  config_options.delimiter = delimiter;
+  return GetStringFromColumnFamilyOptions(config_options, cf_options,
+                                          opt_string);
+}
+
+Status GetStringFromColumnFamilyOptions(const ConfigOptions& config_options,
+                                        const ColumnFamilyOptions& cf_options,
+                                        std::string* opt_string) {
+  return GetStringFromStruct(config_options, &cf_options, cf_options_type_info,
+                             opt_string);
 }
 
 Status GetStringFromCompressionType(std::string* compression_str,
@@ -1073,12 +1095,13 @@ std::vector<CompressionType> GetSupportedCompressions() {
   return supported_compressions;
 }
 
-Status ParseDBOption(const std::string& name,
+static Status ParseDBOption(const ConfigOptions& config_options,
+                            const std::string& name,
                             const std::string& org_value,
-                     DBOptions* new_options,
+                            DBOptions* new_options) {
-                     bool input_strings_escaped = false) {
+  const std::string& value = config_options.input_strings_escaped
-  const std::string& value =
+                                 ? UnescapeOptionString(org_value)
-      input_strings_escaped ? UnescapeOptionString(org_value) : org_value;
+                                 : org_value;
   try {
     if (name == "rate_limiter_bytes_per_sec") {
       new_options->rate_limiter.reset(
@@ -1121,37 +1144,28 @@ Status GetColumnFamilyOptionsFromMap(
     const std::unordered_map<std::string, std::string>& opts_map,
     ColumnFamilyOptions* new_options, bool input_strings_escaped,
     bool ignore_unknown_options) {
-  return GetColumnFamilyOptionsFromMapInternal(
+  ConfigOptions config_options;
-      base_options, opts_map, new_options, input_strings_escaped, nullptr,
+  config_options.ignore_unknown_options = ignore_unknown_options;
-      ignore_unknown_options);
+  config_options.input_strings_escaped = input_strings_escaped;
+  return GetColumnFamilyOptionsFromMap(config_options, base_options, opts_map,
+                                       new_options);
 }
 
-Status GetColumnFamilyOptionsFromMapInternal(
+Status GetColumnFamilyOptionsFromMap(
+    const ConfigOptions& config_options,
     const ColumnFamilyOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    ColumnFamilyOptions* new_options, bool input_strings_escaped,
+    ColumnFamilyOptions* new_options) {
-    std::vector<std::string>* unsupported_options_names,
-    bool ignore_unknown_options) {
   assert(new_options);
   *new_options = base_options;
-  if (unsupported_options_names) {
-    unsupported_options_names->clear();
-  }
   for (const auto& o : opts_map) {
-    auto s = ParseColumnFamilyOption(o.first, o.second, new_options,
+    auto s =
-                                 input_strings_escaped);
+        ParseColumnFamilyOption(config_options, o.first, o.second, new_options);
     if (!s.ok()) {
       if (s.IsNotSupported()) {
-        // If the deserialization of the specified option is not supported
+        continue;
-        // and an output vector of unsupported_options is provided, then
+      } else if (s.IsInvalidArgument() &&
-        // we log the name of the unsupported option and proceed.
+                 config_options.ignore_unknown_options) {
-        if (unsupported_options_names != nullptr) {
-          unsupported_options_names->push_back(o.first);
-        }
-        // Note that we still return Status::OK in such case to maintain
-        // the backward compatibility in the old public API defined in
-        // rocksdb/convenience.h
-      } else if (s.IsInvalidArgument() && ignore_unknown_options) {
         continue;
       } else {
         // Restore "new_options" to the default "base_options".
@@ -1167,13 +1181,25 @@ Status GetColumnFamilyOptionsFromString(
     const ColumnFamilyOptions& base_options,
     const std::string& opts_str,
     ColumnFamilyOptions* new_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+  config_options.ignore_unknown_options = false;
+  return GetColumnFamilyOptionsFromString(config_options, base_options,
+                                          opts_str, new_options);
+}
+
+Status GetColumnFamilyOptionsFromString(const ConfigOptions& config_options,
+                                        const ColumnFamilyOptions& base_options,
+                                        const std::string& opts_str,
+                                        ColumnFamilyOptions* new_options) {
   std::unordered_map<std::string, std::string> opts_map;
   Status s = StringToMap(opts_str, &opts_map);
   if (!s.ok()) {
     *new_options = base_options;
     return s;
   }
-  return GetColumnFamilyOptionsFromMap(base_options, opts_map, new_options);
+  return GetColumnFamilyOptionsFromMap(config_options, base_options, opts_map,
+                                       new_options);
 }
 
 Status GetDBOptionsFromMap(
@@ -1181,25 +1207,33 @@ Status GetDBOptionsFromMap(
     const std::unordered_map<std::string, std::string>& opts_map,
     DBOptions* new_options, bool input_strings_escaped,
     bool ignore_unknown_options) {
-  return GetDBOptionsFromMapInternal(base_options, opts_map, new_options,
+  ConfigOptions config_options;
-                                     input_strings_escaped, nullptr,
+  config_options.input_strings_escaped = input_strings_escaped;
-                                     ignore_unknown_options);
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  return GetDBOptionsFromMap(config_options, base_options, opts_map,
+                             new_options);
+}
+
+Status GetDBOptionsFromMap(
+    const ConfigOptions& config_options, const DBOptions& base_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    DBOptions* new_options) {
+  return GetDBOptionsFromMapInternal(config_options, base_options, opts_map,
+                                     new_options, nullptr);
 }
 
 Status GetDBOptionsFromMapInternal(
-    const DBOptions& base_options,
+    const ConfigOptions& config_options, const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    DBOptions* new_options, bool input_strings_escaped,
+    DBOptions* new_options,
-    std::vector<std::string>* unsupported_options_names,
+    std::vector<std::string>* unsupported_options_names) {
-    bool ignore_unknown_options) {
   assert(new_options);
   *new_options = base_options;
   if (unsupported_options_names) {
     unsupported_options_names->clear();
   }
   for (const auto& o : opts_map) {
-    auto s = ParseDBOption(o.first, o.second,
+    auto s = ParseDBOption(config_options, o.first, o.second, new_options);
-                           new_options, input_strings_escaped);
     if (!s.ok()) {
       if (s.IsNotSupported()) {
         // If the deserialization of the specified option is not supported
@@ -1211,7 +1245,8 @@ Status GetDBOptionsFromMapInternal(
         // Note that we still return Status::OK in such case to maintain
         // the backward compatibility in the old public API defined in
         // rocksdb/convenience.h
-      } else if (s.IsInvalidArgument() && ignore_unknown_options) {
+      } else if (s.IsInvalidArgument() &&
+                 config_options.ignore_unknown_options) {
         continue;
       } else {
         // Restore "new_options" to the default "base_options".
@@ -1223,7 +1258,18 @@ Status GetDBOptionsFromMapInternal(
   return Status::OK();
 }
 
-Status GetDBOptionsFromString(
+Status GetDBOptionsFromString(const DBOptions& base_options,
+                              const std::string& opts_str,
+                              DBOptions* new_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+  config_options.ignore_unknown_options = false;
+
+  return GetDBOptionsFromString(config_options, base_options, opts_str,
+                                new_options);
+}
+
+Status GetDBOptionsFromString(const ConfigOptions& config_options,
                               const DBOptions& base_options,
                               const std::string& opts_str,
                               DBOptions* new_options) {
@@ -1233,11 +1279,23 @@ Status GetDBOptionsFromString(
     *new_options = base_options;
     return s;
   }
-  return GetDBOptionsFromMap(base_options, opts_map, new_options);
+  return GetDBOptionsFromMap(config_options, base_options, opts_map,
+                             new_options);
 }
 
 Status GetOptionsFromString(const Options& base_options,
                             const std::string& opts_str, Options* new_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+  config_options.ignore_unknown_options = false;
+
+  return GetOptionsFromString(config_options, base_options, opts_str,
+                              new_options);
+}
+
+Status GetOptionsFromString(const ConfigOptions& config_options,
+                            const Options& base_options,
+                            const std::string& opts_str, Options* new_options) {
   std::unordered_map<std::string, std::string> opts_map;
   Status s = StringToMap(opts_str, &opts_map);
   if (!s.ok()) {
@@ -1246,9 +1304,11 @@ Status GetOptionsFromString(const Options& base_options,
   DBOptions new_db_options(base_options);
   ColumnFamilyOptions new_cf_options(base_options);
   for (const auto& o : opts_map) {
-    if (ParseDBOption(o.first, o.second, &new_db_options).ok()) {
+    if (ParseDBOption(config_options, o.first, o.second, &new_db_options)
-    } else if (ParseColumnFamilyOption(
+            .ok()) {
-        o.first, o.second, &new_cf_options).ok()) {
+    } else if (ParseColumnFamilyOption(config_options, o.first, o.second,
+                                       &new_cf_options)
+                   .ok()) {
     } else {
       return Status::InvalidArgument("Can't parse option " + o.first);
     }
@@ -1261,13 +1321,22 @@ Status GetTableFactoryFromMap(
     const std::string& factory_name,
     const std::unordered_map<std::string, std::string>& opt_map,
     std::shared_ptr<TableFactory>* table_factory, bool ignore_unknown_options) {
+  ConfigOptions
+      config_options;  // Use default for escaped(true) and check (exact)
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  return GetTableFactoryFromMap(config_options, factory_name, opt_map,
+                                table_factory);
+}
+
+Status GetTableFactoryFromMap(
+    const ConfigOptions& config_options, const std::string& factory_name,
+    const std::unordered_map<std::string, std::string>& opt_map,
+    std::shared_ptr<TableFactory>* table_factory) {
   Status s;
   if (factory_name == BlockBasedTableFactory().Name()) {
     BlockBasedTableOptions bbt_opt;
-    s = GetBlockBasedTableOptionsFromMap(BlockBasedTableOptions(), opt_map,
+    s = GetBlockBasedTableOptionsFromMap(
-                                         &bbt_opt,
+        config_options, BlockBasedTableOptions(), opt_map, &bbt_opt);
-                                         true, /* input_strings_escaped */
-                                         ignore_unknown_options);
     if (!s.ok()) {
       return s;
     }
@@ -1275,9 +1344,8 @@ Status GetTableFactoryFromMap(
     return Status::OK();
   } else if (factory_name == PlainTableFactory().Name()) {
     PlainTableOptions pt_opt;
-    s = GetPlainTableOptionsFromMap(PlainTableOptions(), opt_map, &pt_opt,
+    s = GetPlainTableOptionsFromMap(config_options, PlainTableOptions(),
-                                    true, /* input_strings_escaped */
+                                    opt_map, &pt_opt);
-                                    ignore_unknown_options);
     if (!s.ok()) {
       return s;
     }

options/options_helper.h

@@ -19,6 +19,7 @@
 #include "rocksdb/universal_compaction.h"
 
 namespace ROCKSDB_NAMESPACE {
+struct ConfigOptions;
 
 DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
                          const MutableDBOptions& mutable_db_options);
@@ -28,11 +29,15 @@ ColumnFamilyOptions BuildColumnFamilyOptions(
     const MutableCFOptions& mutable_cf_options);
 
 #ifndef ROCKSDB_LITE
+Status GetStringFromStruct(
+    const ConfigOptions& config_options, const void* const opt_ptr,
+    const std::unordered_map<std::string, OptionTypeInfo>& type_info,
+    std::string* opt_string);
 
-Status ParseColumnFamilyOption(const std::string& name,
+Status ParseColumnFamilyOption(const ConfigOptions& config_options,
+                               const std::string& name,
                                const std::string& org_value,
-                               ColumnFamilyOptions* new_options,
+                               ColumnFamilyOptions* new_options);
-                               bool input_strings_escaped = false);
 
 Status GetMutableOptionsFromStrings(
     const MutableCFOptions& base_options,
@@ -54,6 +59,11 @@ 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,
+    std::shared_ptr<TableFactory>* table_factory);
+
 // A helper function that converts "opt_address" to a std::string
 // based on the specified OptionType.
 bool SerializeSingleOptionHelper(const char* opt_address,
@@ -63,21 +73,10 @@ bool SerializeSingleOptionHelper(const char* opt_address,
 // this further takes an optional output vector "unsupported_options_names",
 // which stores the name of all the unsupported options specified in "opts_map".
 Status GetDBOptionsFromMapInternal(
-    const DBOptions& base_options,
+    const ConfigOptions& config_options, const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    DBOptions* new_options, bool input_strings_escaped,
+    DBOptions* new_options,
-    std::vector<std::string>* unsupported_options_names = nullptr,
+    std::vector<std::string>* unsupported_options_names = nullptr);
-    bool ignore_unknown_options = false);
-
-// In addition to its public version defined in rocksdb/convenience.h,
-// this further takes an optional output vector "unsupported_options_names",
-// which stores the name of all the unsupported options specified in "opts_map".
-Status GetColumnFamilyOptionsFromMapInternal(
-    const ColumnFamilyOptions& base_options,
-    const std::unordered_map<std::string, std::string>& opts_map,
-    ColumnFamilyOptions* new_options, bool input_strings_escaped,
-    std::vector<std::string>* unsupported_options_names = nullptr,
-    bool ignore_unknown_options = false);
 
 bool ParseSliceTransform(
     const std::string& value,
@@ -89,10 +88,6 @@ extern Status StringToMap(
 
 extern bool ParseOptionHelper(char* opt_address, const OptionType& opt_type,
                               const std::string& value);
-Status GetStringFromStruct(
-    std::string* opt_string, const void* const options,
-    const std::unordered_map<std::string, OptionTypeInfo>& type_info,
-    const std::string& delimiter);
 #endif  // !ROCKSDB_LITE
 
 struct OptionsHelper {

options/options_parser.cc

@@ -16,14 +16,15 @@
 #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"
+#include "table/block_based/block_based_table_factory.h"
 #include "test_util/sync_point.h"
 #include "util/cast_util.h"
 #include "util/string_util.h"
 
-#include "port/port.h"
-
 namespace ROCKSDB_NAMESPACE {
 
 static const std::string option_file_header =
@@ -38,6 +39,25 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
                              const std::vector<std::string>& cf_names,
                              const std::vector<ColumnFamilyOptions>& cf_opts,
                              const std::string& file_name, FileSystem* fs) {
+  ConfigOptions
+      config_options;  // Use default for escaped(true) and check (exact)
+  config_options.delimiter = "\n  ";
+  // If a readahead size was set in the input options, use it
+  if (db_opt.log_readahead_size > 0) {
+    config_options.file_readahead_size = db_opt.log_readahead_size;
+  }
+  return PersistRocksDBOptions(config_options, db_opt, cf_names, cf_opts,
+                               file_name, fs);
+}
+
+Status PersistRocksDBOptions(const ConfigOptions& config_options_in,
+                             const DBOptions& db_opt,
+                             const std::vector<std::string>& cf_names,
+                             const std::vector<ColumnFamilyOptions>& cf_opts,
+                             const std::string& file_name, FileSystem* fs) {
+  ConfigOptions config_options = config_options_in;
+  config_options.delimiter = "\n  ";  // Override the default to nl
+
   TEST_SYNC_POINT("PersistRocksDBOptions:start");
   if (cf_names.size() != cf_opts.size()) {
     return Status::InvalidArgument(
@@ -68,7 +88,7 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
   writable->Append("\n[" + opt_section_titles[kOptionSectionDBOptions] +
                    "]\n  ");
 
-  s = GetStringFromDBOptions(&options_file_content, db_opt, "\n  ");
+  s = GetStringFromDBOptions(config_options, db_opt, &options_file_content);
   if (!s.ok()) {
     writable->Close();
     return s;
@@ -79,8 +99,8 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
     // CFOptions section
     writable->Append("\n[" + opt_section_titles[kOptionSectionCFOptions] +
                      " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
-    s = GetStringFromColumnFamilyOptions(&options_file_content, cf_opts[i],
+    s = GetStringFromColumnFamilyOptions(config_options, cf_opts[i],
-                                         "\n  ");
+                                         &options_file_content);
     if (!s.ok()) {
       writable->Close();
       return s;
@@ -93,7 +113,7 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
                        tf->Name() + " \"" + EscapeOptionString(cf_names[i]) +
                        "\"]\n  ");
       options_file_content.clear();
-      s = tf->GetOptionString(&options_file_content, "\n  ");
+      s = tf->GetOptionString(config_options, &options_file_content);
       if (!s.ok()) {
         return s;
       }
@@ -104,7 +124,7 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
   writable->Close();
 
   return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
-      db_opt, cf_names, cf_opts, file_name, fs);
+      config_options, db_opt, cf_names, cf_opts, file_name, fs);
 }
 
 RocksDBOptionsParser::RocksDBOptionsParser() { Reset(); }
@@ -205,7 +225,20 @@ Status RocksDBOptionsParser::ParseStatement(std::string* name,
 Status RocksDBOptionsParser::Parse(const std::string& file_name, FileSystem* fs,
                                    bool ignore_unknown_options,
                                    size_t file_readahead_size) {
+  ConfigOptions
+      config_options;  // Use default for escaped(true) and check (exact)
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  if (file_readahead_size > 0) {
+    config_options.file_readahead_size = file_readahead_size;
+  }
+  return Parse(config_options, file_name, fs);
+}
+
+Status RocksDBOptionsParser::Parse(const ConfigOptions& config_options_in,
+                                   const std::string& file_name,
+                                   FileSystem* fs) {
   Reset();
+  ConfigOptions config_options = config_options_in;
 
   std::unique_ptr<FSSequentialFile> seq_file;
   Status s = fs->NewSequentialFile(file_name, FileOptions(), &seq_file,
@@ -213,9 +246,8 @@ Status RocksDBOptionsParser::Parse(const std::string& file_name, FileSystem* fs,
   if (!s.ok()) {
     return s;
   }
-
   SequentialFileReader sf_reader(std::move(seq_file), file_name,
-                                 file_readahead_size);
+                                 config_options.file_readahead_size);
 
   OptionSection section = kOptionSectionUnknown;
   std::string title;
@@ -235,7 +267,7 @@ Status RocksDBOptionsParser::Parse(const std::string& file_name, FileSystem* fs,
       continue;
     }
     if (IsSection(line)) {
-      s = EndSection(section, title, argument, opt_map, ignore_unknown_options);
+      s = EndSection(config_options, section, title, argument, opt_map);
       opt_map.clear();
       if (!s.ok()) {
         return s;
@@ -243,10 +275,11 @@ Status RocksDBOptionsParser::Parse(const std::string& file_name, FileSystem* fs,
 
       // If the option file is not generated by a higher minor version,
       // there shouldn't be any unknown option.
-      if (ignore_unknown_options && section == kOptionSectionVersion) {
+      if (config_options.ignore_unknown_options &&
+          section == kOptionSectionVersion) {
         if (db_version[0] < ROCKSDB_MAJOR || (db_version[0] == ROCKSDB_MAJOR &&
                                               db_version[1] <= ROCKSDB_MINOR)) {
-          ignore_unknown_options = false;
+          config_options.ignore_unknown_options = false;
         }
       }
 
@@ -265,7 +298,7 @@ Status RocksDBOptionsParser::Parse(const std::string& file_name, FileSystem* fs,
     }
   }
 
-  s = EndSection(section, title, argument, opt_map, ignore_unknown_options);
+  s = EndSection(config_options, section, title, argument, opt_map);
   opt_map.clear();
   if (!s.ok()) {
     return s;
@@ -372,14 +405,12 @@ Status RocksDBOptionsParser::ParseVersionNumber(const std::string& ver_name,
 }
 
 Status RocksDBOptionsParser::EndSection(
-    const OptionSection section, const std::string& section_title,
+    const ConfigOptions& config_options, const OptionSection section,
-    const std::string& section_arg,
+    const std::string& section_title, const std::string& section_arg,
-    const std::unordered_map<std::string, std::string>& opt_map,
+    const std::unordered_map<std::string, std::string>& opt_map) {
-    bool ignore_unknown_options) {
   Status s;
   if (section == kOptionSectionDBOptions) {
-    s = GetDBOptionsFromMap(DBOptions(), opt_map, &db_opt_, true,
+    s = GetDBOptionsFromMap(config_options, DBOptions(), opt_map, &db_opt_);
-                            ignore_unknown_options);
     if (!s.ok()) {
       return s;
     }
@@ -390,9 +421,8 @@ Status RocksDBOptionsParser::EndSection(
     assert(GetCFOptions(section_arg) == nullptr);
     cf_names_.emplace_back(section_arg);
     cf_opts_.emplace_back();
-    s = GetColumnFamilyOptionsFromMap(ColumnFamilyOptions(), opt_map,
+    s = GetColumnFamilyOptionsFromMap(config_options, ColumnFamilyOptions(),
-                                      &cf_opts_.back(), true,
+                                      opt_map, &cf_opts_.back());
-                                      ignore_unknown_options);
     if (!s.ok()) {
       return s;
     }
@@ -409,9 +439,10 @@ Status RocksDBOptionsParser::EndSection(
     }
     // Ignore error as table factory deserialization is optional
     s = GetTableFactoryFromMap(
+        config_options,
         section_title.substr(
             opt_section_titles[kOptionSectionTableOptions].size()),
-        opt_map, &(cf_opt->table_factory), ignore_unknown_options);
+        opt_map, &(cf_opt->table_factory));
     if (!s.ok()) {
       return s;
     }
@@ -652,36 +683,27 @@ bool AreEqualOptions(
 }
 
 Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
-    const DBOptions& db_opt, const std::vector<std::string>& cf_names,
+    const ConfigOptions& config_options, const DBOptions& db_opt,
+    const std::vector<std::string>& cf_names,
     const std::vector<ColumnFamilyOptions>& cf_opts,
-    const std::string& file_name, FileSystem* fs,
+    const std::string& file_name, FileSystem* fs) {
-    OptionsSanityCheckLevel sanity_check_level, bool ignore_unknown_options) {
-  // We infer option file readhead size from log readahead size.
-  // If it is not given, use 512KB.
-  size_t file_readahead_size = db_opt.log_readahead_size;
-  if (file_readahead_size == 0) {
-    const size_t kDefaultOptionFileReadAheadSize = 512 * 1024;
-    file_readahead_size = kDefaultOptionFileReadAheadSize;
-  }
-
   RocksDBOptionsParser parser;
-  Status s =
+  Status s = parser.Parse(config_options, file_name, fs);
-      parser.Parse(file_name, fs, ignore_unknown_options, file_readahead_size);
   if (!s.ok()) {
     return s;
   }
 
   // Verify DBOptions
-  s = VerifyDBOptions(db_opt, *parser.db_opt(), parser.db_opt_map(),
+  s = VerifyDBOptions(config_options, db_opt, *parser.db_opt(),
-                      sanity_check_level);
+                      parser.db_opt_map());
   if (!s.ok()) {
     return s;
   }
 
   // Verify ColumnFamily Name
   if (cf_names.size() != parser.cf_names()->size()) {
-    if (sanity_check_level >=
+    if (config_options.sanity_level >=
-        OptionsSanityCheckLevel::kSanityLevelLooselyCompatible) {
+        ConfigOptions::kSanityLevelLooselyCompatible) {
       return Status::InvalidArgument(
           "[RocksDBOptionParser Error] The persisted options does not have "
           "the same number of column family names as the db instance.");
@@ -703,8 +725,8 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
 
   // Verify Column Family Options
   if (cf_opts.size() != parser.cf_opts()->size()) {
-    if (sanity_check_level >=
+    if (config_options.sanity_level >=
-        OptionsSanityCheckLevel::kSanityLevelLooselyCompatible) {
+        ConfigOptions::kSanityLevelLooselyCompatible) {
       return Status::InvalidArgument(
           "[RocksDBOptionsParser Error]",
           "The persisted options does not have the same number of "
@@ -717,14 +739,13 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
     }
   }
   for (size_t i = 0; i < cf_opts.size(); ++i) {
-    s = VerifyCFOptions(cf_opts[i], parser.cf_opts()->at(i),
+    s = VerifyCFOptions(config_options, cf_opts[i], parser.cf_opts()->at(i),
-                        &(parser.cf_opt_maps()->at(i)), sanity_check_level);
+                        &(parser.cf_opt_maps()->at(i)));
     if (!s.ok()) {
       return s;
     }
-    s = VerifyTableFactory(cf_opts[i].table_factory.get(),
+    s = VerifyTableFactory(config_options, cf_opts[i].table_factory.get(),
-                           parser.cf_opts()->at(i).table_factory.get(),
+                           parser.cf_opts()->at(i).table_factory.get());
-                           sanity_check_level);
     if (!s.ok()) {
       return s;
     }
@@ -734,16 +755,16 @@ Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
 }
 
 Status RocksDBOptionsParser::VerifyDBOptions(
-    const DBOptions& base_opt, const DBOptions& persisted_opt,
+    const ConfigOptions& config_options, const DBOptions& base_opt,
-    const std::unordered_map<std::string, std::string>* /*opt_map*/,
+    const DBOptions& persisted_opt,
-    OptionsSanityCheckLevel sanity_check_level) {
+    const std::unordered_map<std::string, std::string>* /*opt_map*/) {
   for (const auto& pair : db_options_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 (DBOptionSanityCheckLevel(pair.first) <= sanity_check_level) {
+    if (DBOptionSanityCheckLevel(pair.first) <= config_options.sanity_level) {
       if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
                            reinterpret_cast<const char*>(&persisted_opt),
                            pair.second, pair.first, nullptr)) {
@@ -771,17 +792,16 @@ Status RocksDBOptionsParser::VerifyDBOptions(
 }
 
 Status RocksDBOptionsParser::VerifyCFOptions(
-    const ColumnFamilyOptions& base_opt,
+    const ConfigOptions& config_options, const ColumnFamilyOptions& base_opt,
     const ColumnFamilyOptions& persisted_opt,
-    const std::unordered_map<std::string, std::string>* persisted_opt_map,
+    const std::unordered_map<std::string, std::string>* persisted_opt_map) {
-    OptionsSanityCheckLevel sanity_check_level) {
   for (const auto& pair : cf_options_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 (CFOptionSanityCheckLevel(pair.first) <= sanity_check_level) {
+    if (CFOptionSanityCheckLevel(pair.first) <= config_options.sanity_level) {
       if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
                            reinterpret_cast<const char*>(&persisted_opt),
                            pair.second, pair.first, persisted_opt_map)) {
@@ -809,10 +829,10 @@ Status RocksDBOptionsParser::VerifyCFOptions(
 }
 
 Status RocksDBOptionsParser::VerifyTableFactory(
-    const TableFactory* base_tf, const TableFactory* file_tf,
+    const ConfigOptions& config_options, const TableFactory* base_tf,
-    OptionsSanityCheckLevel sanity_check_level) {
+    const TableFactory* file_tf) {
   if (base_tf && file_tf) {
-    if (sanity_check_level > OptionsSanityCheckLevel::kSanityLevelNone &&
+    if (config_options.sanity_level > ConfigOptions::kSanityLevelNone &&
         std::string(base_tf->Name()) != std::string(file_tf->Name())) {
       return Status::Corruption(
           "[RocksDBOptionsParser]: "
@@ -820,11 +840,11 @@ Status RocksDBOptionsParser::VerifyTableFactory(
     }
     if (base_tf->Name() == BlockBasedTableFactory::kName) {
       return VerifyBlockBasedTableFactory(
+          config_options,
           static_cast_with_check<const BlockBasedTableFactory,
                                  const TableFactory>(base_tf),
           static_cast_with_check<const BlockBasedTableFactory,
-                                 const TableFactory>(file_tf),
+                                 const TableFactory>(file_tf));
-          sanity_check_level);
     }
     // TODO(yhchiang): add checks for other table factory types
   } else {

options/options_parser.h

@@ -9,14 +9,15 @@
 #include <string>
 #include <vector>
 
-#include "options/options_sanity_check.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
-#include "table/block_based/block_based_table_factory.h"
 
 namespace ROCKSDB_NAMESPACE {
 
 #ifndef ROCKSDB_LITE
+struct ConfigOptions;
+class OptionTypeInfo;
+class TableFactory;
 
 #define ROCKSDB_OPTION_FILE_MAJOR 1
 #define ROCKSDB_OPTION_FILE_MINOR 1
@@ -36,6 +37,11 @@ Status PersistRocksDBOptions(const DBOptions& db_opt,
                              const std::vector<std::string>& cf_names,
                              const std::vector<ColumnFamilyOptions>& cf_opts,
                              const std::string& file_name, FileSystem* fs);
+Status PersistRocksDBOptions(const ConfigOptions& config_options,
+                             const DBOptions& db_opt,
+                             const std::vector<std::string>& cf_names,
+                             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,
@@ -52,6 +58,10 @@ class RocksDBOptionsParser {
   // If 0 is given, a default value will be used.
   Status Parse(const std::string& file_name, FileSystem* fs,
                bool ignore_unknown_options, size_t file_readahead_size);
+
+  Status Parse(const ConfigOptions& config_options,
+               const std::string& file_name, FileSystem* fs);
+
   static std::string TrimAndRemoveComment(const std::string& line,
                                           const bool trim_only = false);
 
@@ -70,27 +80,26 @@ class RocksDBOptionsParser {
     return GetCFOptionsImpl(name);
   }
   size_t NumColumnFamilies() { return cf_opts_.size(); }
-
   static Status VerifyRocksDBOptionsFromFile(
-      const DBOptions& db_opt, const std::vector<std::string>& cf_names,
+      const ConfigOptions& config_options, const DBOptions& db_opt,
+      const std::vector<std::string>& cf_names,
       const std::vector<ColumnFamilyOptions>& cf_opts,
-      const std::string& file_name, FileSystem* fs,
+      const std::string& file_name, FileSystem* fs);
-      OptionsSanityCheckLevel sanity_check_level = kSanityLevelExactMatch,
-      bool ignore_unknown_options = false);
-
   static Status VerifyDBOptions(
-      const DBOptions& base_opt, const DBOptions& new_opt,
+      const ConfigOptions& config_options, const DBOptions& base_opt,
-      const std::unordered_map<std::string, std::string>* new_opt_map = nullptr,
+      const DBOptions& new_opt,
-      OptionsSanityCheckLevel sanity_check_level = kSanityLevelExactMatch);
+      const std::unordered_map<std::string, std::string>* new_opt_map =
+          nullptr);
 
   static Status VerifyCFOptions(
-      const ColumnFamilyOptions& base_opt, const ColumnFamilyOptions& new_opt,
+      const ConfigOptions& config_options, const ColumnFamilyOptions& base_opt,
-      const std::unordered_map<std::string, std::string>* new_opt_map = nullptr,
+      const ColumnFamilyOptions& new_opt,
-      OptionsSanityCheckLevel sanity_check_level = kSanityLevelExactMatch);
+      const std::unordered_map<std::string, std::string>* new_opt_map =
+          nullptr);
 
-  static Status VerifyTableFactory(
+  static Status VerifyTableFactory(const ConfigOptions& config_options,
-      const TableFactory* base_tf, const TableFactory* file_tf,
+                                   const TableFactory* base_tf,
-      OptionsSanityCheckLevel sanity_check_level = kSanityLevelExactMatch);
+                                   const TableFactory* file_tf);
 
   static Status ExtraParserCheck(const RocksDBOptionsParser& input_parser);
 
@@ -106,10 +115,10 @@ class RocksDBOptionsParser {
   Status ParseStatement(std::string* name, std::string* value,
                         const std::string& line, const int line_num);
 
-  Status EndSection(const OptionSection section, const std::string& title,
+  Status EndSection(
-                    const std::string& section_arg,
+      const ConfigOptions& config_options, const OptionSection section,
-                    const std::unordered_map<std::string, std::string>& opt_map,
+      const std::string& title, const std::string& section_arg,
-                    bool ignore_unknown_options);
+      const std::unordered_map<std::string, std::string>& opt_map);
 
   Status ValidityCheck();
 

options/options_sanity_check.cc

@@ -10,25 +10,26 @@
 namespace ROCKSDB_NAMESPACE {
 
 namespace {
-OptionsSanityCheckLevel SanityCheckLevelHelper(
+ConfigOptions::SanityLevel SanityCheckLevelHelper(
-    const std::unordered_map<std::string, OptionsSanityCheckLevel>& smap,
+    const std::unordered_map<std::string, ConfigOptions::SanityLevel>& smap,
     const std::string& name) {
   auto iter = smap.find(name);
-  return iter != smap.end() ? iter->second : kSanityLevelExactMatch;
+  return iter != smap.end() ? iter->second
+                            : ConfigOptions::kSanityLevelExactMatch;
 }
 }
 
-OptionsSanityCheckLevel DBOptionSanityCheckLevel(
+ConfigOptions::SanityLevel DBOptionSanityCheckLevel(
     const std::string& option_name) {
   return SanityCheckLevelHelper(sanity_level_db_options, option_name);
 }
 
-OptionsSanityCheckLevel CFOptionSanityCheckLevel(
+ConfigOptions::SanityLevel CFOptionSanityCheckLevel(
     const std::string& option_name) {
   return SanityCheckLevelHelper(sanity_level_cf_options, option_name);
 }
 
-OptionsSanityCheckLevel BBTOptionSanityCheckLevel(
+ConfigOptions::SanityLevel BBTOptionSanityCheckLevel(
     const std::string& option_name) {
   return SanityCheckLevelHelper(sanity_level_bbt_options, option_name);
 }

options/options_sanity_check.h

@@ -8,41 +8,36 @@
 #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.
-enum OptionsSanityCheckLevel : unsigned char {
-  // Performs no sanity check at all.
-  kSanityLevelNone = 0x00,
-  // Performs minimum check to ensure the RocksDB instance can be
-  // opened without corrupting / mis-interpreting the data.
-  kSanityLevelLooselyCompatible = 0x01,
-  // Perform exact match sanity check.
-  kSanityLevelExactMatch = 0xFF,
-};
 
 // The sanity check level for DB options
-static const std::unordered_map<std::string, OptionsSanityCheckLevel>
+static const std::unordered_map<std::string, ConfigOptions::SanityLevel>
-    sanity_level_db_options {};
+    sanity_level_db_options{};
 
 // The sanity check level for column-family options
-static const std::unordered_map<std::string, OptionsSanityCheckLevel>
+static const std::unordered_map<std::string, ConfigOptions::SanityLevel>
     sanity_level_cf_options = {
-        {"comparator", kSanityLevelLooselyCompatible},
+        {"comparator",
-        {"table_factory", kSanityLevelLooselyCompatible},
+         ConfigOptions::SanityLevel::kSanityLevelLooselyCompatible},
-        {"merge_operator", 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, OptionsSanityCheckLevel>
+static const std::unordered_map<std::string, ConfigOptions::SanityLevel>
-    sanity_level_bbt_options {};
+    sanity_level_bbt_options{};
 
-OptionsSanityCheckLevel DBOptionSanityCheckLevel(
+ConfigOptions::SanityLevel DBOptionSanityCheckLevel(
     const std::string& options_name);
-OptionsSanityCheckLevel CFOptionSanityCheckLevel(
+ConfigOptions::SanityLevel CFOptionSanityCheckLevel(
     const std::string& options_name);
-OptionsSanityCheckLevel BBTOptionSanityCheckLevel(
+ConfigOptions::SanityLevel BBTOptionSanityCheckLevel(
     const std::string& options_name);
 
 }  // namespace ROCKSDB_NAMESPACE

src.mk

@@ -496,6 +496,7 @@ JNI_NATIVE_SOURCES =                                          \
   java/rocksjni/comparator.cc                                 \
   java/rocksjni/comparatorjnicallback.cc                      \
   java/rocksjni/compression_options.cc                        \
+  java/rocksjni/config_options.cc                             \
   java/rocksjni/env.cc                                        \
   java/rocksjni/env_options.cc                                \
   java/rocksjni/ingest_external_file_options.cc               \

table/block_based/block_based_table_factory.cc

@@ -7,19 +7,22 @@
 // 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 <stdint.h>
+#include "table/block_based/block_based_table_factory.h"
-#include <cinttypes>
 
+#include <stdint.h>
+
+#include <cinttypes>
 #include <memory>
 #include <string>
 
 #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"
 #include "rocksdb/flush_block_policy.h"
 #include "table/block_based/block_based_table_builder.h"
-#include "table/block_based/block_based_table_factory.h"
 #include "table/block_based/block_based_table_reader.h"
 #include "table/format.h"
 #include "util/mutexlock.h"
@@ -536,17 +539,16 @@ std::string BlockBasedTableFactory::GetPrintableTableOptions() const {
 }
 
 #ifndef ROCKSDB_LITE
-
 Status BlockBasedTableFactory::GetOptionString(
-    std::string* opt_string, const std::string& delimiter) const {
+    const ConfigOptions& config_options, std::string* opt_string) const {
   assert(opt_string);
   opt_string->clear();
-  return GetStringFromStruct(opt_string, &table_options_,
+  return GetStringFromStruct(config_options, &table_options_,
-                             block_based_table_type_info, delimiter);
+                             block_based_table_type_info, opt_string);
 }
 #else
 Status BlockBasedTableFactory::GetOptionString(
-    std::string* /*opt_string*/, const std::string& /*delimiter*/) const {
+    const ConfigOptions& /*opts*/, std::string* /*opt_string*/) const {
   return Status::OK();
 }
 #endif  // !ROCKSDB_LITE
@@ -557,14 +559,14 @@ const BlockBasedTableOptions& BlockBasedTableFactory::table_options() const {
 
 #ifndef ROCKSDB_LITE
 namespace {
-std::string ParseBlockBasedTableOption(const std::string& name,
+std::string ParseBlockBasedTableOption(const ConfigOptions& config_options,
+                                       const std::string& name,
                                        const std::string& org_value,
-                                       BlockBasedTableOptions* new_options,
+                                       BlockBasedTableOptions* new_options) {
-                                       bool input_strings_escaped = false,
+  const std::string& value = config_options.input_strings_escaped
-                                       bool ignore_unknown_options = false) {
+                                 ? UnescapeOptionString(org_value)
-  const std::string& value =
+                                 : org_value;
-      input_strings_escaped ? UnescapeOptionString(org_value) : org_value;
+  if (!config_options.input_strings_escaped) {
-  if (!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") {
@@ -615,7 +617,7 @@ std::string ParseBlockBasedTableOption(const std::string& name,
   }
   const auto iter = block_based_table_type_info.find(name);
   if (iter == block_based_table_type_info.end()) {
-    if (ignore_unknown_options) {
+    if (config_options.ignore_unknown_options) {
       return "";
     } else {
       return "Unrecognized option";
@@ -634,14 +636,23 @@ std::string ParseBlockBasedTableOption(const std::string& name,
 Status GetBlockBasedTableOptionsFromString(
     const BlockBasedTableOptions& table_options, const std::string& opts_str,
     BlockBasedTableOptions* new_table_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+  config_options.ignore_unknown_options = false;
+  return GetBlockBasedTableOptionsFromString(config_options, table_options,
+                                             opts_str, new_table_options);
+}
+Status GetBlockBasedTableOptionsFromString(
+    const ConfigOptions& config_options,
+    const BlockBasedTableOptions& table_options, const std::string& opts_str,
+    BlockBasedTableOptions* new_table_options) {
   std::unordered_map<std::string, std::string> opts_map;
   Status s = StringToMap(opts_str, &opts_map);
   if (!s.ok()) {
     return s;
   }
-
+  return GetBlockBasedTableOptionsFromMap(config_options, table_options,
-  return GetBlockBasedTableOptionsFromMap(table_options, opts_map,
+                                          opts_map, new_table_options);
-                                          new_table_options);
 }
 
 Status GetBlockBasedTableOptionsFromMap(
@@ -649,16 +660,29 @@ Status GetBlockBasedTableOptionsFromMap(
     const std::unordered_map<std::string, std::string>& opts_map,
     BlockBasedTableOptions* new_table_options, bool input_strings_escaped,
     bool ignore_unknown_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = input_strings_escaped;
+  config_options.ignore_unknown_options = ignore_unknown_options;
+
+  return GetBlockBasedTableOptionsFromMap(config_options, table_options,
+                                          opts_map, new_table_options);
+}
+
+Status GetBlockBasedTableOptionsFromMap(
+    const ConfigOptions& config_options,
+    const BlockBasedTableOptions& table_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    BlockBasedTableOptions* new_table_options) {
   assert(new_table_options);
   *new_table_options = table_options;
   for (const auto& o : opts_map) {
     auto error_message = ParseBlockBasedTableOption(
-        o.first, o.second, new_table_options, input_strings_escaped,
+        config_options, o.first, o.second, new_table_options);
-        ignore_unknown_options);
     if (error_message != "") {
       const auto iter = block_based_table_type_info.find(o.first);
       if (iter == block_based_table_type_info.end() ||
-          !input_strings_escaped ||  // !input_strings_escaped indicates
+          !config_options
+               .input_strings_escaped ||  // !input_strings_escaped indicates
                                           // the old API, where everything is
                                           // parsable.
           (!iter->second.IsByName() && !iter->second.IsDeprecated())) {
@@ -672,12 +696,11 @@ Status GetBlockBasedTableOptionsFromMap(
   return Status::OK();
 }
 
-Status VerifyBlockBasedTableFactory(
+Status VerifyBlockBasedTableFactory(const ConfigOptions& config_options,
                                     const BlockBasedTableFactory* base_tf,
-    const BlockBasedTableFactory* file_tf,
+                                    const BlockBasedTableFactory* file_tf) {
-    OptionsSanityCheckLevel sanity_check_level) {
   if ((base_tf != nullptr) != (file_tf != nullptr) &&
-      sanity_check_level > OptionsSanityCheckLevel::kSanityLevelNone) {
+      config_options.sanity_level > ConfigOptions::kSanityLevelNone) {
     return Status::Corruption(
         "[RocksDBOptionsParser]: Inconsistent TableFactory class type");
   }
@@ -695,7 +718,7 @@ Status VerifyBlockBasedTableFactory(
       // contain random values since they might not be initialized
       continue;
     }
-    if (BBTOptionSanityCheckLevel(pair.first) <= sanity_check_level) {
+    if (BBTOptionSanityCheckLevel(pair.first) <= config_options.sanity_level) {
       if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
                            reinterpret_cast<const char*>(&file_opt),
                            pair.second, pair.first, nullptr)) {

table/block_based/block_based_table_factory.h

@@ -14,13 +14,11 @@
 #include <string>
 
 #include "db/dbformat.h"
-#include "options/options_helper.h"
-#include "options/options_parser.h"
 #include "rocksdb/flush_block_policy.h"
 #include "rocksdb/table.h"
 
 namespace ROCKSDB_NAMESPACE {
-
+struct ConfigOptions;
 struct EnvOptions;
 
 class BlockBasedTableBuilder;
@@ -66,8 +64,8 @@ class BlockBasedTableFactory : public TableFactory {
 
   std::string GetPrintableTableOptions() const override;
 
-  Status GetOptionString(std::string* opt_string,
+  Status GetOptionString(const ConfigOptions& config_options,
-                         const std::string& delimiter) const override;
+                         std::string* opt_string) const override;
 
   const BlockBasedTableOptions& table_options() const;
 
@@ -89,9 +87,7 @@ extern const std::string kPropFalse;
 
 #ifndef ROCKSDB_LITE
 extern Status VerifyBlockBasedTableFactory(
-    const BlockBasedTableFactory* base_tf,
+    const ConfigOptions& config_options, const BlockBasedTableFactory* base_tf,
-    const BlockBasedTableFactory* file_tf,
+    const BlockBasedTableFactory* file_tf);
-    OptionsSanityCheckLevel sanity_check_level);
-
 #endif  // !ROCKSDB_LITE
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/block_based_table_reader.cc

@@ -7,6 +7,7 @@
 // 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 "table/block_based/block_based_table_reader.h"
+
 #include <algorithm>
 #include <array>
 #include <limits>
@@ -16,10 +17,10 @@
 
 #include "db/dbformat.h"
 #include "db/pinned_iterators_manager.h"
-
 #include "file/file_prefetch_buffer.h"
 #include "file/random_access_file_reader.h"
-
+#include "monitoring/perf_context_imp.h"
+#include "options/options_helper.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/comparator.h"
 #include "rocksdb/env.h"
@@ -30,7 +31,6 @@
 #include "rocksdb/statistics.h"
 #include "rocksdb/table.h"
 #include "rocksdb/table_properties.h"
-
 #include "table/block_based/binary_search_index_reader.h"
 #include "table/block_based/block.h"
 #include "table/block_based/block_based_filter_block.h"

table/cuckoo/cuckoo_table_factory.h

@@ -79,8 +79,8 @@ class CuckooTableFactory : public TableFactory {
 
   void* GetOptions() override { return &table_options_; }
 
-  Status GetOptionString(std::string* /*opt_string*/,
+  Status GetOptionString(const ConfigOptions& /*config_options*/,
-                         const std::string& /*delimiter*/) const override {
+                         std::string* /*opt_string*/) const override {
     return Status::OK();
   }
 

table/plain/plain_table_factory.cc

@@ -7,7 +7,9 @@
 #include "table/plain/plain_table_factory.h"
 
 #include <stdint.h>
+
 #include <memory>
+
 #include "db/dbformat.h"
 #include "options/options_helper.h"
 #include "port/port.h"
@@ -117,12 +119,24 @@ const PlainTableOptions& PlainTableFactory::table_options() const {
 Status GetPlainTableOptionsFromString(const PlainTableOptions& table_options,
                                       const std::string& opts_str,
                                       PlainTableOptions* new_table_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+  config_options.ignore_unknown_options = false;
+  return GetPlainTableOptionsFromString(config_options, table_options, opts_str,
+                                        new_table_options);
+}
+
+Status GetPlainTableOptionsFromString(const ConfigOptions& config_options,
+                                      const PlainTableOptions& table_options,
+                                      const std::string& opts_str,
+                                      PlainTableOptions* new_table_options) {
   std::unordered_map<std::string, std::string> opts_map;
   Status s = StringToMap(opts_str, &opts_map);
   if (!s.ok()) {
     return s;
   }
-  return GetPlainTableOptionsFromMap(table_options, opts_map,
+
+  return GetPlainTableOptionsFromMap(config_options, table_options, opts_map,
                                      new_table_options);
 }
 
@@ -190,16 +204,16 @@ Status GetMemTableRepFactoryFromString(
   return Status::OK();
 }
 
-std::string ParsePlainTableOptions(const std::string& name,
+std::string ParsePlainTableOptions(const ConfigOptions& config_options,
+                                   const std::string& name,
                                    const std::string& org_value,
-                                   PlainTableOptions* new_options,
+                                   PlainTableOptions* new_options) {
-                                   bool input_strings_escaped = false,
+  const std::string& value = config_options.input_strings_escaped
-                                   bool ignore_unknown_options = false) {
+                                 ? UnescapeOptionString(org_value)
-  const std::string& value =
+                                 : org_value;
-      input_strings_escaped ? UnescapeOptionString(org_value) : org_value;
   const auto iter = plain_table_type_info.find(name);
   if (iter == plain_table_type_info.end()) {
-    if (ignore_unknown_options) {
+    if (config_options.ignore_unknown_options) {
       return "";
     } else {
       return "Unrecognized option";
@@ -218,16 +232,28 @@ Status GetPlainTableOptionsFromMap(
     const PlainTableOptions& table_options,
     const std::unordered_map<std::string, std::string>& opts_map,
     PlainTableOptions* new_table_options, bool input_strings_escaped,
-    bool /*ignore_unknown_options*/) {
+    bool ignore_unknown_options) {
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = input_strings_escaped;
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  return GetPlainTableOptionsFromMap(config_options, table_options, opts_map,
+                                     new_table_options);
+}
+
+Status GetPlainTableOptionsFromMap(
+    const ConfigOptions& config_options, const PlainTableOptions& table_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    PlainTableOptions* new_table_options) {
   assert(new_table_options);
   *new_table_options = table_options;
   for (const auto& o : opts_map) {
-    auto error_message = ParsePlainTableOptions(
+    auto error_message = ParsePlainTableOptions(config_options, o.first,
-        o.first, o.second, new_table_options, input_strings_escaped);
+                                                o.second, new_table_options);
     if (error_message != "") {
       const auto iter = plain_table_type_info.find(o.first);
       if (iter == plain_table_type_info.end() ||
-          !input_strings_escaped ||  // !input_strings_escaped indicates
+          !config_options
+               .input_strings_escaped ||  // !input_strings_escaped indicates
                                           // the old API, where everything is
                                           // parsable.
           (!iter->second.IsByName() && !iter->second.IsDeprecated())) {

table/plain/plain_table_factory.h

@@ -184,8 +184,8 @@ class PlainTableFactory : public TableFactory {
 
   void* GetOptions() override { return &table_options_; }
 
-  Status GetOptionString(std::string* /*opt_string*/,
+  Status GetOptionString(const ConfigOptions& /*config_options*/,
-                         const std::string& /*delimiter*/) const override {
+                         std::string* /*opt_string*/) const override {
     return Status::OK();
   }
 

tools/ldb_cmd.cc

@@ -330,7 +330,6 @@ LDBCommand::LDBCommand(const std::map<std::string, std::string>& options,
       is_db_ttl_(false),
       timestamp_(false),
       try_load_options_(false),
-      ignore_unknown_options_(false),
       create_if_missing_(false),
       option_map_(options),
       flags_(flags),
@@ -363,13 +362,15 @@ LDBCommand::LDBCommand(const std::map<std::string, std::string>& options,
   is_db_ttl_ = IsFlagPresent(flags, ARG_TTL);
   timestamp_ = IsFlagPresent(flags, ARG_TIMESTAMP);
   try_load_options_ = IsFlagPresent(flags, ARG_TRY_LOAD_OPTIONS);
-  ignore_unknown_options_ = IsFlagPresent(flags, ARG_IGNORE_UNKNOWN_OPTIONS);
+  config_options_.ignore_unknown_options =
+      IsFlagPresent(flags, ARG_IGNORE_UNKNOWN_OPTIONS);
 }
 
 void LDBCommand::OpenDB() {
   if (!create_if_missing_ && try_load_options_) {
-    Status s = LoadLatestOptions(db_path_, options_.env, &options_,
+    config_options_.env = options_.env;
-                                 &column_families_, ignore_unknown_options_);
+    Status s = LoadLatestOptions(config_options_, db_path_, &options_,
+                                 &column_families_);
     if (!s.ok() && !s.IsNotFound()) {
       // Option file exists but load option file error.
       std::string msg = s.ToString();

utilities/options/options_util.cc

@@ -10,7 +10,9 @@
 #include "env/composite_env_wrapper.h"
 #include "file/filename.h"
 #include "options/options_parser.h"
+#include "rocksdb/convenience.h"
 #include "rocksdb/options.h"
+#include "table/block_based/block_based_table_factory.h"
 
 namespace ROCKSDB_NAMESPACE {
 Status LoadOptionsFromFile(const std::string& file_name, Env* env,
@@ -18,10 +20,22 @@ Status LoadOptionsFromFile(const std::string& file_name, Env* env,
                            std::vector<ColumnFamilyDescriptor>* cf_descs,
                            bool ignore_unknown_options,
                            std::shared_ptr<Cache>* cache) {
+  ConfigOptions config_options;
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  config_options.input_strings_escaped = true;
+  config_options.env = env;
+
+  return LoadOptionsFromFile(config_options, file_name, db_options, cf_descs,
+                             cache);
+}
+
+Status LoadOptionsFromFile(const ConfigOptions& config_options,
+                           const std::string& file_name, DBOptions* db_options,
+                           std::vector<ColumnFamilyDescriptor>* cf_descs,
+                           std::shared_ptr<Cache>* cache) {
   RocksDBOptionsParser parser;
-  LegacyFileSystemWrapper fs(env);
+  LegacyFileSystemWrapper fs(config_options.env);
-  Status s = parser.Parse(file_name, &fs, ignore_unknown_options,
+  Status s = parser.Parse(config_options, file_name, &fs);
-                          0 /* file_readahead_size */);
   if (!s.ok()) {
     return s;
   }
@@ -76,21 +90,48 @@ Status LoadLatestOptions(const std::string& dbpath, Env* env,
                          std::vector<ColumnFamilyDescriptor>* cf_descs,
                          bool ignore_unknown_options,
                          std::shared_ptr<Cache>* cache) {
+  ConfigOptions config_options;
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  config_options.input_strings_escaped = true;
+  config_options.env = env;
+
+  return LoadLatestOptions(config_options, dbpath, db_options, cf_descs, cache);
+}
+
+Status LoadLatestOptions(const ConfigOptions& config_options,
+                         const std::string& dbpath, DBOptions* db_options,
+                         std::vector<ColumnFamilyDescriptor>* cf_descs,
+                         std::shared_ptr<Cache>* cache) {
   std::string options_file_name;
-  Status s = GetLatestOptionsFileName(dbpath, env, &options_file_name);
+  Status s =
+      GetLatestOptionsFileName(dbpath, config_options.env, &options_file_name);
   if (!s.ok()) {
     return s;
   }
-  return LoadOptionsFromFile(dbpath + "/" + options_file_name, env, db_options,
+  return LoadOptionsFromFile(config_options, dbpath + "/" + options_file_name,
-                             cf_descs, ignore_unknown_options, cache);
+                             db_options, cf_descs, cache);
 }
 
 Status CheckOptionsCompatibility(
     const std::string& dbpath, Env* env, const DBOptions& db_options,
     const std::vector<ColumnFamilyDescriptor>& cf_descs,
     bool ignore_unknown_options) {
+  ConfigOptions config_options;
+  config_options.sanity_level = ConfigOptions::kSanityLevelLooselyCompatible;
+  config_options.ignore_unknown_options = ignore_unknown_options;
+  config_options.input_strings_escaped = true;
+  config_options.env = env;
+  return CheckOptionsCompatibility(config_options, dbpath, db_options,
+                                   cf_descs);
+}
+
+Status CheckOptionsCompatibility(
+    const ConfigOptions& config_options, const std::string& dbpath,
+    const DBOptions& db_options,
+    const std::vector<ColumnFamilyDescriptor>& cf_descs) {
   std::string options_file_name;
-  Status s = GetLatestOptionsFileName(dbpath, env, &options_file_name);
+  Status s =
+      GetLatestOptionsFileName(dbpath, config_options.env, &options_file_name);
   if (!s.ok()) {
     return s;
   }
@@ -102,12 +143,12 @@ Status CheckOptionsCompatibility(
     cf_opts.push_back(cf_desc.options);
   }
 
-  const OptionsSanityCheckLevel kDefaultLevel = kSanityLevelLooselyCompatible;
+  LegacyFileSystemWrapper fs(config_options.env);
-  LegacyFileSystemWrapper fs(env);
 
   return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
-      db_options, cf_names, cf_opts, dbpath + "/" + options_file_name, &fs,
+
-      kDefaultLevel, ignore_unknown_options);
+      config_options, db_options, cf_names, cf_opts,
+      dbpath + "/" + options_file_name, &fs);
 }
 
 }  // namespace ROCKSDB_NAMESPACE

utilities/options/options_util_test.cc

@@ -5,15 +5,16 @@
 
 #ifndef ROCKSDB_LITE
 
-#include <cinttypes>
+#include "rocksdb/utilities/options_util.h"
 
 #include <cctype>
+#include <cinttypes>
 #include <unordered_map>
 
 #include "options/options_parser.h"
+#include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 #include "rocksdb/table.h"
-#include "rocksdb/utilities/options_util.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 #include "util/random.h"
@@ -67,23 +68,26 @@ TEST_F(OptionsUtilTest, SaveAndLoad) {
   std::vector<ColumnFamilyDescriptor> loaded_cf_descs;
   ASSERT_OK(LoadOptionsFromFile(kFileName, env_.get(), &loaded_db_opt,
                                 &loaded_cf_descs));
-
+  ConfigOptions exact;
-  ASSERT_OK(RocksDBOptionsParser::VerifyDBOptions(db_opt, loaded_db_opt));
+  exact.sanity_level = ConfigOptions::kSanityLevelExactMatch;
+  ASSERT_OK(
+      RocksDBOptionsParser::VerifyDBOptions(exact, db_opt, loaded_db_opt));
   test::RandomInitDBOptions(&db_opt, &rnd_);
-  ASSERT_NOK(RocksDBOptionsParser::VerifyDBOptions(db_opt, loaded_db_opt));
+  ASSERT_NOK(
+      RocksDBOptionsParser::VerifyDBOptions(exact, db_opt, loaded_db_opt));
 
   for (size_t i = 0; i < kCFCount; ++i) {
     ASSERT_EQ(cf_names[i], loaded_cf_descs[i].name);
     ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(
-        cf_opts[i], loaded_cf_descs[i].options));
+        exact, cf_opts[i], loaded_cf_descs[i].options));
     if (IsBlockBasedTableFactory(cf_opts[i].table_factory.get())) {
       ASSERT_OK(RocksDBOptionsParser::VerifyTableFactory(
-          cf_opts[i].table_factory.get(),
+          exact, cf_opts[i].table_factory.get(),
           loaded_cf_descs[i].options.table_factory.get()));
     }
     test::RandomInitCFOptions(&cf_opts[i], db_opt, &rnd_);
     ASSERT_NOK(RocksDBOptionsParser::VerifyCFOptions(
-        cf_opts[i], loaded_cf_descs[i].options));
+        exact, cf_opts[i], loaded_cf_descs[i].options));
   }
 
   for (size_t i = 0; i < kCFCount; ++i) {
@@ -125,6 +129,25 @@ TEST_F(OptionsUtilTest, SaveAndLoadWithCacheCheck) {
   PersistRocksDBOptions(db_opt, cf_names, cf_opts, kFileName, fs_.get());
   DBOptions loaded_db_opt;
   std::vector<ColumnFamilyDescriptor> loaded_cf_descs;
+
+  ConfigOptions config_options;
+  config_options.ignore_unknown_options = false;
+  config_options.input_strings_escaped = true;
+  config_options.env = env_.get();
+  ASSERT_OK(LoadOptionsFromFile(config_options, kFileName, &loaded_db_opt,
+                                &loaded_cf_descs, &cache));
+  for (size_t i = 0; i < loaded_cf_descs.size(); i++) {
+    if (IsBlockBasedTableFactory(cf_opts[i].table_factory.get())) {
+      auto* loaded_bbt_opt = reinterpret_cast<BlockBasedTableOptions*>(
+          loaded_cf_descs[i].options.table_factory->GetOptions());
+      // Expect the same cache will be loaded
+      if (loaded_bbt_opt != nullptr) {
+        ASSERT_EQ(loaded_bbt_opt->block_cache.get(), cache.get());
+      }
+    }
+  }
+
+  // Test the old interface
   ASSERT_OK(LoadOptionsFromFile(kFileName, env_.get(), &loaded_db_opt,
                                 &loaded_cf_descs, false, &cache));
   for (size_t i = 0; i < loaded_cf_descs.size(); i++) {
@@ -170,8 +193,8 @@ class DummyTableFactory : public TableFactory {
 
   std::string GetPrintableTableOptions() const override { return ""; }
 
-  Status GetOptionString(std::string* /*opt_string*/,
+  Status GetOptionString(const ConfigOptions& /*opts*/,
-                         const std::string& /*delimiter*/) const override {
+                         std::string* /*opt_string*/) const override {
     return Status::OK();
   }
 };
@@ -248,9 +271,13 @@ TEST_F(OptionsUtilTest, SanityCheck) {
   }
   delete db;
 
+  ConfigOptions config_options;
+  config_options.ignore_unknown_options = false;
+  config_options.input_strings_escaped = true;
+  config_options.sanity_level = ConfigOptions::kSanityLevelLooselyCompatible;
   // perform sanity check
   ASSERT_OK(
-      CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+      CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
   ASSERT_GE(kCFCount, 5);
   // merge operator
@@ -261,15 +288,15 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     ASSERT_NE(merge_op.get(), nullptr);
     cf_descs[0].options.merge_operator.reset();
     ASSERT_NOK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[0].options.merge_operator.reset(new DummyMergeOperator());
     ASSERT_NOK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[0].options.merge_operator = merge_op;
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
   }
 
   // prefix extractor
@@ -281,15 +308,15 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     ASSERT_NE(prefix_extractor, nullptr);
     cf_descs[1].options.prefix_extractor.reset();
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[1].options.prefix_extractor.reset(new DummySliceTransform());
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[1].options.prefix_extractor = prefix_extractor;
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
   }
 
   // prefix extractor nullptr case
@@ -301,16 +328,16 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     ASSERT_EQ(prefix_extractor, nullptr);
     cf_descs[0].options.prefix_extractor.reset();
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     // It's okay to change prefix_extractor from nullptr to non-nullptr
     cf_descs[0].options.prefix_extractor.reset(new DummySliceTransform());
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[0].options.prefix_extractor = prefix_extractor;
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
   }
 
   // comparator
@@ -320,11 +347,11 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     auto* prev_comparator = cf_descs[2].options.comparator;
     cf_descs[2].options.comparator = &comparator;
     ASSERT_NOK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[2].options.comparator = prev_comparator;
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
   }
 
   // table factory
@@ -335,11 +362,11 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     ASSERT_NE(table_factory, nullptr);
     cf_descs[3].options.table_factory.reset(new DummyTableFactory());
     ASSERT_NOK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
 
     cf_descs[3].options.table_factory = table_factory;
     ASSERT_OK(
-        CheckOptionsCompatibility(dbname_, Env::Default(), db_opt, cf_descs));
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
   }
 }