Add MaxInputLevel() to CompactionPicker

Summary:
Having if-then branch for different compaction strategies is considered
hacky and make CompactionPicker less pluggable.  This diff removes two
of such if-then branches in version_set.cc by adding MaxInputLevel() to
CompactionPicker.

    // Given the current number of levels, returns the lowest allowed level
    // for compaction input.
    virtual int MaxInputLevel(int current_num_levels) const;

Test Plan:
make db_test
export ROCKSDB_TESTS=Compaction
./db_test

Reviewers: igor, sdong, ljin

Reviewed By: ljin

Subscribers: leveldb

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

db/compaction_picker.h

@@ -52,6 +52,10 @@ class CompactionPicker {
                                    const InternalKey* end,
                                    InternalKey** compaction_end);
 
+  // Given the current number of levels, returns the lowest allowed level
+  // for compaction input.
+  virtual int MaxInputLevel(int current_num_levels) const = 0;
+
   // Free up the files that participated in a compaction
   void ReleaseCompactionFiles(Compaction* c, Status status);
 
@@ -135,6 +139,11 @@ class UniversalCompactionPicker : public CompactionPicker {
   virtual Compaction* PickCompaction(Version* version,
                                      LogBuffer* log_buffer) override;
 
+  // The maxinum allowed input level.  Always return 0.
+  virtual int MaxInputLevel(int current_num_levels) const override {
+    return 0;
+  }
+
  private:
   // Pick Universal compaction to limit read amplification
   Compaction* PickCompactionUniversalReadAmp(Version* version, double score,
@@ -159,6 +168,12 @@ class LevelCompactionPicker : public CompactionPicker {
   virtual Compaction* PickCompaction(Version* version,
                                      LogBuffer* log_buffer) override;
 
+  // Returns current_num_levels - 2, meaning the last level cannot be
+  // compaction input level.
+  virtual int MaxInputLevel(int current_num_levels) const override {
+    return current_num_levels - 2;
+  }
+
  private:
   // For the specfied level, pick a compaction.
   // Returns nullptr if there is no compaction to be done.
@@ -180,6 +195,11 @@ class FIFOCompactionPicker : public CompactionPicker {
                                    int output_level, const InternalKey* begin,
                                    const InternalKey* end,
                                    InternalKey** compaction_end) override;
+
+  // The maxinum allowed input level.  Always return 0.
+  virtual int MaxInputLevel(int current_num_levels) const override {
+    return 0;
+  }
 };
 
 }  // namespace rocksdb

db/version_set.cc

@@ -929,13 +929,10 @@ void Version::ComputeCompactionScore(
   double max_score = 0;
   int max_score_level = 0;
 
-  int num_levels_to_check =
-      (cfd_->options()->compaction_style != kCompactionStyleUniversal &&
-       cfd_->options()->compaction_style != kCompactionStyleFIFO)
-          ? NumberLevels() - 1
-          : 1;
+  int max_input_level =
+      cfd_->compaction_picker()->MaxInputLevel(NumberLevels());
 
-  for (int level = 0; level < num_levels_to_check; level++) {
+  for (int level = 0; level <= max_input_level; level++) {
     double score;
     if (level == 0) {
       // We treat level-0 specially by bounding the number of files
@@ -1084,12 +1081,10 @@ bool Version::NeedsCompaction() const {
   // ending up with nothing to do. We can improve it later.
   // TODO(sdong): improve this function to be accurate for universal
   //              compactions.
-  int num_levels_to_check =
-      (cfd_->options()->compaction_style != kCompactionStyleUniversal &&
-       cfd_->options()->compaction_style != kCompactionStyleFIFO)
-          ? NumberLevels() - 1
-          : 1;
-  for (int i = 0; i < num_levels_to_check; i++) {
+  int max_input_level =
+      cfd_->compaction_picker()->MaxInputLevel(NumberLevels());
+
+  for (int i = 0; i <= max_input_level; i++) {
     if (compaction_score_[i] >= 1) {
       return true;
     }