Make FIFO compaction options dynamically configurable

Summary:
ColumnFamilyOptions::compaction_options_fifo and all its sub-fields can be set dynamically now.

Some of the ways in which the fifo compaction options can be set are:
- `SetOptions({{"compaction_options_fifo", "{max_table_files_size=1024}"}})`
- `SetOptions({{"compaction_options_fifo", "{ttl=600;}"}})`
- `SetOptions({{"compaction_options_fifo", "{max_table_files_size=1024;ttl=600;}"}})`
- `SetOptions({{"compaction_options_fifo", "{max_table_files_size=51;ttl=49;allow_compaction=true;}"}})`

Most of the code has been made generic enough so that it could be reused later to make universal options (and other such nested defined-types) dynamic with very few lines of parsing/serializing code changes.
Introduced a few new functions like `ParseStruct`, `SerializeStruct` and `GetStringFromStruct`.
The duplicate code in `GetStringFromDBOptions` and `GetStringFromColumnFamilyOptions` has been moved into `GetStringFromStruct`. So they become just simple wrappers now.
Closes https://github.com/facebook/rocksdb/pull/3006

Differential Revision: D6058619

Pulled By: sagar0

fbshipit-source-id: 1e8f78b3374ca5249bb4f3be8a6d3bb4cbc52f92

HISTORY.md

@@ -6,6 +6,7 @@
 ### New Features
 * `DBOptions::bytes_per_sync` and `DBOptions::wal_bytes_per_sync` can now be changed dynamically, `DBOptions::wal_bytes_per_sync` will flush all memtables and switch to a new WAL file.
 * Support dynamic adjustment of rate limit according to demand for background I/O. It can be enabled by passing `true` to the `auto_tuned` parameter in `NewGenericRateLimiter()`. The value passed as `rate_bytes_per_sec` will still be respected as an upper-bound.
+* Support dynamically changing `ColumnFamilyOptions::compaction_options_fifo`.
 
 ### Bug Fixes
 * Fix a potential data inconsistency issue during point-in-time recovery. `DB:Open()` will abort if column family inconsistency is found during PIT recovery.

db/compaction_picker.cc

@@ -1282,19 +1282,19 @@ uint32_t LevelCompactionBuilder::GetPathId(
         current_path_size -= level_size;
         if (cur_level > 0) {
           if (ioptions.level_compaction_dynamic_level_bytes) {
-            // Currently, level_compaction_dynamic_level_bytes is ignored when 
+            // Currently, level_compaction_dynamic_level_bytes is ignored when
             // multiple db paths are specified. https://github.com/facebook/
-            // rocksdb/blob/master/db/column_family.cc. 
+            // rocksdb/blob/master/db/column_family.cc.
-            // Still, adding this check to avoid accidentally using 
+            // Still, adding this check to avoid accidentally using
             // max_bytes_for_level_multiplier_additional
-            level_size = static_cast<uint64_t>( 
+            level_size = static_cast<uint64_t>(
-              level_size * mutable_cf_options.max_bytes_for_level_multiplier);
+                level_size * mutable_cf_options.max_bytes_for_level_multiplier);
           } else {
             level_size = static_cast<uint64_t>(
-              level_size * mutable_cf_options.max_bytes_for_level_multiplier 
+                level_size * mutable_cf_options.max_bytes_for_level_multiplier *
-                * mutable_cf_options.MaxBytesMultiplerAdditional(cur_level));
+                mutable_cf_options.MaxBytesMultiplerAdditional(cur_level));
           }
-	}
+        }
         cur_level++;
         continue;
       }
