Added a parameter to limit the maximum space amplification for universal compaction.

Summary:
Added a new field called max_size_amplification_ratio in the
CompactionOptionsUniversal structure. This determines the maximum
percentage overhead of space amplification.

The size amplification is defined to be the ratio between the size of
the oldest file to the sum of the sizes of all other files. If the
size amplification exceeds the specified value, then min_merge_width
and max_merge_width are ignored and a full compaction of all files is done.
A value of 10 means that the size a database that stores 100 bytes
of user data could occupy 110 bytes of physical storage.

Test Plan: Unit test DBTest.UniversalCompactionSpaceAmplification added.

Reviewers: haobo, emayanke, xjin

Reviewed By: haobo

CC: leveldb

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

db/db_bench.cc

@@ -155,11 +155,17 @@ static leveldb::CompactionStyle FLAGS_compaction_style = leveldb::kCompactionSty
 
 // Percentage flexibility while comparing file size
 // (for universal compaction only).
-static int FLAGS_universal_size_ratio = 1;
+static int FLAGS_universal_size_ratio = 0;
 
 // The minimum number of files in a single compaction run
 // (for universal compaction only).
-static int FLAGS_compaction_universal_min_merge_width = 2;
+static int FLAGS_universal_min_merge_width = 0;
+
+// The max number of files to compact in universal style compaction
+static unsigned int FLAGS_universal_max_merge_width = 0;
+
+// The max size amplification for universal style compaction
+static unsigned int FLAGS_universal_max_size_amplification_percent = 0;
 
 // Number of bytes to use as a cache of uncompressed data.
 // Negative means use default settings.
