Refactored to remove the ChecksLowerBound() and ChecksUpperBound() fu…

…nctions from the Iterator API as requested by @rambacher

db/arena_wrapped_db_iter.h

@@ -68,13 +68,7 @@ class ArenaWrappedDBIter : public Iterator {
   bool IsBlob() const { return db_iter_->IsBlob(); }
 
   Status GetProperty(std::string prop_name, std::string* prop) override;
-  bool ChecksLowerBound() const override {
-    return db_iter_->ChecksLowerBound();
-  }
   const Slice* lower_bound() const override { return db_iter_->lower_bound(); }
-  bool ChecksUpperBound() const override {
-    return db_iter_->ChecksUpperBound();
-  }
   const Slice* upper_bound() const override { return db_iter_->upper_bound(); }
 
   Status Refresh() override;

db/db_iter.h

@@ -198,9 +198,7 @@ class DBIter final : public Iterator {
   }
 
   Status GetProperty(std::string prop_name, std::string* prop) override;
-  bool ChecksLowerBound() const override { return true; }
   const Slice* lower_bound() const override { return iterate_lower_bound_; }
-  bool ChecksUpperBound() const override { return true; }
   const Slice* upper_bound() const override { return iterate_upper_bound_; }
 
   void Next() final override;

include/rocksdb/iterator.h

@@ -116,43 +116,23 @@ class Iterator : public Cleanable {
     return Slice();
   }
 
-  /**
-   * true if this iterator is already
-   * checking ReadOptions::iterate_lower_bound.
-   * false indicates that either the iterator does
-   * not check the lower bound, or that it does not
-   * report that it is performing the check.
-   */
-  virtual bool ChecksLowerBound() const = 0;
-
   /**
    * Returns the lower bound if this iterator has a
    * ReadOptions::iterate_lower_bound set, else
    * returns nullptr.
    *
-   * Just because a lower bound is present, it does
-   * not mean that it is checked. This can however
-   * be determined by calling Iterator::ChecksLowerBound().
+   * If a lower bound is present, it should be assumed
+   * that the iterator adheres to it.
    */
   virtual const Slice* lower_bound() const = 0;
 
-  /**
-   * true if this iterator is already
-   * checking ReadOptions::iterate_upper_bound.
-   * false indicates that either the iterator does
-   * not check the upper bound, or that it does not
-   * report that it is performing the check.
-   */
-  virtual bool ChecksUpperBound() const = 0;
-
   /**
    * Returns the upper bound if this iterator has a
    * ReadOptions::iterate_upper_bound set, else
    * returns nullptr.
    *
-   * Just because an upper bound is present, it does
-   * not mean that it is checked. This can however
-   * be determined by calling Iterator::ChecksUpperBound().
+   * If an upper bound is present, it should be assumed
+   * that the iterator adheres to it.
    */
   virtual const Slice* upper_bound() const = 0;
 };

table/iterator.cc

@@ -126,9 +126,7 @@ class EmptyIterator : public Iterator {
     return Slice();
   }
   Status status() const override { return status_; }
-  bool ChecksLowerBound() const override { return false; }
   const Slice* lower_bound() const override { return nullptr; }
-  bool ChecksUpperBound() const override { return false; }
   const Slice* upper_bound() const override { return nullptr; }
 
  private:

utilities/blob_db/blob_db_iterator.h

