handle last level

db/db_impl.cc

@@ -602,12 +602,7 @@ void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
 void DBImpl::TEST_CompactRange(int level, const Slice* begin,
                                const Slice* end) {
   assert(level >= 0);
-  assert(level <= config::kNumLevels);
-
-  if (level == config::kNumLevels) {
-    // TTL: TODO: update files in the last level
-    return;
-  }
+  assert(level < config::kNumLevels);
 
   InternalKey begin_storage, end_storage;
 
@@ -896,8 +891,14 @@ 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];
-    compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size,
-                                         out.smallest, out.largest);
+    if (level < config::kNumLevels - 1) {
+      compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size,
+                                           out.smallest, out.largest);
+    } else {
+      // TTL: outputs of last level compaction should be writen to last level itself
+      compact->compaction->edit()->AddFile(level, out.number, out.file_size,
+                                           out.smallest, out.largest);
+    }
   }
   return versions_->LogAndApply(compact->compaction->edit(), &mutex_);
 }
@@ -1058,7 +1059,12 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
   }
 
   mutex_.Lock();
-  stats_[compact->compaction->level() + 1].Add(stats);
+  if (compact->compaction->level() + 1 < config::kNumLevels) {
+    stats_[compact->compaction->level() + 1].Add(stats);
+  } else {
+    // TTL: compaction for last level
+    stats_[compact->compaction->level()].Add(stats);
+  }
 
   if (status.ok()) {
     status = InstallCompactionResults(compact);

db/version_set.cc

@@ -1389,9 +1389,12 @@ void VersionSet::SetupOtherInputs(Compaction* c) {
   AddBoundaryInputs(icmp_, current_->files_[level], &c->inputs_[0]);
   GetRange(c->inputs_[0], &smallest, &largest);
 
-  current_->GetOverlappingInputs(level + 1, &smallest, &largest,
-                                 &c->inputs_[1]);
-  AddBoundaryInputs(icmp_, current_->files_[level + 1], &c->inputs_[1]);
+  // TTL: manual compaction for last level shouldn't have inputs[1]
+  if (level + 1 < config::kNumLevels) {
+    current_->GetOverlappingInputs(level + 1, &smallest, &largest,
+                                   &c->inputs_[1]);
+    AddBoundaryInputs(icmp_, current_->files_[level + 1], &c->inputs_[1]);
+  }
 
   // Get entire range covered by compaction
   InternalKey all_start, all_limit;

test/ttl_test.cc

@@ -92,11 +92,9 @@ TEST(TestTTL, CompactionTTL) {
     uint64_t sizes[1];
     db->GetApproximateSizes(ranges, 1, sizes);
     ASSERT_GT(sizes[0], 0);
-
     Env::Default()->SleepForMicroseconds(ttl * 1000000);
 
     db->CompactRange(nullptr, nullptr);
-
     leveldb::Range ranges1[1];
     ranges1[0] = leveldb::Range("-", "A");
     uint64_t sizes1[1];
@@ -106,6 +104,44 @@ TEST(TestTTL, CompactionTTL) {
     delete db;
 }
 
+// Time-consuming when kNumLevels is 7, so run this test when kNumLevels is set to 3,
+// the codes on branch light_ver
+// In this case level 2 is the last level with files
+TEST(TestTTL, LastLevelCompaction) {
+  DB *db;
+
+  if(OpenDB("testdb", &db).ok() == false) {
+    std::cerr << "open db failed" << std::endl;
+    abort();
+  }
+
+  uint64_t ttl = 20;
+  InsertData(db, ttl);
+
+  leveldb::Range ranges[1];
+  ranges[0] = leveldb::Range("-", "A");
+  uint64_t sizes[1];
+  db->GetApproximateSizes(ranges, 1, sizes);
+  ASSERT_GT(sizes[0], 0);
+
+  std::string last_level_file_num;
+  std::string last_level = "2";
+  db->GetProperty("leveldb.num-files-at-level" + last_level, &last_level_file_num);
+  std::cout << "File nums in last level: " << last_level_file_num << std::endl;
+  ASSERT_GT(std::atoi(last_level_file_num.c_str()), 0);
+
+  Env::Default()->SleepForMicroseconds(ttl * 1000000);
+
+  db->CompactRange(nullptr, nullptr);
+  leveldb::Range ranges1[1];
+  ranges1[0] = leveldb::Range("-", "A");
+  uint64_t sizes1[1];
+  db->GetApproximateSizes(ranges1, 1, sizes1);
+  ASSERT_EQ(sizes1[0], 0);
+
+  delete db;
+}
+
 
 int main(int argc, char** argv) {
   // All tests currently run with the same read-only file limits.