add support for nested BlockBasedTableOptions in config string

Summary:
Add support to allow nested config for block-based table factory. The format looks like this:

"write_buffer_size=1024;block_based_table_factory={block_size=4k};max_write_buffer_num=2"

Test Plan: unit test

Reviewers: yhchiang, rven, igor, ljin, jonahcohen

Reviewed By: jonahcohen

Subscribers: jonahcohen, dhruba, leveldb

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

HISTORY.md

@@ -15,6 +15,8 @@
 * Add rocksdb::GetThreadList(), which in the future will return the current status of all
   rocksdb-related threads.  We will have more code instruments in the following RocksDB
   releases.
+* Change convert function in rocksdb/utilities/convenience.h to return Status instead of boolean.
+  Also add support for nested options in convert function
 
 ### Public API changes
 * New API to create a checkpoint added. Given a directory name, creates a new

Makefile

@@ -536,7 +536,7 @@ thread_list_test: util/thread_list_test.o $(LIBOBJECTS) $(TESTHARNESS)
 compactor_test: utilities/compaction/compactor_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) utilities/compaction/compactor_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
 
-options_test: util/options_test.o $(LIBOBJECTS) $(TESTHARNESS)
+options_test: util/options_test.o util/options_helper.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) util/options_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
 
 $(MEMENVLIBRARY) : $(MEMENVOBJECTS)

include/rocksdb/utilities/convenience.h

@@ -8,35 +8,51 @@
 #include <unordered_map>
 #include <string>
 #include "rocksdb/options.h"
+#include "rocksdb/table.h"
 
 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