@@ -1185,9 +1191,6 @@ class Benchmark {
       FLAGS_min_write_buffer_number_to_merge;
     options.max_background_compactions = FLAGS_max_background_compactions;
     options.compaction_style = FLAGS_compaction_style;
-    options.compaction_options_universal.size_ratio = FLAGS_universal_size_ratio;
-    options.compaction_options_universal.min_merge_width =
-      FLAGS_compaction_universal_min_merge_width;
     options.block_size = FLAGS_block_size;
     options.filter_policy = filter_policy_;
     options.prefix_extractor = FLAGS_use_prefix_blooms ? prefix_extractor_
@@ -1290,6 +1293,24 @@ class Benchmark {
       exit(1);
     }
 
+    // set universal style compaction configurations, if applicable
+    if (FLAGS_universal_size_ratio != 0) {
+      options.compaction_options_universal.size_ratio =
+        FLAGS_universal_size_ratio;
+    }
+    if (FLAGS_universal_min_merge_width != 0) {
+      options.compaction_options_universal.min_merge_width =
+        FLAGS_universal_min_merge_width;
+    }
+    if (FLAGS_universal_max_merge_width != 0) {
+      options.compaction_options_universal.max_merge_width =
+        FLAGS_universal_max_merge_width;
+    }
+    if (FLAGS_universal_max_size_amplification_percent != 0) {
+      options.compaction_options_universal.max_size_amplification_percent =
+        FLAGS_universal_max_size_amplification_percent;
+    }
+
     Status s;
     if(FLAGS_read_only) {
       s = DB::OpenForReadOnly(options, FLAGS_db, &db_);
@@ -2242,10 +2263,6 @@ int main(int argc, char** argv) {
   FLAGS_max_background_compactions =
     leveldb::Options().max_background_compactions;
   FLAGS_compaction_style = leveldb::Options().compaction_style;
-  FLAGS_universal_size_ratio =
-    leveldb::Options().compaction_options_universal.size_ratio;
-  FLAGS_compaction_universal_min_merge_width =
-    leveldb::Options().compaction_options_universal.min_merge_width;
   // Compression test code above refers to FLAGS_block_size
   FLAGS_block_size = leveldb::Options().block_size;
   FLAGS_use_os_buffer = leveldb::EnvOptions().use_os_buffer;
@@ -2315,11 +2332,6 @@ int main(int argc, char** argv) {
       FLAGS_max_background_compactions = n;
     } else if (sscanf(argv[i], "--compaction_style=%d%c", &n, &junk) == 1) {
       FLAGS_compaction_style = (leveldb::CompactionStyle)n;
-    } else if (sscanf(argv[i], "--universal_size_ratio=%d%c", &n, &junk) == 1) {
-      FLAGS_universal_size_ratio = n;
-    } else if (sscanf(argv[i], "--universal_min_merge_width=%d%c",
-               &n, &junk) == 1) {
-      FLAGS_compaction_universal_min_merge_width = n;
     } else if (sscanf(argv[i], "--cache_size=%ld%c", &l, &junk) == 1) {
       FLAGS_cache_size = l;
     } else if (sscanf(argv[i], "--block_size=%d%c", &n, &junk) == 1) {
@@ -2525,6 +2537,19 @@ int main(int argc, char** argv) {
     } else if (sscanf(argv[i], "--purge_log_after_memtable_flush=%d%c", &n, &junk)
                == 1 && (n == 0 || n ==1 )) {
       FLAGS_purge_log_after_memtable_flush = n;
+    } else if (sscanf(argv[i], "--universal_size_ratio=%d%c",
+                      &n, &junk) == 1) {
+      FLAGS_universal_size_ratio = n;
+    } else if (sscanf(argv[i], "--universal_min_merge_width=%d%c",
+                      &n, &junk) == 1) {
+      FLAGS_universal_min_merge_width = n;
+    } else if (sscanf(argv[i], "--universal_max_merge_width=%d%c",
+                      &n, &junk) == 1) {
+      FLAGS_universal_max_merge_width = n;
+    } else if (sscanf(argv[i],
+               "--universal_max_size_amplification_percent=%d%c",
+               &n, &junk) == 1) {
+      FLAGS_universal_max_size_amplification_percent = n;
     } else {
       fprintf(stderr, "Invalid flag '%s'\n", argv[i]);
       exit(1);

db/db_test.cc

@@ -1728,6 +1728,43 @@ TEST(DBTest, UniversalCompactionTrigger) {
   }
 }
 
+TEST(DBTest, UniversalCompactionSizeAmplification) {
+  Options options = CurrentOptions();
+  options.compaction_style = kCompactionStyleUniversal;
+  options.write_buffer_size = 100<<10; //100KB
+  options.level0_file_num_compaction_trigger = 2;
+
+  // Trigger compaction if size amplification exceeds 110%
+  options.compaction_options_universal.
+    max_size_amplification_percent = 110;
+  Reopen(&options);
+
+  Random rnd(301);
+  int key_idx = 0;
+
+  //   Generate two files in Level 0. Both files are approx the same size.
+  for (int num = 0;
+       num < options.level0_file_num_compaction_trigger;
+       num++) {
+    // Write 120KB (12 values, each 10K)
+    for (int i = 0; i < 12; i++) {
+      ASSERT_OK(Put(Key(key_idx), RandomString(&rnd, 10000)));
+      key_idx++;
+    }
+    dbfull()->TEST_WaitForCompactMemTable();
+    ASSERT_EQ(NumTableFilesAtLevel(0), num + 1);
+  }
+  ASSERT_EQ(NumTableFilesAtLevel(0), 2);
+
+  // Flush whatever is remaining in memtable. This is typically
+  // small, which should not trigger size ratio based compaction
+  // but will instead trigger size amplification.
+  dbfull()->Flush(FlushOptions());
+
+  // Verify that size amplification did occur
+  ASSERT_EQ(NumTableFilesAtLevel(0), 1);
+}
+
 TEST(DBTest, ConvertCompactionStyle) {
   Random rnd(301);
   int max_key_level_insert = 200;

db/version_set.cc

@@ -2159,16 +2159,214 @@ void VersionSet::SizeBeingCompacted(std::vector<uint64_t>& sizes) {
   }
 }
 
-Compaction* VersionSet::PickCompactionUniversal(int level, double score) {
+//
+// Look at overall size amplification. If size amplification
+// exceeeds the configured value, then do a compaction
+// of the candidate files all the way upto the earliest
+// base file (overrides configured values of file-size ratios,
+// min_merge_width and max_merge_width).
+//
+Compaction* VersionSet::PickCompactionUniversalSizeAmp(
+    int level, double score) {
   assert (level == 0);
 
-  // percentage flexibilty while comparing file sizes
-  uint64_t ratio = options_->compaction_options_universal.size_ratio;
+  // percentage flexibilty while reducing size amplification
+  uint64_t ratio = options_->compaction_options_universal.
+                     max_size_amplification_percent;
+
+  // The files are sorted from newest first to oldest last.
+  std::vector<int>& file_by_time = current_->files_by_size_[level];
+  assert(file_by_time.size() == current_->files_[level].size());
+
+  unsigned int candidate_count = 0;
+  uint64_t candidate_size = 0;
+  unsigned int start_index = 0;
+  FileMetaData* f = nullptr;
+
+  // Skip files that are already being compacted
+  for (unsigned int loop = 0; loop < file_by_time.size() - 1; loop++) {
+    int index = file_by_time[loop];
+    f = current_->files_[level][index];
+    if (!f->being_compacted) {
+      start_index = loop;         // Consider this as the first candidate.
+      break;
+    }
+    Log(options_->info_log, "Universal: skipping file %ld[%d] compacted %s",
+        f->number, loop, " cannot be a candidate to reduce size amp.\n");
+    f = nullptr;
+  }
+  if (f == nullptr) {
+    return nullptr;             // no candidate files
+  }
+
+  Log(options_->info_log, "Universal: First candidate file %ld[%d] %s",
+      f->number, start_index, " to reduce size amp.\n");
+
+  // keep adding up all the remaining files
+  for (unsigned int loop = start_index; loop < file_by_time.size() - 1;
+       loop++) {
+    int index = file_by_time[loop];
+    f = current_->files_[level][index];
+    if (f->being_compacted) {
+      Log(options_->info_log,
+          "Universal: Possible candidate file %ld[%d] %s.", f->number, loop,
+          " is already being compacted. No size amp reduction possible.\n");
+      return nullptr;
+    }
+    candidate_size += f->file_size;
+    candidate_count++;
+  }
+  if (candidate_count == 0) {
+    return nullptr;
+  }
+
+  // size of earliest file
+  int index = file_by_time[file_by_time.size() - 1];
+  uint64_t earliest_file_size = current_->files_[level][index]->file_size;
+
+  // size amplification = percentage of additional size
+  if (candidate_size * 100 < ratio * earliest_file_size) {
+    Log(options_->info_log,
+        "Universal: size amp not needed. newer-files-total-size %ld "
+        "earliest-file-size %ld",
+        candidate_size, earliest_file_size);
+    return nullptr;
+  } else {
+    Log(options_->info_log,
+        "Universal: size amp needed. newer-files-total-size %ld "
+        "earliest-file-size %ld",
+        candidate_size, earliest_file_size);
+  }
+  assert(start_index >= 0 && start_index < file_by_time.size() - 1);
+
+  // create a compaction request
+  Compaction* c = new Compaction(level, level, MaxFileSizeForLevel(level),
+                                 LLONG_MAX, NumberLevels());
+  c->score_ = score;
+  for (unsigned int loop = start_index; loop < file_by_time.size(); loop++) {
+    int index = file_by_time[loop];
+    f = current_->files_[level][index];
+    c->inputs_[0].push_back(f);
+    Log(options_->info_log,
+        "Universal: size amp picking file %ld[%d] with size %ld",
+        f->number, index, f->file_size);
+  }
+  return c;
+}
+
+//
+// Consider compaction files based on their size differences with
+// the next file in time order.
+//
+Compaction* VersionSet::PickCompactionUniversalReadAmp(
+    int level, double score, unsigned int ratio,
+    unsigned int max_number_of_files_to_compact) {
+
   unsigned int min_merge_width =
     options_->compaction_options_universal.min_merge_width;
   unsigned int max_merge_width =
     options_->compaction_options_universal.max_merge_width;
 
+  // The files are sorted from newest first to oldest last.
+  std::vector<int>& file_by_time = current_->files_by_size_[level];
+  FileMetaData* f = nullptr;
+  bool done = false;
+  int start_index = 0;
+  unsigned int candidate_count;
+  assert(file_by_time.size() == current_->files_[level].size());
+
+  unsigned int max_files_to_compact = std::min(max_merge_width,
+                                       max_number_of_files_to_compact);
+  min_merge_width = std::max(min_merge_width, 2U);
+
+  // Considers a candidate file only if it is smaller than the
+  // total size accumulated so far.
+  for (unsigned int loop = 0; loop < file_by_time.size(); loop++) {
+
+    candidate_count = 0;
+
+    // Skip files that are already being compacted
+    for (f = nullptr; loop < file_by_time.size(); loop++) {
+      int index = file_by_time[loop];
+      f = current_->files_[level][index];
+
+      if (!f->being_compacted) {
+        candidate_count = 1;
+        break;
+      }
+      Log(options_->info_log,
+          "Universal: file %ld[%d] being compacted, skipping",
+          f->number, loop);
+      f = nullptr;
+    }
+
+    // This file is not being compacted. Consider it as the
+    // first candidate to be compacted.
+    uint64_t candidate_size =  f != nullptr? f->file_size : 0;
+    if (f != nullptr) {
+      Log(options_->info_log, "Universal: Possible candidate file %ld[%d].",
+          f->number, loop);
+    }
+
+    // Check if the suceeding files need compaction.
+    for (unsigned int i = loop+1;
+         candidate_count < max_files_to_compact && i < file_by_time.size();
+         i++) {
+      int index = file_by_time[i];
+      FileMetaData* f = current_->files_[level][index];
+      if (f->being_compacted) {
+        break;
+      }
+      // pick files if the total candidate file size (increased by the
+      // specified ratio) is still larger than the next candidate file.
+      uint64_t sz = (candidate_size * (100L + ratio)) /100;
+      if (sz < f->file_size) {
+        break;
+      }
+      candidate_count++;
+      candidate_size += f->file_size;
+    }
+
+    // Found a series of consecutive files that need compaction.
+    if (candidate_count >= (unsigned int)min_merge_width) {
+      start_index = loop;
+      done = true;
+      break;
+    } else {
+      for (unsigned int i = loop;
+           i < loop + candidate_count && i < file_by_time.size(); i++) {
+       int index = file_by_time[i];
+       FileMetaData* f = current_->files_[level][index];
+       Log(options_->info_log,
+           "Universal: Skipping file %ld[%d] with size %ld %d\n",
+           f->number, i, f->file_size, f->being_compacted);
+      }
+    }
+  }
+  if (!done || candidate_count <= 1) {
+    return nullptr;
+  }
+  Compaction* c = new Compaction(level, level, MaxFileSizeForLevel(level),
+                                 LLONG_MAX, NumberLevels());
+  c->score_ = score;
+
+  for (unsigned int i = start_index; i < start_index + candidate_count; i++) {
+    int index = file_by_time[i];
+    FileMetaData* f = current_->files_[level][index];
+    c->inputs_[0].push_back(f);
+    Log(options_->info_log, "Universal: Picking file %ld[%d] with size %ld\n",
+        f->number, i, f->file_size);
+  }
+  return c;
+}
+
+//
+// Universal style of compaction. Pick files that are contiguous in
+// time-range to compact.
+//
+Compaction* VersionSet::PickCompactionUniversal(int level, double score) {
+  assert (level == 0);
+
   if ((current_->files_[level].size() <=
       (unsigned int)options_->level0_file_num_compaction_trigger)) {
     Log(options_->info_log, "Universal: nothing to do\n");
@@ -2179,127 +2377,29 @@ Compaction* VersionSet::PickCompactionUniversal(int level, double score) {
       current_->files_[level].size(),
       LevelFileSummary(&tmp, 0));
 
-  Compaction* c = nullptr;
-  c = new Compaction(level, level, MaxFileSizeForLevel(level),
-                     LLONG_MAX, NumberLevels());
-  c->score_ = score;
+  // Check for size amplification first.
+  Compaction* c = PickCompactionUniversalSizeAmp(level, score);
+  if (c == nullptr) {
 
-  // The files are sorted from newest first to oldest last.
-  std::vector<int>& file_by_time = current_->files_by_size_[level];
-  FileMetaData* f = nullptr;
-  bool done = false;
-  assert(file_by_time.size() == current_->files_[level].size());
+    // Size amplification is within limits. Try reducing read
+    // amplification while maintaining file size ratios.
+    unsigned int ratio = options_->compaction_options_universal.size_ratio;
+    c = PickCompactionUniversalReadAmp(level, score, ratio, UINT_MAX);
 
-  unsigned int max_files_to_compact = std::min(max_merge_width, UINT_MAX);
-
-  // Make two pass. The first pass considers a candidate file
-  // only if it is smaller than the total size accumulated so far.
-  // The second pass does not look at the slope of the
-  // file-size  curve to decide what to pick for compaction.
-  for (int iter = 0; !done && iter < 2; iter++) {
-
-    for (unsigned int loop = 0; loop < file_by_time.size(); ) {
-
-      // Skip files that are already being compacted
-      for (f = nullptr; loop < file_by_time.size(); loop++) {
-        int index = file_by_time[loop];
-        f = current_->files_[level][index];
-
-        if (!f->being_compacted) {
-          break;
-        }
-        Log(options_->info_log, "Universal: file %ld[%d] being compacted, skipping",
-            f->number, loop);
-        f = nullptr;
-      }
-
-      // This file is not being compacted. Consider it as the
-      // first candidate to be compacted.
-      unsigned int candidate_count = 1;
-      uint64_t candidate_size =  f != nullptr? f->file_size : 0;
-      if (f != nullptr) {
-        Log(options_->info_log, "Universal: Possible candidate file %ld[%d] %s.",
-            f->number, loop, iter == 0? "" : "forced ");
-      }
-
-      // Check if the suceeding files need compaction.
-      for (unsigned int i = loop+1;
-           candidate_count < max_files_to_compact && i < file_by_time.size();
-           i++) {
-        int index = file_by_time[i];
-        FileMetaData* f = current_->files_[level][index];
-        if (f->being_compacted) {
-          break;
-        }
-        // If this is the first iteration, then we pick files if the
-        // total candidate file size (increased by the specified ratio)
-        // is still larger than the next candidate file.
-        if (iter == 0) {
-          uint64_t sz = (candidate_size * (100 + ratio)) /100;
-          if (sz < f->file_size) {
-            break;
-          }
-        }
-        candidate_count++;
-        candidate_size += f->file_size;
-      }
-
-      // Found a series of consecutive files that need compaction.
-      if (candidate_count >= (unsigned int)min_merge_width) {
-        for (unsigned int i = loop; i < loop + candidate_count; i++) {
-          int index = file_by_time[i];
-          FileMetaData* f = current_->files_[level][index];
-          c->inputs_[0].push_back(f);
-          Log(options_->info_log, "Universal: Picking file %ld[%d] with size %ld %s",
-              f->number, i, f->file_size,
-              (iter == 0 ? "" : "forced"));
-        }
-        done = true;
-        break;
-      } else {
-        for (unsigned int i = loop;
-             i < loop + candidate_count && i < file_by_time.size(); i++) {
-         int index = file_by_time[i];
-         FileMetaData* f = current_->files_[level][index];
-         Log(options_->info_log, "Universal: Skipping file %ld[%d] with size %ld %d %s",
-             f->number, i, f->file_size, f->being_compacted,
-              (iter == 0 ? "" : "forced"));
-        }
-      }
-      loop += candidate_count;
-    }
-    assert(done || c->inputs_[0].size() == 0);
-
-    // If we are unable to find a normal compaction run and we are still
-    // above the compaction threshold, iterate again to pick compaction
-    // candidates, this time without considering their size differences.
-    if (!done) {
-      int files_not_in_compaction = 0;
-      for (unsigned int i = 0; i < current_->files_[level].size(); i++) {
-        f = current_->files_[level][i];
-        if (!f->being_compacted) {
-          files_not_in_compaction++;
-        }
-      }
-      int expected_num_files = files_not_in_compaction +
-                compactions_in_progress_[level].size();
-      if (expected_num_files <=
-          options_->level0_file_num_compaction_trigger + 1) {
-        done = true;     // nothing more to do
-      } else {
-        max_files_to_compact = std::min((int)max_merge_width,
-          expected_num_files - options_->level0_file_num_compaction_trigger);
-        Log(options_->info_log, "Universal: second loop with maxfiles %d",
-            max_files_to_compact);
-      }
+    // Size amplification and file size ratios are within configured limits.
+    // If max read amplification is exceeding configured limits, then force
+    // compaction without looking at filesize ratios and try to reduce
+    // the number of files to fewer than level0_file_num_compaction_trigger.
+    if (c == nullptr) {
+      unsigned int num_files = current_->files_[level].size() -
+                               options_->level0_file_num_compaction_trigger;
+      c = PickCompactionUniversalReadAmp(level, score, UINT_MAX, num_files);
     }
   }
-  if (c->inputs_[0].size() <= 1) {
-    Log(options_->info_log, "Universal: only %ld files, nothing to do.\n",
-        c->inputs_[0].size());
-    delete c;
+  if (c == nullptr) {
     return nullptr;
   }
+  assert(c->inputs_[0].size() > 1);
 
   // validate that all the chosen files are non overlapping in time
   FileMetaData* newerfile __attribute__((unused)) = nullptr;
@@ -2311,6 +2411,9 @@ Compaction* VersionSet::PickCompactionUniversal(int level, double score) {
     newerfile = f;
   }
 
+  // The files are sorted from newest first to oldest last.
+  std::vector<int>& file_by_time = current_->files_by_size_[level];
+
   // 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];

db/version_set.h

@@ -383,6 +383,13 @@ class VersionSet {
   // Pick files to compact in Universal mode
   Compaction* PickCompactionUniversal(int level, double score);
 
+  // Pick Universal compaction to limit read amplification
+  Compaction* PickCompactionUniversalReadAmp(int level, double score,
+                unsigned int ratio, unsigned int num_files);
+
+  // Pick Universal compaction to limit space amplification.
+  Compaction* PickCompactionUniversalSizeAmp(int level, double score);
+
   // Free up the files that were participated in a compaction
   void ReleaseCompactionFiles(Compaction* c, Status status);