From d7d43ae21a4c4a07e7c08b01d07ace03b302d45d Mon Sep 17 00:00:00 2001
From: Dhruba Borthakur <dhruba@fb.com>
Date: Sun, 30 Dec 2012 22:18:52 -0800
Subject: [PATCH] ExtendOverlappingInputs too slow for large databases.

Summary:
There was a bug in the ExtendOverlappingInputs method so that
the terminating condition for the backward search was incorrect.

Test Plan: make clean check

Reviewers: sheki, emayanke, MarkCallaghan

Reviewed By: MarkCallaghan

CC: leveldb

Differential Revision: https://reviews.facebook.net/D7725
---
 db/version_set.cc | 20 ++++++++++++++++----
 1 file changed, 16 insertions(+), 4 deletions(-)

diff --git a/db/version_set.cc b/db/version_set.cc
index 03bd2a757b..4493abc527 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -589,27 +589,39 @@ void Version::ExtendOverlappingInputs(
     }
   }
 #endif
+  int startIndex = midIndex + 1;
+  int endIndex = midIndex;
+  int count = 0;
 
   // check backwards from 'mid' to lower indices
-  for (size_t i = midIndex; i < files_[level].size(); i--) {
+  for (int i = midIndex; i >= 0 ; i--) {
     FileMetaData* f = files_[level][i];
     const Slice file_limit = f->largest.user_key();
     if (user_cmp->Compare(file_limit, user_begin) >= 0) {
-      inputs->insert(inputs->begin(), f); // insert into beginning of vector
+      startIndex = i;
+      assert((count++, true));
     } else {
       break;
     }
   }
   // check forward from 'mid+1' to higher indices
-  for (size_t i = midIndex+1; i < files_[level].size(); i++) {
+  for (unsigned int i = midIndex+1; i < files_[level].size(); i++) {
     FileMetaData* f = files_[level][i];
     const Slice file_start = f->smallest.user_key();
     if (user_cmp->Compare(file_start, user_end) <= 0) {
-      inputs->push_back(f); // insert into end of vector
+      assert((count++, true));
+      endIndex = i;
     } else {
       break;
     }
   }
+  assert(count == endIndex - startIndex + 1);
+
+  // insert overlapping files into vector
+  for (int i = startIndex; i <= endIndex; i++) {
+    FileMetaData* f = files_[level][i];
+    inputs->push_back(f); 
+  }
 }
 
 std::string Version::DebugString(bool hex) const {