dynamic max_mem_compact_level

Summary: as title

Test Plan: unit test

Reviewers: sdong, yhchiang, rven, igor

Reviewed By: igor

Subscribers: leveldb

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

db/db_impl.cc

@@ -1863,12 +1863,16 @@ int DBImpl::NumberLevels(ColumnFamilyHandle* column_family) {
 
 int DBImpl::MaxMemCompactionLevel(ColumnFamilyHandle* column_family) {
   auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
-  return cfh->cfd()->options()->max_mem_compaction_level;
+  MutexLock l(&mutex_);
+  return cfh->cfd()->GetSuperVersion()->
+      mutable_cf_options.max_mem_compaction_level;
 }
 
 int DBImpl::Level0StopWriteTrigger(ColumnFamilyHandle* column_family) {
   auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
-  return cfh->cfd()->options()->level0_stop_writes_trigger;
+  MutexLock l(&mutex_);
+  return cfh->cfd()->GetSuperVersion()->
+      mutable_cf_options.level0_stop_writes_trigger;
 }
 
 Status DBImpl::Flush(const FlushOptions& flush_options,

db/db_test.cc

@@ -8914,6 +8914,45 @@ TEST(DBTest, DynamicCompactionOptions) {
   }));
   dbfull()->TEST_FlushMemTable(true);
   ASSERT_TRUE(Put(Key(count), RandomString(&rnd, 1024), wo).ok());
+
+  // Test max_mem_compaction_level.
+  // Destory DB and start from scratch
+  options.max_background_compactions = 1;
+  options.max_background_flushes = 0;
+  options.max_mem_compaction_level = 2;
+  DestroyAndReopen(&options);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(2), 0);
+
+  ASSERT_TRUE(Put("max_mem_compaction_level_key",
+              RandomString(&rnd, 8)).ok());
+  dbfull()->TEST_FlushMemTable(true);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(2), 1);
+
+  ASSERT_TRUE(Put("max_mem_compaction_level_key",
+              RandomString(&rnd, 8)).ok());
+  // Set new value and it becomes effective in this flush
+  ASSERT_TRUE(dbfull()->SetOptions({
+    {"max_mem_compaction_level", "1"}
+  }));
+  dbfull()->TEST_FlushMemTable(true);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
+  ASSERT_EQ(NumTableFilesAtLevel(2), 1);
+
+  ASSERT_TRUE(Put("max_mem_compaction_level_key",
+              RandomString(&rnd, 8)).ok());
+  // Set new value and it becomes effective in this flush
+  ASSERT_TRUE(dbfull()->SetOptions({
+    {"max_mem_compaction_level", "0"}
+  }));
+  dbfull()->TEST_FlushMemTable(true);
+  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
+  ASSERT_EQ(NumTableFilesAtLevel(2), 1);
 }
 
 TEST(DBTest, DynamicMiscOptions) {

db/version_set.cc

@@ -895,7 +895,7 @@ void Version::ComputeCompactionScore(
       }
       if (cfd_->ioptions()->compaction_style == kCompactionStyleFIFO) {
         score = static_cast<double>(total_size) /
-                cfd_->options()->compaction_options_fifo.max_table_files_size;
+                cfd_->ioptions()->compaction_options_fifo.max_table_files_size;
       } else if (numfiles >= mutable_cf_options.level0_stop_writes_trigger) {
         // If we are slowing down writes, then we better compact that first
         score = 1000000;
@@ -1051,8 +1051,8 @@ int Version::PickLevelForMemTableOutput(
     InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek);
     InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0));
     std::vector<FileMetaData*> overlaps;
-    int max_mem_compact_level = cfd_->options()->max_mem_compaction_level;
+    while (mutable_cf_options.max_mem_compaction_level > 0 &&
-    while (max_mem_compact_level > 0 && level < max_mem_compact_level) {
+           level < mutable_cf_options.max_mem_compaction_level) {
       if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) {
         break;
       }

util/mutable_cf_options.cc

@@ -92,6 +92,8 @@ void MutableCFOptions::Dump(Logger* log) const {
       max_successive_merges);
   Log(log, "                           filter_deletes: %d",
       filter_deletes);
+  Log(log, "                 inplace_update_num_locks: %zu",
+      inplace_update_num_locks);
   Log(log, "                 disable_auto_compactions: %d",
       disable_auto_compactions);
   Log(log, "                          soft_rate_limit: %lf",
@@ -126,6 +128,10 @@ void MutableCFOptions::Dump(Logger* log) const {
   }
   result.resize(result.size() - 2);
   Log(log, "max_bytes_for_level_multiplier_additional: %s", result.c_str());
+  Log(log, "                 max_mem_compaction_level: %d",
+      max_mem_compaction_level);
+  Log(log, "        max_sequential_skip_in_iterations: %" PRIu64,
+      max_sequential_skip_in_iterations);
 }
 
 }  // namespace rocksdb

util/mutable_cf_options.h

@@ -38,6 +38,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),
+      max_mem_compaction_level(options.max_mem_compaction_level),
       max_sequential_skip_in_iterations(
           options.max_sequential_skip_in_iterations)
   {
@@ -65,6 +66,7 @@ struct MutableCFOptions {
       target_file_size_multiplier(0),
       max_bytes_for_level_base(0),
       max_bytes_for_level_multiplier(0),
+      max_mem_compaction_level(0),
       max_sequential_skip_in_iterations(0)
   {}
 
@@ -108,6 +110,7 @@ struct MutableCFOptions {
   uint64_t max_bytes_for_level_base;
   int max_bytes_for_level_multiplier;
   std::vector<int> max_bytes_for_level_multiplier_additional;
+  int max_mem_compaction_level;
 
   // Misc options
   uint64_t max_sequential_skip_in_iterations;

util/options_helper.cc

@@ -144,6 +144,8 @@ bool ParseCompactionOptions(const std::string& name, const std::string& value,
         start = end + 1;
       }
     }
+  } else if (name == "max_mem_compaction_level") {
+    new_options->max_mem_compaction_level = ParseInt(value);
   } else {
     return false;
   }
@@ -283,8 +285,6 @@ bool GetColumnFamilyOptionsFromMap(
             ParseInt(o.second.substr(start, o.second.size() - start));
       } else if (o.first == "num_levels") {
         new_options->num_levels = ParseInt(o.second);
-      } else if (o.first == "max_mem_compaction_level") {
-        new_options->max_mem_compaction_level = ParseInt(o.second);
       } else if (o.first == "purge_redundant_kvs_while_flush") {
         new_options->purge_redundant_kvs_while_flush =
           ParseBoolean(o.first, o.second);