New ldb command to convert compaction style

Summary:
Add new command "change_compaction_style" to ldb tool. For
universal->level, it shows "nothing to do". For level->universal, it
compacts all files into a single one and moves the file to level 0.

Also add check for number of files at level 1+ when opening db with
universal compaction style.

Test Plan:
'make all check'. New unit test for internal convertion function. Also manully test various
cmd like:

./ldb change_compaction_style --old_compaction_style=0
--new_compaction_style=1 --db=/tmp/leveldbtest-3088/db_test

Reviewers: haobo, dhruba

Reviewed By: haobo

CC: vamsi, emayanke

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

db/db_impl.cc

@@ -946,7 +946,7 @@ Status DBImpl::CompactMemTable(bool* madeProgress) {
 }
 
 void DBImpl::CompactRange(const Slice* begin, const Slice* end,
-                          bool reduce_level) {
+                          bool reduce_level, int target_level) {
   int max_level_with_files = 1;
   {
     MutexLock l(&mutex_);
@@ -963,7 +963,7 @@ void DBImpl::CompactRange(const Slice* begin, const Slice* end,
   }
 
   if (reduce_level) {
-    ReFitLevel(max_level_with_files);
+    ReFitLevel(max_level_with_files, target_level);
   }
 }
 
@@ -983,7 +983,7 @@ int DBImpl::FindMinimumEmptyLevelFitting(int level) {
   return minimum_level;
 }
 
-void DBImpl::ReFitLevel(int level) {
+void DBImpl::ReFitLevel(int level, int target_level) {
   assert(level < NumberLevels());
 
   MutexLock l(&mutex_);
@@ -1005,7 +1005,10 @@ void DBImpl::ReFitLevel(int level) {
   }
 
   // move to a smaller level
-  int to_level = FindMinimumEmptyLevelFitting(level);
+  int to_level = target_level;
+  if (target_level < 0) {
+    to_level = FindMinimumEmptyLevelFitting(level);
+  }
 
   assert(to_level <= level);
 
@@ -3107,6 +3110,20 @@ Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {
     }
   }
   impl->mutex_.Unlock();
+
+  if (options.compaction_style == kCompactionStyleUniversal) {
+    std::string property;
+    int num_files;
+    for (int i = 1; i < impl->NumberLevels(); i++) {
+      num_files = impl->versions_->NumLevelFiles(i);
+      if (num_files > 0) {
+        s = Status::InvalidArgument("Not all files are at level 0. Cannot "
+          "open with universal compaction style.");
+        break;
+      }
+    }
+  }
+
   if (s.ok()) {
     *dbptr = impl;
   } else {

db/db_impl.h

@@ -64,7 +64,7 @@ class DBImpl : public DB {
   virtual bool GetProperty(const Slice& property, std::string* value);
   virtual void GetApproximateSizes(const Range* range, int n, uint64_t* sizes);
   virtual void CompactRange(const Slice* begin, const Slice* end,
-                            bool reduce_level = false);
+                            bool reduce_level = false, int target_level = -1);
   virtual int NumberLevels();
   virtual int MaxMemCompactionLevel();
   virtual int Level0StopWriteTrigger();
@@ -241,9 +241,10 @@ class DBImpl : public DB {
   // input level. Return the input level, if such level could not be found.
   int FindMinimumEmptyLevelFitting(int level);
 
-  // Move the files in the input level to the minimum level that could hold
-  // the data set.
-  void ReFitLevel(int level);
+  // Move the files in the input level to the target level.
+  // If target_level < 0, automatically calculate the minimum level that could
+  // hold the data set.
+  void ReFitLevel(int level, int target_level = -1);
 
   // Constant after construction
   const InternalFilterPolicy internal_filter_policy_;

db/db_impl_readonly.h

@@ -51,7 +51,7 @@ public:
    return Status::NotSupported("Not supported operation in read only mode.");
  }
  virtual void CompactRange(const Slice* begin, const Slice* end,
-                           bool reduce_level = false) {
+                           bool reduce_level = false, int target_level = -1) {
  }
  virtual Status DisableFileDeletions() {
    return Status::NotSupported("Not supported operation in read only mode.");

db/db_test.cc

@@ -1620,6 +1620,100 @@ TEST(DBTest, UniversalCompactionTrigger) {
   }
 }
 
+TEST(DBTest, ConvertCompactionStyle) {
+  Random rnd(301);
+  int max_key_level_insert = 200;
+  int max_key_universal_insert = 600;
+
+  // Stage 1: generate a db with level compaction
+  Options options = CurrentOptions();
+  options.write_buffer_size = 100<<10; //100KB
+  options.num_levels = 4;
+  options.level0_file_num_compaction_trigger = 3;
+  options.max_bytes_for_level_base = 500<<10; // 500KB
+  options.max_bytes_for_level_multiplier = 1;
+  options.target_file_size_base = 200<<10; // 200KB
+  options.target_file_size_multiplier = 1;
+  Reopen(&options);
+
+  for (int i = 0; i <= max_key_level_insert; i++) {
+    // each value is 10K
+    ASSERT_OK(Put(Key(i), RandomString(&rnd, 10000)));
+  }
+  dbfull()->Flush(FlushOptions());
+  dbfull()->TEST_WaitForCompact();
+
+  ASSERT_GT(TotalTableFiles(), 1);
+  int non_level0_num_files = 0;
+  for (int i = 1; i < dbfull()->NumberLevels(); i++) {
+    non_level0_num_files += NumTableFilesAtLevel(i);
+  }
+  ASSERT_GT(non_level0_num_files, 0);
+
+  // Stage 2: reopen with universal compaction - should fail
+  options = CurrentOptions();
+  options.compaction_style = kCompactionStyleUniversal;
+  Status s = TryReopen(&options);
+  ASSERT_TRUE(s.IsInvalidArgument());
+
+  // Stage 3: compact into a single file and move the file to level 0
+  options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  options.target_file_size_base = INT_MAX;
+  options.target_file_size_multiplier = 1;
+  options.max_bytes_for_level_base = INT_MAX;
+  options.max_bytes_for_level_multiplier = 1;
+  Reopen(&options);
+
+  dbfull()->CompactRange(nullptr, nullptr,
+                         true /* reduce level */,
+                         0    /* reduce to level 0 */);
+
+  for (int i = 0; i < dbfull()->NumberLevels(); i++) {
+    int num = NumTableFilesAtLevel(i);
+    if (i == 0) {
+      ASSERT_EQ(num, 1);
+    } else {
+      ASSERT_EQ(num, 0);
+    }
+  }
+
+  // Stage 4: re-open in universal compaction style and do some db operations
+  options = CurrentOptions();
+  options.compaction_style = kCompactionStyleUniversal;
+  options.write_buffer_size = 100<<10; //100KB
+  options.level0_file_num_compaction_trigger = 3;
+  Reopen(&options);
+
+  for (int i = max_key_level_insert / 2; i <= max_key_universal_insert; i++) {
+    ASSERT_OK(Put(Key(i), RandomString(&rnd, 10000)));
+  }
+  dbfull()->Flush(FlushOptions());
+  dbfull()->TEST_WaitForCompact();
+
+  for (int i = 1; i < dbfull()->NumberLevels(); i++) {
+    ASSERT_EQ(NumTableFilesAtLevel(i), 0);
+  }
+
+  // verify keys inserted in both level compaction style and universal
+  // compaction style
+  std::string keys_in_db;
+  Iterator* iter = dbfull()->NewIterator(ReadOptions());
+  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+    keys_in_db.append(iter->key().ToString());
+    keys_in_db.push_back(',');
+  }
+  delete iter;
+
+  std::string expected_keys;
+  for (int i = 0; i <= max_key_universal_insert; i++) {
+    expected_keys.append(Key(i));
+    expected_keys.push_back(',');
+  }
+
+  ASSERT_EQ(keys_in_db, expected_keys);
+}
+
 void MinLevelHelper(DBTest* self, Options& options) {
   Random rnd(301);
 
@@ -3530,7 +3624,7 @@ class ModelDB: public DB {
     }
   }
   virtual void CompactRange(const Slice* start, const Slice* end,
-                            bool reduce_level ) {
+                            bool reduce_level, int target_level) {
   }
 
   virtual int NumberLevels()

include/rocksdb/db.h

@@ -209,9 +209,10 @@ class DB {
   // after compaction is reduced, that level might not be appropriate for
   // hosting all the files. In this case, client could set reduce_level
   // to true, to move the files back to the minimum level capable of holding
-  // the data set.
+  // the data set or a given level (specified by non-negative target_level).
   virtual void CompactRange(const Slice* begin, const Slice* end,
-                            bool reduce_level = false) = 0;
+                            bool reduce_level = false,
+                            int target_level = -1) = 0;
 
   // Number of levels used for this DB.
   virtual int NumberLevels() = 0;

include/utilities/stackable_db.h

@@ -104,8 +104,9 @@ class StackableDB : public DB {
   }
 
   virtual void CompactRange(const Slice* begin, const Slice* end,
-                            bool reduce_level = false) override {
-    return sdb_->CompactRange(begin, end, reduce_level);
+                            bool reduce_level = false,
+                            int target_level = -1) override {
+    return sdb_->CompactRange(begin, end, reduce_level, target_level);
   }
 
   virtual int NumberLevels() override {

util/ldb_cmd.cc

@@ -143,6 +143,8 @@ LDBCommand* LDBCommand::SelectCommand(
     return new WALDumperCommand(cmdParams, option_map, flags);
   } else if (cmd == ReduceDBLevelsCommand::Name()) {
     return new ReduceDBLevelsCommand(cmdParams, option_map, flags);
+  } else if (cmd == ChangeCompactionStyleCommand::Name()) {
+    return new ChangeCompactionStyleCommand(cmdParams, option_map, flags);
   } else if (cmd == DBDumperCommand::Name()) {
     return new DBDumperCommand(cmdParams, option_map, flags);
   } else if (cmd == DBLoaderCommand::Name()) {
@@ -996,6 +998,138 @@ void ReduceDBLevelsCommand::DoCommand() {
   }
 }
 
+const string ChangeCompactionStyleCommand::ARG_OLD_COMPACTION_STYLE =
+  "old_compaction_style";
+const string ChangeCompactionStyleCommand::ARG_NEW_COMPACTION_STYLE =
+  "new_compaction_style";
+
+ChangeCompactionStyleCommand::ChangeCompactionStyleCommand(
+      const vector<string>& params, const map<string, string>& options,
+      const vector<string>& flags) :
+    LDBCommand(options, flags, false,
+               BuildCmdLineOptions({ARG_OLD_COMPACTION_STYLE,
+                                    ARG_NEW_COMPACTION_STYLE})),
+    old_compaction_style_(-1),
+    new_compaction_style_(-1) {
+
+  ParseIntOption(option_map_, ARG_OLD_COMPACTION_STYLE, old_compaction_style_,
+    exec_state_);
+  if (old_compaction_style_ != kCompactionStyleLevel &&
+     old_compaction_style_ != kCompactionStyleUniversal) {
+    exec_state_ = LDBCommandExecuteResult::FAILED(
+      "Use --" + ARG_OLD_COMPACTION_STYLE + " to specify old compaction " +
+      "style. Check ldb help for proper compaction style value.\n");
+    return;
+  }
+
+  ParseIntOption(option_map_, ARG_NEW_COMPACTION_STYLE, new_compaction_style_,
+    exec_state_);
+  if (new_compaction_style_ != kCompactionStyleLevel &&
+     new_compaction_style_ != kCompactionStyleUniversal) {
+    exec_state_ = LDBCommandExecuteResult::FAILED(
+      "Use --" + ARG_NEW_COMPACTION_STYLE + " to specify new compaction " +
+      "style. Check ldb help for proper compaction style value.\n");
+    return;
+  }
+
+  if (new_compaction_style_ == old_compaction_style_) {
+    exec_state_ = LDBCommandExecuteResult::FAILED(
+      "Old compaction style is the same as new compaction style. "
+      "Nothing to do.\n");
+    return;
+  }
+
+  if (old_compaction_style_ == kCompactionStyleUniversal &&
+      new_compaction_style_ == kCompactionStyleLevel) {
+    exec_state_ = LDBCommandExecuteResult::FAILED(
+      "Convert from universal compaction to level compaction. "
+      "Nothing to do.\n");
+    return;
+  }
+}
+
+void ChangeCompactionStyleCommand::Help(string& ret) {
+  ret.append("  ");
+  ret.append(ChangeCompactionStyleCommand::Name());
+  ret.append(" --" + ARG_OLD_COMPACTION_STYLE + "=<Old compaction style: 0 " +
+             "for level compaction, 1 for universal compaction>");
+  ret.append(" --" + ARG_NEW_COMPACTION_STYLE + "=<New compaction style: 0 " +
+             "for level compaction, 1 for universal compaction>");
+  ret.append("\n");
+}
+
+Options ChangeCompactionStyleCommand::PrepareOptionsForOpenDB() {
+  Options opt = LDBCommand::PrepareOptionsForOpenDB();
+
+  if (old_compaction_style_ == kCompactionStyleLevel &&
+      new_compaction_style_ == kCompactionStyleUniversal) {
+    // In order to convert from level compaction to universal compaction, we
+    // need to compact all data into a single file and move it to level 0.
+    opt.disable_auto_compactions = true;
+    opt.target_file_size_base = INT_MAX;
+    opt.target_file_size_multiplier = 1;
+    opt.max_bytes_for_level_base = INT_MAX;
+    opt.max_bytes_for_level_multiplier = 1;
+  }
+
+  return opt;
+}
+
+void ChangeCompactionStyleCommand::DoCommand() {
+  // print db stats before we have made any change
+  std::string property;
+  std::string files_per_level;
+  for (int i = 0; i < db_->NumberLevels(); i++) {
+    db_->GetProperty("leveldb.num-files-at-level" + NumberToString(i),
+                     &property);
+
+    // format print string
+    char buf[100];
+    snprintf(buf, sizeof(buf), "%s%s", (i ? "," : ""), property.c_str());
+    files_per_level += buf;
+  }
+  fprintf(stdout, "files per level before compaction: %s\n",
+          files_per_level.c_str());
+
+  // manual compact into a single file and move the file to level 0
+  db_->CompactRange(nullptr, nullptr,
+                    true /* reduce level */,
+                    0    /* reduce to level 0 */);
+
+  // verify compaction result
+  files_per_level = "";
+  int num_files = 0;
+  for (int i = 0; i < db_->NumberLevels(); i++) {
+    db_->GetProperty("leveldb.num-files-at-level" + NumberToString(i),
+                     &property);
+
+    // format print string
+    char buf[100];
+    snprintf(buf, sizeof(buf), "%s%s", (i ? "," : ""), property.c_str());
+    files_per_level += buf;
+
+    num_files = atoi(property.c_str());
+
+    // level 0 should have only 1 file
+    if (i == 0 && num_files != 1) {
+      exec_state_ = LDBCommandExecuteResult::FAILED("Number of db files at "
+        "level 0 after compaction is " + std::to_string(num_files) +
+        ", not 1.\n");
+      return;
+    }
+    // other levels should have no file
+    if (i > 0 && num_files != 0) {
+      exec_state_ = LDBCommandExecuteResult::FAILED("Number of db files at "
+        "level " + std::to_string(i) + " after compaction is " +
+        std::to_string(num_files) + ", not 0.\n");
+      return;
+    }
+  }
+
+  fprintf(stdout, "files per level after compaction: %s\n",
+          files_per_level.c_str());
+}
+
 class InMemoryHandler : public WriteBatch::Handler {
  public:
 
@@ -1050,8 +1184,8 @@ void WALDumperCommand::Help(string& ret) {
   ret.append("  ");
   ret.append(WALDumperCommand::Name());
   ret.append(" --" + ARG_WAL_FILE + "=<write_ahead_log_file_path>");
-  ret.append(" --[" + ARG_PRINT_HEADER + "] ");
-  ret.append(" --[ " + ARG_PRINT_VALUE + "] ");
+  ret.append(" [--" + ARG_PRINT_HEADER + "] ");
+  ret.append(" [--" + ARG_PRINT_VALUE + "] ");
   ret.append("\n");
 }
 

util/ldb_cmd.h

@@ -509,6 +509,27 @@ private:
   Status GetOldNumOfLevels(Options& opt, int* levels);
 };
 
+class ChangeCompactionStyleCommand : public LDBCommand {
+public:
+  static string Name() { return "change_compaction_style"; }
+
+  ChangeCompactionStyleCommand(const vector<string>& params,
+      const map<string, string>& options, const vector<string>& flags);
+
+  virtual Options PrepareOptionsForOpenDB();
+
+  virtual void DoCommand();
+
+  static void Help(string& msg);
+
+private:
+  int old_compaction_style_;
+  int new_compaction_style_;
+
+  static const string ARG_OLD_COMPACTION_STYLE;
+  static const string ARG_NEW_COMPACTION_STYLE;
+};
+
 class WALDumperCommand : public LDBCommand {
 public:
   static string Name() { return "dump_wal"; }

util/ldb_tool.cc

@@ -56,6 +56,7 @@ public:
     WALDumperCommand::Help(ret);
     CompactorCommand::Help(ret);
     ReduceDBLevelsCommand::Help(ret);
+    ChangeCompactionStyleCommand::Help(ret);
     DBDumperCommand::Help(ret);
     DBLoaderCommand::Help(ret);
     ManifestDumpCommand::Help(ret);

utilities/ttl/db_ttl.cc

@@ -236,8 +236,8 @@ void DBWithTTL::GetApproximateSizes(const Range* r, int n, uint64_t* sizes) {
 }
 
 void DBWithTTL::CompactRange(const Slice* begin, const Slice* end,
-                             bool reduce_level) {
-  db_->CompactRange(begin, end, reduce_level);
+                             bool reduce_level, int target_level) {
+  db_->CompactRange(begin, end, reduce_level, target_level);
 }
 
 int DBWithTTL::NumberLevels() {

utilities/ttl/db_ttl.h

@@ -61,7 +61,7 @@ class DBWithTTL : public StackableDB {
   virtual void GetApproximateSizes(const Range* r, int n, uint64_t* sizes);
 
   virtual void CompactRange(const Slice* begin, const Slice* end,
-                            bool reduce_level = false);
+                            bool reduce_level = false, int target_level = -1);
 
   virtual int NumberLevels();