@@ -1425,7 +1425,7 @@ uint64_t GetTotalFilesSize(
 Compaction* FIFOCompactionPicker::PickTTLCompaction(
     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
     VersionStorageInfo* vstorage, LogBuffer* log_buffer) {
-  assert(ioptions_.compaction_options_fifo.ttl > 0);
+  assert(mutable_cf_options.compaction_options_fifo.ttl > 0);
 
   const int kLevel0 = 0;
   const std::vector<FileMetaData*>& level_files = vstorage->LevelFiles(kLevel0);
@@ -1454,7 +1454,7 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction(
           f->fd.table_reader->GetTableProperties()->creation_time;
       if (creation_time == 0 ||
           creation_time >=
-              (current_time - ioptions_.compaction_options_fifo.ttl)) {
+              (current_time - mutable_cf_options.compaction_options_fifo.ttl)) {
         break;
       }
       total_size -= f->compensated_file_size;
@@ -1467,7 +1467,8 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction(
   // 2. there are a few files older than ttl, but deleting them will not bring
   //    the total size to be less than max_table_files_size threshold.
   if (inputs[0].files.empty() ||
-      total_size > ioptions_.compaction_options_fifo.max_table_files_size) {
+      total_size >
+          mutable_cf_options.compaction_options_fifo.max_table_files_size) {
     return nullptr;
   }
 
@@ -1493,10 +1494,11 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
   const std::vector<FileMetaData*>& level_files = vstorage->LevelFiles(kLevel0);
   uint64_t total_size = GetTotalFilesSize(level_files);
 
-  if (total_size <= ioptions_.compaction_options_fifo.max_table_files_size ||
+  if (total_size <=
+          mutable_cf_options.compaction_options_fifo.max_table_files_size ||
       level_files.size() == 0) {
     // total size not exceeded
-    if (ioptions_.compaction_options_fifo.allow_compaction &&
+    if (mutable_cf_options.compaction_options_fifo.allow_compaction &&
         level_files.size() > 0) {
       CompactionInputFiles comp_inputs;
       if (FindIntraL0Compaction(
@@ -1516,11 +1518,12 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
       }
     }
 
-    ROCKS_LOG_BUFFER(log_buffer,
+    ROCKS_LOG_BUFFER(
-                     "[%s] FIFO compaction: nothing to do. Total size %" PRIu64
+        log_buffer,
-                     ", max size %" PRIu64 "\n",
+        "[%s] FIFO compaction: nothing to do. Total size %" PRIu64
-                     cf_name.c_str(), total_size,
+        ", max size %" PRIu64 "\n",
-                     ioptions_.compaction_options_fifo.max_table_files_size);
+        cf_name.c_str(), total_size,
+        mutable_cf_options.compaction_options_fifo.max_table_files_size);
     return nullptr;
   }
 
@@ -1547,7 +1550,8 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
                      "[%s] FIFO compaction: picking file %" PRIu64
                      " with size %s for deletion",
                      cf_name.c_str(), f->fd.GetNumber(), tmp_fsize);
-    if (total_size <= ioptions_.compaction_options_fifo.max_table_files_size) {
+    if (total_size <=
+        mutable_cf_options.compaction_options_fifo.max_table_files_size) {
       break;
     }
   }
@@ -1565,7 +1569,7 @@ Compaction* FIFOCompactionPicker::PickCompaction(
   assert(vstorage->num_levels() == 1);
 
   Compaction* c = nullptr;
-  if (ioptions_.compaction_options_fifo.ttl > 0) {
+  if (mutable_cf_options.compaction_options_fifo.ttl > 0) {
     c = PickTTLCompaction(cf_name, mutable_cf_options, vstorage, log_buffer);
   }
   if (c == nullptr) {

db/compaction_picker_test.cc

@@ -496,7 +496,7 @@ TEST_F(CompactionPickerTest, NeedsCompactionFIFO) {
   const uint64_t kMaxSize = kFileSize * kFileCount / 2;
 
   fifo_options_.max_table_files_size = kMaxSize;
-  ioptions_.compaction_options_fifo = fifo_options_;
+  mutable_cf_options_.compaction_options_fifo = fifo_options_;
   FIFOCompactionPicker fifo_compaction_picker(ioptions_, &icmp_);
   UpdateVersionStorageInfo();
   // must return false when there's no files.

db/db_options_test.cc

@@ -532,6 +532,114 @@ TEST_F(DBOptionsTest, SanitizeDelayedWriteRate) {
   ASSERT_EQ(31 * 1024 * 1024, dbfull()->GetDBOptions().delayed_write_rate);
 }
 
+TEST_F(DBOptionsTest, SetFIFOCompactionOptions) {
+  Options options;
+  options.compaction_style = kCompactionStyleFIFO;
+  options.write_buffer_size = 10 << 10;  // 10KB
+  options.arena_block_size = 4096;
+  options.compression = kNoCompression;
+  options.create_if_missing = true;
+  options.compaction_options_fifo.allow_compaction = false;
+  env_->time_elapse_only_sleep_ = false;
+  options.env = env_;
+
+  // Test dynamically changing compaction_options_fifo.ttl
+  env_->addon_time_.store(0);
+  options.compaction_options_fifo.ttl = 1 * 60 * 60;  // 1 hour
+  ASSERT_OK(TryReopen(options));
+
+  Random rnd(301);
+  for (int i = 0; i < 10; i++) {
+    // Generate and flush a file about 10KB.
+    for (int j = 0; j < 10; j++) {
+      ASSERT_OK(Put(ToString(i * 20 + j), RandomString(&rnd, 980)));
+    }
+    Flush();
+  }
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+
+  // Add 61 seconds to the time.
+  env_->addon_time_.fetch_add(61);
+
+  // No files should be compacted as ttl is set to 1 hour.
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 3600);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+
+  // Set ttl to 1 minute. So all files should get deleted.
+  ASSERT_OK(dbfull()->SetOptions({{"compaction_options_fifo", "{ttl=60;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 60);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+
+  // Test dynamically changing compaction_options_fifo.max_table_files_size
+  env_->addon_time_.store(0);
+  options.compaction_options_fifo.max_table_files_size = 500 << 10;  // 00KB
+  options.compaction_options_fifo.ttl = 0;
+  DestroyAndReopen(options);
+
+  for (int i = 0; i < 10; i++) {
+    // Generate and flush a file about 10KB.
+    for (int j = 0; j < 10; j++) {
+      ASSERT_OK(Put(ToString(i * 20 + j), RandomString(&rnd, 980)));
+    }
+    Flush();
+  }
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+
+  // No files should be compacted as max_table_files_size is set to 500 KB.
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            500 << 10);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+
+  // Set max_table_files_size to 12 KB. So only 1 file should remain now.
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"compaction_options_fifo", "{max_table_files_size=12288;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            12 << 10);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
+
+  // Test dynamically changing compaction_options_fifo.allow_compaction
+  options.compaction_options_fifo.max_table_files_size = 500 << 10;  // 500KB
+  options.compaction_options_fifo.ttl = 0;
+  options.compaction_options_fifo.allow_compaction = false;
+  options.level0_file_num_compaction_trigger = 6;
+  DestroyAndReopen(options);
+
+  for (int i = 0; i < 10; i++) {
+    // Generate and flush a file about 10KB.
+    for (int j = 0; j < 10; j++) {
+      ASSERT_OK(Put(ToString(i * 20 + j), RandomString(&rnd, 980)));
+    }
+    Flush();
+  }
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+
+  // No files should be compacted as max_table_files_size is set to 500 KB and
+  // allow_compaction is false
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            false);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+
+  // Set allow_compaction to true. So number of files should be between 1 and 5.
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"compaction_options_fifo", "{allow_compaction=true;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            true);
+  dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_GE(NumTableFilesAtLevel(0), 1);
+  ASSERT_LE(NumTableFilesAtLevel(0), 5);
+}
+
 #endif  // ROCKSDB_LITE
 
 }  // namespace rocksdb

db/db_test.cc

@@ -4292,6 +4292,72 @@ TEST_F(DBTest, DynamicCompactionOptions) {
 }
 #endif  // ROCKSDB_LITE
 
+// Test dynamic FIFO copmaction options.
+// This test covers just option parsing and makes sure that the options are
+// correctly assigned. Also look at DBOptionsTest.SetFIFOCompactionOptions
+// test which makes sure that the FIFO compaction funcionality is working
+// as expected on dynamically changing the options.
+// Even more FIFOCompactionTests are at DBTest.FIFOCompaction* .
+TEST_F(DBTest, DynamicFIFOCompactionOptions) {
+  Options options;
+  options.create_if_missing = true;
+  DestroyAndReopen(options);
+
+  // Initial defaults
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            1024 * 1024 * 1024);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 0);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            false);
+
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"compaction_options_fifo", "{max_table_files_size=23;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            23);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 0);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            false);
+
+  ASSERT_OK(dbfull()->SetOptions({{"compaction_options_fifo", "{ttl=97}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            23);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 97);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            false);
+
+  ASSERT_OK(dbfull()->SetOptions({{"compaction_options_fifo", "{ttl=203;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            23);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 203);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            false);
+
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"compaction_options_fifo", "{allow_compaction=true;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            23);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 203);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            true);
+
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"compaction_options_fifo", "{max_table_files_size=31;ttl=19;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            31);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 19);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            true);
+
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"compaction_options_fifo",
+        "{max_table_files_size=51;ttl=49;allow_compaction=true;}"}}));
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.max_table_files_size,
+            51);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.ttl, 49);
+  ASSERT_EQ(dbfull()->GetOptions().compaction_options_fifo.allow_compaction,
+            true);
+}
+
 TEST_F(DBTest, FileCreationRandomFailure) {
   Options options;
   options.env = env_;

db/version_set.cc

@@ -1328,6 +1328,7 @@ void VersionStorageInfo::EstimateCompactionBytesNeeded(
 
 namespace {
 uint32_t GetExpiredTtlFilesCount(const ImmutableCFOptions& ioptions,
+                                 const MutableCFOptions& mutable_cf_options,
                                  const std::vector<FileMetaData*>& files) {
   uint32_t ttl_expired_files_count = 0;
 
@@ -1341,8 +1342,8 @@ uint32_t GetExpiredTtlFilesCount(const ImmutableCFOptions& ioptions,
         auto creation_time =
             f->fd.table_reader->GetTableProperties()->creation_time;
         if (creation_time > 0 &&
-            creation_time <
+            creation_time < (current_time -
-                (current_time - ioptions.compaction_options_fifo.ttl)) {
+                             mutable_cf_options.compaction_options_fifo.ttl)) {
           ttl_expired_files_count++;
         }
       }
@@ -1389,19 +1390,19 @@ void VersionStorageInfo::ComputeCompactionScore(
       }
 
       if (compaction_style_ == kCompactionStyleFIFO) {
-        score =
+        score = static_cast<double>(total_size) /
-            static_cast<double>(total_size) /
+                mutable_cf_options.compaction_options_fifo.max_table_files_size;
-            immutable_cf_options.compaction_options_fifo.max_table_files_size;
+        if (mutable_cf_options.compaction_options_fifo.allow_compaction) {
-        if (immutable_cf_options.compaction_options_fifo.allow_compaction) {
           score = std::max(
               static_cast<double>(num_sorted_runs) /
                   mutable_cf_options.level0_file_num_compaction_trigger,
               score);
         }
-        if (immutable_cf_options.compaction_options_fifo.ttl > 0) {
+        if (mutable_cf_options.compaction_options_fifo.ttl > 0) {
-          score = std::max(static_cast<double>(GetExpiredTtlFilesCount(
+          score = std::max(
-                               immutable_cf_options, files_[level])),
+              static_cast<double>(GetExpiredTtlFilesCount(
-                           score);
+                  immutable_cf_options, mutable_cf_options, files_[level])),
+              score);
         }
 
       } else {

include/rocksdb/advanced_options.h

@@ -460,6 +460,10 @@ struct AdvancedColumnFamilyOptions {
   CompactionOptionsUniversal compaction_options_universal;
 
   // The options for FIFO compaction style
+  //
+  // Dynamically changeable through SetOptions() API
+  // Dynamic change example:
+  // SetOption("compaction_options_fifo", "{max_table_files_size=100;ttl=2;}")
   CompactionOptionsFIFO compaction_options_fifo;
 
   // An iteration->Next() sequentially skips over keys with the same

options/cf_options.cc

@@ -28,7 +28,6 @@ ImmutableCFOptions::ImmutableCFOptions(const ImmutableDBOptions& db_options,
     : compaction_style(cf_options.compaction_style),
       compaction_pri(cf_options.compaction_pri),
       compaction_options_universal(cf_options.compaction_options_universal),
-      compaction_options_fifo(cf_options.compaction_options_fifo),
       prefix_extractor(cf_options.prefix_extractor.get()),
       user_comparator(cf_options.comparator),
       internal_comparator(InternalKeyComparator(cf_options.comparator)),

options/cf_options.h

@@ -31,7 +31,6 @@ struct ImmutableCFOptions {
   CompactionPri compaction_pri;
 
   CompactionOptionsUniversal compaction_options_universal;
-  CompactionOptionsFIFO compaction_options_fifo;
 
   const SliceTransform* prefix_extractor;
 
@@ -149,6 +148,7 @@ struct MutableCFOptions {
         max_bytes_for_level_multiplier(options.max_bytes_for_level_multiplier),
         max_bytes_for_level_multiplier_additional(
             options.max_bytes_for_level_multiplier_additional),
+        compaction_options_fifo(options.compaction_options_fifo),
         max_sequential_skip_in_iterations(
             options.max_sequential_skip_in_iterations),
         paranoid_file_checks(options.paranoid_file_checks),
@@ -176,6 +176,7 @@ struct MutableCFOptions {
         target_file_size_multiplier(0),
         max_bytes_for_level_base(0),
         max_bytes_for_level_multiplier(0),
+        compaction_options_fifo(),
         max_sequential_skip_in_iterations(0),
         paranoid_file_checks(false),
         report_bg_io_stats(false),
@@ -222,6 +223,7 @@ struct MutableCFOptions {
   uint64_t max_bytes_for_level_base;
   double max_bytes_for_level_multiplier;
   std::vector<int> max_bytes_for_level_multiplier_additional;
+  CompactionOptionsFIFO compaction_options_fifo;
 
   // Misc options
   uint64_t max_sequential_skip_in_iterations;

options/options_helper.cc

@@ -165,6 +165,8 @@ ColumnFamilyOptions BuildColumnFamilyOptions(
     cf_opts.max_bytes_for_level_multiplier_additional.emplace_back(value);
   }
 
+  cf_opts.compaction_options_fifo = mutable_cf_options.compaction_options_fifo;
+
   // Misc options
   cf_opts.max_sequential_skip_in_iterations =
       mutable_cf_options.max_sequential_skip_in_iterations;
@@ -181,6 +183,12 @@ ColumnFamilyOptions BuildColumnFamilyOptions(
 
 #ifndef ROCKSDB_LITE
 
+template <typename T>
+Status GetStringFromStruct(
+    std::string* opt_string, const T& options,
+    const std::unordered_map<std::string, OptionTypeInfo> type_info,
+    const std::string& delimiter);
+
 namespace {
 template <typename T>
 bool ParseEnum(const std::unordered_map<std::string, T>& type_map,
@@ -255,6 +263,76 @@ bool ParseVectorCompressionType(
   return true;
 }
 
+// This is to handle backward compatibility, where compaction_options_fifo
+// could be assigned a single scalar value, say, like "23", which would be
+// assigned to max_table_files_size.
+bool FIFOCompactionOptionsSpecialCase(const std::string& opt_str,
+                                      CompactionOptionsFIFO* options) {
+  if (opt_str.find("=") != std::string::npos) {
+    // New format. Go do your new parsing using ParseStructOptions.
+    return false;
+  }
+
+  // Old format. Parse just a single uint64_t value.
+  options->max_table_files_size = ParseUint64(opt_str);
+  return true;
+}
+
+template <typename T>
+bool SerializeStruct(
+    const T& options, std::string* value,
+    std::unordered_map<std::string, OptionTypeInfo> type_info_map) {
+  std::string opt_str;
+  Status s = GetStringFromStruct(&opt_str, options, type_info_map, ";");
+  if (!s.ok()) {
+    return false;
+  }
+  *value = "{" + opt_str + "}";
+  return true;
+}
+
+template <typename T>
+bool ParseSingleStructOption(
+    const std::string& opt_val_str, T* options,
+    std::unordered_map<std::string, OptionTypeInfo> type_info_map) {
+  size_t end = opt_val_str.find('=');
+  std::string key = opt_val_str.substr(0, end);
+  std::string value = opt_val_str.substr(end + 1);
+  auto iter = type_info_map.find(key);
+  if (iter == type_info_map.end()) {
+    return false;
+  }
+  const auto& opt_info = iter->second;
+  return ParseOptionHelper(
+      reinterpret_cast<char*>(options) + opt_info.mutable_offset, opt_info.type,
+      value);
+}
+
+template <typename T>
+bool ParseStructOptions(
+    const std::string& opt_str, T* options,
+    std::unordered_map<std::string, OptionTypeInfo> type_info_map) {
+  assert(!opt_str.empty());
+
+  size_t start = 0;
+  if (opt_str[0] == '{') {
+    start++;
+  }
+  while ((start != std::string::npos) && (start < opt_str.size())) {
+    if (opt_str[start] == '}') {
+      break;
+    }
+    size_t end = opt_str.find(';', start);
+    size_t len = (end == std::string::npos) ? end : end - start;
+    if (!ParseSingleStructOption(opt_str.substr(start, len), options,
+                                 type_info_map)) {
+      return false;
+    }
+    start = (end == std::string::npos) ? end : end + 1;
+  }
+  return true;
+}
+
 bool ParseSliceTransformHelper(
     const std::string& kFixedPrefixName, const std::string& kCappedPrefixName,
     const std::string& value,
@@ -379,6 +457,15 @@ bool ParseOptionHelper(char* opt_address, const OptionType& opt_type,
       return ParseEnum<InfoLogLevel>(
           info_log_level_string_map, value,
           reinterpret_cast<InfoLogLevel*>(opt_address));
+    case OptionType::kCompactionOptionsFIFO: {
+      if (!FIFOCompactionOptionsSpecialCase(
+              value, reinterpret_cast<CompactionOptionsFIFO*>(opt_address))) {
+        return ParseStructOptions<CompactionOptionsFIFO>(
+            value, reinterpret_cast<CompactionOptionsFIFO*>(opt_address),
+            fifo_compaction_options_type_info);
+      }
+      return true;
+    }
     default:
       return false;
   }
@@ -543,6 +630,11 @@ bool SerializeSingleOptionHelper(const char* opt_address,
       return SerializeEnum<InfoLogLevel>(
           info_log_level_string_map,
           *reinterpret_cast<const InfoLogLevel*>(opt_address), value);
+    case OptionType::kCompactionOptionsFIFO: {
+      return SerializeStruct<CompactionOptionsFIFO>(
+          *reinterpret_cast<const CompactionOptionsFIFO*>(opt_address), value,
+          fifo_compaction_options_type_info);
+    }
     default:
       return false;
   }
@@ -768,9 +860,6 @@ Status ParseColumnFamilyOption(const std::string& name,
         new_options->compression_opts.max_dict_bytes =
             ParseInt(value.substr(start, value.size() - start));
       }
-    } else if (name == "compaction_options_fifo") {
-      new_options->compaction_options_fifo.max_table_files_size =
-          ParseUint64(value);
     } else {
       auto iter = cf_options_type_info.find(name);
       if (iter == cf_options_type_info.end()) {
@@ -805,17 +894,18 @@ Status ParseColumnFamilyOption(const std::string& name,
   return Status::OK();
 }
 
-bool SerializeSingleDBOption(std::string* opt_string,
+template <typename T>
-                             const DBOptions& db_options,
+bool SerializeSingleStructOption(
-                             const std::string& name,
+    std::string* opt_string, const T& options,
-                             const std::string& delimiter) {
+    const std::unordered_map<std::string, OptionTypeInfo> type_info,
-  auto iter = db_options_type_info.find(name);
+    const std::string& name, const std::string& delimiter) {
-  if (iter == db_options_type_info.end()) {
+  auto iter = type_info.find(name);
+  if (iter == type_info.end()) {
     return false;
   }
   auto& opt_info = iter->second;
   const char* opt_address =
-      reinterpret_cast<const char*>(&db_options) + opt_info.offset;
+      reinterpret_cast<const char*>(&options) + opt_info.offset;
   std::string value;
   bool result = SerializeSingleOptionHelper(opt_address, opt_info.type, &value);
   if (result) {
@@ -824,72 +914,45 @@ bool SerializeSingleDBOption(std::string* opt_string,
   return result;
 }
 
-Status GetStringFromDBOptions(std::string* opt_string,
+template <typename T>
-                              const DBOptions& db_options,
+Status GetStringFromStruct(
-                              const std::string& delimiter) {
+    std::string* opt_string, const T& options,
+    const std::unordered_map<std::string, OptionTypeInfo> type_info,
+    const std::string& delimiter) {
   assert(opt_string);
   opt_string->clear();
-  for (auto iter = db_options_type_info.begin();
+  for (auto iter = type_info.begin(); iter != type_info.end(); ++iter) {
-       iter != db_options_type_info.end(); ++iter) {
     if (iter->second.verification == OptionVerificationType::kDeprecated) {
       // If the option is no longer used in rocksdb and marked as deprecated,
       // we skip it in the serialization.
       continue;
     }
     std::string single_output;
-    bool result = SerializeSingleDBOption(&single_output, db_options,
+    bool result = SerializeSingleStructOption<T>(
-                                          iter->first, delimiter);
+        &single_output, options, type_info, iter->first, delimiter);
-    assert(result);
     if (result) {
       opt_string->append(single_output);
+    } else {
+      return Status::InvalidArgument("failed to serialize %s\n",
+                                     iter->first.c_str());
     }
+    assert(result);
   }
   return Status::OK();
 }
 
-bool SerializeSingleColumnFamilyOption(std::string* opt_string,
+Status GetStringFromDBOptions(std::string* opt_string,
-                                       const ColumnFamilyOptions& cf_options,
+                              const DBOptions& db_options,
-                                       const std::string& name,
+                              const std::string& delimiter) {
-                                       const std::string& delimiter) {
+  return GetStringFromStruct<DBOptions>(opt_string, db_options,
-  auto iter = cf_options_type_info.find(name);
+                                        db_options_type_info, delimiter);
-  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 + delimiter;
-  }
-  return result;
 }
 
 Status GetStringFromColumnFamilyOptions(std::string* opt_string,
                                         const ColumnFamilyOptions& cf_options,
                                         const std::string& delimiter) {
-  assert(opt_string);
+  return GetStringFromStruct<ColumnFamilyOptions>(
-  opt_string->clear();
+      opt_string, cf_options, cf_options_type_info, delimiter);
-  for (auto iter = cf_options_type_info.begin();
-       iter != cf_options_type_info.end(); ++iter) {
-    if (iter->second.verification == OptionVerificationType::kDeprecated) {
-      // If the option is no longer used in rocksdb and marked as deprecated,
-      // we skip it in the serialization.
-      continue;
-    }
-    std::string single_output;
-    bool result = SerializeSingleColumnFamilyOption(&single_output, cf_options,
-                                                    iter->first, delimiter);
-    if (result) {
-      opt_string->append(single_output);
-    } else {
-      return Status::InvalidArgument("failed to serialize %s\n",
-                                     iter->first.c_str());
-    }
-    assert(result);
-  }
-  return Status::OK();
 }
 
 Status GetStringFromCompressionType(std::string* compression_str,

options/options_helper.h

@@ -82,6 +82,7 @@ enum class OptionType {
   kComparator,
   kCompactionFilter,
   kCompactionFilterFactory,
+  kCompactionOptionsFIFO,
   kMergeOperator,
   kMemTableRepFactory,
   kBlockBasedTableIndexType,
@@ -587,7 +588,26 @@ static std::unordered_map<std::string, OptionTypeInfo> cf_options_type_info = {
       OptionType::kCompactionStyle, OptionVerificationType::kNormal, false, 0}},
     {"compaction_pri",
      {offset_of(&ColumnFamilyOptions::compaction_pri),
-      OptionType::kCompactionPri, OptionVerificationType::kNormal, false, 0}}};
+      OptionType::kCompactionPri, OptionVerificationType::kNormal, false, 0}},
+    {"compaction_options_fifo",
+     {offset_of(&ColumnFamilyOptions::compaction_options_fifo),
+      OptionType::kCompactionOptionsFIFO, OptionVerificationType::kNormal, true,
+      offsetof(struct MutableCFOptions, compaction_options_fifo)}}};
+
+static std::unordered_map<std::string, OptionTypeInfo>
+    fifo_compaction_options_type_info = {
+        {"max_table_files_size",
+         {offset_of(&CompactionOptionsFIFO::max_table_files_size),
+          OptionType::kUInt64T, OptionVerificationType::kNormal, true,
+          offsetof(struct CompactionOptionsFIFO, max_table_files_size)}},
+        {"ttl",
+         {offset_of(&CompactionOptionsFIFO::ttl), OptionType::kUInt64T,
+          OptionVerificationType::kNormal, true,
+          offsetof(struct CompactionOptionsFIFO, ttl)}},
+        {"allow_compaction",
+         {offset_of(&CompactionOptionsFIFO::allow_compaction),
+          OptionType::kBoolean, OptionVerificationType::kNormal, true,
+          offsetof(struct CompactionOptionsFIFO, allow_compaction)}}};
 
 static std::unordered_map<std::string, CompressionType>
     compression_type_string_map = {

options/options_parser.cc

@@ -587,6 +587,17 @@ bool AreEqualOptions(
     case OptionType::kInfoLogLevel:
       return (*reinterpret_cast<const InfoLogLevel*>(offset1) ==
               *reinterpret_cast<const InfoLogLevel*>(offset2));
+    case OptionType::kCompactionOptionsFIFO: {
+      CompactionOptionsFIFO lhs =
+          *reinterpret_cast<const CompactionOptionsFIFO*>(offset1);
+      CompactionOptionsFIFO rhs =
+          *reinterpret_cast<const CompactionOptionsFIFO*>(offset2);
+      if (lhs.max_table_files_size == rhs.max_table_files_size &&
+          lhs.ttl == rhs.ttl && lhs.allow_compaction == rhs.allow_compaction) {
+        return true;
+      }
+      return false;
+    }
     default:
       if (type_info.verification == OptionVerificationType::kByName ||
           type_info.verification ==

options/options_settable_test.cc

@@ -371,7 +371,6 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
   options->hard_rate_limit = 0;
   options->soft_rate_limit = 0;
   options->purge_redundant_kvs_while_flush = false;
-  options->compaction_options_fifo = CompactionOptionsFIFO();
   options->max_mem_compaction_level = 0;
 
   char* new_options_ptr = new char[sizeof(ColumnFamilyOptions)];
@@ -427,7 +426,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "compaction_pri=kMinOverlappingRatio;"
       "hard_pending_compaction_bytes_limit=0;"
       "disable_auto_compactions=false;"
-      "report_bg_io_stats=true;",
+      "report_bg_io_stats=true;"
+      "compaction_options_fifo={max_table_files_size=3;ttl=100;allow_compaction=false;};",
       new_options));
 
   ASSERT_EQ(unset_bytes_base,

util/testutil.cc

@@ -310,6 +310,7 @@ void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, Random* rnd) {
   cf_opt->paranoid_file_checks = rnd->Uniform(2);
   cf_opt->purge_redundant_kvs_while_flush = rnd->Uniform(2);
   cf_opt->force_consistency_checks = rnd->Uniform(2);
+  cf_opt->compaction_options_fifo.allow_compaction = rnd->Uniform(2);
 
   // double options
   cf_opt->hard_rate_limit = static_cast<double>(rnd->Uniform(10000)) / 13;
@@ -352,6 +353,9 @@ void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, Random* rnd) {
   cf_opt->target_file_size_base = uint_max + rnd->Uniform(10000);
   cf_opt->max_compaction_bytes =
       cf_opt->target_file_size_base * rnd->Uniform(100);
+  cf_opt->compaction_options_fifo.max_table_files_size =
+      uint_max + rnd->Uniform(10000);
+  cf_opt->compaction_options_fifo.ttl = uint_max + rnd->Uniform(10000);
 
   // unsigned int options
   cf_opt->rate_limit_delay_max_milliseconds = rnd->Uniform(10000);