update new implementation

db/corruption_test.cc

@@ -228,8 +228,8 @@ TEST_F(CorruptionTest, TableFile) {
   Build(100);
   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
   dbi->TEST_CompactMemTable();
-  dbi->TEST_CompactRange(0, nullptr, nullptr);
-  dbi->TEST_CompactRange(1, nullptr, nullptr);
+  dbi->TEST_CompactRange(0, nullptr, nullptr, false);
+  dbi->TEST_CompactRange(1, nullptr, nullptr, false);
 
   Corrupt(kTableFile, 100, 1);
   Check(90, 99);
@@ -242,8 +242,8 @@ TEST_F(CorruptionTest, TableFileRepair) {
   Build(100);
   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
   dbi->TEST_CompactMemTable();
-  dbi->TEST_CompactRange(0, nullptr, nullptr);
-  dbi->TEST_CompactRange(1, nullptr, nullptr);
+  dbi->TEST_CompactRange(0, nullptr, nullptr, false);
+  dbi->TEST_CompactRange(1, nullptr, nullptr, false);
 
   Corrupt(kTableFile, 100, 1);
   RepairDB();
@@ -293,7 +293,7 @@ TEST_F(CorruptionTest, CorruptedDescriptor) {
   ASSERT_LEVELDB_OK(db_->Put(WriteOptions(), "foo", "hello"));
   DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
   dbi->TEST_CompactMemTable();
-  dbi->TEST_CompactRange(0, nullptr, nullptr);
+  dbi->TEST_CompactRange(0, nullptr, nullptr, false);
 
   Corrupt(kDescriptorFile, 0, 1000);
   Status s = TryReopen();

db/db_impl.cc

@@ -595,12 +595,12 @@ void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
   }
   TEST_CompactMemTable();  // TODO(sanjay): Skip if memtable does not overlap
   for (int level = 0; level < max_level_with_files + 1; level++) {
-    TEST_CompactRange(level, begin, end);
+    TEST_CompactRange(level, begin, end, level == max_level_with_files);
   }
 }
 
 void DBImpl::TEST_CompactRange(int level, const Slice* begin,
-                               const Slice* end) {
+                               const Slice* end, bool is_last_level) {
   assert(level >= 0);
   assert(level < config::kNumLevels);
 
@@ -608,6 +608,7 @@ void DBImpl::TEST_CompactRange(int level, const Slice* begin,
 
   ManualCompaction manual;
   manual.level = level;
+  manual.is_last_level = is_last_level;
   manual.done = false;
   if (begin == nullptr) {
     manual.begin = nullptr;
@@ -720,7 +721,7 @@ void DBImpl::BackgroundCompaction() {
   InternalKey manual_end;
   if (is_manual) {
     ManualCompaction* m = manual_compaction_;
-    c = versions_->CompactRange(m->level, m->begin, m->end);
+    c = versions_->CompactRange(m->level, m->begin, m->end, m->is_last_level);
     m->done = (c == nullptr);
     if (c != nullptr) {
       manual_end = c->input(0, c->num_input_files(0) - 1)->largest;
@@ -891,7 +892,7 @@ Status DBImpl::InstallCompactionResults(CompactionState* compact) {
   const int level = compact->compaction->level();
   for (size_t i = 0; i < compact->outputs.size(); i++) {
     const CompactionState::Output& out = compact->outputs[i];
-    if (level < config::kNumLevels - 1) {
+    if (!compact->compaction->is_last_level()) {
       compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size,
                                            out.smallest, out.largest);
     } else {
@@ -1059,7 +1060,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
   }
 
   mutex_.Lock();
-  if (compact->compaction->level() + 1 < config::kNumLevels) {
+  if (!compact->compaction->is_last_level()) {
     stats_[compact->compaction->level() + 1].Add(stats);
   } else {
     // TTL: compaction for last level

db/db_impl.h

@@ -54,7 +54,8 @@ class DBImpl : public DB {
   // Extra methods (for testing) that are not in the public DB interface
 
   // Compact any files in the named level that overlap [*begin,*end]
-  void TEST_CompactRange(int level, const Slice* begin, const Slice* end);
+  void TEST_CompactRange(int level, const Slice* begin, const Slice* end,
+                         bool is_last_level);
 
   // Force current memtable contents to be compacted.
   Status TEST_CompactMemTable();
@@ -81,6 +82,7 @@ class DBImpl : public DB {
   // Information for a manual compaction
   struct ManualCompaction {
     int level;
+    bool is_last_level;  // TTL: Used to check if last level with files
     bool done;
     const InternalKey* begin;  // null means beginning of key range
     const InternalKey* end;    // null means end of key range

db/db_test.cc

@@ -786,7 +786,7 @@ TEST_F(DBTest, GetEncountersEmptyLevel) {
     }
 
     // Step 2: clear level 1 if necessary.
-    dbfull()->TEST_CompactRange(1, nullptr, nullptr);
+    dbfull()->TEST_CompactRange(1, nullptr, nullptr, false);
     ASSERT_EQ(NumTableFilesAtLevel(0), 1);
     ASSERT_EQ(NumTableFilesAtLevel(1), 0);
     ASSERT_EQ(NumTableFilesAtLevel(2), 1);
@@ -1145,7 +1145,7 @@ TEST_F(DBTest, CompactionsGenerateMultipleFiles) {
 
   // Reopening moves updates to level-0
   Reopen(&options);
-  dbfull()->TEST_CompactRange(0, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(0, nullptr, nullptr, false);
 
   ASSERT_EQ(NumTableFilesAtLevel(0), 0);
   ASSERT_GT(NumTableFilesAtLevel(1), 1);
@@ -1196,7 +1196,7 @@ TEST_F(DBTest, SparseMerge) {
   }
   Put("C", "vc");
   dbfull()->TEST_CompactMemTable();
-  dbfull()->TEST_CompactRange(0, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(0, nullptr, nullptr, false);
 
   // Make sparse update
   Put("A", "va2");
@@ -1207,9 +1207,9 @@ TEST_F(DBTest, SparseMerge) {
   // Compactions should not cause us to create a situation where
   // a file overlaps too much data at the next level.
   ASSERT_LE(dbfull()->TEST_MaxNextLevelOverlappingBytes(), 20 * 1048576);
-  dbfull()->TEST_CompactRange(0, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(0, nullptr, nullptr, false);
   ASSERT_LE(dbfull()->TEST_MaxNextLevelOverlappingBytes(), 20 * 1048576);
-  dbfull()->TEST_CompactRange(1, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(1, nullptr, nullptr, false);
   ASSERT_LE(dbfull()->TEST_MaxNextLevelOverlappingBytes(), 20 * 1048576);
 }
 
@@ -1273,7 +1273,7 @@ TEST_F(DBTest, ApproximateSizes) {
         std::string cend_str = Key(compact_start + 9);
         Slice cstart = cstart_str;
         Slice cend = cend_str;
-        dbfull()->TEST_CompactRange(0, &cstart, &cend);
+        dbfull()->TEST_CompactRange(0, &cstart, &cend, false);
       }
 
       ASSERT_EQ(NumTableFilesAtLevel(0), 0);
@@ -1320,7 +1320,7 @@ TEST_F(DBTest, ApproximateSizes_MixOfSmallAndLarge) {
 
       ASSERT_TRUE(Between(Size(Key(3), Key(5)), 110000, 111000));
 
-      dbfull()->TEST_CompactRange(0, nullptr, nullptr);
+      dbfull()->TEST_CompactRange(0, nullptr, nullptr, false);
     }
   } while (ChangeOptions());
 }
@@ -1397,11 +1397,11 @@ TEST_F(DBTest, HiddenValuesAreRemoved) {
     db_->ReleaseSnapshot(snapshot);
     ASSERT_EQ(AllEntriesFor("foo"), "[ tiny, " + big + " ]");
     Slice x("x");
-    dbfull()->TEST_CompactRange(0, nullptr, &x);
+    dbfull()->TEST_CompactRange(0, nullptr, &x, false);
     ASSERT_EQ(AllEntriesFor("foo"), "[ tiny ]");
     ASSERT_EQ(NumTableFilesAtLevel(0), 0);
     ASSERT_GE(NumTableFilesAtLevel(1), 1);
-    dbfull()->TEST_CompactRange(1, nullptr, &x);
+    dbfull()->TEST_CompactRange(1, nullptr, &x, false);
     ASSERT_EQ(AllEntriesFor("foo"), "[ tiny ]");
 
     ASSERT_TRUE(Between(Size("", "pastfoo"), 0, 1000));
@@ -1427,11 +1427,11 @@ TEST_F(DBTest, DeletionMarkers1) {
   ASSERT_LEVELDB_OK(dbfull()->TEST_CompactMemTable());  // Moves to level last-2
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2, DEL, v1 ]");
   Slice z("z");
-  dbfull()->TEST_CompactRange(last - 2, nullptr, &z);
+  dbfull()->TEST_CompactRange(last - 2, nullptr, &z, false);
   // DEL eliminated, but v1 remains because we aren't compacting that level
   // (DEL can be eliminated because v2 hides v1).
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2, v1 ]");
-  dbfull()->TEST_CompactRange(last - 1, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(last - 1, nullptr, nullptr, false);
   // Merging last-1 w/ last, so we are the base level for "foo", so
   // DEL is removed.  (as is v1).
   ASSERT_EQ(AllEntriesFor("foo"), "[ v2 ]");
@@ -1454,10 +1454,10 @@ TEST_F(DBTest, DeletionMarkers2) {
   ASSERT_EQ(AllEntriesFor("foo"), "[ DEL, v1 ]");
   ASSERT_LEVELDB_OK(dbfull()->TEST_CompactMemTable());  // Moves to level last-2
   ASSERT_EQ(AllEntriesFor("foo"), "[ DEL, v1 ]");
-  dbfull()->TEST_CompactRange(last - 2, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(last - 2, nullptr, nullptr, false);
   // DEL kept: "last" file overlaps
   ASSERT_EQ(AllEntriesFor("foo"), "[ DEL, v1 ]");
-  dbfull()->TEST_CompactRange(last - 1, nullptr, nullptr);
+  dbfull()->TEST_CompactRange(last - 1, nullptr, nullptr, false);
   // Merging last-1 w/ last, so we are the base level for "foo", so
   // DEL is removed.  (as is v1).
   ASSERT_EQ(AllEntriesFor("foo"), "[ ]");
@@ -1491,8 +1491,8 @@ TEST_F(DBTest, OverlapInLevel0) {
     ASSERT_EQ("2,1,1", FilesPerLevel());
 
     // Compact away the placeholder files we created initially
-    dbfull()->TEST_CompactRange(1, nullptr, nullptr);
-    dbfull()->TEST_CompactRange(2, nullptr, nullptr);
+    dbfull()->TEST_CompactRange(1, nullptr, nullptr, false);
+    dbfull()->TEST_CompactRange(2, nullptr, nullptr, false);
     ASSERT_EQ("2", FilesPerLevel());
 
     // Do a memtable compaction.  Before bug-fix, the compaction would
@@ -1787,7 +1787,7 @@ TEST_F(DBTest, NoSpace) {
   env_->no_space_.store(true, std::memory_order_release);
   for (int i = 0; i < 10; i++) {
     for (int level = 0; level < config::kNumLevels - 1; level++) {
-      dbfull()->TEST_CompactRange(level, nullptr, nullptr);
+      dbfull()->TEST_CompactRange(level, nullptr, nullptr, false);
     }
   }
   env_->no_space_.store(false, std::memory_order_release);
@@ -1876,7 +1876,7 @@ TEST_F(DBTest, ManifestWriteError) {
 
     // Merging compaction (will fail)
     error_type->store(true, std::memory_order_release);
-    dbfull()->TEST_CompactRange(last, nullptr, nullptr);  // Should fail
+    dbfull()->TEST_CompactRange(last, nullptr, nullptr, false);  // Should fail
     ASSERT_EQ("bar", Get("foo"));
 
     // Recovery: should not lose data

db/version_set.cc

@@ -1388,7 +1388,7 @@ void VersionSet::SetupOtherInputs(Compaction* c) {
   GetRange(c->inputs_[0], &smallest, &largest);
 
   // TTL: manual compaction for last level shouldn't have inputs[1]
-  if (level + 1 < config::kNumLevels) {
+  if (!c->is_last_level()) {
     current_->GetOverlappingInputs(level + 1, &smallest, &largest,
                                    &c->inputs_[1]);
     AddBoundaryInputs(icmp_, current_->files_[level + 1], &c->inputs_[1]);
@@ -1447,7 +1447,7 @@ void VersionSet::SetupOtherInputs(Compaction* c) {
 }
 
 Compaction* VersionSet::CompactRange(int level, const InternalKey* begin,
-                                     const InternalKey* end) {
+                                     const InternalKey* end, bool is_last_level) {
   std::vector<FileMetaData*> inputs;
   current_->GetOverlappingInputs(level, begin, end, &inputs);
   if (inputs.empty()) {
@@ -1471,7 +1471,7 @@ Compaction* VersionSet::CompactRange(int level, const InternalKey* begin,
     }
   }
 
-  Compaction* c = new Compaction(options_, level);
+  Compaction* c = new Compaction(options_, level, is_last_level);
   c->input_version_ = current_;
   c->input_version_->Ref();
   c->inputs_[0] = inputs;
@@ -1479,8 +1479,22 @@ Compaction* VersionSet::CompactRange(int level, const InternalKey* begin,
   return c;
 }
 
+Compaction::Compaction(const Options* options, int level, bool is_last_level)
+    : level_(level),
+      is_last_level_(is_last_level),
+      max_output_file_size_(MaxFileSizeForLevel(options, level)),
+      input_version_(nullptr),
+      grandparent_index_(0),
+      seen_key_(false),
+      overlapped_bytes_(0) {
+  for (int i = 0; i < config::kNumLevels; i++) {
+    level_ptrs_[i] = 0;
+  }
+}
+
 Compaction::Compaction(const Options* options, int level)
     : level_(level),
+      is_last_level_(false),
       max_output_file_size_(MaxFileSizeForLevel(options, level)),
       input_version_(nullptr),
       grandparent_index_(0),

db/version_set.h

@@ -238,7 +238,7 @@ class VersionSet {
   // level that overlaps the specified range.  Caller should delete
   // the result.
   Compaction* CompactRange(int level, const InternalKey* begin,
-                           const InternalKey* end);
+                           const InternalKey* end, bool is_last_level);
 
   // Return the maximum overlapping data (in bytes) at next level for any
   // file at a level >= 1.
@@ -324,6 +324,8 @@ class Compaction {
   // and "level+1" will be merged to produce a set of "level+1" files.
   int level() const { return level_; }
 
+  bool is_last_level() const {return is_last_level_; }
+
   // Return the object that holds the edits to the descriptor done
   // by this compaction.
   VersionEdit* edit() { return &edit_; }
@@ -362,8 +364,10 @@ class Compaction {
   friend class VersionSet;
 
   Compaction(const Options* options, int level);
+  Compaction(const Options* options, int level, bool is_last_level);
 
   int level_;
+  bool is_last_level_;  // TTL: info whether the last level to compact
   uint64_t max_output_file_size_;
   Version* input_version_;
   VersionEdit edit_;