From 28fe86c48a6c26b6926fa47a1377f348c0345072 Mon Sep 17 00:00:00 2001
From: Kosie van der Merwe <kosie.vandermerwe@gmail.com>
Date: Tue, 15 Jan 2013 12:43:09 -0800
Subject: [PATCH] Fixed bug with seek compactions on Level 0

Summary: Due to how the code handled compactions in Level 0 in `PickCompaction()` it could be the case that two compactions on level 0 ran that produced tables in level 1 that overlap. However, this case seems like it would only occur on a seek compaction which is unlikely on level 0. Furthermore, level 0 and level 1 had to have a certain arrangement of files.

Test Plan:
make check

Reviewers: dhruba, vamsi

Reviewed By: dhruba

CC: leveldb, sheki

Differential Revision: https://reviews.facebook.net/D7923
---
 db/version_set.cc | 22 +++++++++++-----------
 db/version_set.h  |  2 ++
 2 files changed, 13 insertions(+), 11 deletions(-)

diff --git a/db/version_set.cc b/db/version_set.cc
index eb45fdc95..c4abdeadc 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -1936,9 +1936,13 @@ Compaction* VersionSet::PickCompaction() {
   // Find compactions needed by seeks
   if (c == NULL && (current_->file_to_compact_ != NULL)) {
     level = current_->file_to_compact_level_;
-    c = new Compaction(level, MaxFileSizeForLevel(level),
-    MaxGrandParentOverlapBytes(level), NumberLevels(), true);
-    c->inputs_[0].push_back(current_->file_to_compact_);
+
+    // Only allow one level 0 compaction at a time.
+    if (level != 0 || compactions_in_progress_[0].empty()) {
+      c = new Compaction(level, MaxFileSizeForLevel(level),
+      MaxGrandParentOverlapBytes(level), NumberLevels(), true);
+      c->inputs_[0].push_back(current_->file_to_compact_);
+    }
   }
 
   if (c == NULL) {
@@ -1949,26 +1953,22 @@ Compaction* VersionSet::PickCompaction() {
   c->input_version_->Ref();
 
   // Files in level 0 may overlap each other, so pick up all overlapping ones
+  // Two level 0 compaction won't run at the same time, so don't need to worry
+  // about files on level 0 being compacted.
   if (level == 0) {
+    assert(compactions_in_progress_[0].empty());
     InternalKey smallest, largest;
     GetRange(c->inputs_[0], &smallest, &largest);
     // Note that the next call will discard the file we placed in
     // c->inputs_[0] earlier and replace it with an overlapping set
     // which will include the picked file.
     c->inputs_[0].clear();
-    std::vector<FileMetaData*> more;
-    current_->GetOverlappingInputs(0, &smallest, &largest, &more);
+    current_->GetOverlappingInputs(0, &smallest, &largest, &c->inputs_[0]);
     if (ParentRangeInCompaction(&smallest, &largest,
                                 level, &c->parent_index_)) {
       delete c;
       return NULL;
     }
-    for (unsigned int i = 0; i < more.size(); i++) {
-      FileMetaData* f = more[i];
-      if (!f->being_compacted) {
-        c->inputs_[0].push_back(f);
-      }
-    }
     assert(!c->inputs_[0].empty());
   }
 
diff --git a/db/version_set.h b/db/version_set.h
index 4ee006c0c..b6a620468 100644
--- a/db/version_set.h
+++ b/db/version_set.h
@@ -345,6 +345,8 @@ class VersionSet {
 
   // For the specfied level, pick a compaction.
   // Returns NULL if there is no compaction to be done.
+  // If level is 0 and there is already a compaction on that level, this
+  // function will return NULL.
   Compaction* PickCompactionBySize(int level, double score);
 
   // Free up the files that were participated in a compaction