RocksDB Options file format and its serialization / deserialization.

Summary:
This patch defines the format of RocksDB options file, which
follows the INI file format, and implements functions for its
serialization and deserialization.  An example RocksDB options
file can be found in examples/rocksdb_option_file_example.ini.

A typical RocksDB options file has three sections, which are
Version, DBOptions, and more than one CFOptions.  The RocksDB
options file in general follows the basic INI file format
with the following extensions / modifications:
 * Escaped characters
   We escaped the following characters:
    - \n -- line feed - new line
    - \r -- carriage return
    - \\ -- backslash \
    - \: -- colon symbol :
    - \# -- hash tag #
 * Comments
   We support # style comments.  Comments can appear at the ending
   part of a line.
 * Statements
   A statement is of the form option_name = value.
   Each statement contains a '=', where extra white-spaces
   are supported. However, we don't support multi-lined statement.
   Furthermore, each line can only contain at most one statement.
 * Section
   Sections are of the form [SecitonTitle "SectionArgument"],
   where section argument is optional.
 * List
   We use colon-separated string to represent a list.
   For instance, n1:n2:n3:n4 is a list containing four values.

Below is an example of a RocksDB options file:

[Version]
  rocksdb_version=4.0.0
  options_file_version=1.0
[DBOptions]
  max_open_files=12345
  max_background_flushes=301
[CFOptions "default"]
[CFOptions "the second column family"]
[CFOptions "the third column family"]

Test Plan: Added many tests in options_test.cc

Reviewers: igor, IslamAbdelRahman, sdong, anthony

Reviewed By: anthony

Subscribers: maykov, dhruba, leveldb

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

CMakeLists.txt