@@ -117,12 +117,8 @@ class BlobDBIterator : public Iterator {
 
   // Iterator::Refresh() not supported.
 
-  bool ChecksLowerBound() const override { return iter_->ChecksLowerBound(); }
-
   const Slice* lower_bound() const override { return iter_->lower_bound(); }
 
-  bool ChecksUpperBound() const override { return iter_->ChecksUpperBound(); }
-
   const Slice* upper_bound() const override { return iter_->upper_bound(); }
 
  private:

utilities/ttl/db_ttl_impl.h

@@ -146,12 +146,8 @@ class TtlIterator : public Iterator {
 
   Status status() const override { return iter_->status(); }
 
-  bool ChecksLowerBound() const override { return iter_->ChecksLowerBound(); }
-
   const Slice* lower_bound() const override { return iter_->lower_bound(); }
 
-  bool ChecksUpperBound() const override { return iter_->ChecksUpperBound(); }
-
   const Slice* upper_bound() const override { return iter_->upper_bound(); }
 
  private:

utilities/write_batch_with_index/write_batch_with_index_internal.cc

@@ -30,7 +30,47 @@ BaseDeltaIterator::BaseDeltaIterator(Iterator* base_iterator,
       base_iterator_(base_iterator),
       delta_iterator_(delta_iterator),
       comparator_(comparator),
-      read_options_(read_options) {}
+      read_options_(read_options) {
+
+  // We have to consider the upper_bound constraint of both
+  // the base_iterator and the read_options provdied
+  // to us. We use the tightest constraint, i.e. the
+  // lower of the two.
+  const Slice* base_iterator_upper_bound = base_iterator_->upper_bound();
+  const Slice* read_iterate_upper_bound = read_options == nullptr ? nullptr : read_options->iterate_upper_bound;
+  calc_bound(upper_bound_, upper_bound_equals_base_upper_bound_, base_iterator_upper_bound, read_iterate_upper_bound, [&comparator](const Slice* const base_upper, const Slice* const read_upper){
+    return comparator->Compare(*base_upper, *read_upper) <= 0;
+  });
+
+  // We have to consider the lower_bound constraint of both
+  // the base_iterator and the read_options provdied
+  // to us. We use the tightest constraint, i.e. the
+  // higher of the two.
+  const Slice* base_iterator_lower_bound = base_iterator_->lower_bound();
+  const Slice* read_iterate_lower_bound = read_options == nullptr ? nullptr : read_options->iterate_lower_bound;
+  calc_bound(lower_bound_, lower_bound_equals_base_lower_bound_, base_iterator_lower_bound, read_iterate_lower_bound, [&comparator](const Slice* const base_lower, const Slice* const read_lower){
+    return comparator->Compare(*base_lower, *read_lower) >= 0;
+  });
+}
+
+void BaseDeltaIterator::calc_bound(const Slice*& out_bound, bool& out_bound_equals_base_bound, const Slice* const base_iterator_bound, const Slice* const read_iterate_bound,
+    const std::function<bool(const Slice* const, const Slice* const)> use_base_bound) {
+  if (base_iterator_bound == nullptr) {
+    out_bound = read_iterate_bound;
+    out_bound_equals_base_bound = false;
+  } else if (read_iterate_bound == nullptr) {
+    out_bound = base_iterator_bound;
+    out_bound_equals_base_bound = true;
+  } else {
+    if (use_base_bound(base_iterator_bound, read_iterate_bound)) {
+      out_bound = base_iterator_bound;
+      out_bound_equals_base_bound = true;
+    } else {
+      out_bound = read_iterate_bound;
+      out_bound_equals_base_bound = false;
+    }
+  }
+}
 
 bool BaseDeltaIterator::Valid() const {
   return status_.ok() ? (current_at_base_ ? BaseValid() : DeltaValid()) : false;
@@ -45,13 +85,13 @@ void BaseDeltaIterator::SeekToFirst() {
 
 void BaseDeltaIterator::SeekToLast() {
   progress_ = Progress::SEEK_TO_LAST;
-  // is there an upper bound constraint on base_iterator_?
-  const Slice* base_upper_bound = base_iterator_upper_bound();
-  if (base_upper_bound != nullptr) {
-    // yes, and is base_iterator already constrained by an upper_bound?
-    if (!base_iterator_->ChecksUpperBound()) {
+
+  // is there an upper bound constraint?
+  if (upper_bound_ != nullptr) {
+    // yes, and is base_iterator already constrained by the same upper_bound?
+    if (!upper_bound_equals_base_upper_bound_) {
       // no, so we have to seek it to before base_upper_bound
-      base_iterator_->Seek(*(base_upper_bound));
+      base_iterator_->Seek(*upper_bound_);
       if (base_iterator_->Valid()) {
         base_iterator_->Prev();  // upper bound should be exclusive!
       } else {
@@ -60,7 +100,7 @@ void BaseDeltaIterator::SeekToLast() {
         base_iterator_->SeekToLast();
       }
     } else {
-      // yes, so the base_iterator will take care of base_upper_bound
+      // yes, so the base_iterator will take care of its own upper_bound
       base_iterator_->SeekToLast();
     }
   } else {
@@ -69,6 +109,7 @@ void BaseDeltaIterator::SeekToLast() {
   }
 
   // is there an upper bound constraint on delta_iterator_?
+  // TODO(AR) should we use the same upper_bound_ here?
   if (read_options_ != nullptr &&
       read_options_->iterate_upper_bound != nullptr) {
     // delta iterator does not itself support iterate_upper_bound,
@@ -219,16 +260,14 @@ Status BaseDeltaIterator::status() const {
   return delta_iterator_->status();
 }
 
-bool BaseDeltaIterator::ChecksLowerBound() const { return false; }
-
 const Slice* BaseDeltaIterator::lower_bound() const {
-  return base_iterator_lower_bound();
+  // NOTE: BaseDeltaIterator does not correctly check lower bound yet
+  // and so we can only return nullptr
+  return nullptr;
 }
 
-bool BaseDeltaIterator::ChecksUpperBound() const { return true; }
-
 const Slice* BaseDeltaIterator::upper_bound() const {
-  return base_iterator_upper_bound();
+  return upper_bound_;
 }
 
 void BaseDeltaIterator::Invalidate(Status s) { status_ = s; }
@@ -320,7 +359,7 @@ bool BaseDeltaIterator::BaseValid() const {
   // base_iterator if the base iterator has an
   // upper_bounds_check already
   return base_iterator_->Valid() &&
-         (base_iterator_->ChecksUpperBound() ? true : BaseIsWithinBounds());
+         (upper_bound_equals_base_upper_bound_ ? true : BaseIsWithinBounds());
 }
 
 bool BaseDeltaIterator::DeltaValid() const {
@@ -411,34 +450,16 @@ void BaseDeltaIterator::UpdateCurrent() {
 #endif  // __clang_analyzer__
 }
 
-inline const Slice* BaseDeltaIterator::base_iterator_upper_bound() const {
-  const Slice* upper = base_iterator_->upper_bound();
-  if (upper == nullptr && read_options_ != nullptr) {
-    return read_options_->iterate_upper_bound;
-  }
-  return upper;
-}
-
-inline const Slice* BaseDeltaIterator::base_iterator_lower_bound() const {
-  const Slice* lower = base_iterator_->lower_bound();
-  if (lower == nullptr && read_options_ != nullptr) {
-    return read_options_->iterate_lower_bound;
-  }
-  return lower;
-}
-
 bool BaseDeltaIterator::BaseIsWithinBounds() const {
   if (IsMovingBackward()) {
-    const Slice* lower = base_iterator_lower_bound();
-    if (lower != nullptr) {
-      return comparator_->Compare(base_iterator_->key(), *lower) >= 0;
+    if (lower_bound_ != nullptr) {
+      return comparator_->Compare(base_iterator_->key(), *lower_bound_) >= 0;
     }
   }
 
   if (IsMovingForward()) {
-    const Slice* upper = base_iterator_upper_bound();
-    if (upper != nullptr) {
-      return comparator_->Compare(base_iterator_->key(), *upper) < 0;
+    if (upper_bound_ != nullptr) {
+      return comparator_->Compare(base_iterator_->key(), *upper_bound_) < 0;
     }
   }
 

utilities/write_batch_with_index/write_batch_with_index_internal.h

@@ -6,6 +6,7 @@
 
 #ifndef ROCKSDB_LITE
 
+#include <functional>
 #include <limits>
 #include <string>
 #include <vector>
@@ -49,13 +50,13 @@ class BaseDeltaIterator : public Iterator {
   Slice key() const override;
   Slice value() const override;
   Status status() const override;
-  bool ChecksLowerBound() const override;
   const Slice* lower_bound() const override;
-  bool ChecksUpperBound() const override;
   const Slice* upper_bound() const override;
   void Invalidate(Status s);
 
  private:
+  static void calc_bound(const Slice*& out_bound, bool& out_bound_equals_base_bound, const Slice* const base_iterator_bound, const Slice* const read_iterate_bound,
+    const std::function<bool(const Slice* const, const Slice* const)> use_base_bound);
   void AssertInvariants();
   void Advance();
   void AdvanceDelta();
@@ -64,25 +65,6 @@ class BaseDeltaIterator : public Iterator {
   bool DeltaValid() const;
   void UpdateCurrent();
 
-  /**
-   * Returns the upper bound for the base iterator,
-   * or nullptr if there is no upper bound.
-   *
-   * The base iterator may have its own upper bound,
-   * if not not we use the upper bound from this
-   * iterator's ReadOptions (if present).
-   */
-  inline const Slice* base_iterator_upper_bound() const;
-
-  /**
-   * Returns the lower bound for the base iterator,
-   * or nullptr if there is no lower bound.
-   *
-   * The base iterator may have its own lower bound,
-   * if not not we use the lower bound from this
-   * iterator's ReadOptions (if present).
-   */
-  inline const Slice* base_iterator_lower_bound() const;
   bool BaseIsWithinBounds() const;
   bool DeltaIsWithinBounds() const;
 
@@ -124,6 +106,10 @@ class BaseDeltaIterator : public Iterator {
   std::unique_ptr<WBWIIterator> delta_iterator_;
   const Comparator* comparator_;  // not owned
   const ReadOptions* read_options_;  // not owned
+  const Slice* lower_bound_;  // not owned
+  bool lower_bound_equals_base_lower_bound_;
+  const Slice* upper_bound_;  // not owned
+  bool upper_bound_equals_base_upper_bound_;
 };
 
 // Key used by skip list, as the binary searchable index of WriteBatchWithIndex.

utilities/write_batch_with_index/write_batch_with_index_test.cc

@@ -561,28 +561,18 @@ class KVIter : public Iterator {
   Slice value() const override { return iter_->second; }
   Status status() const override { return Status::OK(); }
 
-  bool ChecksLowerBound() const override {
-    return read_options_ != nullptr &&
-           read_options_->iterate_lower_bound != nullptr;
-  }
-
   const Slice* lower_bound() const override {
-    if (ChecksLowerBound()) {
-      return read_options_->iterate_lower_bound;
+    if (read_options_ == nullptr) {
+      return nullptr;
     }
-    return nullptr;
-  }
-
-  bool ChecksUpperBound() const override {
-    return read_options_ != nullptr &&
-           read_options_->iterate_upper_bound != nullptr;
+    return read_options_->iterate_lower_bound;
   }
 
   const Slice* upper_bound() const override {
-    if (ChecksUpperBound()) {
-      return read_options_->iterate_upper_bound;
+    if (read_options_ == nullptr) {
+      return nullptr;
     }
-    return nullptr;
+    return read_options_->iterate_upper_bound;
   }
 
  private: