From 17b5d5a064a56a502c0733e48136a0e88c6f17b6 Mon Sep 17 00:00:00 2001
From: luohaha <18810541851@163.com>
Date: Mon, 19 Oct 2020 00:23:39 +0800
Subject: [PATCH] use concatenating iterator for level-0 if there is no overlap

---
 db/version_set.cc | 22 ++++++++++++++++------
 db/version_set.h  |  8 +++++++-
 2 files changed, 23 insertions(+), 7 deletions(-)

diff --git a/db/version_set.cc b/db/version_set.cc
index a459587..d775879 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -228,10 +228,16 @@ Iterator* Version::NewConcatenatingIterator(const ReadOptions& options,
 
 void Version::AddIterators(const ReadOptions& options,
                            std::vector<Iterator*>* iters) {
-  // Merge all level zero files together since they may overlap
-  for (size_t i = 0; i < files_[0].size(); i++) {
-    iters->push_back(vset_->table_cache_->NewIterator(
-        options, files_[0][i]->number, files_[0][i]->file_size));
+  if (OverlapInLevel0()) {
+    // Merge all level zero files together since they overlap
+    for (size_t i = 0; i < files_[0].size(); i++) {
+      iters->push_back(vset_->table_cache_->NewIterator(
+          options, files_[0][i]->number, files_[0][i]->file_size));
+    }
+  } else {
+    if (!files_[0].empty()) {
+      iters->push_back(NewConcatenatingIterator(options, 0));
+    }
   }
 
   // For levels > 0, we can use a concatenating iterator that sequentially
@@ -725,6 +731,11 @@ class VersionSet::Builder {
                                     f->smallest) < 0);
       }
       f->refs++;
+      if (level == 0 && !files->empty() && !v->OverlapInLevel0()) {
+        if (vset_->icmp_.Compare(files->at(files->size() - 1)->largest, f->smallest) >= 0) {
+          v->SetOverlapInLevel0();
+        }
+      }
       files->push_back(f);
     }
   }
@@ -1218,13 +1229,12 @@ Iterator* VersionSet::MakeInputIterator(Compaction* c) {
 
   // Level-0 files have to be merged together.  For other levels,
   // we will make a concatenating iterator per level.
-  // TODO(opt): use concatenating iterator for level-0 if there is no overlap
   const int space = (c->level() == 0 ? c->inputs_[0].size() + 1 : 2);
   Iterator** list = new Iterator*[space];
   int num = 0;
   for (int which = 0; which < 2; which++) {
     if (!c->inputs_[which].empty()) {
-      if (c->level() + which == 0) {
+      if (c->level() + which == 0 && c->input_version_->OverlapInLevel0()) {
         const std::vector<FileMetaData*>& files = c->inputs_[which];
         for (size_t i = 0; i < files.size(); i++) {
           list[num++] = table_cache_->NewIterator(options, files[i]->number,
diff --git a/db/version_set.h b/db/version_set.h
index 69f3d70..b1e6e95 100644
--- a/db/version_set.h
+++ b/db/version_set.h
@@ -114,6 +114,10 @@ class Version {
   // Return a human readable string that describes this version's contents.
   std::string DebugString() const;
 
+  bool OverlapInLevel0() const { return overlap_in_level0_; }
+
+  void SetOverlapInLevel0() { overlap_in_level0_ = true; }
+
  private:
   friend class Compaction;
   friend class VersionSet;
@@ -128,7 +132,8 @@ class Version {
         file_to_compact_(nullptr),
         file_to_compact_level_(-1),
         compaction_score_(-1),
-        compaction_level_(-1) {}
+        compaction_level_(-1),
+        overlap_in_level0_(false) {}
 
   Version(const Version&) = delete;
   Version& operator=(const Version&) = delete;
@@ -162,6 +167,7 @@ class Version {
   // are initialized by Finalize().
   double compaction_score_;
   int compaction_level_;
+  bool overlap_in_level0_;
 };
 
 class VersionSet {