-bool GetColumnFamilyOptionsFromMap(
+Status GetColumnFamilyOptionsFromMap(
     const ColumnFamilyOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
     ColumnFamilyOptions* new_options);
 
-bool GetDBOptionsFromMap(
+Status GetDBOptionsFromMap(
     const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
     DBOptions* new_options);
 
+Status GetBlockBasedTableOptionsFromMap(
+    const BlockBasedTableOptions& table_options,
+    const std::unordered_map<std::string, std::string>& opts_map,
+    BlockBasedTableOptions* new_table_options);
+
 // Take a string representation of option names and  values, apply them into the
 // base_options, and return the new options as a result. The string has the
 // following format:
 //   "write_buffer_size=1024;max_write_buffer_number=2"
-bool GetColumnFamilyOptionsFromString(
+// Nested options config is also possible. For example, you can define
+// 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"
+Status GetColumnFamilyOptionsFromString(
     const ColumnFamilyOptions& base_options,
     const std::string& opts_str,
     ColumnFamilyOptions* new_options);
 
-bool GetDBOptionsFromString(
+Status GetDBOptionsFromString(
     const DBOptions& base_options,
     const std::string& opts_str,
     DBOptions* new_options);
+
+Status GetBlockBasedTableOptionsFromString(
+    const BlockBasedTableOptions& table_options,
+    const std::string& opts_str,
+    BlockBasedTableOptions* new_table_options);
+
 #endif  // ROCKSDB_LITE
 
 }  // namespace rocksdb

java/rocksjni/options.cc

@@ -1801,11 +1801,11 @@ jlong Java_org_rocksdb_ColumnFamilyOptions_getColumnFamilyOptionsFromProps(
   rocksdb::ColumnFamilyOptions* cf_options =
       new rocksdb::ColumnFamilyOptions();
   const char* opt_string = env->GetStringUTFChars(jopt_string, 0);
-  bool status = rocksdb::GetColumnFamilyOptionsFromString(
+  rocksdb::Status status = rocksdb::GetColumnFamilyOptionsFromString(
       rocksdb::ColumnFamilyOptions(), opt_string, cf_options);
   env->ReleaseStringUTFChars(jopt_string, opt_string);
   // Check if ColumnFamilyOptions creation was possible.
-  if (status) {
+  if (status.ok()) {
     ret_value = reinterpret_cast<jlong>(cf_options);
   } else {
     // if operation failed the ColumnFamilyOptions need to be deleted
@@ -2803,11 +2803,11 @@ jlong Java_org_rocksdb_DBOptions_getDBOptionsFromProps(
   rocksdb::DBOptions* db_options =
       new rocksdb::DBOptions();
   const char* opt_string = env->GetStringUTFChars(jopt_string, 0);
-  bool status = rocksdb::GetDBOptionsFromString(
+  rocksdb::Status status = rocksdb::GetDBOptionsFromString(
       rocksdb::DBOptions(), opt_string, db_options);
   env->ReleaseStringUTFChars(jopt_string, opt_string);
   // Check if DBOptions creation was possible.
-  if (status) {
+  if (status.ok()) {
     ret_value = reinterpret_cast<jlong>(db_options);
   } else {
     // if operation failed the DBOptions need to be deleted

util/options_helper.cc

@@ -6,7 +6,10 @@
 #include <cassert>
 #include <cctype>
 #include <unordered_set>
+#include "rocksdb/cache.h"
+#include "rocksdb/filter_policy.h"
 #include "rocksdb/options.h"
+#include "rocksdb/table.h"
 #include "rocksdb/utilities/convenience.h"
 #include "util/options_helper.h"
 
@@ -29,19 +32,40 @@ CompressionType ParseCompressionType(const std::string& type) {
   } else if (type == "kLZ4HCCompression") {
     return kLZ4HCCompression;
   } else {
-    throw "unknown compression type: " + type;
+    throw std::invalid_argument("Unknown compression type: " + type);
   }
   return kNoCompression;
 }
 
+BlockBasedTableOptions::IndexType ParseBlockBasedTableIndexType(
+    const std::string& type) {
+  if (type == "kBinarySearch") {
+    return BlockBasedTableOptions::kBinarySearch;
+  } else if (type == "kHashSearch") {
+    return BlockBasedTableOptions::kHashSearch;
+  }
+  throw std::invalid_argument("Unknown index type: " + type);
+}
+
+ChecksumType ParseBlockBasedTableChecksumType(
+    const std::string& type) {
+  if (type == "kNoChecksum") {
+    return kNoChecksum;
+  } else if (type == "kCRC32c") {
+    return kCRC32c;
+  } else if (type == "kxxHash") {
+    return kxxHash;
+  }
+  throw std::invalid_argument("Unknown checksum type: " + type);
+}
+
 bool ParseBoolean(const std::string& type, const std::string& value) {
   if (value == "true" || value == "1") {
     return true;
   } else if (value == "false" || value == "0") {
     return false;
-  } else {
-    throw type;
   }
+  throw std::invalid_argument(type);
 }
 
 uint64_t ParseUint64(const std::string& value) {
@@ -105,7 +129,7 @@ CompactionStyle ParseCompactionStyle(const std::string& type) {
   } else if (type == "kCompactionStyleFIFO") {
     return kCompactionStyleFIFO;
   } else {
-    throw "unknown compaction style: " + type;
+    throw std::invalid_argument("unknown compaction style: " + type);
   }
   return kCompactionStyleLevel;
 }
@@ -172,7 +196,7 @@ bool ParseCompactionOptions(const std::string& name, const std::string& value,
     new_options->max_bytes_for_level_multiplier_additional.clear();
     size_t start = 0;
     while (true) {
-      size_t end = value.find_first_of(':', start);
+      size_t end = value.find(':', start);
       if (end == std::string::npos) {
         new_options->max_bytes_for_level_multiplier_additional.push_back(
             ParseInt(value.substr(start)));
@@ -210,8 +234,8 @@ Status GetMutableOptionsFromStrings(
     MutableCFOptions* new_options) {
   assert(new_options);
   *new_options = base_options;
-  try {
   for (const auto& o : options_map) {
+    try {
       if (ParseMemtableOptions(o.first, o.second, new_options)) {
       } else if (ParseCompactionOptions(o.first, o.second, new_options)) {
       } else if (ParseMiscOptions(o.first, o.second, new_options)) {
@@ -219,9 +243,10 @@ Status GetMutableOptionsFromStrings(
         return Status::InvalidArgument(
             "unsupported dynamic option: " + o.first);
       }
-    }
     } catch (std::exception& e) {
-    return Status::InvalidArgument("error parsing " + std::string(e.what()));
+      return Status::InvalidArgument("error parsing " + o.first + ":" +
+                                     std::string(e.what()));
+    }
   }
   return Status::OK();
 }
@@ -243,38 +268,165 @@ std::string trim(const std::string& str) {
   return std::string();
 }
 
-bool StringToMap(const std::string& opts_str,
+}  // anonymous namespace
+
+Status StringToMap(const std::string& opts_str,
                    std::unordered_map<std::string, std::string>* opts_map) {
   assert(opts_map);
   // Example:
-  //   opts_str = "write_buffer_size=1024;max_write_buffer_number=2"
+  //   opts_str = "write_buffer_size=1024;max_write_buffer_number=2;"
+  //              "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100"
   size_t pos = 0;
-
   std::string opts = trim(opts_str);
   while (pos < opts.size()) {
     size_t eq_pos = opts.find('=', pos);
     if (eq_pos == std::string::npos) {
-      return false;
+      return Status::InvalidArgument("Mismatched key value pair, '=' expected");
     }
     std::string key = trim(opts.substr(pos, eq_pos - pos));
+    if (key.empty()) {
+      return Status::InvalidArgument("Empty key found");
+    }
 
-    size_t sc_pos = opts.find(';', eq_pos + 1);
+    // skip space after '=' and look for '{' for possible nested options
+    pos = eq_pos + 1;
+    while (pos < opts.size() && isspace(opts[pos])) {
+      ++pos;
+    }
+    // Empty value at the end
+    if (pos >= opts.size()) {
+      (*opts_map)[key] = "";
+      break;
+    }
+    if (opts[pos] == '{') {
+      int count = 1;
+      size_t brace_pos = pos + 1;
+      while (brace_pos < opts.size()) {
+        if (opts[brace_pos] == '{') {
+          ++count;
+        } else if (opts[brace_pos] == '}') {
+          --count;
+          if (count == 0) {
+            break;
+          }
+        }
+        ++brace_pos;
+      }
+      // found the matching closing brace
+      if (count == 0) {
+        (*opts_map)[key] = trim(opts.substr(pos + 1, brace_pos - pos - 1));
+        // skip all whitespace and move to the next ';'
+        // brace_pos points to the next position after the matching '}'
+        pos = brace_pos + 1;
+        while (pos < opts.size() && isspace(opts[pos])) {
+          ++pos;
+        }
+        if (pos < opts.size() && opts[pos] != ';') {
+          return Status::InvalidArgument(
+              "Unexpected chars after nested options");
+        }
+        ++pos;
+      } else {
+        return Status::InvalidArgument(
+            "Mismatched curly braces for nested options");
+      }
+    } else {
+      size_t sc_pos = opts.find(';', pos);
       if (sc_pos == std::string::npos) {
-      (*opts_map)[key] = trim(opts.substr(eq_pos + 1));
+        (*opts_map)[key] = trim(opts.substr(pos));
         // It either ends with a trailing semi-colon or the last key-value pair
         break;
       } else {
-      (*opts_map)[key] = trim(opts.substr(eq_pos + 1, sc_pos - eq_pos - 1));
+        (*opts_map)[key] = trim(opts.substr(pos, sc_pos - pos));
       }
       pos = sc_pos + 1;
     }
+  }
 
-  return true;
+  return Status::OK();
 }
 
-}  // anonymous namespace
 
-bool GetColumnFamilyOptionsFromMap(
+Status GetBlockBasedTableOptionsFromMap(
+    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) {
+    try {
+      if (o.first == "cache_index_and_filter_blocks") {
+        new_table_options->cache_index_and_filter_blocks =
+          ParseBoolean(o.first, o.second);
+      } else if (o.first == "index_type") {
+        new_table_options->index_type = ParseBlockBasedTableIndexType(o.second);
+      } else if (o.first == "hash_index_allow_collision") {
+        new_table_options->hash_index_allow_collision =
+          ParseBoolean(o.first, o.second);
+      } else if (o.first == "checksum") {
+        new_table_options->checksum =
+          ParseBlockBasedTableChecksumType(o.second);
+      } else if (o.first == "no_block_cache") {
+        new_table_options->no_block_cache = ParseBoolean(o.first, o.second);
+      } else if (o.first == "block_cache") {
+        new_table_options->block_cache = NewLRUCache(ParseSizeT(o.second));
+      } else if (o.first == "block_cache_compressed") {
+        new_table_options->block_cache_compressed =
+          NewLRUCache(ParseSizeT(o.second));
+      } else if (o.first == "block_size") {
+        new_table_options->block_size = ParseSizeT(o.second);
+      } else if (o.first == "block_size_deviation") {
+        new_table_options->block_size_deviation = ParseInt(o.second);
+      } else if (o.first == "block_restart_interval") {
+        new_table_options->block_restart_interval = ParseInt(o.second);
+      } else if (o.first == "filter_policy") {
+        // Expect the following format
+        // bloomfilter:int:bool
+        const std::string kName = "bloomfilter:";
+        if (o.second.compare(0, kName.size(), kName) != 0) {
+          return Status::InvalidArgument("Invalid filter policy name");
+        }
+        size_t pos = o.second.find(':', kName.size());
+        if (pos == std::string::npos) {
+          return Status::InvalidArgument("Invalid filter policy config, "
+                                         "missing bits_per_key");
+        }
+        int bits_per_key = ParseInt(
+            trim(o.second.substr(kName.size(), pos - kName.size())));
+        bool use_block_based_builder =
+          ParseBoolean("use_block_based_builder",
+                       trim(o.second.substr(pos + 1)));
+        new_table_options->filter_policy.reset(
+            NewBloomFilterPolicy(bits_per_key, use_block_based_builder));
+      } else if (o.first == "whole_key_filtering") {
+        new_table_options->whole_key_filtering =
+          ParseBoolean(o.first, o.second);
+      } else {
+        return Status::InvalidArgument("Unrecognized option: " + o.first);
+      }
+    } catch (std::exception& e) {
+      return Status::InvalidArgument("error parsing " + o.first + ":" +
+                                     std::string(e.what()));
+    }
+  }
+  return Status::OK();
+}
+
+Status GetBlockBasedTableOptionsFromString(
+    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(table_options, opts_map,
+                                          new_table_options);
+}
+
+Status GetColumnFamilyOptionsFromMap(
     const ColumnFamilyOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
     ColumnFamilyOptions* new_options) {
@@ -285,6 +437,15 @@ bool GetColumnFamilyOptionsFromMap(
       if (ParseMemtableOptions(o.first, o.second, new_options)) {
       } else if (ParseCompactionOptions(o.first, o.second, new_options)) {
       } else if (ParseMiscOptions(o.first, o.second, new_options)) {
+      } else if (o.first == "block_based_table_factory") {
+        // Nested options
+        BlockBasedTableOptions table_opt;
+        Status table_opt_s = GetBlockBasedTableOptionsFromString(
+            BlockBasedTableOptions(), o.second, &table_opt);
+        if (!table_opt_s.ok()) {
+          return table_opt_s;
+        }
+        new_options->table_factory.reset(NewBlockBasedTableFactory(table_opt));
       } else if (o.first == "min_write_buffer_number_to_merge") {
         new_options->min_write_buffer_number_to_merge = ParseInt(o.second);
       } else if (o.first == "compression") {
@@ -293,7 +454,7 @@ bool GetColumnFamilyOptionsFromMap(
         new_options->compression_per_level.clear();
         size_t start = 0;
         while (true) {
-          size_t end = o.second.find_first_of(':', start);
+          size_t end = o.second.find(':', start);
           if (end == std::string::npos) {
             new_options->compression_per_level.push_back(
                 ParseCompressionType(o.second.substr(start)));
@@ -306,22 +467,25 @@ bool GetColumnFamilyOptionsFromMap(
         }
       } else if (o.first == "compression_opts") {
         size_t start = 0;
-        size_t end = o.second.find_first_of(':');
+        size_t end = o.second.find(':');
         if (end == std::string::npos) {
-          throw o.first;
+          return Status::InvalidArgument("invalid config value for: "
+                                         + o.first);
         }
         new_options->compression_opts.window_bits =
             ParseInt(o.second.substr(start, end - start));
         start = end + 1;
-        end = o.second.find_first_of(':', start);
+        end = o.second.find(':', start);
         if (end == std::string::npos) {
-          throw o.first;
+          return Status::InvalidArgument("invalid config value for: "
+                                         + o.first);
         }
         new_options->compression_opts.level =
             ParseInt(o.second.substr(start, end - start));
         start = end + 1;
         if (start >= o.second.size()) {
-          throw o.first;
+          return Status::InvalidArgument("invalid config value for: "
+                                         + o.first);
         }
         new_options->compression_opts.strategy =
             ParseInt(o.second.substr(start, o.second.size() - start));
@@ -334,7 +498,7 @@ bool GetColumnFamilyOptionsFromMap(
         new_options->compaction_style = ParseCompactionStyle(o.second);
       } else if (o.first == "compaction_options_universal") {
         // TODO(ljin): add support
-        throw o.first;
+        return Status::NotSupported("Not supported: " + o.first);
       } else if (o.first == "compaction_options_fifo") {
         new_options->compaction_options_fifo.max_table_files_size
           = ParseUint64(o.second);
@@ -345,27 +509,29 @@ bool GetColumnFamilyOptionsFromMap(
       } else if (o.first == "inplace_update_support") {
         new_options->inplace_update_support = ParseBoolean(o.first, o.second);
       } else {
-        return false;
+        return Status::InvalidArgument("Unrecognized option: " + o.first);
       }
-    } catch (std::exception) {
-      return false;
+    } catch (std::exception& e) {
+      return Status::InvalidArgument("error parsing " + o.first + ":" +
+                                     std::string(e.what()));
     }
   }
-  return true;
+  return Status::OK();
 }
 
-bool GetColumnFamilyOptionsFromString(
+Status GetColumnFamilyOptionsFromString(
     const ColumnFamilyOptions& base_options,
     const std::string& opts_str,
     ColumnFamilyOptions* new_options) {
   std::unordered_map<std::string, std::string> opts_map;
-  if (!StringToMap(opts_str, &opts_map)) {
-    return false;
+  Status s = StringToMap(opts_str, &opts_map);
+  if (!s.ok()) {
+    return s;
   }
   return GetColumnFamilyOptionsFromMap(base_options, opts_map, new_options);
 }
 
-bool GetDBOptionsFromMap(
+Status GetDBOptionsFromMap(
     const DBOptions& base_options,
     const std::unordered_map<std::string, std::string>& opts_map,
     DBOptions* new_options) {
@@ -392,7 +558,7 @@ bool GetDBOptionsFromMap(
         new_options->use_fsync = ParseBoolean(o.first, o.second);
       } else if (o.first == "db_paths") {
         // TODO(ljin): add support
-        throw o.first;
+        return Status::NotSupported("Not supported: " + o.first);
       } else if (o.first == "db_log_dir") {
         new_options->db_log_dir = o.second;
       } else if (o.first == "wal_dir") {
@@ -444,22 +610,24 @@ bool GetDBOptionsFromMap(
       } else if (o.first == "bytes_per_sync") {
         new_options->bytes_per_sync = ParseUint64(o.second);
       } else {
-        return false;
+        return Status::InvalidArgument("Unrecognized option: " + o.first);
       }
-    } catch (std::exception) {
-      return false;
+    } catch (std::exception& e) {
+      return Status::InvalidArgument("error parsing " + o.first + ":" +
+                                     std::string(e.what()));
     }
   }
-  return true;
+  return Status::OK();
 }
 
-bool GetDBOptionsFromString(
+Status GetDBOptionsFromString(
     const DBOptions& base_options,
     const std::string& opts_str,
     DBOptions* new_options) {
   std::unordered_map<std::string, std::string> opts_map;
-  if (!StringToMap(opts_str, &opts_map)) {
-    return false;
+  Status s = StringToMap(opts_str, &opts_map);
+  if (!s.ok()) {
+    return s;
   }
   return GetDBOptionsFromMap(base_options, opts_map, new_options);
 }

util/options_test.cc

@@ -14,13 +14,13 @@
 #include <unordered_map>
 #include <inttypes.h>
 
+#include "rocksdb/cache.h"
 #include "rocksdb/options.h"
 #include "rocksdb/table.h"
+#include "rocksdb/utilities/convenience.h"
+#include "rocksdb/utilities/leveldb_options.h"
 #include "table/block_based_table_factory.h"
 #include "util/testharness.h"
-#include "rocksdb/cache.h"
-#include "rocksdb/utilities/leveldb_options.h"
-#include "rocksdb/utilities/convenience.h"
 
 #ifndef GFLAGS
 bool FLAGS_enable_print = false;
@@ -168,7 +168,7 @@ TEST(OptionsTest, GetOptionsFromMapTest) {
 
   ColumnFamilyOptions base_cf_opt;
   ColumnFamilyOptions new_cf_opt;
-  ASSERT_TRUE(GetColumnFamilyOptionsFromMap(
+  ASSERT_OK(GetColumnFamilyOptionsFromMap(
             base_cf_opt, cf_options_map, &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 1U);
   ASSERT_EQ(new_cf_opt.max_write_buffer_number, 2);
@@ -222,18 +222,18 @@ TEST(OptionsTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.min_partial_merge_operands, 31U);
 
   cf_options_map["write_buffer_size"] = "hello";
-  ASSERT_TRUE(!GetColumnFamilyOptionsFromMap(
+  ASSERT_NOK(GetColumnFamilyOptionsFromMap(
              base_cf_opt, cf_options_map, &new_cf_opt));
   cf_options_map["write_buffer_size"] = "1";
-  ASSERT_TRUE(GetColumnFamilyOptionsFromMap(
+  ASSERT_OK(GetColumnFamilyOptionsFromMap(
             base_cf_opt, cf_options_map, &new_cf_opt));
   cf_options_map["unknown_option"] = "1";
-  ASSERT_TRUE(!GetColumnFamilyOptionsFromMap(
+  ASSERT_NOK(GetColumnFamilyOptionsFromMap(
              base_cf_opt, cf_options_map, &new_cf_opt));
 
   DBOptions base_db_opt;
   DBOptions new_db_opt;
-  ASSERT_TRUE(GetDBOptionsFromMap(base_db_opt, db_options_map, &new_db_opt));
+  ASSERT_OK(GetDBOptionsFromMap(base_db_opt, db_options_map, &new_db_opt));
   ASSERT_EQ(new_db_opt.create_if_missing, false);
   ASSERT_EQ(new_db_opt.create_missing_column_families, true);
   ASSERT_EQ(new_db_opt.error_if_exists, false);
@@ -271,63 +271,331 @@ TEST(OptionsTest, GetOptionsFromMapTest) {
 TEST(OptionsTest, GetOptionsFromStringTest) {
   ColumnFamilyOptions base_cf_opt;
   ColumnFamilyOptions new_cf_opt;
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt, "", &new_cf_opt));
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  base_cf_opt.table_factory.reset();
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt, "", &new_cf_opt));
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "write_buffer_size=5", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 5U);
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_TRUE(new_cf_opt.table_factory == nullptr);
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "write_buffer_size=6;", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 6U);
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "  write_buffer_size =  7  ", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 7U);
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "  write_buffer_size =  8 ; ", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 8U);
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "write_buffer_size=9;max_write_buffer_number=10", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 9U);
   ASSERT_EQ(new_cf_opt.max_write_buffer_number, 10);
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "write_buffer_size=11; max_write_buffer_number  =  12 ;",
             &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 11U);
   ASSERT_EQ(new_cf_opt.max_write_buffer_number, 12);
   // Wrong name "max_write_buffer_number_"
-  ASSERT_TRUE(!GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
              "write_buffer_size=13;max_write_buffer_number_=14;",
               &new_cf_opt));
   // Wrong key/value pair
-  ASSERT_TRUE(!GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
              "write_buffer_size=13;max_write_buffer_number;", &new_cf_opt));
   // Error Paring value
-  ASSERT_TRUE(!GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
              "write_buffer_size=13;max_write_buffer_number=;", &new_cf_opt));
   // Missing option name
-  ASSERT_TRUE(!GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
              "write_buffer_size=13; =100;", &new_cf_opt));
   // Units (k)
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "memtable_prefix_bloom_bits=14k;max_write_buffer_number=-15K",
             &new_cf_opt));
   ASSERT_EQ(new_cf_opt.memtable_prefix_bloom_bits, 14UL*1024UL);
   ASSERT_EQ(new_cf_opt.max_write_buffer_number, -15*1024);
   // Units (m)
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "max_write_buffer_number=16m;inplace_update_num_locks=17M",
             &new_cf_opt));
   ASSERT_EQ(new_cf_opt.max_write_buffer_number, 16*1024*1024);
   ASSERT_EQ(new_cf_opt.inplace_update_num_locks, 17*1024UL*1024UL);
   // Units (g)
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "write_buffer_size=18g;arena_block_size=19G", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 18*1024UL*1024UL*1024UL);
   ASSERT_EQ(new_cf_opt.arena_block_size, 19*1024UL*1024UL*1024UL);
   // Units (t)
-  ASSERT_TRUE(GetColumnFamilyOptionsFromString(base_cf_opt,
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
             "write_buffer_size=20t;arena_block_size=21T", &new_cf_opt));
   ASSERT_EQ(new_cf_opt.write_buffer_size, 20*1024UL*1024UL*1024UL*1024UL);
   ASSERT_EQ(new_cf_opt.arena_block_size, 21*1024UL*1024UL*1024UL*1024UL);
+
+  // Nested block based table options
+  // Emtpy
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
+            "write_buffer_size=10;max_write_buffer_number=16;"
+            "block_based_table_factory={};arena_block_size=1024",
+            &new_cf_opt));
+  ASSERT_TRUE(new_cf_opt.table_factory != nullptr);
+  // Non-empty
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
+            "write_buffer_size=10;max_write_buffer_number=16;"
+            "block_based_table_factory={block_cache=1M;block_size=4;};"
+            "arena_block_size=1024",
+            &new_cf_opt));
+  ASSERT_TRUE(new_cf_opt.table_factory != nullptr);
+  // Last one
+  ASSERT_OK(GetColumnFamilyOptionsFromString(base_cf_opt,
+            "write_buffer_size=10;max_write_buffer_number=16;"
+            "block_based_table_factory={block_cache=1M;block_size=4;}",
+            &new_cf_opt));
+  ASSERT_TRUE(new_cf_opt.table_factory != nullptr);
+  // Mismatch curly braces
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
+             "write_buffer_size=10;max_write_buffer_number=16;"
+             "block_based_table_factory={{{block_size=4;};"
+             "arena_block_size=1024",
+             &new_cf_opt));
+  // Unexpected chars after closing curly brace
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
+             "write_buffer_size=10;max_write_buffer_number=16;"
+             "block_based_table_factory={block_size=4;}};"
+             "arena_block_size=1024",
+             &new_cf_opt));
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
+             "write_buffer_size=10;max_write_buffer_number=16;"
+             "block_based_table_factory={block_size=4;}xdfa;"
+             "arena_block_size=1024",
+             &new_cf_opt));
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
+             "write_buffer_size=10;max_write_buffer_number=16;"
+             "block_based_table_factory={block_size=4;}xdfa",
+             &new_cf_opt));
+  // Invalid block based table option
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(base_cf_opt,
+             "write_buffer_size=10;max_write_buffer_number=16;"
+             "block_based_table_factory={xx_block_size=4;}",
+             &new_cf_opt));
+}
+
+TEST(OptionsTest, GetBlockBasedTableOptionsFromString) {
+  BlockBasedTableOptions table_opt;
+  BlockBasedTableOptions new_opt;
+  // make sure default values are overwritten by something else
+  ASSERT_OK(GetBlockBasedTableOptionsFromString(table_opt,
+            "cache_index_and_filter_blocks=1;index_type=kHashSearch;"
+            "checksum=kxxHash;hash_index_allow_collision=1;no_block_cache=1;"
+            "block_cache=1M;block_cache_compressed=1k;block_size=1024;"
+            "block_size_deviation=8;block_restart_interval=4;"
+            "filter_policy=bloomfilter:4:true;whole_key_filtering=1",
+            &new_opt));
+  ASSERT_TRUE(new_opt.cache_index_and_filter_blocks);
+  ASSERT_EQ(new_opt.index_type, BlockBasedTableOptions::kHashSearch);
+  ASSERT_EQ(new_opt.checksum, ChecksumType::kxxHash);
+  ASSERT_TRUE(new_opt.hash_index_allow_collision);
+  ASSERT_TRUE(new_opt.no_block_cache);
+  ASSERT_TRUE(new_opt.block_cache != nullptr);
+  ASSERT_EQ(new_opt.block_cache->GetCapacity(), 1024UL*1024UL);
+  ASSERT_TRUE(new_opt.block_cache_compressed != nullptr);
+  ASSERT_EQ(new_opt.block_cache_compressed->GetCapacity(), 1024UL);
+  ASSERT_EQ(new_opt.block_size, 1024UL);
+  ASSERT_EQ(new_opt.block_size_deviation, 8);
+  ASSERT_EQ(new_opt.block_restart_interval, 4);
+  ASSERT_TRUE(new_opt.filter_policy != nullptr);
+
+  // unknown option
+  ASSERT_NOK(GetBlockBasedTableOptionsFromString(table_opt,
+             "cache_index_and_filter_blocks=1;index_type=kBinarySearch;"
+             "bad_option=1",
+             &new_opt));
+
+  // unrecognized index type
+  ASSERT_NOK(GetBlockBasedTableOptionsFromString(table_opt,
+             "cache_index_and_filter_blocks=1;index_type=kBinarySearchXX",
+             &new_opt));
+
+  // unrecognized checksum type
+  ASSERT_NOK(GetBlockBasedTableOptionsFromString(table_opt,
+             "cache_index_and_filter_blocks=1;checksum=kxxHashXX",
+             &new_opt));
+
+  // unrecognized filter policy name
+  ASSERT_NOK(GetBlockBasedTableOptionsFromString(table_opt,
+             "cache_index_and_filter_blocks=1;"
+             "filter_policy=bloomfilterxx:4:true",
+             &new_opt));
+  // unrecognized filter policy config
+  ASSERT_NOK(GetBlockBasedTableOptionsFromString(table_opt,
+             "cache_index_and_filter_blocks=1;"
+             "filter_policy=bloomfilter:4",
+             &new_opt));
+}
+
+Status StringToMap(
+    const std::string& opts_str,
+    std::unordered_map<std::string, std::string>* opts_map);
+
+TEST(OptionsTest, StringToMapTest) {
+  std::unordered_map<std::string, std::string> opts_map;
+  // Regular options
+  ASSERT_OK(StringToMap("k1=v1;k2=v2;k3=v3", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "v2");
+  ASSERT_EQ(opts_map["k3"], "v3");
+  // Value with '='
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1==v1;k2=v2=;", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "=v1");
+  ASSERT_EQ(opts_map["k2"], "v2=");
+  // Overwrriten option
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k1=v2;k3=v3", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v2");
+  ASSERT_EQ(opts_map["k3"], "v3");
+  // Empty value
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2=;k3=v3;k4=", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_TRUE(opts_map.find("k2") != opts_map.end());
+  ASSERT_EQ(opts_map["k2"], "");
+  ASSERT_EQ(opts_map["k3"], "v3");
+  ASSERT_TRUE(opts_map.find("k4") != opts_map.end());
+  ASSERT_EQ(opts_map["k4"], "");
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2=;k3=v3;k4=   ", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_TRUE(opts_map.find("k2") != opts_map.end());
+  ASSERT_EQ(opts_map["k2"], "");
+  ASSERT_EQ(opts_map["k3"], "v3");
+  ASSERT_TRUE(opts_map.find("k4") != opts_map.end());
+  ASSERT_EQ(opts_map["k4"], "");
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2=;k3=", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_TRUE(opts_map.find("k2") != opts_map.end());
+  ASSERT_EQ(opts_map["k2"], "");
+  ASSERT_TRUE(opts_map.find("k3") != opts_map.end());
+  ASSERT_EQ(opts_map["k3"], "");
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2=;k3=;", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_TRUE(opts_map.find("k2") != opts_map.end());
+  ASSERT_EQ(opts_map["k2"], "");
+  ASSERT_TRUE(opts_map.find("k3") != opts_map.end());
+  ASSERT_EQ(opts_map["k3"], "");
+  // Regular nested options
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2={nk1=nv1;nk2=nv2};k3=v3", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2=nv2");
+  ASSERT_EQ(opts_map["k3"], "v3");
+  // Multi-level nested options
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2={nk1=nv1;nk2={nnk1=nnk2}};"
+                        "k3={nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}};k4=v4",
+                        &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2={nnk1=nnk2}");
+  ASSERT_EQ(opts_map["k3"], "nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}");
+  ASSERT_EQ(opts_map["k4"], "v4");
+  // Garbage inside curly braces
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2={dfad=};k3={=};k4=v4",
+                        &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "dfad=");
+  ASSERT_EQ(opts_map["k3"], "=");
+  ASSERT_EQ(opts_map["k4"], "v4");
+  // Empty nested options
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2={};", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "");
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2={{{{}}}{}{}};", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "{{{}}}{}{}");
+  // With random spaces
+  opts_map.clear();
+  ASSERT_OK(StringToMap("  k1 =  v1 ; k2= {nk1=nv1; nk2={nnk1=nnk2}}  ; "
+                        "k3={  {   } }; k4= v4  ",
+                        &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "nk1=nv1; nk2={nnk1=nnk2}");
+  ASSERT_EQ(opts_map["k3"], "{   }");
+  ASSERT_EQ(opts_map["k4"], "v4");
+
+  // Empty key
+  ASSERT_NOK(StringToMap("k1=v1;k2=v2;=", &opts_map));
+  ASSERT_NOK(StringToMap("=v1;k2=v2", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2v2;", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2=v2;fadfa", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2=v2;;", &opts_map));
+  // Mismatch curly braces
+  ASSERT_NOK(StringToMap("k1=v1;k2={;k3=v3", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{};k3=v3", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={}};k3=v3", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{}{}}};k3=v3", &opts_map));
+  // However this is valid!
+  opts_map.clear();
+  ASSERT_OK(StringToMap("k1=v1;k2=};k3=v3", &opts_map));
+  ASSERT_EQ(opts_map["k1"], "v1");
+  ASSERT_EQ(opts_map["k2"], "}");
+  ASSERT_EQ(opts_map["k3"], "v3");
+
+  // Invalid chars after closing curly brace
+  ASSERT_NOK(StringToMap("k1=v1;k2={{}}{};k3=v3", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{}}cfda;k3=v3", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{}}  cfda;k3=v3", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{}}  cfda", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{}}{}", &opts_map));
+  ASSERT_NOK(StringToMap("k1=v1;k2={{dfdl}adfa}{}", &opts_map));
+}
+
+TEST(OptionsTest, StringToMapRandomTest) {
+  std::unordered_map<std::string, std::string> opts_map;
+  // Make sure segfault is not hit by semi-random strings
+
+  std::vector<std::string> bases = {
+      "a={aa={};tt={xxx={}}};c=defff",
+      "a={aa={};tt={xxx={}}};c=defff;d={{}yxx{}3{xx}}",
+      "abc={{}{}{}{{{}}}{{}{}{}{}{}{}{}"};
+
+  for (std::string base : bases) {
+    for (int rand_seed = 301; rand_seed < 401; rand_seed++) {
+      Random rnd(rand_seed);
+      for (int attempt = 0; attempt < 10; attempt++) {
+        std::string str = base;
+        // Replace random position to space
+        size_t pos = static_cast<size_t>(
+            rnd.Uniform(static_cast<int>(base.size())));
+        str[pos] = ' ';
+        Status s = StringToMap(str, &opts_map);
+        ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+        opts_map.clear();
+      }
+    }
+  }
+
+  // Random Construct a string
+  std::vector<char> chars = {'{', '}', ' ', '=', ';', 'c'};
+  for (int rand_seed = 301; rand_seed < 1301; rand_seed++) {
+    Random rnd(rand_seed);
+    int len = rnd.Uniform(30);
+    std::string str = "";
+    for (int attempt = 0; attempt < len; attempt++) {
+      // Add a random character
+      size_t pos = static_cast<size_t>(
+          rnd.Uniform(static_cast<int>(chars.size())));
+      str.append(1, chars[pos]);
+    }
+    Status s = StringToMap(str, &opts_map);
+    ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+    s = StringToMap("name=" + str, &opts_map);
+    ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+    opts_map.clear();
+  }
 }
 
 TEST(OptionsTest, ConvertOptionsTest) {

util/testharness.h

@@ -84,6 +84,14 @@ class Tester {
     return *this;
   }
 
+  Tester& IsNotOk(const Status& s) {
+    if (s.ok()) {
+      ss_ << " Error status expected";
+      ok_ = false;
+    }
+    return *this;
+  }
+
 #define BINARY_OP(name,op)                              \
   template <class X, class Y>                           \
   Tester& name(const X& x, const Y& y) {                \
@@ -114,6 +122,7 @@ class Tester {
 
 #define ASSERT_TRUE(c) ::rocksdb::test::Tester(__FILE__, __LINE__).Is((c), #c)
 #define ASSERT_OK(s) ::rocksdb::test::Tester(__FILE__, __LINE__).IsOk((s))
+#define ASSERT_NOK(s) ::rocksdb::test::Tester(__FILE__, __LINE__).IsNotOk((s))
 #define ASSERT_EQ(a,b) ::rocksdb::test::Tester(__FILE__, __LINE__).IsEq((a),(b))
 #define ASSERT_NE(a,b) ::rocksdb::test::Tester(__FILE__, __LINE__).IsNe((a),(b))
 #define ASSERT_GE(a,b) ::rocksdb::test::Tester(__FILE__, __LINE__).IsGe((a),(b))