@@ -199,6 +199,7 @@ set(SOURCES
         util/options.cc
         util/options_builder.cc
         util/options_helper.cc
+        util/options_parser.cc
         util/perf_context.cc
         util/perf_level.cc
         util/rate_limiter.cc

examples/rocksdb_option_file_example.ini

@@ -0,0 +1,53 @@
+# This is a RocksDB option file.
+#
+# A typical RocksDB options file has three sections, which are
+# Version, DBOptions, and more than one CFOptions.  The RocksDB
+# options file in general follows the basic INI file format
+# with the following extensions / modifications:
+#
+#  * Escaped characters
+#    We escaped the following characters:
+#     - \n -- line feed - new line
+#     - \r -- carriage return
+#     - \\ -- backslash \
+#     - \: -- colon symbol :
+#     - \# -- hash tag #
+#  * Comments
+#    We support # style comments.  Comments can appear at the ending
+#    part of a line.
+#  * Statements
+#    A statement is of the form option_name = value.
+#    Each statement contains a '=', where extra white-spaces
+#    are supported. However, we don't support multi-lined statement.
+#    Furthermore, each line can only contain at most one statement.
+#  * Section
+#    Sections are of the form [SecitonTitle "SectionArgument"],
+#    where section argument is optional.
+#  * List
+#    We use colon-separated string to represent a list.
+#    For instance, n1:n2:n3:n4 is a list containing four values.
+#
+# Below is an example of a RocksDB options file:
+[Version]
+  # The Version section stores the version information about rocksdb
+  # and option file.  This is used for handling potential format
+  # change in the future.
+  rocksdb_version=4.0.0  # We support "#" style comment.
+  options_file_version=1.0
+[DBOptions]
+  # Followed by the Version section is the DBOptions section.
+  # The value of an options can be assigned using a statement.
+  # Note that for those options that is not set in the options file,
+  # we will use the default value.
+  max_open_files=12345
+  max_background_flushes=301
+[CFOptions "default"]
+  # ColumnFamilyOptions section must follow the format of
+  # [CFOptions "cf name"].  If a rocksdb instance
+  # has multiple column families, then its CFOptions must be
+  # specified in the same order as column family creation order.
+[CFOptions "the second column family"]
+  # Each column family must have one section in the RocksDB option
+  # file even all the options of this column family are set to
+  # default value.
+[CFOptions "the third column family"]

include/rocksdb/convenience.h

@@ -14,16 +14,28 @@ namespace rocksdb {
 
 #ifndef ROCKSDB_LITE
 // Take a map of option name and option value, apply them into the
-// base_options, and return the new options as a result
+// base_options, and return the new options as a result.
+//
+// If input_strings_escaped is set to true, then each escaped characters
+// prefixed by '\' in the the values of the opts_map will be further
+// converted back to the raw string before assigning to the associated
+// options.
 Status GetColumnFamilyOptionsFromMap(
     const ColumnFamilyOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    ColumnFamilyOptions* new_options);
+    ColumnFamilyOptions* new_options, bool input_strings_escaped = false);
 
+// Take a map of option name and option value, apply them into the
+// base_options, and return the new options as a result.
+//
+// If input_strings_escaped is set to true, then each escaped characters
+// prefixed by '\' in the the values of the opts_map will be further
+// converted back to the raw string before assigning to the associated
+// options.
 Status GetDBOptionsFromMap(
     const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    DBOptions* new_options);
+    DBOptions* new_options, bool input_strings_escaped = false);
 
 Status GetBlockBasedTableOptionsFromMap(
     const BlockBasedTableOptions& table_options,
@@ -48,11 +60,13 @@ Status GetDBOptionsFromString(
     const std::string& opts_str,
     DBOptions* new_options);
 
-Status GetStringFromDBOptions(const DBOptions& db_options,
+Status GetStringFromDBOptions(std::string* opts_str,
-                              std::string* opts_str);
+                              const DBOptions& db_options,
+                              const std::string& delimiter = ";  ");
 
-Status GetStringFromColumnFamilyOptions(const ColumnFamilyOptions& db_options,
+Status GetStringFromColumnFamilyOptions(std::string* opts_str,
-                                        std::string* opts_str);
+                                        const ColumnFamilyOptions& db_options,
+                                        const std::string& delimiter = ";  ");
 
 Status GetBlockBasedTableOptionsFromString(
     const BlockBasedTableOptions& table_options,

src.mk

@@ -140,8 +140,9 @@ LIB_SOURCES =                                                   \
   util/options_builder.cc                                       \
   util/options.cc                                               \
   util/options_helper.cc                                        \
+  util/options_parser.cc                                        \
   util/perf_context.cc                                          \
-  util/perf_level.cc                                          \
+  util/perf_level.cc                                            \
   util/rate_limiter.cc                                          \
   util/skiplistrep.cc                                           \
   util/slice.cc                                                 \

util/options_helper.cc

@@ -22,6 +22,67 @@
 namespace rocksdb {
 
 #ifndef ROCKSDB_LITE
+bool isSpecialChar(const char c) {
+  if (c == '\\' || c == '#' || c == ':' || c == '\r' || c == '\n') {
+    return true;
+  }
+  return false;
+}
+
+char UnescapeChar(const char c) {
+  static const std::unordered_map<char, char> convert_map = {{'r', '\r'},
+                                                             {'n', '\n'}};
+
+  auto iter = convert_map.find(c);
+  if (iter == convert_map.end()) {
+    return c;
+  }
+  return iter->second;
+}
+
+char EscapeChar(const char c) {
+  static const std::unordered_map<char, char> convert_map = {{'\n', 'n'},
+                                                             {'\r', 'r'}};
+
+  auto iter = convert_map.find(c);
+  if (iter == convert_map.end()) {
+    return c;
+  }
+  return iter->second;
+}
+
+std::string EscapeOptionString(const std::string& raw_string) {
+  std::string output;
+  for (auto c : raw_string) {
+    if (isSpecialChar(c)) {
+      output += '\\';
+      output += EscapeChar(c);
+    } else {
+      output += c;
+    }
+  }
+
+  return output;
+}
+
+std::string UnescapeOptionString(const std::string& escaped_string) {
+  bool escaped = false;
+  std::string output;
+
+  for (auto c : escaped_string) {
+    if (escaped) {
+      output += UnescapeChar(c);
+      escaped = false;
+    } else {
+      if (c == '\\') {
+        escaped = true;
+        continue;
+      }
+      output += c;
+    }
+  }
+  return output;
+}
 
 namespace {
 CompressionType ParseCompressionType(const std::string& type) {
@@ -232,7 +293,8 @@ bool SerializeSingleOptionHelper(const char* opt_address,
       *value = ToString(*(reinterpret_cast<const double*>(opt_address)));
       break;
     case OptionType::kString:
-      *value = *(reinterpret_cast<const std::string*>(opt_address));
+      *value = EscapeOptionString(
+          *(reinterpret_cast<const std::string*>(opt_address)));
       break;
     case OptionType::kCompactionStyle:
       *value = CompactionStyleToString(
@@ -461,8 +523,12 @@ Status StringToMap(const std::string& opts_str,
   return Status::OK();
 }
 
-bool ParseColumnFamilyOption(const std::string& name, const std::string& value,
+bool ParseColumnFamilyOption(const std::string& name,
-                             ColumnFamilyOptions* new_options) {
+                             const std::string& org_value,
+                             ColumnFamilyOptions* new_options,
+                             bool input_string_escaped = false) {
+  const std::string& value =
+      input_string_escaped ? UnescapeOptionString(org_value) : org_value;
   try {
     if (name == "max_bytes_for_level_multiplier_additional") {
       new_options->max_bytes_for_level_multiplier_additional.clear();
@@ -573,8 +639,10 @@ bool ParseColumnFamilyOption(const std::string& name, const std::string& value,
   return true;
 }
 
-bool SerializeSingleDBOption(const DBOptions& db_options,
+bool SerializeSingleDBOption(std::string* opt_string,
-                             const std::string& name, std::string* opt_string) {
+                             const DBOptions& db_options,
+                             const std::string& name,
+                             const std::string& delimiter) {
   auto iter = db_options_type_info.find(name);
   if (iter == db_options_type_info.end()) {
     return false;
@@ -585,20 +653,21 @@ bool SerializeSingleDBOption(const DBOptions& db_options,
   std::string value;
   bool result = SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
   if (result) {
-    *opt_string = name + " = " + value + ";  ";
+    *opt_string = name + "=" + value + delimiter;
   }
   return result;
 }
 
-Status GetStringFromDBOptions(const DBOptions& db_options,
+Status GetStringFromDBOptions(std::string* opt_string,
-                              std::string* opt_string) {
+                              const DBOptions& db_options,
+                              const std::string& delimiter) {
   assert(opt_string);
   opt_string->clear();
   for (auto iter = db_options_type_info.begin();
        iter != db_options_type_info.end(); ++iter) {
     std::string single_output;
-    bool result =
+    bool result = SerializeSingleDBOption(&single_output, db_options,
-        SerializeSingleDBOption(db_options, iter->first, &single_output);
+                                          iter->first, delimiter);
     assert(result);
     if (result) {
       opt_string->append(single_output);
@@ -607,9 +676,10 @@ Status GetStringFromDBOptions(const DBOptions& db_options,
   return Status::OK();
 }
 
-bool SerializeSingleColumnFamilyOption(const ColumnFamilyOptions& cf_options,
+bool SerializeSingleColumnFamilyOption(std::string* opt_string,
+                                       const ColumnFamilyOptions& cf_options,
                                        const std::string& name,
-                                       std::string* opt_string) {
+                                       const std::string& delimiter) {
   auto iter = cf_options_type_info.find(name);
   if (iter == cf_options_type_info.end()) {
     return false;
@@ -620,32 +690,36 @@ bool SerializeSingleColumnFamilyOption(const ColumnFamilyOptions& cf_options,
   std::string value;
   bool result = SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
   if (result) {
-    *opt_string = name + " = " + value + ";  ";
+    *opt_string = name + "=" + value + delimiter;
   }
   return result;
 }
 
-Status GetStringFromColumnFamilyOptions(const ColumnFamilyOptions& cf_options,
+Status GetStringFromColumnFamilyOptions(std::string* opt_string,
-                                        std::string* opt_string) {
+                                        const ColumnFamilyOptions& cf_options,
+                                        const std::string& delimiter) {
   assert(opt_string);
   opt_string->clear();
   for (auto iter = cf_options_type_info.begin();
        iter != cf_options_type_info.end(); ++iter) {
     std::string single_output;
-    bool result = SerializeSingleColumnFamilyOption(cf_options, iter->first,
+    bool result = SerializeSingleColumnFamilyOption(&single_output, cf_options,
-                                                    &single_output);
+                                                    iter->first, delimiter);
     if (result) {
       opt_string->append(single_output);
     } else {
-      printf("failed to serialize %s\n", iter->first.c_str());
+      return Status::InvalidArgument("failed to serialize %s\n",
+                                     iter->first.c_str());
     }
     assert(result);
   }
   return Status::OK();
 }
 
-bool ParseDBOption(const std::string& name, const std::string& value,
+bool ParseDBOption(const std::string& name, const std::string& org_value,
-                   DBOptions* new_options) {
+                   DBOptions* new_options, bool input_string_escaped = false) {
+  const std::string& value =
+      input_string_escaped ? UnescapeOptionString(org_value) : org_value;
   try {
     if (name == "rate_limiter_bytes_per_sec") {
       new_options->rate_limiter.reset(
@@ -791,11 +865,12 @@ Status GetPlainTableOptionsFromMap(
 Status GetColumnFamilyOptionsFromMap(
     const ColumnFamilyOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    ColumnFamilyOptions* new_options) {
+    ColumnFamilyOptions* new_options, bool input_strings_escaped) {
   assert(new_options);
   *new_options = base_options;
   for (const auto& o : opts_map) {
-    if (!ParseColumnFamilyOption(o.first, o.second, new_options)) {
+    if (!ParseColumnFamilyOption(o.first, o.second, new_options,
+                                 input_strings_escaped)) {
       return Status::InvalidArgument("Can't parse option " + o.first);
     }
   }
@@ -817,11 +892,11 @@ Status GetColumnFamilyOptionsFromString(
 Status GetDBOptionsFromMap(
     const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
-    DBOptions* new_options) {
+    DBOptions* new_options, bool input_strings_escaped) {
   assert(new_options);
   *new_options = base_options;
   for (const auto& o : opts_map) {
-    if (!ParseDBOption(o.first, o.second, new_options)) {
+    if (!ParseDBOption(o.first, o.second, new_options, input_strings_escaped)) {
       return Status::InvalidArgument("Can't parse option " + o.first);
     }
   }
@@ -860,5 +935,5 @@ Status GetOptionsFromString(const Options& base_options,
   return Status::OK();
 }
 
-#endif  // ROCKSDB_LITE
+#endif  // !ROCKSDB_LITE
 }  // namespace rocksdb

util/options_helper.h

@@ -11,8 +11,45 @@
 #include "rocksdb/status.h"
 #include "util/mutable_cf_options.h"
 
+#ifndef ROCKSDB_LITE
 namespace rocksdb {
 
+// Returns true if the input char "c" is considered as a special character
+// that will be escaped when EscapeOptionString() is called.
+//
+// @param c the input char
+// @return true if the input char "c" is considered as a special character.
+// @see EscapeOptionString
+bool isSpecialChar(const char c);
+
+// If the input char is an escaped char, it will return the its
+// associated raw-char.  Otherwise, the function will simply return
+// the original input char.
+char UnescapeChar(const char c);
+
+// If the input char is a control char, it will return the its
+// associated escaped char.  Otherwise, the function will simply return
+// the original input char.
+char EscapeChar(const char c);
+
+// Converts a raw string to an escaped string.  Escaped-characters are
+// defined via the isSpecialChar() function.  When a char in the input
+// string "raw_string" is classified as a special characters, then it
+// will be prefixed by '\' in the output.
+//
+// It's inverse function is UnescapeOptionString().
+// @param raw_string the input string
+// @return the '\' escaped string of the input "raw_string"
+// @see isSpecialChar, UnescapeOptionString
+std::string EscapeOptionString(const std::string& raw_string);
+
+// The inverse function of EscapeOptionString.  It converts
+// an '\' escaped string back to a raw string.
+//
+// @param escaped_string the input '\' escaped string
+// @return the raw string of the input "escaped_string"
+std::string UnescapeOptionString(const std::string& escaped_string);
+
 Status GetMutableOptionsFromStrings(
     const MutableCFOptions& base_options,
     const std::unordered_map<std::string, std::string>& options_map,
@@ -286,3 +323,5 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
       OptionType::kCompactionStyle}}};
 
 }  // namespace rocksdb
+
+#endif  // !ROCKSDB_LITE

util/options_parser.cc

@@ -0,0 +1,563 @@
+// Copyright (c) 2014, Facebook, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+
+#ifndef ROCKSDB_LITE
+
+#include "util/options_parser.h"
+
+#include <cmath>
+#include <map>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "rocksdb/convenience.h"
+#include "rocksdb/db.h"
+#include "util/options_helper.h"
+#include "util/string_util.h"
+
+namespace rocksdb {
+
+static const std::string option_file_header =
+    "# This is a RocksDB option file.\n"
+    "#\n"
+    "# For detailed file format spec, please refer to the example file\n"
+    "# in examples/rocksdb_option_file_example.ini\n"
+    "#\n"
+    "\n";
+
+Status PersistRocksDBOptions(const DBOptions& db_opt,
+                             const std::vector<std::string>& cf_names,
+                             const std::vector<ColumnFamilyOptions>& cf_opts,
+                             const std::string& file_name, Env* env) {
+  if (cf_names.size() != cf_opts.size()) {
+    return Status::InvalidArgument(
+        "cf_names.size() and cf_opts.size() must be the same");
+  }
+  std::unique_ptr<WritableFile> writable;
+
+  Status s = env->NewWritableFile(file_name, &writable, EnvOptions());
+  if (!s.ok()) {
+    return s;
+  }
+  std::string options_file_content;
+
+  writable->Append(option_file_header + "[" +
+                   opt_section_titles[kOptionSectionVersion] +
+                   "]\n"
+                   "  rocksdb_version=" +
+                   ToString(ROCKSDB_MAJOR) + "." + ToString(ROCKSDB_MINOR) +
+                   "." + ToString(ROCKSDB_PATCH) + "\n");
+  writable->Append("  options_file_version=" +
+                   ToString(ROCKSDB_OPTION_FILE_MAJOR) + "." +
+                   ToString(ROCKSDB_OPTION_FILE_MINOR) + "\n");
+  writable->Append("\n[" + opt_section_titles[kOptionSectionDBOptions] +
+                   "]\n  ");
+
+  s = GetStringFromDBOptions(&options_file_content, db_opt, "\n  ");
+  if (!s.ok()) {
+    writable->Close();
+    return s;
+  }
+  writable->Append(options_file_content + "\n");
+
+  for (size_t i = 0; i < cf_opts.size(); ++i) {
+    writable->Append("\n[" + opt_section_titles[kOptionSectionCFOptions] +
+                     " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
+    s = GetStringFromColumnFamilyOptions(&options_file_content, cf_opts[i],
+                                         "\n  ");
+    if (!s.ok()) {
+      writable->Close();
+      return s;
+    }
+    writable->Append(options_file_content + "\n");
+  }
+  writable->Flush();
+  writable->Fsync();
+  writable->Close();
+
+  return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
+      db_opt, cf_names, cf_opts, file_name, env);
+}
+
+RocksDBOptionsParser::RocksDBOptionsParser() { Reset(); }
+
+void RocksDBOptionsParser::Reset() {
+  db_opt_ = DBOptions();
+  cf_names_.clear();
+  cf_opts_.clear();
+  has_version_section_ = false;
+  has_db_options_ = false;
+  has_default_cf_options_ = false;
+  for (int i = 0; i < 3; ++i) {
+    db_version[i] = 0;
+    opt_file_version[i] = 0;
+  }
+}
+
+bool RocksDBOptionsParser::IsSection(const std::string& line) {
+  if (line.size() < 2) {
+    return false;
+  }
+  if (line[0] != '[' || line[line.size() - 1] != ']') {
+    return false;
+  }
+  return true;
+}
+
+Status RocksDBOptionsParser::ParseSection(OptionSection* section,
+                                          std::string* argument,
+                                          const std::string& line,
+                                          const int line_num) {
+  *section = kOptionSectionUnknown;
+  std::string sec_string;
+  // A section is of the form [<SectionName> "<SectionArg>"], where
+  // "<SectionArg>" is optional.
+  size_t arg_start_pos = line.find("\"");
+  size_t arg_end_pos = line.rfind("\"");
+  // The following if-then check tries to identify whether the input
+  // section has the optional section argument.
+  if (arg_start_pos != std::string::npos && arg_start_pos != arg_end_pos) {
+    sec_string = TrimAndRemoveComment(line.substr(1, arg_start_pos - 1), true);
+    *argument = UnescapeOptionString(
+        line.substr(arg_start_pos + 1, arg_end_pos - arg_start_pos - 1));
+  } else {
+    sec_string = TrimAndRemoveComment(line.substr(1, line.size() - 2), true);
+    *argument = "";
+  }
+  for (int i = 0; i < kOptionSectionUnknown; ++i) {
+    if (opt_section_titles[i] == sec_string) {
+      *section = static_cast<OptionSection>(i);
+      return CheckSection(*section, *argument, line_num);
+    }
+  }
+  return Status::InvalidArgument(std::string("Unknown section ") + line);
+}
+
+Status RocksDBOptionsParser::InvalidArgument(const int line_num,
+                                             const std::string& message) {
+  return Status::InvalidArgument(
+      "[RocksDBOptionsParser Error] ",
+      message + " (at line " + ToString(line_num) + ")");
+}
+
+Status RocksDBOptionsParser::ParseStatement(std::string* name,
+                                            std::string* value,
+                                            const std::string& line,
+                                            const int line_num) {
+  size_t eq_pos = line.find("=");
+  if (eq_pos == std::string::npos) {
+    return InvalidArgument(line_num, "A valid statement must have a '='.");
+  }
+
+  *name = TrimAndRemoveComment(line.substr(0, eq_pos), true);
+  *value =
+      TrimAndRemoveComment(line.substr(eq_pos + 1, line.size() - eq_pos - 1));
+  if (name->empty()) {
+    return InvalidArgument(line_num,
+                           "A valid statement must have a variable name.");
+  }
+  return Status::OK();
+}
+
+namespace {
+bool ReadOneLine(std::istringstream* iss, SequentialFile* seq_file,
+                 std::string* output, bool* has_data, Status* result) {
+  const int kBufferSize = 4096;
+  char buffer[kBufferSize + 1];
+  Slice input_slice;
+
+  std::string line;
+  bool has_complete_line = false;
+  while (!has_complete_line) {
+    if (std::getline(*iss, line)) {
+      has_complete_line = !iss->eof();
+    } else {
+      has_complete_line = false;
+    }
+    if (!has_complete_line) {
+      // if we're not sure whether we have a complete line,
+      // further read from the file.
+      if (*has_data) {
+        *result = seq_file->Read(kBufferSize, &input_slice, buffer);
+      }
+      if (input_slice.size() == 0) {
+        // meaning we have read all the data
+        *has_data = false;
+        break;
+      } else {
+        iss->str(line + input_slice.ToString());
+        // reset the internal state of iss so that we can keep reading it.
+        iss->clear();
+        *has_data = (input_slice.size() == kBufferSize);
+        continue;
+      }
+    }
+  }
+  *output = line;
+  return *has_data || has_complete_line;
+}
+}  // namespace
+
+Status RocksDBOptionsParser::Parse(const std::string& file_name, Env* env) {
+  Reset();
+
+  std::unique_ptr<SequentialFile> seq_file;
+  Status s = env->NewSequentialFile(file_name, &seq_file, EnvOptions());
+  if (!s.ok()) {
+    return s;
+  }
+
+  OptionSection section = kOptionSectionUnknown;
+  std::string argument;
+  std::unordered_map<std::string, std::string> opt_map;
+  std::istringstream iss;
+  std::string line;
+  bool has_data = true;
+  // we only support single-lined statement.
+  for (int line_num = 1;
+       ReadOneLine(&iss, seq_file.get(), &line, &has_data, &s); ++line_num) {
+    if (!s.ok()) {
+      return s;
+    }
+    line = TrimAndRemoveComment(line);
+    if (line.empty()) {
+      continue;
+    }
+    if (IsSection(line)) {
+      s = EndSection(section, argument, opt_map);
+      opt_map.clear();
+      if (!s.ok()) {
+        return s;
+      }
+      s = ParseSection(&section, &argument, line, line_num);
+      if (!s.ok()) {
+        return s;
+      }
+    } else {
+      std::string name;
+      std::string value;
+      s = ParseStatement(&name, &value, line, line_num);
+      if (!s.ok()) {
+        return s;
+      }
+      opt_map.insert({name, value});
+    }
+  }
+
+  s = EndSection(section, argument, opt_map);
+  opt_map.clear();
+  if (!s.ok()) {
+    return s;
+  }
+  return ValidityCheck();
+}
+
+Status RocksDBOptionsParser::CheckSection(const OptionSection section,
+                                          const std::string& section_arg,
+                                          const int line_num) {
+  if (section == kOptionSectionDBOptions) {
+    if (has_db_options_) {
+      return InvalidArgument(
+          line_num,
+          "More than one DBOption section found in the option config file");
+    }
+    has_db_options_ = true;
+  } else if (section == kOptionSectionCFOptions) {
+    bool is_default_cf = (section_arg == kDefaultColumnFamilyName);
+    if (cf_opts_.size() == 0 && !is_default_cf) {
+      return InvalidArgument(
+          line_num,
+          "Default column family must be the first CFOptions section "
+          "in the option config file");
+    } else if (cf_opts_.size() != 0 && is_default_cf) {
+      return InvalidArgument(
+          line_num,
+          "Default column family must be the first CFOptions section "
+          "in the option config file");
+    } else if (GetCFOptions(section_arg) != nullptr) {
+      return InvalidArgument(
+          line_num,
+          "Two identical column families found in option config file");
+    }
+    has_default_cf_options_ |= is_default_cf;
+  } else if (section == kOptionSectionVersion) {
+    if (has_version_section_) {
+      return InvalidArgument(
+          line_num,
+          "More than one Version section found in the option config file.");
+    }
+    has_version_section_ = true;
+  }
+  return Status::OK();
+}
+
+Status RocksDBOptionsParser::ParseVersionNumber(const std::string& ver_name,
+                                                const std::string& ver_string,
+                                                const int max_count,
+                                                int* version) {
+  int version_index = 0;
+  int current_number = 0;
+  int current_digit_count = 0;
+  bool has_dot = false;
+  for (int i = 0; i < max_count; ++i) {
+    version[i] = 0;
+  }
+  const int kBufferSize = 200;
+  char buffer[kBufferSize];
+  for (size_t i = 0; i < ver_string.size(); ++i) {
+    if (ver_string[i] == '.') {
+      if (version_index >= max_count - 1) {
+        snprintf(buffer, sizeof(buffer) - 1,
+                 "A valid %s can only contains at most %d dots.",
+                 ver_name.c_str(), max_count - 1);
+        return Status::InvalidArgument(buffer);
+      }
+      if (current_digit_count == 0) {
+        snprintf(buffer, sizeof(buffer) - 1,
+                 "A valid %s must have at least one digit before each dot.",
+                 ver_name.c_str());
+        return Status::InvalidArgument(buffer);
+      }
+      version[version_index++] = current_number;
+      current_number = 0;
+      current_digit_count = 0;
+      has_dot = true;
+    } else if (isdigit(ver_string[i])) {
+      current_number = current_number * 10 + (ver_string[i] - '0');
+      current_digit_count++;
+    } else {
+      snprintf(buffer, sizeof(buffer) - 1,
+               "A valid %s can only contains dots and numbers.",
+               ver_name.c_str());
+      return Status::InvalidArgument(buffer);
+    }
+  }
+  version[version_index] = current_number;
+  if (has_dot && current_digit_count == 0) {
+    snprintf(buffer, sizeof(buffer) - 1,
+             "A valid %s must have at least one digit after each dot.",
+             ver_name.c_str());
+    return Status::InvalidArgument(buffer);
+  }
+  return Status::OK();
+}
+
+Status RocksDBOptionsParser::EndSection(
+    const OptionSection section, const std::string& section_arg,
+    const std::unordered_map<std::string, std::string>& opt_map) {
+  Status s;
+  if (section == kOptionSectionDBOptions) {
+    s = GetDBOptionsFromMap(DBOptions(), opt_map, &db_opt_, true);
+    if (!s.ok()) {
+      return s;
+    }
+  } else if (section == kOptionSectionCFOptions) {
+    // This condition should be ensured earlier in ParseSection
+    // so we make an assertion here.
+    assert(GetCFOptions(section_arg) == nullptr);
+    cf_names_.emplace_back(section_arg);
+    cf_opts_.emplace_back();
+    s = GetColumnFamilyOptionsFromMap(ColumnFamilyOptions(), opt_map,
+                                      &cf_opts_.back(), true);
+    if (!s.ok()) {
+      return s;
+    }
+  } else if (section == kOptionSectionVersion) {
+    for (const auto pair : opt_map) {
+      if (pair.first == "rocksdb_version") {
+        s = ParseVersionNumber(pair.first, pair.second, 3, db_version);
+        if (!s.ok()) {
+          return s;
+        }
+      } else if (pair.first == "options_file_version") {
+        s = ParseVersionNumber(pair.first, pair.second, 2, opt_file_version);
+        if (!s.ok()) {
+          return s;
+        }
+        if (opt_file_version[0] < 1) {
+          return Status::InvalidArgument(
+              "A valid options_file_version must be at least 1.");
+        }
+      }
+    }
+  }
+  return Status::OK();
+}
+
+Status RocksDBOptionsParser::ValidityCheck() {
+  if (!has_db_options_) {
+    return Status::Corruption(
+        "A RocksDB Option file must have a single DBOptions section");
+  }
+  if (!has_default_cf_options_) {
+    return Status::Corruption(
+        "A RocksDB Option file must have a single CFOptions:default section");
+  }
+
+  return Status::OK();
+}
+
+std::string RocksDBOptionsParser::TrimAndRemoveComment(const std::string& line,
+                                                       bool trim_only) {
+  size_t start = 0;
+  size_t end = line.size();
+
+  // we only support "#" style comment
+  if (!trim_only) {
+    size_t search_pos = 0;
+    while (search_pos < line.size()) {
+      size_t comment_pos = line.find('#', search_pos);
+      if (comment_pos == std::string::npos) {
+        break;
+      }
+      if (comment_pos == 0 || line[comment_pos - 1] != '\\') {
+        end = comment_pos;
+        break;
+      }
+      search_pos = comment_pos + 1;
+    }
+  }
+
+  while (start < end && isspace(line[start]) != 0) {
+    ++start;
+  }
+
+  // start < end implies end > 0.
+  while (start < end && isspace(line[end - 1]) != 0) {
+    --end;
+  }
+
+  if (start < end) {
+    return line.substr(start, end - start);
+  }
+
+  return "";
+}
+
+namespace {
+bool AreEqualDoubles(const double a, const double b) {
+  return (fabs(a - b) < 0.00001);
+}
+
+bool AreEqualOptions(const char* opt1, const char* opt2,
+                     const OptionTypeInfo& type_info) {
+  const char* offset1 = opt1 + type_info.offset;
+  const char* offset2 = opt2 + type_info.offset;
+  switch (type_info.type) {
+    case OptionType::kBoolean:
+      return (*reinterpret_cast<const bool*>(offset1) ==
+              *reinterpret_cast<const bool*>(offset2));
+    case OptionType::kInt:
+      return (*reinterpret_cast<const int*>(offset1) ==
+              *reinterpret_cast<const int*>(offset2));
+    case OptionType::kUInt:
+      return (*reinterpret_cast<const unsigned int*>(offset1) ==
+              *reinterpret_cast<const unsigned int*>(offset2));
+    case OptionType::kUInt32T:
+      return (*reinterpret_cast<const uint32_t*>(offset1) ==
+              *reinterpret_cast<const uint32_t*>(offset2));
+    case OptionType::kUInt64T:
+      return (*reinterpret_cast<const uint64_t*>(offset1) ==
+              *reinterpret_cast<const uint64_t*>(offset2));
+    case OptionType::kSizeT:
+      return (*reinterpret_cast<const size_t*>(offset1) ==
+              *reinterpret_cast<const size_t*>(offset2));
+    case OptionType::kString:
+      return (*reinterpret_cast<const std::string*>(offset1) ==
+              *reinterpret_cast<const std::string*>(offset2));
+    case OptionType::kDouble:
+      return AreEqualDoubles(*reinterpret_cast<const double*>(offset1),
+                             *reinterpret_cast<const double*>(offset2));
+    case OptionType::kCompactionStyle:
+      return (*reinterpret_cast<const CompactionStyle*>(offset1) ==
+              *reinterpret_cast<const CompactionStyle*>(offset2));
+    default:
+      return false;
+  }
+}
+
+}  // namespace
+
+Status RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
+    const DBOptions& db_opt, const std::vector<std::string>& cf_names,
+    const std::vector<ColumnFamilyOptions>& cf_opts,
+    const std::string& file_name, Env* env) {
+  RocksDBOptionsParser parser;
+  std::unique_ptr<SequentialFile> seq_file;
+  Status s = parser.Parse(file_name, env);
+  if (!s.ok()) {
+    return s;
+  }
+
+  // Verify DBOptions
+  s = VerifyDBOptions(db_opt, *parser.db_opt());
+  if (!s.ok()) {
+    return s;
+  }
+
+  // Verify ColumnFamily Name
+  if (cf_names.size() != parser.cf_names()->size()) {
+    return Status::Corruption(
+        "[RocksDBOptionParser Error] The persisted options does not have"
+        "the same number of column family names as the db instance.");
+  }
+  for (size_t i = 0; i < cf_names.size(); ++i) {
+    if (cf_names[i] != parser.cf_names()->at(i)) {
+      return Status::Corruption(
+          "[RocksDBOptionParser Error] The persisted options and the db"
+          "instance does not have the same name for column family ",
+          ToString(i));
+    }
+  }
+
+  // Verify Column Family Options
+  if (cf_opts.size() != parser.cf_opts()->size()) {
+    return Status::Corruption(
+        "[RocksDBOptionParser Error] The persisted options does not have"
+        "the same number of column families as the db instance.");
+  }
+  for (size_t i = 0; i < cf_opts.size(); ++i) {
+    s = VerifyCFOptions(cf_opts[i], parser.cf_opts()->at(i));
+    if (!s.ok()) {
+      return s;
+    }
+  }
+
+  return Status::OK();
+}
+
+Status RocksDBOptionsParser::VerifyDBOptions(const DBOptions& base_opt,
+                                             const DBOptions& new_opt) {
+  for (auto pair : db_options_type_info) {
+    if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
+                         reinterpret_cast<const char*>(&new_opt),
+                         pair.second)) {
+      return Status::Corruption(
+          "[RocksDBOptionsParser]: "
+          "failed the verification on DBOptions::",
+          pair.first);
+    }
+  }
+  return Status::OK();
+}
+
+Status RocksDBOptionsParser::VerifyCFOptions(
+    const ColumnFamilyOptions& base_opt, const ColumnFamilyOptions& new_opt) {
+  for (auto& pair : cf_options_type_info) {
+    if (!AreEqualOptions(reinterpret_cast<const char*>(&base_opt),
+                         reinterpret_cast<const char*>(&new_opt),
+                         pair.second)) {
+      return Status::Corruption(
+          "[RocksDBOptionsParser]: "
+          "failed the verification on ColumnFamilyOptions::",
+          pair.first);
+    }
+  }
+  return Status::OK();
+}
+}  // namespace rocksdb
+
+#endif  // !ROCKSDB_LITE

util/options_parser.h

@@ -0,0 +1,111 @@
+// Copyright (c) 2014, Facebook, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+
+#pragma once
+
+#include <map>
+#include <string>
+#include <vector>
+
+#include "rocksdb/env.h"
+#include "rocksdb/options.h"
+
+namespace rocksdb {
+
+#ifndef ROCKSDB_LITE
+
+#define ROCKSDB_OPTION_FILE_MAJOR 1
+#define ROCKSDB_OPTION_FILE_MINOR 0
+
+enum OptionSection : char {
+  kOptionSectionVersion = 0,
+  kOptionSectionDBOptions,
+  kOptionSectionCFOptions,
+  kOptionSectionUnknown
+};
+
+static const std::string opt_section_titles[] = {"Version", "DBOptions",
+                                                 "CFOptions", "Unknown"};
+
+Status PersistRocksDBOptions(const DBOptions& db_opt,
+                             const std::vector<std::string>& cf_names,
+                             const std::vector<ColumnFamilyOptions>& cf_opts,
+                             const std::string& file_name, Env* env);
+
+class RocksDBOptionsParser {
+ public:
+  explicit RocksDBOptionsParser();
+  ~RocksDBOptionsParser() {}
+  void Reset();
+
+  Status Parse(const std::string& file_name, Env* env);
+  static std::string TrimAndRemoveComment(const std::string& line,
+                                          const bool trim_only = false);
+
+  const DBOptions* db_opt() const { return &db_opt_; }
+  const std::vector<ColumnFamilyOptions>* cf_opts() const { return &cf_opts_; }
+  const std::vector<std::string>* cf_names() const { return &cf_names_; }
+
+  const ColumnFamilyOptions* GetCFOptions(const std::string& name) const {
+    assert(cf_names_.size() == cf_opts_.size());
+    for (size_t i = 0; i < cf_names_.size(); ++i) {
+      if (cf_names_[i] == name) {
+        return &cf_opts_[i];
+      }
+    }
+    return nullptr;
+  }
+  size_t NumColumnFamilies() { return cf_opts_.size(); }
+
+  static Status VerifyRocksDBOptionsFromFile(
+      const DBOptions& db_opt, const std::vector<std::string>& cf_names,
+      const std::vector<ColumnFamilyOptions>& cf_opts,
+      const std::string& file_name, Env* env);
+
+  static Status VerifyDBOptions(const DBOptions& base_opt,
+                                const DBOptions& new_opt);
+
+  static Status VerifyCFOptions(const ColumnFamilyOptions& base_opt,
+                                const ColumnFamilyOptions& new_opt);
+
+  static Status ExtraParserCheck(const RocksDBOptionsParser& input_parser);
+
+ protected:
+  bool IsSection(const std::string& line);
+  Status ParseSection(OptionSection* section, std::string* argument,
+                      const std::string& line, const int line_num);
+
+  Status CheckSection(const OptionSection section,
+                      const std::string& section_arg, const int line_num);
+
+  Status ParseStatement(std::string* name, std::string* value,
+                        const std::string& line, const int line_num);
+
+  Status EndSection(
+      const OptionSection section, const std::string& section_arg,
+      const std::unordered_map<std::string, std::string>& opt_map);
+
+  Status ValidityCheck();
+
+  Status InvalidArgument(const int line_num, const std::string& message);
+
+  Status ParseVersionNumber(const std::string& ver_name,
+                            const std::string& ver_string, const int max_count,
+                            int* version);
+
+ private:
+  DBOptions db_opt_;
+  std::vector<std::string> cf_names_;
+  std::vector<ColumnFamilyOptions> cf_opts_;
+  bool has_version_section_;
+  bool has_db_options_;
+  bool has_default_cf_options_;
+  int db_version[3];
+  int opt_file_version[3];
+};
+
+#endif  // !ROCKSDB_LITE
+
+}  // namespace rocksdb

util/options_test.cc

@@ -11,6 +11,7 @@
 #define __STDC_FORMAT_MACROS
 #endif
 
+#include <cctype>
 #include <unordered_map>
 #include <inttypes.h>
 
@@ -20,8 +21,11 @@
 #include "rocksdb/table.h"
 #include "rocksdb/utilities/leveldb_options.h"
 #include "table/block_based_table_factory.h"
+#include "util/options_helper.h"
+#include "util/options_parser.h"
 #include "util/random.h"
 #include "util/testharness.h"
+#include "util/testutil.h"
 
 #ifndef GFLAGS
 bool FLAGS_enable_print = false;
@@ -33,8 +37,6 @@ DEFINE_bool(enable_print, false, "Print options generated to console.");
 
 namespace rocksdb {
 
-class OptionsTest : public testing::Test {};
-
 class StderrLogger : public Logger {
  public:
   using Logger::Logv;
@@ -69,6 +71,165 @@ Options PrintAndGetOptions(size_t total_write_buffer_limit,
   return options;
 }
 
+class StringEnv : public EnvWrapper {
+ public:
+  class SeqStringSource : public SequentialFile {
+   public:
+    explicit SeqStringSource(const std::string& data)
+        : data_(data), offset_(0) {}
+    ~SeqStringSource() {}
+    Status Read(size_t n, Slice* result, char* scratch) override {
+      std::string output;
+      if (offset_ < data_.size()) {
+        n = std::min(data_.size() - offset_, n);
+        memcpy(scratch, data_.data() + offset_, n);
+        offset_ += n;
+        *result = Slice(scratch, n);
+      } else {
+        return Status::InvalidArgument(
+            "Attemp to read when it already reached eof.");
+      }
+      return Status::OK();
+    }
+    Status Skip(uint64_t n) {
+      if (offset_ >= data_.size()) {
+        return Status::InvalidArgument(
+            "Attemp to read when it already reached eof.");
+      }
+      // TODO(yhchiang): Currently doesn't handle the overflow case.
+      offset_ += n;
+      return Status::OK();
+    }
+
+   private:
+    std::string data_;
+    size_t offset_;
+  };
+
+  class StringSink : public WritableFile {
+   public:
+    explicit StringSink(std::string* contents)
+        : WritableFile(), contents_(contents) {}
+    virtual Status Truncate(uint64_t size) override {
+      contents_->resize(size);
+      return Status::OK();
+    }
+    virtual Status Close() override { return Status::OK(); }
+    virtual Status Flush() override { return Status::OK(); }
+    virtual Status Sync() override { return Status::OK(); }
+    virtual Status Append(const Slice& slice) override {
+      contents_->append(slice.data(), slice.size());
+      return Status::OK();
+    }
+
+   private:
+    std::string* contents_;
+  };
+
+  explicit StringEnv(Env* t) : EnvWrapper(t) {}
+  virtual ~StringEnv() {}
+
+  const std::string& GetContent(const std::string& f) { return files_[f]; }
+
+  const Status WriteToNewFile(const std::string& file_name,
+                              const std::string& content) {
+    unique_ptr<WritableFile> r;
+    auto s = NewWritableFile(file_name, &r, EnvOptions());
+    if (!s.ok()) {
+      return s;
+    }
+    r->Append(content);
+    r->Flush();
+    r->Close();
+    assert(files_[file_name] == content);
+    return Status::OK();
+  }
+
+  // The following text is boilerplate that forwards all methods to target()
+  Status NewSequentialFile(const std::string& f, unique_ptr<SequentialFile>* r,
+                           const EnvOptions& options) override {
+    auto iter = files_.find(f);
+    if (iter == files_.end()) {
+      return Status::NotFound("The specified file does not exist", f);
+    }
+    r->reset(new SeqStringSource(iter->second));
+    return Status::OK();
+  }
+  Status NewRandomAccessFile(const std::string& f,
+                             unique_ptr<RandomAccessFile>* r,
+                             const EnvOptions& options) override {
+    return Status::NotSupported();
+  }
+  Status NewWritableFile(const std::string& f, unique_ptr<WritableFile>* r,
+                         const EnvOptions& options) override {
+    auto iter = files_.find(f);
+    if (iter != files_.end()) {
+      return Status::IOError("The specified file already exists", f);
+    }
+    r->reset(new StringSink(&files_[f]));
+    return Status::OK();
+  }
+  virtual Status NewDirectory(const std::string& name,
+                              unique_ptr<Directory>* result) override {
+    return Status::NotSupported();
+  }
+  Status FileExists(const std::string& f) override {
+    if (files_.find(f) == files_.end()) {
+      return Status::NotFound();
+    }
+    return Status::OK();
+  }
+  Status GetChildren(const std::string& dir,
+                     std::vector<std::string>* r) override {
+    return Status::NotSupported();
+  }
+  Status DeleteFile(const std::string& f) override {
+    files_.erase(f);
+    return Status::OK();
+  }
+  Status CreateDir(const std::string& d) override {
+    return Status::NotSupported();
+  }
+  Status CreateDirIfMissing(const std::string& d) override {
+    return Status::NotSupported();
+  }
+  Status DeleteDir(const std::string& d) override {
+    return Status::NotSupported();
+  }
+  Status GetFileSize(const std::string& f, uint64_t* s) override {
+    auto iter = files_.find(f);
+    if (iter == files_.end()) {
+      return Status::NotFound("The specified file does not exist:", f);
+    }
+    *s = iter->second.size();
+    return Status::OK();
+  }
+
+  Status GetFileModificationTime(const std::string& fname,
+                                 uint64_t* file_mtime) override {
+    return Status::NotSupported();
+  }
+
+  Status RenameFile(const std::string& s, const std::string& t) override {
+    return Status::NotSupported();
+  }
+
+  Status LinkFile(const std::string& s, const std::string& t) override {
+    return Status::NotSupported();
+  }
+
+  Status LockFile(const std::string& f, FileLock** l) override {
+    return Status::NotSupported();
+  }
+
+  Status UnlockFile(FileLock* l) override { return Status::NotSupported(); }
+
+ protected:
+  std::unordered_map<std::string, std::string> files_;
+};
+
+class OptionsTest : public testing::Test {};
+
 TEST_F(OptionsTest, LooseCondition) {
   Options options;
   PrintAndGetOptions(static_cast<size_t>(10) * 1024 * 1024 * 1024, 100, 100);
@@ -512,66 +673,60 @@ TEST_F(OptionsTest, GetOptionsFromStringTest) {
 }
 
 namespace {
-void VerifyDBOptions(const DBOptions& base_opt, const DBOptions& new_opt) {
+void RandomInitDBOptions(DBOptions* db_opt, Random* rnd) {
   // boolean options
-  ASSERT_EQ(base_opt.advise_random_on_open, new_opt.advise_random_on_open);
+  db_opt->advise_random_on_open = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.allow_mmap_reads, new_opt.allow_mmap_reads);
+  db_opt->allow_mmap_reads = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.allow_mmap_writes, new_opt.allow_mmap_writes);
+  db_opt->allow_mmap_writes = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.allow_os_buffer, new_opt.allow_os_buffer);
+  db_opt->allow_os_buffer = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.create_if_missing, new_opt.create_if_missing);
+  db_opt->create_if_missing = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.create_missing_column_families,
+  db_opt->create_missing_column_families = rnd->Uniform(2);
-            new_opt.create_missing_column_families);
+  db_opt->disableDataSync = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.disableDataSync, new_opt.disableDataSync);
+  db_opt->enable_thread_tracking = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.enable_thread_tracking, new_opt.enable_thread_tracking);
+  db_opt->error_if_exists = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.error_if_exists, new_opt.error_if_exists);
+  db_opt->is_fd_close_on_exec = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.is_fd_close_on_exec, new_opt.is_fd_close_on_exec);
+  db_opt->paranoid_checks = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.paranoid_checks, new_opt.paranoid_checks);
+  db_opt->skip_log_error_on_recovery = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.skip_log_error_on_recovery,
+  db_opt->skip_stats_update_on_db_open = rnd->Uniform(2);
-            new_opt.skip_log_error_on_recovery);
+  db_opt->use_adaptive_mutex = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.skip_stats_update_on_db_open,
+  db_opt->use_fsync = rnd->Uniform(2);
-            new_opt.skip_stats_update_on_db_open);
-  ASSERT_EQ(base_opt.use_adaptive_mutex, new_opt.use_adaptive_mutex);
-  ASSERT_EQ(base_opt.use_fsync, new_opt.use_fsync);
 
   // int options
-  ASSERT_EQ(base_opt.max_background_compactions,
+  db_opt->max_background_compactions = rnd->Uniform(100);
-            new_opt.max_background_compactions);
+  db_opt->max_background_flushes = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.max_background_flushes, new_opt.max_background_flushes);
+  db_opt->max_file_opening_threads = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.max_file_opening_threads,
+  db_opt->max_open_files = rnd->Uniform(100);
-            new_opt.max_file_opening_threads);
+  db_opt->table_cache_numshardbits = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.max_open_files, new_opt.max_open_files);
-  ASSERT_EQ(base_opt.table_cache_numshardbits,
-            new_opt.table_cache_numshardbits);
 
   // size_t options
-  ASSERT_EQ(base_opt.db_write_buffer_size, new_opt.db_write_buffer_size);
+  db_opt->db_write_buffer_size = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.keep_log_file_num, new_opt.keep_log_file_num);
+  db_opt->keep_log_file_num = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.log_file_time_to_roll, new_opt.log_file_time_to_roll);
+  db_opt->log_file_time_to_roll = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.manifest_preallocation_size,
+  db_opt->manifest_preallocation_size = rnd->Uniform(10000);
-            new_opt.manifest_preallocation_size);
+  db_opt->max_log_file_size = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.max_log_file_size, new_opt.max_log_file_size);
 
   // std::string options
-  ASSERT_EQ(base_opt.db_log_dir, new_opt.db_log_dir);
+  db_opt->db_log_dir = "path/to/db_log_dir";
-  ASSERT_EQ(base_opt.wal_dir, new_opt.wal_dir);
+  db_opt->wal_dir = "path/to/wal_dir";
 
   // uint32_t options
-  ASSERT_EQ(base_opt.max_subcompactions, new_opt.max_subcompactions);
+  db_opt->max_subcompactions = rnd->Uniform(100000);
 
   // uint64_t options
-  ASSERT_EQ(base_opt.WAL_size_limit_MB, new_opt.WAL_size_limit_MB);
+  static const uint64_t uint_max = static_cast<uint64_t>(UINT_MAX);
-  ASSERT_EQ(base_opt.WAL_ttl_seconds, new_opt.WAL_ttl_seconds);
+  db_opt->WAL_size_limit_MB = uint_max + rnd->Uniform(100000);
-  ASSERT_EQ(base_opt.bytes_per_sync, new_opt.bytes_per_sync);
+  db_opt->WAL_ttl_seconds = uint_max + rnd->Uniform(100000);
-  ASSERT_EQ(base_opt.delayed_write_rate, new_opt.delayed_write_rate);
+  db_opt->bytes_per_sync = uint_max + rnd->Uniform(100000);
-  ASSERT_EQ(base_opt.delete_obsolete_files_period_micros,
+  db_opt->delayed_write_rate = uint_max + rnd->Uniform(100000);
-            new_opt.delete_obsolete_files_period_micros);
+  db_opt->delete_obsolete_files_period_micros = uint_max + rnd->Uniform(100000);
-  ASSERT_EQ(base_opt.max_manifest_file_size, new_opt.max_manifest_file_size);
+  db_opt->max_manifest_file_size = uint_max + rnd->Uniform(100000);
-  ASSERT_EQ(base_opt.max_total_wal_size, new_opt.max_total_wal_size);
+  db_opt->max_total_wal_size = uint_max + rnd->Uniform(100000);
-  ASSERT_EQ(base_opt.wal_bytes_per_sync, new_opt.wal_bytes_per_sync);
+  db_opt->wal_bytes_per_sync = uint_max + rnd->Uniform(100000);
 
   // unsigned int options
-  ASSERT_EQ(base_opt.stats_dump_period_sec, new_opt.stats_dump_period_sec);
+  db_opt->stats_dump_period_sec = rnd->Uniform(100000);
 }
+
 }  // namespace
 
 TEST_F(OptionsTest, DBOptionsSerialization) {
@@ -579,154 +734,77 @@ TEST_F(OptionsTest, DBOptionsSerialization) {
   Random rnd(301);
 
   // Phase 1: Make big change in base_options
-  // boolean options
+  RandomInitDBOptions(&base_options, &rnd);
-  base_options.advise_random_on_open = rnd.Uniform(2);
-  base_options.allow_mmap_reads = rnd.Uniform(2);
-  base_options.allow_mmap_writes = rnd.Uniform(2);
-  base_options.allow_os_buffer = rnd.Uniform(2);
-  base_options.create_if_missing = rnd.Uniform(2);
-  base_options.create_missing_column_families = rnd.Uniform(2);
-  base_options.disableDataSync = rnd.Uniform(2);
-  base_options.enable_thread_tracking = rnd.Uniform(2);
-  base_options.error_if_exists = rnd.Uniform(2);
-  base_options.is_fd_close_on_exec = rnd.Uniform(2);
-  base_options.paranoid_checks = rnd.Uniform(2);
-  base_options.skip_log_error_on_recovery = rnd.Uniform(2);
-  base_options.skip_stats_update_on_db_open = rnd.Uniform(2);
-  base_options.use_adaptive_mutex = rnd.Uniform(2);
-  base_options.use_fsync = rnd.Uniform(2);
-
-  // int options
-  base_options.max_background_compactions = rnd.Uniform(100);
-  base_options.max_background_flushes = rnd.Uniform(100);
-  base_options.max_file_opening_threads = rnd.Uniform(100);
-  base_options.max_open_files = rnd.Uniform(100);
-  base_options.table_cache_numshardbits = rnd.Uniform(100);
-
-  // size_t options
-  base_options.db_write_buffer_size = rnd.Uniform(10000);
-  base_options.keep_log_file_num = rnd.Uniform(10000);
-  base_options.log_file_time_to_roll = rnd.Uniform(10000);
-  base_options.manifest_preallocation_size = rnd.Uniform(10000);
-  base_options.max_log_file_size = rnd.Uniform(10000);
-
-  // std::string options
-  base_options.db_log_dir = "path/to/db_log_dir";
-  base_options.wal_dir = "path/to/wal_dir";
-
-  // uint32_t options
-  base_options.max_subcompactions = rnd.Uniform(100000);
-
-  // uint64_t options
-  static const uint64_t uint_max = static_cast<uint64_t>(UINT_MAX);
-  base_options.WAL_size_limit_MB = uint_max + rnd.Uniform(100000);
-  base_options.WAL_ttl_seconds = uint_max + rnd.Uniform(100000);
-  base_options.bytes_per_sync = uint_max + rnd.Uniform(100000);
-  base_options.delayed_write_rate = uint_max + rnd.Uniform(100000);
-  base_options.delete_obsolete_files_period_micros =
-      uint_max + rnd.Uniform(100000);
-  base_options.max_manifest_file_size = uint_max + rnd.Uniform(100000);
-  base_options.max_total_wal_size = uint_max + rnd.Uniform(100000);
-  base_options.wal_bytes_per_sync = uint_max + rnd.Uniform(100000);
-
-  // unsigned int options
-  base_options.stats_dump_period_sec = rnd.Uniform(100000);
 
   // Phase 2: obtain a string from base_option
-  std::string base_opt_string;
+  std::string base_options_file_content;
-  ASSERT_OK(GetStringFromDBOptions(base_options, &base_opt_string));
+  ASSERT_OK(GetStringFromDBOptions(&base_options_file_content, base_options));
 
   // Phase 3: Set new_options from the derived string and expect
   //          new_options == base_options
-  ASSERT_OK(GetDBOptionsFromString(DBOptions(), base_opt_string, &new_options));
+  ASSERT_OK(GetDBOptionsFromString(DBOptions(), base_options_file_content,
-  VerifyDBOptions(base_options, new_options);
+                                   &new_options));
+  ASSERT_OK(RocksDBOptionsParser::VerifyDBOptions(base_options, new_options));
 }
 
 namespace {
-void VerifyDouble(double a, double b) { ASSERT_LT(fabs(a - b), 0.00001); }
 
-void VerifyColumnFamilyOptions(const ColumnFamilyOptions& base_opt,
+void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, Random* rnd) {
-                               const ColumnFamilyOptions& new_opt) {
+  cf_opt->compaction_style = (CompactionStyle)(rnd->Uniform(4));
-  // custom type options
-  ASSERT_EQ(base_opt.compaction_style, new_opt.compaction_style);
 
   // boolean options
-  ASSERT_EQ(base_opt.compaction_measure_io_stats,
+  cf_opt->compaction_measure_io_stats = rnd->Uniform(2);
-            new_opt.compaction_measure_io_stats);
+  cf_opt->disable_auto_compactions = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.disable_auto_compactions,
+  cf_opt->filter_deletes = rnd->Uniform(2);
-            new_opt.disable_auto_compactions);
+  cf_opt->inplace_update_support = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.filter_deletes, new_opt.filter_deletes);
+  cf_opt->level_compaction_dynamic_level_bytes = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.inplace_update_support, new_opt.inplace_update_support);
+  cf_opt->optimize_filters_for_hits = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.level_compaction_dynamic_level_bytes,
+  cf_opt->paranoid_file_checks = rnd->Uniform(2);
-            new_opt.level_compaction_dynamic_level_bytes);
+  cf_opt->purge_redundant_kvs_while_flush = rnd->Uniform(2);
-  ASSERT_EQ(base_opt.optimize_filters_for_hits,
+  cf_opt->verify_checksums_in_compaction = rnd->Uniform(2);
-            new_opt.optimize_filters_for_hits);
-  ASSERT_EQ(base_opt.paranoid_file_checks, new_opt.paranoid_file_checks);
-  ASSERT_EQ(base_opt.purge_redundant_kvs_while_flush,
-            new_opt.purge_redundant_kvs_while_flush);
-  ASSERT_EQ(base_opt.verify_checksums_in_compaction,
-            new_opt.verify_checksums_in_compaction);
 
   // double options
-  ASSERT_EQ(base_opt.hard_pending_compaction_bytes_limit,
+  cf_opt->hard_rate_limit = static_cast<double>(rnd->Uniform(10000)) / 13;
-            new_opt.hard_pending_compaction_bytes_limit);
+  cf_opt->soft_rate_limit = static_cast<double>(rnd->Uniform(10000)) / 13;
-  VerifyDouble(base_opt.soft_rate_limit, new_opt.soft_rate_limit);
 
   // int options
-  ASSERT_EQ(base_opt.expanded_compaction_factor,
+  cf_opt->expanded_compaction_factor = rnd->Uniform(100);
-            new_opt.expanded_compaction_factor);
+  cf_opt->level0_file_num_compaction_trigger = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.level0_file_num_compaction_trigger,
+  cf_opt->level0_slowdown_writes_trigger = rnd->Uniform(100);
-            new_opt.level0_file_num_compaction_trigger);
+  cf_opt->level0_stop_writes_trigger = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.level0_slowdown_writes_trigger,
+  cf_opt->max_bytes_for_level_multiplier = rnd->Uniform(100);
-            new_opt.level0_slowdown_writes_trigger);
+  cf_opt->max_grandparent_overlap_factor = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.level0_stop_writes_trigger,
+  cf_opt->max_mem_compaction_level = rnd->Uniform(100);
-            new_opt.level0_stop_writes_trigger);
+  cf_opt->max_write_buffer_number = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.max_bytes_for_level_multiplier,
+  cf_opt->max_write_buffer_number_to_maintain = rnd->Uniform(100);
-            new_opt.max_bytes_for_level_multiplier);
+  cf_opt->min_write_buffer_number_to_merge = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.max_grandparent_overlap_factor,
+  cf_opt->num_levels = rnd->Uniform(100);
-            new_opt.max_grandparent_overlap_factor);
+  cf_opt->source_compaction_factor = rnd->Uniform(100);
-  ASSERT_EQ(base_opt.max_mem_compaction_level,
+  cf_opt->target_file_size_multiplier = rnd->Uniform(100);
-            new_opt.max_mem_compaction_level);
-  ASSERT_EQ(base_opt.max_write_buffer_number, new_opt.max_write_buffer_number);
-  ASSERT_EQ(base_opt.max_write_buffer_number_to_maintain,
-            new_opt.max_write_buffer_number_to_maintain);
-  ASSERT_EQ(base_opt.min_write_buffer_number_to_merge,
-            new_opt.min_write_buffer_number_to_merge);
-  ASSERT_EQ(base_opt.num_levels, new_opt.num_levels);
-  ASSERT_EQ(base_opt.source_compaction_factor,
-            new_opt.source_compaction_factor);
-  ASSERT_EQ(base_opt.target_file_size_multiplier,
-            new_opt.target_file_size_multiplier);
 
   // size_t options
-  ASSERT_EQ(base_opt.arena_block_size, new_opt.arena_block_size);
+  cf_opt->arena_block_size = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.inplace_update_num_locks,
+  cf_opt->inplace_update_num_locks = rnd->Uniform(10000);
-            new_opt.inplace_update_num_locks);
+  cf_opt->max_successive_merges = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.max_successive_merges, new_opt.max_successive_merges);
+  cf_opt->memtable_prefix_bloom_huge_page_tlb_size = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.memtable_prefix_bloom_huge_page_tlb_size,
+  cf_opt->write_buffer_size = rnd->Uniform(10000);
-            new_opt.memtable_prefix_bloom_huge_page_tlb_size);
-  ASSERT_EQ(base_opt.write_buffer_size, new_opt.write_buffer_size);
 
   // uint32_t options
-  ASSERT_EQ(base_opt.bloom_locality, new_opt.bloom_locality);
+  cf_opt->bloom_locality = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.memtable_prefix_bloom_bits,
+  cf_opt->memtable_prefix_bloom_bits = rnd->Uniform(10000);
-            new_opt.memtable_prefix_bloom_bits);
+  cf_opt->memtable_prefix_bloom_probes = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.memtable_prefix_bloom_probes,
+  cf_opt->min_partial_merge_operands = rnd->Uniform(10000);
-            new_opt.memtable_prefix_bloom_probes);
+  cf_opt->max_bytes_for_level_base = rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.min_partial_merge_operands,
-            new_opt.min_partial_merge_operands);
-  ASSERT_EQ(base_opt.max_bytes_for_level_base,
-            new_opt.max_bytes_for_level_base);
 
   // uint64_t options
