From 2a3969990082ee6b2dd3a80ae8c921e1e321bbbe Mon Sep 17 00:00:00 2001
From: Dhruba Borthakur <dhruba@fb.com>
Date: Mon, 26 Nov 2012 01:49:50 -0800
Subject: [PATCH] Assertion failure while running with unit tests with OPT=-g

Summary:
When we expand the range of keys for a level 0 compaction, we
need to invoke ParentFilesInCompaction() only once for the
entire range of keys that is being compacted. We were invoking
it for each file that was being compacted, but this triggers
an assertion because each file's range were contiguous but
non-overlapping.

I renamed ParentFilesInCompaction to ParentRangeInCompaction
to adequately represent that it is the range-of-keys and
not individual files that we compact in a single compaction run.

Here is the assertion that is fixed by this patch.
db_test: db/version_set.cc:585: void leveldb::Version::ExtendOverlappingInputs(int, const leveldb::Slice&, const leveldb::Slice&, std::vector<leveldb::FileMetaData*, std::allocator<leveldb::FileMetaData*> >*, int): Assertion `user_cmp->Compare(flimit, user_begin) >= 0' failed.

Test Plan: make clean check OPT=-g

Reviewers: sheki

Reviewed By: sheki

CC: MarkCallaghan, emayanke, leveldb

Differential Revision: https://reviews.facebook.net/D6963
---
 db/version_set.cc | 23 ++++++++++++++++-------
 db/version_set.h  |  3 ++-
 2 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/db/version_set.cc b/db/version_set.cc
index dd33fe21b..1bd6dab70 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -474,6 +474,9 @@ void Version::GetOverlappingInputs(
   if (end != NULL) {
     user_end = end->user_key();
   }
+  if (file_index) {
+    *file_index = -1;
+  }
   const Comparator* user_cmp = vset_->icmp_.user_comparator();
   if (begin != NULL && end != NULL && level > 0) {
     GetOverlappingInputsBinarySearch(level, user_begin, user_end, inputs,
@@ -503,7 +506,7 @@ void Version::GetOverlappingInputs(
           i = 0;
         }
       } else if (file_index) {
-        *file_index = i;
+        *file_index = i-1;
       }
     }
   }
@@ -1835,7 +1838,8 @@ Compaction* VersionSet::PickCompactionBySize(int level) {
     // Do not pick this file if its parents at level+1 are being compacted.
     // Maybe we can avoid redoing this work in SetupOtherInputs
     int parent_index = -1;
-    if (ParentFilesInCompaction(f, level, &parent_index)) {
+    if (ParentRangeInCompaction(&f->smallest, &f->largest, level, 
+                                &parent_index)) {
       continue;
     }
     c->inputs_[0].push_back(f);
@@ -1904,10 +1908,14 @@ Compaction* VersionSet::PickCompaction() {
     c->inputs_[0].clear();
     std::vector<FileMetaData*> more;
     current_->GetOverlappingInputs(0, &smallest, &largest, &more);
+    if (ParentRangeInCompaction(&smallest, &largest,
+                                level, &c->parent_index_)) {
+      delete c;
+      return NULL;
+    }
     for (unsigned int i = 0; i < more.size(); i++) {
       FileMetaData* f = more[i];
-      if (!f->being_compacted &&
-          !ParentFilesInCompaction(f, level, &c->parent_index_)) {
+      if (!f->being_compacted) {
         c->inputs_[0].push_back(f);
       }
     }
@@ -1926,10 +1934,11 @@ Compaction* VersionSet::PickCompaction() {
 }
 
 // Returns true if any one of the parent files are being compacted
-bool VersionSet::ParentFilesInCompaction(FileMetaData* f, int level,
-    int* parent_index) {
+bool VersionSet::ParentRangeInCompaction(const InternalKey* smallest,
+  const InternalKey* largest, int level, int* parent_index) {
   std::vector<FileMetaData*> inputs;
-  current_->GetOverlappingInputs(level+1, &f->smallest, &f->largest,
+  
+  current_->GetOverlappingInputs(level+1, smallest, largest,
                                  &inputs, *parent_index, parent_index);
   return FilesInCompaction(inputs);
 }
diff --git a/db/version_set.h b/db/version_set.h
index 1bb133477..70a4cb699 100644
--- a/db/version_set.h
+++ b/db/version_set.h
@@ -448,7 +448,8 @@ class VersionSet {
   uint64_t SizeBeingCompacted(int level);
 
   // Returns true if any one of the parent files are being compacted
-  bool ParentFilesInCompaction(FileMetaData* f, int level, int* index);
+  bool ParentRangeInCompaction(const InternalKey* smallest, 
+    const InternalKey* largest, int level, int* index);
 
   // Returns true if any one of the specified files are being compacted
   bool FilesInCompaction(std::vector<FileMetaData*>& files);
-- 
GitLab