ColumnFamilyOptions serialization / deserialization.

Summary:
This patch adds GetStringFromColumnFamilyOptions(), the inverse function
of the existing GetColumnFamilyOptionsFromString(), and improves
the implementation of GetColumnFamilyOptionsFromString().

Test Plan: Add a test in options_test.cc

Reviewers: igor, sdong, anthony, IslamAbdelRahman

Reviewed By: IslamAbdelRahman

Subscribers: noetzli, dhruba, leveldb

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

include/rocksdb/convenience.h

@@ -51,6 +51,9 @@ Status GetDBOptionsFromString(
 Status GetStringFromDBOptions(const DBOptions& db_options,
                               std::string* opts_str);
 
+Status GetStringFromColumnFamilyOptions(const ColumnFamilyOptions& db_options,
+                                        std::string* opts_str);
+
 Status GetBlockBasedTableOptionsFromString(
     const BlockBasedTableOptions& table_options,
     const std::string& opts_str,

util/options_helper.cc

@@ -143,18 +143,105 @@ double ParseDouble(const std::string& value) {
 #endif
 }
 
+static const std::unordered_map<char, std::string>
+    compaction_style_to_string_map = {
+        {kCompactionStyleLevel, "kCompactionStyleLevel"},
+        {kCompactionStyleUniversal, "kCompactionStyleUniversal"},
+        {kCompactionStyleFIFO, "kCompactionStyleFIFO"},
+        {kCompactionStyleNone, "kCompactionStyleNone"}};
+
 CompactionStyle ParseCompactionStyle(const std::string& type) {
-  if (type == "kCompactionStyleLevel") {
-    return kCompactionStyleLevel;
-  } else if (type == "kCompactionStyleUniversal") {
-    return kCompactionStyleUniversal;
-  } else if (type == "kCompactionStyleFIFO") {
-    return kCompactionStyleFIFO;
-  } else {
-    throw std::invalid_argument("unknown compaction style: " + type);
+  for (auto const& entry : compaction_style_to_string_map) {
+    if (entry.second == type) {
+      return static_cast<CompactionStyle>(entry.first);
+    }
   }
+  throw std::invalid_argument("unknown compaction style: " + type);
   return kCompactionStyleLevel;
 }
+
+std::string CompactionStyleToString(const CompactionStyle style) {
+  auto iter = compaction_style_to_string_map.find(style);
+  assert(iter != compaction_style_to_string_map.end());
+  return iter->second;
+}
+
+bool ParseOptionHelper(char* opt_address, const OptionType& opt_type,
+                       const std::string& value) {
+  switch (opt_type) {
+    case OptionType::kBoolean:
+      *reinterpret_cast<bool*>(opt_address) = ParseBoolean("", value);
+      break;
+    case OptionType::kInt:
+      *reinterpret_cast<int*>(opt_address) = ParseInt(value);
+      break;
+    case OptionType::kUInt:
+      *reinterpret_cast<unsigned int*>(opt_address) = ParseUint32(value);
+      break;
+    case OptionType::kUInt32T:
+      *reinterpret_cast<uint32_t*>(opt_address) = ParseUint32(value);
+      break;
+    case OptionType::kUInt64T:
+      *reinterpret_cast<uint64_t*>(opt_address) = ParseUint64(value);
+      break;
+    case OptionType::kSizeT:
+      *reinterpret_cast<size_t*>(opt_address) = ParseSizeT(value);
+      break;
+    case OptionType::kString:
+      *reinterpret_cast<std::string*>(opt_address) = value;
+      break;
+    case OptionType::kDouble:
+      *reinterpret_cast<double*>(opt_address) = ParseDouble(value);
+      break;
+    case OptionType::kCompactionStyle:
+      *reinterpret_cast<CompactionStyle*>(opt_address) =
+          ParseCompactionStyle(value);
+      break;
+    default:
+      return false;
+  }
+  return true;
+}
+
+bool SerializeSingleOptionHelper(const char* opt_address,
+                                 const OptionType opt_type,
+                                 std::string* value) {
+  assert(value);
+  switch (opt_type) {
+    case OptionType::kBoolean:
+      *value = *(reinterpret_cast<const bool*>(opt_address)) ? "true" : "false";
+      break;
+    case OptionType::kInt:
+      *value = ToString(*(reinterpret_cast<const int*>(opt_address)));
+      break;
+    case OptionType::kUInt:
+      *value = ToString(*(reinterpret_cast<const unsigned int*>(opt_address)));
+      break;
+    case OptionType::kUInt32T:
+      *value = ToString(*(reinterpret_cast<const uint32_t*>(opt_address)));
+      break;
+    case OptionType::kUInt64T:
+      *value = ToString(*(reinterpret_cast<const uint64_t*>(opt_address)));
+      break;
+    case OptionType::kSizeT:
+      *value = ToString(*(reinterpret_cast<const size_t*>(opt_address)));
+      break;
+    case OptionType::kDouble:
+      *value = ToString(*(reinterpret_cast<const double*>(opt_address)));
+      break;
+    case OptionType::kString:
+      *value = *(reinterpret_cast<const std::string*>(opt_address));
+      break;
+    case OptionType::kCompactionStyle:
+      *value = CompactionStyleToString(
+          *(reinterpret_cast<const CompactionStyle*>(opt_address)));
+      break;
+    default:
+      return false;
+  }
+  return true;
+}
+
 }  // anonymouse namespace
 
 template<typename OptionsType>
@@ -372,9 +459,21 @@ Status StringToMap(const std::string& opts_str,
 bool ParseColumnFamilyOption(const std::string& name, const std::string& value,
                              ColumnFamilyOptions* new_options) {
   try {
-    if (ParseMemtableOptions(name, value, new_options)) {
-    } else if (ParseCompactionOptions(name, value, new_options)) {
-    } else if (ParseMiscOptions(name, value, new_options)) {
+    if (name == "max_bytes_for_level_multiplier_additional") {
+      new_options->max_bytes_for_level_multiplier_additional.clear();
+      size_t start = 0;
+      while (true) {
+        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)));
+          break;
+        } else {
+          new_options->max_bytes_for_level_multiplier_additional.push_back(
+              ParseInt(value.substr(start, end - start)));
+          start = end + 1;
+        }
+      }
     } else if (name == "block_based_table_factory") {
       // Nested options
       BlockBasedTableOptions table_opt, base_table_options;
@@ -389,10 +488,6 @@ bool ParseColumnFamilyOption(const std::string& name, const std::string& value,
         return false;
       }
       new_options->table_factory.reset(NewBlockBasedTableFactory(table_opt));
-    } else if (name == "min_write_buffer_number_to_merge") {
-      new_options->min_write_buffer_number_to_merge = ParseInt(value);
-    } else if (name == "max_write_buffer_number_to_maintain") {
-      new_options->max_write_buffer_number_to_maintain = ParseInt(value);
     } else if (name == "compression") {
       new_options->compression = ParseCompressionType(value);
     } else if (name == "compression_per_level") {
@@ -431,27 +526,12 @@ bool ParseColumnFamilyOption(const std::string& name, const std::string& value,
       }
       new_options->compression_opts.strategy =
           ParseInt(value.substr(start, value.size() - start));
-    } else if (name == "num_levels") {
-      new_options->num_levels = ParseInt(value);
-    } else if (name == "level_compaction_dynamic_level_bytes") {
-      new_options->level_compaction_dynamic_level_bytes =
-          ParseBoolean(name, value);
-    } else if (name == "compaction_style") {
-      new_options->compaction_style = ParseCompactionStyle(value);
     } else if (name == "compaction_options_universal") {
       // TODO(ljin): add support
       return false;
     } else if (name == "compaction_options_fifo") {
       new_options->compaction_options_fifo.max_table_files_size =
           ParseUint64(value);
-    } else if (name == "bloom_locality") {
-      new_options->bloom_locality = ParseUint32(value);
-    } else if (name == "min_partial_merge_operands") {
-      new_options->min_partial_merge_operands = ParseUint32(value);
-    } else if (name == "inplace_update_support") {
-      new_options->inplace_update_support = ParseBoolean(name, value);
-    } else if (name == "compaction_measure_io_stats") {
-      new_options->compaction_measure_io_stats = ParseBoolean(name, value);
     } else if (name == "prefix_extractor") {
       const std::string kFixedPrefixName = "fixed:";
       const std::string kCappedPrefixName = "capped:";
@@ -472,13 +552,17 @@ bool ParseColumnFamilyOption(const std::string& name, const std::string& value,
       } else {
         return false;
       }
-    } else if (name == "optimize_filters_for_hits") {
-      new_options->optimize_filters_for_hits = ParseBoolean(name, value);
     } else {
-      return false;
+      auto iter = cf_options_type_info.find(name);
+      if (iter == cf_options_type_info.end()) {
+        return false;
+      }
+      const auto& opt_info = iter->second;
+      return ParseOptionHelper(
+          reinterpret_cast<char*>(new_options) + opt_info.offset, opt_info.type,
+          value);
     }
-  }
-  catch (std::exception& e) {
+  } catch (std::exception& e) {
     return false;
   }
   return true;
@@ -494,35 +578,11 @@ bool SerializeSingleDBOption(const DBOptions& db_options,
   const char* opt_address =
       reinterpret_cast<const char*>(&db_options) + opt_info.offset;
   std::string value;
-
-  switch (opt_info.type) {
-    case OptionType::kBoolean:
-      value = *(reinterpret_cast<const bool*>(opt_address)) ? "true" : "false";
-      break;
-    case OptionType::kInt:
-      value = ToString(*(reinterpret_cast<const int*>(opt_address)));
-      break;
-    case OptionType::kUInt:
-      value = ToString(*(reinterpret_cast<const unsigned int*>(opt_address)));
-      break;
-    case OptionType::kUInt32T:
-      value = ToString(*(reinterpret_cast<const uint32_t*>(opt_address)));
-      break;
-    case OptionType::kUInt64T:
-      value = ToString(*(reinterpret_cast<const uint64_t*>(opt_address)));
-      break;
-    case OptionType::kSizeT:
-      value = ToString(*(reinterpret_cast<const size_t*>(opt_address)));
-      break;
-    case OptionType::kString:
-      value = *(reinterpret_cast<const std::string*>(opt_address));
-      break;
-    default:
-      return false;
+  bool result = SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
+  if (result) {
+    *opt_string = name + " = " + value + ";  ";
   }
-
-  *opt_string = name + " = " + value + ";  ";
-  return true;
+  return result;
 }
 
 Status GetStringFromDBOptions(const DBOptions& db_options,
@@ -542,6 +602,43 @@ Status GetStringFromDBOptions(const DBOptions& db_options,
   return Status::OK();
 }
 
+bool SerializeSingleColumnFamilyOption(const ColumnFamilyOptions& cf_options,
+                                       const std::string& name,
+                                       std::string* opt_string) {
+  auto iter = cf_options_type_info.find(name);
+  if (iter == cf_options_type_info.end()) {
+    return false;
+  }
+  auto& opt_info = iter->second;
+  const char* opt_address =
+      reinterpret_cast<const char*>(&cf_options) + opt_info.offset;
+  std::string value;
+  bool result = SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
+  if (result) {
+    *opt_string = name + " = " + value + ";  ";
+  }
+  return result;
+}
+
+Status GetStringFromColumnFamilyOptions(const ColumnFamilyOptions& cf_options,
+                                        std::string* opt_string) {
+  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,
+                                                    &single_output);
+    if (result) {
+      opt_string->append(single_output);
+    } else {
+      printf("failed to serialize %s\n", iter->first.c_str());
+    }
+    assert(result);
+  }
+  return Status::OK();
+}
+
 bool ParseDBOption(const std::string& name, const std::string& value,
                    DBOptions* new_options) {
   try {
@@ -553,34 +650,10 @@ bool ParseDBOption(const std::string& name, const std::string& value,
       if (iter == db_options_type_info.end()) {
         return false;
       }
-      auto& opt_info = iter->second;
-      char* opt_address =
-          reinterpret_cast<char*>(new_options) + opt_info.offset;
-      switch (opt_info.type) {
-        case OptionType::kBoolean:
-          *reinterpret_cast<bool*>(opt_address) = ParseBoolean("", value);
-          break;
-        case OptionType::kInt:
-          *reinterpret_cast<int*>(opt_address) = ParseInt(value);
-          break;
-        case OptionType::kUInt:
-          *reinterpret_cast<unsigned int*>(opt_address) = ParseUint32(value);
-          break;
-        case OptionType::kUInt32T:
-          *reinterpret_cast<uint32_t*>(opt_address) = ParseUint32(value);
-          break;
-        case OptionType::kUInt64T:
-          *reinterpret_cast<uint64_t*>(opt_address) = ParseUint64(value);
-          break;
-        case OptionType::kSizeT:
-          *reinterpret_cast<size_t*>(opt_address) = ParseUint32(value);
-          break;
-        case OptionType::kString:
-          *reinterpret_cast<std::string*>(opt_address) = value;
-          break;
-        default:
-          return false;
-      }
+      const auto& opt_info = iter->second;
+      return ParseOptionHelper(
+          reinterpret_cast<char*>(new_options) + opt_info.offset, opt_info.type,
+          value);
     }
   } catch (const std::exception& e) {
     return false;

util/options_helper.h

@@ -7,8 +7,9 @@
 
 #include <string>
 #include <stdexcept>
-#include "util/mutable_cf_options.h"
+#include "rocksdb/options.h"
 #include "rocksdb/status.h"
+#include "util/mutable_cf_options.h"
 
 namespace rocksdb {
 
@@ -25,6 +26,8 @@ enum class OptionType {
   kUInt64T,
   kSizeT,
   kString,
+  kDouble,
+  kCompactionStyle,
   kUnknown
 };
 
@@ -140,4 +143,143 @@ static std::unordered_map<std::string, OptionTypeInfo> db_options_type_info = {
     {"stats_dump_period_sec",
      {offsetof(struct DBOptions, stats_dump_period_sec), OptionType::kUInt}}};
 
+static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
+    /* not yet supported
+    CompactionOptionsFIFO compaction_options_fifo;
+    CompactionOptionsUniversal compaction_options_universal;
+    CompressionOptions compression_opts;
+    CompressionType compression;
+    TablePropertiesCollectorFactories table_properties_collector_factories;
+    typedef std::vector<std::shared_ptr<TablePropertiesCollectorFactory>>
+        TablePropertiesCollectorFactories;
+    UpdateStatus (*inplace_callback)(char* existing_value,
+                                     uint34_t* existing_value_size,
+                                     Slice delta_value,
+                                     std::string* merged_value);
+    const CompactionFilter* compaction_filter;
+    const Comparator* comparator;
+    std::shared_ptr<CompactionFilterFactory> compaction_filter_factory;
+    std::shared_ptr<MemTableRepFactory> memtable_factory;
+    std::shared_ptr<MergeOperator> merge_operator;
+    std::shared_ptr<TableFactory> table_factory;
+    std::shared_ptr<const SliceTransform> prefix_extractor;
+    std::vector<CompressionType> compression_per_level;
+    std::vector<int> max_bytes_for_level_multiplier_additional;
+     */
+    {"compaction_measure_io_stats",
+     {offsetof(struct ColumnFamilyOptions, compaction_measure_io_stats),
+      OptionType::kBoolean}},
+    {"disable_auto_compactions",
+     {offsetof(struct ColumnFamilyOptions, disable_auto_compactions),
+      OptionType::kBoolean}},
+    {"filter_deletes",
+     {offsetof(struct ColumnFamilyOptions, filter_deletes),
+      OptionType::kBoolean}},
+    {"inplace_update_support",
+     {offsetof(struct ColumnFamilyOptions, inplace_update_support),
+      OptionType::kBoolean}},
+    {"level_compaction_dynamic_level_bytes",
+     {offsetof(struct ColumnFamilyOptions,
+               level_compaction_dynamic_level_bytes),
+      OptionType::kBoolean}},
+    {"optimize_filters_for_hits",
+     {offsetof(struct ColumnFamilyOptions, optimize_filters_for_hits),
+      OptionType::kBoolean}},
+    {"paranoid_file_checks",
+     {offsetof(struct ColumnFamilyOptions, paranoid_file_checks),
+      OptionType::kBoolean}},
+    {"purge_redundant_kvs_while_flush",
+     {offsetof(struct ColumnFamilyOptions, purge_redundant_kvs_while_flush),
+      OptionType::kBoolean}},
+    {"verify_checksums_in_compaction",
+     {offsetof(struct ColumnFamilyOptions, verify_checksums_in_compaction),
+      OptionType::kBoolean}},
+    {"hard_rate_limit",
+     {offsetof(struct ColumnFamilyOptions, hard_rate_limit),
+      OptionType::kDouble}},
+    {"soft_rate_limit",
+     {offsetof(struct ColumnFamilyOptions, soft_rate_limit),
+      OptionType::kDouble}},
+    {"expanded_compaction_factor",
+     {offsetof(struct ColumnFamilyOptions, expanded_compaction_factor),
+      OptionType::kInt}},
+    {"level0_file_num_compaction_trigger",
+     {offsetof(struct ColumnFamilyOptions, level0_file_num_compaction_trigger),
+      OptionType::kInt}},
+    {"level0_slowdown_writes_trigger",
+     {offsetof(struct ColumnFamilyOptions, level0_slowdown_writes_trigger),
+      OptionType::kInt}},
+    {"level0_stop_writes_trigger",
+     {offsetof(struct ColumnFamilyOptions, level0_stop_writes_trigger),
+      OptionType::kInt}},
+    {"max_bytes_for_level_multiplier",
+     {offsetof(struct ColumnFamilyOptions, max_bytes_for_level_multiplier),
+      OptionType::kInt}},
+    {"max_grandparent_overlap_factor",
+     {offsetof(struct ColumnFamilyOptions, max_grandparent_overlap_factor),
+      OptionType::kInt}},
+    {"max_mem_compaction_level",
+     {offsetof(struct ColumnFamilyOptions, max_mem_compaction_level),
+      OptionType::kInt}},
+    {"max_write_buffer_number",
+     {offsetof(struct ColumnFamilyOptions, max_write_buffer_number),
+      OptionType::kInt}},
+    {"max_write_buffer_number_to_maintain",
+     {offsetof(struct ColumnFamilyOptions, max_write_buffer_number_to_maintain),
+      OptionType::kInt}},
+    {"min_write_buffer_number_to_merge",
+     {offsetof(struct ColumnFamilyOptions, min_write_buffer_number_to_merge),
+      OptionType::kInt}},
+    {"num_levels",
+     {offsetof(struct ColumnFamilyOptions, num_levels), OptionType::kInt}},
+    {"source_compaction_factor",
+     {offsetof(struct ColumnFamilyOptions, source_compaction_factor),
+      OptionType::kInt}},
+    {"target_file_size_multiplier",
+     {offsetof(struct ColumnFamilyOptions, target_file_size_multiplier),
+      OptionType::kInt}},
+    {"arena_block_size",
+     {offsetof(struct ColumnFamilyOptions, arena_block_size),
+      OptionType::kSizeT}},
+    {"inplace_update_num_locks",
+     {offsetof(struct ColumnFamilyOptions, inplace_update_num_locks),
+      OptionType::kSizeT}},
+    {"max_successive_merges",
+     {offsetof(struct ColumnFamilyOptions, max_successive_merges),
+      OptionType::kSizeT}},
+    {"memtable_prefix_bloom_huge_page_tlb_size",
+     {offsetof(struct ColumnFamilyOptions,
+               memtable_prefix_bloom_huge_page_tlb_size),
+      OptionType::kSizeT}},
+    {"write_buffer_size",
+     {offsetof(struct ColumnFamilyOptions, write_buffer_size),
+      OptionType::kSizeT}},
+    {"bloom_locality",
+     {offsetof(struct ColumnFamilyOptions, bloom_locality),
+      OptionType::kUInt32T}},
+    {"memtable_prefix_bloom_bits",
+     {offsetof(struct ColumnFamilyOptions, memtable_prefix_bloom_bits),
+      OptionType::kUInt32T}},
+    {"memtable_prefix_bloom_probes",
+     {offsetof(struct ColumnFamilyOptions, memtable_prefix_bloom_probes),
+      OptionType::kUInt32T}},
+    {"min_partial_merge_operands",
+     {offsetof(struct ColumnFamilyOptions, min_partial_merge_operands),
+      OptionType::kUInt32T}},
+    {"max_bytes_for_level_base",
+     {offsetof(struct ColumnFamilyOptions, max_bytes_for_level_base),
+      OptionType::kUInt64T}},
+    {"max_sequential_skip_in_iterations",
+     {offsetof(struct ColumnFamilyOptions, max_sequential_skip_in_iterations),
+      OptionType::kUInt64T}},
+    {"target_file_size_base",
+     {offsetof(struct ColumnFamilyOptions, target_file_size_base),
+      OptionType::kUInt64T}},
+    {"rate_limit_delay_max_milliseconds",
+     {offsetof(struct ColumnFamilyOptions, rate_limit_delay_max_milliseconds),
+      OptionType::kUInt}},
+    {"compaction_style",
+     {offsetof(struct ColumnFamilyOptions, compaction_style),
+      OptionType::kCompactionStyle}}};
+
 }  // namespace rocksdb

util/options_test.cc

@@ -638,6 +638,163 @@ TEST_F(OptionsTest, DBOptionsSerialization) {
   ASSERT_OK(GetDBOptionsFromString(DBOptions(), base_opt_string, &new_options));
   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,
+                               const ColumnFamilyOptions& new_opt) {
+  // custom type options
+  ASSERT_EQ(base_opt.compaction_style, new_opt.compaction_style);
+
+  // boolean options
+  ASSERT_EQ(base_opt.compaction_measure_io_stats,
+            new_opt.compaction_measure_io_stats);
+  ASSERT_EQ(base_opt.disable_auto_compactions,
+            new_opt.disable_auto_compactions);
+  ASSERT_EQ(base_opt.filter_deletes, new_opt.filter_deletes);
+  ASSERT_EQ(base_opt.inplace_update_support, new_opt.inplace_update_support);
+  ASSERT_EQ(base_opt.level_compaction_dynamic_level_bytes,
+            new_opt.level_compaction_dynamic_level_bytes);
+  ASSERT_EQ(base_opt.optimize_filters_for_hits,
+            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
+  VerifyDouble(base_opt.hard_rate_limit, new_opt.hard_rate_limit);
+  VerifyDouble(base_opt.soft_rate_limit, new_opt.soft_rate_limit);
+
+  // int options
+  ASSERT_EQ(base_opt.expanded_compaction_factor,
+            new_opt.expanded_compaction_factor);
+  ASSERT_EQ(base_opt.level0_file_num_compaction_trigger,
+            new_opt.level0_file_num_compaction_trigger);
+  ASSERT_EQ(base_opt.level0_slowdown_writes_trigger,
+            new_opt.level0_slowdown_writes_trigger);
+  ASSERT_EQ(base_opt.level0_stop_writes_trigger,
+            new_opt.level0_stop_writes_trigger);
+  ASSERT_EQ(base_opt.max_bytes_for_level_multiplier,
+            new_opt.max_bytes_for_level_multiplier);
+  ASSERT_EQ(base_opt.max_grandparent_overlap_factor,
+            new_opt.max_grandparent_overlap_factor);
+  ASSERT_EQ(base_opt.max_mem_compaction_level,
+            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);
+  ASSERT_EQ(base_opt.inplace_update_num_locks,
+            new_opt.inplace_update_num_locks);
+  ASSERT_EQ(base_opt.max_successive_merges, new_opt.max_successive_merges);
+  ASSERT_EQ(base_opt.memtable_prefix_bloom_huge_page_tlb_size,
+            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);
+  ASSERT_EQ(base_opt.memtable_prefix_bloom_bits,
+            new_opt.memtable_prefix_bloom_bits);
+  ASSERT_EQ(base_opt.memtable_prefix_bloom_probes,
+            new_opt.memtable_prefix_bloom_probes);
+  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,
+            new_opt.max_sequential_skip_in_iterations);
+  ASSERT_EQ(base_opt.target_file_size_base, new_opt.target_file_size_base);
+
+  // unsigned int options
+  ASSERT_EQ(base_opt.rate_limit_delay_max_milliseconds,
+            new_opt.rate_limit_delay_max_milliseconds);
+}
+}  // namespace
+
+TEST_F(OptionsTest, ColumnFamilyOptionsSerialization) {
+  ColumnFamilyOptions base_opt, new_opt;
+  Random rnd(302);
+  // Phase 1: randomly assign base_opt
+  // custom type options
+  base_opt.compaction_style = (CompactionStyle)(rnd.Uniform(4));
+
+  // 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.hard_rate_limit = static_cast<double>(rnd.Uniform(10000)) / 13;
+  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);
+
+  // 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;
+  ASSERT_OK(GetStringFromColumnFamilyOptions(base_opt, &base_opt_string));
+
+  // Phase 3: Set new_opt from the derived string and expect
+  //          new_opt == base_opt
+  ASSERT_OK(GetColumnFamilyOptionsFromString(ColumnFamilyOptions(),
+                                             base_opt_string, &new_opt));
+  VerifyColumnFamilyOptions(base_opt, new_opt);
+}
+
 #endif  // !ROCKSDB_LITE