Universal Compaction should keep DeleteMarkers unless it is the earliest file.

Summary:
The pre-existing code was purging a DeleteMarker if thay key did not
exist in deeper levels.  But in the Universal Compaction Style, all
files are in Level0. For compaction runs that did not include the
earliest file, we were erroneously purging the DeleteMarkers.

The fix is to purge DeleteMarkers only if the compaction includes
the earlist file.

Test Plan: DBTest.Randomized triggers this code path.

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

db/db_impl.cc

@@ -1761,14 +1761,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
   }
 
   // Is this compaction producing files at the bottommost level?
-  bool bottommost_level = true;
-  for (int i = compact->compaction->level() + 2;
-       i < versions_->NumberLevels(); i++) {
-    if (versions_->NumLevelFiles(i) > 0) {
-      bottommost_level = false;
-      break;
-    }
-  }
+  bool bottommost_level = compact->compaction->BottomMostLevel();
 
   // Allocate the output file numbers before we release the lock
   AllocateCompactionOutputFileNumbers(compact);

db/db_test.cc

@@ -3539,9 +3539,7 @@ TEST(DBTest, Randomized) {
     }
     if (model_snap != nullptr) model.ReleaseSnapshot(model_snap);
     if (db_snap != nullptr) db_->ReleaseSnapshot(db_snap);
-  // TODO (xjin): remove kSkipUniversalCompaction after bug in
-  //              IsBaseLevelForKey() is fixed.
-  } while (ChangeOptions(kSkipDeletesFilterFirst | kSkipUniversalCompaction));
+  } while (ChangeOptions(kSkipDeletesFilterFirst));
 }
 
 TEST(DBTest, MultiGetSimple) {

db/version_set.cc

@@ -2266,6 +2266,13 @@ Compaction* VersionSet::PickCompactionUniversal(int level, double score) {
     newerfile = f;
   }
 
+  // Is the earliest file part of this compaction?
+  int last_index = file_by_time[file_by_time.size()-1];
+  FileMetaData* last_file = current_->files_[level][last_index];
+  if (c->inputs_[0][c->inputs_[0].size()-1] == last_file) {
+    c->bottommost_level_ = true;
+  }
+
   // update statistics
   if (options_->statistics != nullptr) {
     options_->statistics->measureTime(NUM_FILES_IN_SINGLE_COMPACTION,
@@ -2444,13 +2451,15 @@ Compaction* VersionSet::PickCompaction() {
     assert(!c->inputs_[0].empty());
   }
 
-
   // Setup "level+1" files (inputs_[1])
   SetupOtherInputs(c);
 
   // mark all the files that are being compacted
   c->MarkFilesBeingCompacted(true);
 
+  // Is this compaction creating a file at the bottommost level
+  c->SetupBottomMostLevel(false);
+
   // remember this currently undergoing compaction
   compactions_in_progress_[level].insert(c);
 
@@ -2624,6 +2633,13 @@ Compaction* VersionSet::CompactRange(
     const InternalKey* begin,
     const InternalKey* end) {
   std::vector<FileMetaData*> inputs;
+
+  // All files are 'overlapping' in universal style compaction.
+  // We have to compact the entire range in one shot.
+  if (options_->compaction_style == kCompactionStyleUniversal) {
+    begin = nullptr;
+    end = nullptr;
+  }
   current_->GetOverlappingInputs(level, begin, end, &inputs);
   if (inputs.empty()) {
     return nullptr;
@@ -2667,6 +2683,9 @@ Compaction* VersionSet::CompactRange(
   // upon other files because manual compactions are processed when
   // the system has a max of 1 background compaction thread.
   c->MarkFilesBeingCompacted(true);
+
+  // Is this compaction creating a file at the bottommost level
+  c->SetupBottomMostLevel(true);
   return c;
 }
 
@@ -2686,6 +2705,7 @@ Compaction::Compaction(int level, int out_level, uint64_t target_file_size,
       base_index_(-1),
       parent_index_(-1),
       score_(0),
+      bottommost_level_(false),
       level_ptrs_(std::vector<size_t>(number_levels)) {
   edit_ = new VersionEdit(number_levels_);
   for (int i = 0; i < number_levels_; i++) {
@@ -2718,6 +2738,10 @@ void Compaction::AddInputDeletions(VersionEdit* edit) {
 }
 
 bool Compaction::IsBaseLevelForKey(const Slice& user_key) {
+  if (input_version_->vset_->options_->compaction_style ==
+      kCompactionStyleUniversal) {
+    return bottommost_level_;
+  }
   // Maybe use binary search to find right entry instead of linear search?
   const Comparator* user_cmp = input_version_->vset_->icmp_.user_comparator();
   for (int lvl = level_ + 2; lvl < number_levels_; lvl++) {
@@ -2776,6 +2800,31 @@ void Compaction::MarkFilesBeingCompacted(bool value) {
   }
 }
 
+// Is this compaction producing files at the bottommost level?
+void Compaction::SetupBottomMostLevel(bool isManual) {
+  if (input_version_->vset_->options_->compaction_style  ==
+         kCompactionStyleUniversal) {
+    // If universal compaction style is used and manual
+    // compaction is occuring, then we are guaranteed that
+    // all files will be picked in a single compaction
+    // run. We can safely set bottommost_level_ = true.
+    // If it is not manual compaction, then bottommost_level_
+    // is already set when the Compaction was created.
+    if (isManual) {
+      bottommost_level_ = true;
+    }
+    return;
+  }
+  bottommost_level_ = true;
+  int num_levels = input_version_->vset_->NumberLevels();
+  for (int i = level() + 2; i < num_levels; i++) {
+    if (input_version_->vset_->NumLevelFiles(i) > 0) {
+      bottommost_level_ = false;
+      break;
+    }
+  }
+}
+
 void Compaction::ReleaseInputs() {
   if (input_version_ != nullptr) {
     input_version_->Unref();

db/version_set.h

@@ -557,6 +557,9 @@ class Compaction {
   // Return the score that was used to pick this compaction run.
   double score() const { return score_; }
 
+  // Is this compaction creating a file in the bottom most level?
+  bool BottomMostLevel() { return bottommost_level_; }
+
  private:
   friend class Version;
   friend class VersionSet;
@@ -589,7 +592,8 @@ class Compaction {
   int parent_index_; // index of some file with same range in files_[level_+1]
   double score_;     // score that was used to pick this compaction.
 
-  // State for implementing IsBaseLevelForKey
+  // Is this compaction creating a file in the bottom most level?
+  bool bottommost_level_;
 
   // level_ptrs_ holds indices into input_version_->levels_: our state
   // is that we are positioned at one of the file ranges for each
@@ -600,6 +604,9 @@ class Compaction {
   // mark (or clear) all files that are being compacted
   void MarkFilesBeingCompacted(bool);
 
+  // Initialize whether compaction producing files at the bottommost level
+  void SetupBottomMostLevel(bool isManual);
+
   // In case of compaction error, reset the nextIndex that is used
   // to pick up the next file to be compacted from files_by_size_
   void ResetNextCompactionIndex();