-  ASSERT_EQ(base_opt.max_sequential_skip_in_iterations,
+  static const uint64_t uint_max = static_cast<uint64_t>(UINT_MAX);
-            new_opt.max_sequential_skip_in_iterations);
+  cf_opt->max_sequential_skip_in_iterations = uint_max + rnd->Uniform(10000);
-  ASSERT_EQ(base_opt.target_file_size_base, new_opt.target_file_size_base);
+  cf_opt->target_file_size_base = uint_max + rnd->Uniform(10000);
 
   // unsigned int options
-  ASSERT_EQ(base_opt.rate_limit_delay_max_milliseconds,
+  cf_opt->rate_limit_delay_max_milliseconds = rnd->Uniform(10000);
-            new_opt.rate_limit_delay_max_milliseconds);
 }
+
 }  // namespace
 
 TEST_F(OptionsTest, ColumnFamilyOptionsSerialization) {
@@ -734,69 +812,18 @@ TEST_F(OptionsTest, ColumnFamilyOptionsSerialization) {
   Random rnd(302);
   // Phase 1: randomly assign base_opt
   // custom type options
-  base_opt.compaction_style = (CompactionStyle)(rnd.Uniform(4));
+  RandomInitCFOptions(&base_opt, &rnd);
-
-  // boolean options
-  base_opt.compaction_measure_io_stats = rnd.Uniform(2);
-  base_opt.disable_auto_compactions = rnd.Uniform(2);
-  base_opt.filter_deletes = rnd.Uniform(2);
-  base_opt.inplace_update_support = rnd.Uniform(2);
-  base_opt.level_compaction_dynamic_level_bytes = rnd.Uniform(2);
-  base_opt.optimize_filters_for_hits = rnd.Uniform(2);
-  base_opt.paranoid_file_checks = rnd.Uniform(2);
-  base_opt.purge_redundant_kvs_while_flush = rnd.Uniform(2);
-  base_opt.verify_checksums_in_compaction = rnd.Uniform(2);
-
-  // double options
-  base_opt.soft_rate_limit = static_cast<double>(rnd.Uniform(10000)) / 13;
-
-  // int options
-  base_opt.expanded_compaction_factor = rnd.Uniform(100);
-  base_opt.level0_file_num_compaction_trigger = rnd.Uniform(100);
-  base_opt.level0_slowdown_writes_trigger = rnd.Uniform(100);
-  base_opt.level0_stop_writes_trigger = rnd.Uniform(100);
-  base_opt.max_bytes_for_level_multiplier = rnd.Uniform(100);
-  base_opt.max_grandparent_overlap_factor = rnd.Uniform(100);
-  base_opt.max_mem_compaction_level = rnd.Uniform(100);
-  base_opt.max_write_buffer_number = rnd.Uniform(100);
-  base_opt.max_write_buffer_number_to_maintain = rnd.Uniform(100);
-  base_opt.min_write_buffer_number_to_merge = rnd.Uniform(100);
-  base_opt.num_levels = rnd.Uniform(100);
-  base_opt.source_compaction_factor = rnd.Uniform(100);
-  base_opt.target_file_size_multiplier = rnd.Uniform(100);
-
-  // size_t options
-  base_opt.arena_block_size = rnd.Uniform(10000);
-  base_opt.inplace_update_num_locks = rnd.Uniform(10000);
-  base_opt.max_successive_merges = rnd.Uniform(10000);
-  base_opt.memtable_prefix_bloom_huge_page_tlb_size = rnd.Uniform(10000);
-  base_opt.write_buffer_size = rnd.Uniform(10000);
-
-  // uint32_t options
-  base_opt.bloom_locality = rnd.Uniform(10000);
-  base_opt.memtable_prefix_bloom_bits = rnd.Uniform(10000);
-  base_opt.memtable_prefix_bloom_probes = rnd.Uniform(10000);
-  base_opt.min_partial_merge_operands = rnd.Uniform(10000);
-  base_opt.max_bytes_for_level_base = rnd.Uniform(10000);
-
-  // uint64_t options
-  static const uint64_t uint_max = static_cast<uint64_t>(UINT_MAX);
-  base_opt.max_sequential_skip_in_iterations = uint_max + rnd.Uniform(10000);
-  base_opt.target_file_size_base = uint_max + rnd.Uniform(10000);
-  base_opt.hard_pending_compaction_bytes_limit = uint_max + rnd.Uniform(10000);
-
-  // unsigned int options
-  base_opt.rate_limit_delay_max_milliseconds = rnd.Uniform(10000);
 
   // Phase 2: obtain a string from base_opt
-  std::string base_opt_string;
+  std::string base_options_file_content;
-  ASSERT_OK(GetStringFromColumnFamilyOptions(base_opt, &base_opt_string));
+  ASSERT_OK(
+      GetStringFromColumnFamilyOptions(&base_options_file_content, base_opt));
 
   // Phase 3: Set new_opt from the derived string and expect
   //          new_opt == base_opt
-  ASSERT_OK(GetColumnFamilyOptionsFromString(ColumnFamilyOptions(),
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
-                                             base_opt_string, &new_opt));
+      ColumnFamilyOptions(), base_options_file_content, &new_opt));
-  VerifyColumnFamilyOptions(base_opt, new_opt);
+  ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(base_opt, new_opt));
 }
 
 #endif  // !ROCKSDB_LITE
