merger.cc raw

   1  // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   2  // Use of this source code is governed by a BSD-style license that can be
   3  // found in the LICENSE file. See the AUTHORS file for names of contributors.
   4  
   5  #include "table/merger.h"
   6  
   7  #include "leveldb/comparator.h"
   8  #include "leveldb/iterator.h"
   9  #include "table/iterator_wrapper.h"
  10  
  11  namespace leveldb {
  12  
  13  namespace {
  14  class MergingIterator : public Iterator {
  15   public:
  16    MergingIterator(const Comparator* comparator, Iterator** children, int n)
  17        : comparator_(comparator),
  18          children_(new IteratorWrapper[n]),
  19          n_(n),
  20          current_(nullptr),
  21          direction_(kForward) {
  22      for (int i = 0; i < n; i++) {
  23        children_[i].Set(children[i]);
  24      }
  25    }
  26  
  27    ~MergingIterator() override { delete[] children_; }
  28  
  29    bool Valid() const override { return (current_ != nullptr); }
  30  
  31    void SeekToFirst() override {
  32      for (int i = 0; i < n_; i++) {
  33        children_[i].SeekToFirst();
  34      }
  35      FindSmallest();
  36      direction_ = kForward;
  37    }
  38  
  39    void SeekToLast() override {
  40      for (int i = 0; i < n_; i++) {
  41        children_[i].SeekToLast();
  42      }
  43      FindLargest();
  44      direction_ = kReverse;
  45    }
  46  
  47    void Seek(const Slice& target) override {
  48      for (int i = 0; i < n_; i++) {
  49        children_[i].Seek(target);
  50      }
  51      FindSmallest();
  52      direction_ = kForward;
  53    }
  54  
  55    void Next() override {
  56      assert(Valid());
  57  
  58      // Ensure that all children are positioned after key().
  59      // If we are moving in the forward direction, it is already
  60      // true for all of the non-current_ children since current_ is
  61      // the smallest child and key() == current_->key().  Otherwise,
  62      // we explicitly position the non-current_ children.
  63      if (direction_ != kForward) {
  64        for (int i = 0; i < n_; i++) {
  65          IteratorWrapper* child = &children_[i];
  66          if (child != current_) {
  67            child->Seek(key());
  68            if (child->Valid() &&
  69                comparator_->Compare(key(), child->key()) == 0) {
  70              child->Next();
  71            }
  72          }
  73        }
  74        direction_ = kForward;
  75      }
  76  
  77      current_->Next();
  78      FindSmallest();
  79    }
  80  
  81    void Prev() override {
  82      assert(Valid());
  83  
  84      // Ensure that all children are positioned before key().
  85      // If we are moving in the reverse direction, it is already
  86      // true for all of the non-current_ children since current_ is
  87      // the largest child and key() == current_->key().  Otherwise,
  88      // we explicitly position the non-current_ children.
  89      if (direction_ != kReverse) {
  90        for (int i = 0; i < n_; i++) {
  91          IteratorWrapper* child = &children_[i];
  92          if (child != current_) {
  93            child->Seek(key());
  94            if (child->Valid()) {
  95              // Child is at first entry >= key().  Step back one to be < key()
  96              child->Prev();
  97            } else {
  98              // Child has no entries >= key().  Position at last entry.
  99              child->SeekToLast();
 100            }
 101          }
 102        }
 103        direction_ = kReverse;
 104      }
 105  
 106      current_->Prev();
 107      FindLargest();
 108    }
 109  
 110    Slice key() const override {
 111      assert(Valid());
 112      return current_->key();
 113    }
 114  
 115    Slice value() const override {
 116      assert(Valid());
 117      return current_->value();
 118    }
 119  
 120    Status status() const override {
 121      Status status;
 122      for (int i = 0; i < n_; i++) {
 123        status = children_[i].status();
 124        if (!status.ok()) {
 125          break;
 126        }
 127      }
 128      return status;
 129    }
 130  
 131   private:
 132    // Which direction is the iterator moving?
 133    enum Direction { kForward, kReverse };
 134  
 135    void FindSmallest();
 136    void FindLargest();
 137  
 138    // We might want to use a heap in case there are lots of children.
 139    // For now we use a simple array since we expect a very small number
 140    // of children in leveldb.
 141    const Comparator* comparator_;
 142    IteratorWrapper* children_;
 143    int n_;
 144    IteratorWrapper* current_;
 145    Direction direction_;
 146  };
 147  
 148  void MergingIterator::FindSmallest() {
 149    IteratorWrapper* smallest = nullptr;
 150    for (int i = 0; i < n_; i++) {
 151      IteratorWrapper* child = &children_[i];
 152      if (child->Valid()) {
 153        if (smallest == nullptr) {
 154          smallest = child;
 155        } else if (comparator_->Compare(child->key(), smallest->key()) < 0) {
 156          smallest = child;
 157        }
 158      }
 159    }
 160    current_ = smallest;
 161  }
 162  
 163  void MergingIterator::FindLargest() {
 164    IteratorWrapper* largest = nullptr;
 165    for (int i = n_ - 1; i >= 0; i--) {
 166      IteratorWrapper* child = &children_[i];
 167      if (child->Valid()) {
 168        if (largest == nullptr) {
 169          largest = child;
 170        } else if (comparator_->Compare(child->key(), largest->key()) > 0) {
 171          largest = child;
 172        }
 173      }
 174    }
 175    current_ = largest;
 176  }
 177  }  // namespace
 178  
 179  Iterator* NewMergingIterator(const Comparator* comparator, Iterator** children,
 180                               int n) {
 181    assert(n >= 0);
 182    if (n == 0) {
 183      return NewEmptyIterator();
 184    } else if (n == 1) {
 185      return children[0];
 186    } else {
 187      return new MergingIterator(comparator, children, n);
 188    }
 189  }
 190  
 191  }  // namespace leveldb
 192