From 3754f2f4ff23ddec70f26e5002a3f80151924caf Mon Sep 17 00:00:00 2001
From: Dhruba Borthakur <dhruba@fb.com>
Date: Tue, 20 Nov 2012 09:08:11 -0800
Subject: [PATCH] A major bug that was not considering the compaction score of
 the n-1 level.

Summary:
The method Finalize() recomputes the compaction score of each
level and then sorts these score from largest to smallest. The
idea is that the level with the largest compaction score will
be a better candidate for compaction.  There are usually very
few levels, and a bubble sort code was used to sort these
compaction scores. There existed a bug in the sorting code that
skipped looking at the score for the n-1 level. This meant that
even if the compaction score of the n-1 level is large, it will
not be picked for compaction.

This patch fixes the bug and also introduces "asserts" in the
code to detect any possible inconsistencies caused by future bugs.

This bug existed in the very first code change that introduced
multi-threaded compaction to the leveldb code. That version of
code was committed on Oct 19th via
https://github.com/facebook/leveldb/commit/1ca0584345af85d2dccc434f451218119626d36e

Test Plan: make clean check OPT=-g

Reviewers: emayanke, sheki, MarkCallaghan

Reviewed By: sheki

CC: leveldb

Differential Revision: https://reviews.facebook.net/D6837
---
 db/version_set.cc | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/db/version_set.cc b/db/version_set.cc
index 9b9c768c9..47d82c7b0 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -1416,7 +1416,7 @@ void VersionSet::Finalize(Version* v) {
   // sort all the levels based on their score. Higher scores get listed
   // first. Use bubble sort because the number of entries are small.
   for(int i = 0; i <  NumberLevels()-2; i++) {
-    for (int j = i+1; j < NumberLevels()-2; j++) {
+    for (int j = i+1; j < NumberLevels()-1; j++) {
       if (v->compaction_score_[i] < v->compaction_score_[j]) {
         double score = v->compaction_score_[i];
         int level = v->compaction_level_[i];
@@ -1868,6 +1868,8 @@ Compaction* VersionSet::PickCompaction() {
   //
   // Find the compactions by size on all levels.
   for (int i = 0; i < NumberLevels()-1; i++) {
+    assert(i == 0 || current_->compaction_score_[i] <=
+                     current_->compaction_score_[i-1]);
     level = current_->compaction_level_[i];
     if ((current_->compaction_score_[i] >= 1)) {
       c = PickCompactionBySize(level);
@@ -2127,6 +2129,10 @@ bool Compaction::ShouldStopBefore(const Slice& internal_key) {
     if (seen_key_) {
       overlapped_bytes_ += grandparents_[grandparent_index_]->file_size;
     }
+    assert(grandparent_index_ + 1 >= grandparents_.size() ||
+           icmp->Compare(grandparents_[grandparent_index_]->largest.Encode(),
+                         grandparents_[grandparent_index_+1]->smallest.Encode())
+                         < 0);
     grandparent_index_++;
   }
   seen_key_ = true;