@@ -1000,6 +1027,365 @@ TEST_F(OptionsTest, ConvertOptionsTest) {
   ASSERT_EQ(table_opt.filter_policy.get(), leveldb_opt.filter_policy);
 }
 
+#ifndef ROCKSDB_LITE
+class OptionsParserTest : public testing::Test {
+ public:
+  OptionsParserTest() { env_.reset(new StringEnv(Env::Default())); }
+
+ protected:
+  std::unique_ptr<StringEnv> env_;
+};
+
+TEST_F(OptionsParserTest, Comment) {
+  DBOptions db_opt;
+  db_opt.max_open_files = 12345;
+  db_opt.max_background_flushes = 301;
+  db_opt.max_total_wal_size = 1024;
+  ColumnFamilyOptions cf_opt;
+
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.14.0\n"
+      "  options_file_version=1\n"
+      "[ DBOptions ]\n"
+      "  # note that we don't support space around \"=\"\n"
+      "  max_open_files=12345;\n"
+      "  max_background_flushes=301  # comment after a statement is fine\n"
+      "  # max_background_flushes=1000  # this line would be ignored\n"
+      "  # max_background_compactions=2000 # so does this one\n"
+      "  max_total_wal_size=1024  # keep_log_file_num=1000\n"
+      "[CFOptions   \"default\"]  # column family must be specified\n"
+      "                     # in the correct order\n"
+      "  # if a section is blank, we will use the default\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_OK(parser.Parse(kTestFileName, env_.get()));
+
+  ASSERT_OK(RocksDBOptionsParser::VerifyDBOptions(*parser.db_opt(), db_opt));
+  ASSERT_EQ(parser.NumColumnFamilies(), 1U);
+  ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(
+      *parser.GetCFOptions("default"), cf_opt));
+}
+
+TEST_F(OptionsParserTest, ExtraSpace) {
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[      Version   ]\n"
+      "  rocksdb_version     = 3.14.0      \n"
+      "  options_file_version=1   # some comment\n"
+      "[DBOptions  ]  # some comment\n"
+      "max_open_files=12345   \n"
+      "    max_background_flushes   =    301   \n"
+      " max_total_wal_size     =   1024  # keep_log_file_num=1000\n"
+      "        [CFOptions      \"default\"     ]\n"
+      "  # if a section is blank, we will use the default\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_OK(parser.Parse(kTestFileName, env_.get()));
+}
+
+TEST_F(OptionsParserTest, MissingDBOptions) {
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.14.0\n"
+      "  options_file_version=1\n"
+      "[CFOptions \"default\"]\n"
+      "  # if a section is blank, we will use the default\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_NOK(parser.Parse(kTestFileName, env_.get()));
+}
+
+TEST_F(OptionsParserTest, DoubleDBOptions) {
+  DBOptions db_opt;
+  db_opt.max_open_files = 12345;
+  db_opt.max_background_flushes = 301;
+  db_opt.max_total_wal_size = 1024;
+  ColumnFamilyOptions cf_opt;
+
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.14.0\n"
+      "  options_file_version=1\n"
+      "[DBOptions]\n"
+      "  max_open_files=12345\n"
+      "  max_background_flushes=301\n"
+      "  max_total_wal_size=1024  # keep_log_file_num=1000\n"
+      "[DBOptions]\n"
+      "[CFOptions \"default\"]\n"
+      "  # if a section is blank, we will use the default\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_NOK(parser.Parse(kTestFileName, env_.get()));
+}
+
+TEST_F(OptionsParserTest, NoDefaultCFOptions) {
+  DBOptions db_opt;
+  db_opt.max_open_files = 12345;
+  db_opt.max_background_flushes = 301;
+  db_opt.max_total_wal_size = 1024;
+  ColumnFamilyOptions cf_opt;
+
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.14.0\n"
+      "  options_file_version=1\n"
+      "[DBOptions]\n"
+      "  max_open_files=12345\n"
+      "  max_background_flushes=301\n"
+      "  max_total_wal_size=1024  # keep_log_file_num=1000\n"
+      "[CFOptions \"something_else\"]\n"
+      "  # if a section is blank, we will use the default\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_NOK(parser.Parse(kTestFileName, env_.get()));
+}
+
+TEST_F(OptionsParserTest, DefaultCFOptionsMustBeTheFirst) {
+  DBOptions db_opt;
+  db_opt.max_open_files = 12345;
+  db_opt.max_background_flushes = 301;
+  db_opt.max_total_wal_size = 1024;
+  ColumnFamilyOptions cf_opt;
+
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.14.0\n"
+      "  options_file_version=1\n"
+      "[DBOptions]\n"
+      "  max_open_files=12345\n"
+      "  max_background_flushes=301\n"
+      "  max_total_wal_size=1024  # keep_log_file_num=1000\n"
+      "[CFOptions \"something_else\"]\n"
+      "  # if a section is blank, we will use the default\n"
+      "[CFOptions \"default\"]\n"
+      "  # if a section is blank, we will use the default\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_NOK(parser.Parse(kTestFileName, env_.get()));
+}
+
+TEST_F(OptionsParserTest, DuplicateCFOptions) {
+  DBOptions db_opt;
+  db_opt.max_open_files = 12345;
+  db_opt.max_background_flushes = 301;
+  db_opt.max_total_wal_size = 1024;
+  ColumnFamilyOptions cf_opt;
+
+  std::string options_file_content =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.14.0\n"
+      "  options_file_version=1\n"
+      "[DBOptions]\n"
+      "  max_open_files=12345\n"
+      "  max_background_flushes=301\n"
+      "  max_total_wal_size=1024  # keep_log_file_num=1000\n"
+      "[CFOptions \"default\"]\n"
+      "[CFOptions \"something_else\"]\n"
+      "[CFOptions \"something_else\"]\n";
+
+  const std::string kTestFileName = "test-rocksdb-options.ini";
+  env_->WriteToNewFile(kTestFileName, options_file_content);
+  RocksDBOptionsParser parser;
+  ASSERT_NOK(parser.Parse(kTestFileName, env_.get()));
+}
+
+TEST_F(OptionsParserTest, ParseVersion) {
+  DBOptions db_opt;
+  db_opt.max_open_files = 12345;
+  db_opt.max_background_flushes = 301;
+  db_opt.max_total_wal_size = 1024;
+  ColumnFamilyOptions cf_opt;
+
+  std::string file_template =
+      "# This is a testing option string.\n"
+      "# Currently we only support \"#\" styled comment.\n"
+      "\n"
+      "[Version]\n"
+      "  rocksdb_version=3.13.1\n"
+      "  options_file_version=%s\n"
+      "[DBOptions]\n"
+      "[CFOptions \"default\"]\n";
+  const int kLength = 1000;
+  char buffer[kLength];
+  RocksDBOptionsParser parser;
+
+  const std::vector<std::string> invalid_versions = {
+      "a.b.c", "3.2.2b", "3.-12", "3. 1",  // only digits and dots are allowed
+      "1.2.3.4",
+      "1.2.3"  // can only contains at most one dot.
+      "0",     // options_file_version must be at least one
+      "3..2",
+      ".", ".1.2",             // must have at least one digit before each dot
+      "1.2.", "1.", "2.34."};  // must have at least one digit after each dot
+  for (auto iv : invalid_versions) {
+    snprintf(buffer, kLength - 1, file_template.c_str(), iv.c_str());
+
+    parser.Reset();
+    env_->WriteToNewFile(iv, buffer);
+    ASSERT_NOK(parser.Parse(iv, env_.get()));
+  }
+
+  const std::vector<std::string> valid_versions = {
+      "1.232", "100", "3.12", "1", "12.3  ", "  1.25  "};
+  for (auto vv : valid_versions) {
+    snprintf(buffer, kLength - 1, file_template.c_str(), vv.c_str());
+    parser.Reset();
+    env_->WriteToNewFile(vv, buffer);
+    ASSERT_OK(parser.Parse(vv, env_.get()));
+  }
+}
+
+TEST_F(OptionsParserTest, DumpAndParse) {
+  DBOptions base_db_opt;
+  std::vector<ColumnFamilyOptions> base_cf_opts;
+  std::vector<std::string> cf_names = {
+      // special characters are also included.
+      "default", "p\\i\\k\\a\\chu\\\\\\", "###rocksdb#1-testcf#2###"};
+  const int num_cf = static_cast<int>(cf_names.size());
+  Random rnd(302);
+  RandomInitDBOptions(&base_db_opt, &rnd);
+  base_db_opt.db_log_dir += "/#odd #but #could #happen #path #/\\\\#OMG";
+  for (int c = 0; c < num_cf; ++c) {
+    ColumnFamilyOptions cf_opt;
+    Random cf_rnd(0xFB + c);
+    RandomInitCFOptions(&cf_opt, &cf_rnd);
+    base_cf_opts.emplace_back(cf_opt);
+  }
+
+  const std::string kOptionsFileName = "test-persisted-options.ini";
+  ASSERT_OK(PersistRocksDBOptions(base_db_opt, cf_names, base_cf_opts,
+                                  kOptionsFileName, env_.get()));
+
+  RocksDBOptionsParser parser;
+  ASSERT_OK(parser.Parse(kOptionsFileName, env_.get()));
+
+  ASSERT_OK(RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
+      base_db_opt, cf_names, base_cf_opts, kOptionsFileName, env_.get()));
+
+  ASSERT_OK(
+      RocksDBOptionsParser::VerifyDBOptions(*parser.db_opt(), base_db_opt));
+  for (int c = 0; c < num_cf; ++c) {
+    const auto* cf_opt = parser.GetCFOptions(cf_names[c]);
+    ASSERT_NE(cf_opt, nullptr);
+    ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(*cf_opt, base_cf_opts[c]));
+  }
+  ASSERT_EQ(parser.GetCFOptions("does not exist"), nullptr);
+
+  base_db_opt.max_open_files++;
+  ASSERT_NOK(RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
+      base_db_opt, cf_names, base_cf_opts, kOptionsFileName, env_.get()));
+}
+
+namespace {
+bool IsEscapedString(const std::string& str) {
+  for (size_t i = 0; i < str.size(); ++i) {
+    if (str[i] == '\\') {
+      // since we already handle those two consecutive '\'s in
+      // the next if-then branch, any '\' appear at the end
+      // of an escaped string in such case is not valid.
+      if (i == str.size() - 1) {
+        return false;
+      }
+      if (str[i + 1] == '\\') {
+        // if there're two consecutive '\'s, skip the second one.
+        i++;
+        continue;
+      }
+      switch (str[i + 1]) {
+        case ':':
+        case '\\':
+        case '#':
+          continue;
+        default:
+          // if true, '\' together with str[i + 1] is not a valid escape.
+          if (UnescapeChar(str[i + 1]) == str[i + 1]) {
+            return false;
+          }
+      }
+    } else if (isSpecialChar(str[i]) && (i == 0 || str[i - 1] != '\\')) {
+      return false;
+    }
+  }
+  return true;
+}
+}  // namespace
+
+TEST_F(OptionsParserTest, EscapeOptionString) {
+  ASSERT_EQ(UnescapeOptionString(
+                "This is a test string with \\# \\: and \\\\ escape chars."),
+            "This is a test string with # : and \\ escape chars.");
+
+  ASSERT_EQ(
+      EscapeOptionString("This is a test string with # : and \\ escape chars."),
+      "This is a test string with \\# \\: and \\\\ escape chars.");
+
+  std::string readible_chars =
+      "A String like this \"1234567890-=_)(*&^%$#@!ertyuiop[]{POIU"
+      "YTREWQasdfghjkl;':LKJHGFDSAzxcvbnm,.?>"
+      "<MNBVCXZ\\\" should be okay to \\#\\\\\\:\\#\\#\\#\\ "
+      "be serialized and deserialized";
+
+  std::string escaped_string = EscapeOptionString(readible_chars);
+  ASSERT_TRUE(IsEscapedString(escaped_string));
+  // This two transformations should be canceled and should output
+  // the original input.
+  ASSERT_EQ(UnescapeOptionString(escaped_string), readible_chars);
+
+  std::string all_chars;
+  for (unsigned char c = 0;; ++c) {
+    all_chars += c;
+    if (c == 255) {
+      break;
+    }
+  }
+  escaped_string = EscapeOptionString(all_chars);
+  ASSERT_TRUE(IsEscapedString(escaped_string));
+  ASSERT_EQ(UnescapeOptionString(escaped_string), all_chars);
+
+  ASSERT_EQ(RocksDBOptionsParser::TrimAndRemoveComment(
+                "     A simple statement with a comment.  # like this :)"),
+            "A simple statement with a comment.");
+
+  ASSERT_EQ(RocksDBOptionsParser::TrimAndRemoveComment(
+                "Escape \\# and # comment together   ."),
+            "Escape \\# and");
+}
+
+#endif  // !ROCKSDB_LITE
+
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {