make forward/reverse iteration a template parameter (#9726)

arangod/RocksDBEngine/RocksDBIterators.cpp

@@ -295,9 +295,8 @@ bool RocksDBAnyIndexIterator::outOfRange() const {
 }
 
 RocksDBGenericIterator::RocksDBGenericIterator(rocksdb::ReadOptions& options,
-                                               RocksDBKeyBounds const& bounds, bool reverse)
-    : _reverse(reverse),
-      _bounds(bounds),
+                                               RocksDBKeyBounds const& bounds)
+    : _bounds(bounds),
       _options(options),
       _iterator(arangodb::rocksutils::globalRocksDB()->NewIterator(_options,
                                                                    _bounds.columnFamily())),
@@ -310,19 +309,11 @@ bool RocksDBGenericIterator::hasMore() const {
 }
 
 bool RocksDBGenericIterator::outOfRange() const {
-  if (_reverse) {
-    return _cmp->Compare(_iterator->key(), _bounds.start()) < 0;
-  } else {
-    return _cmp->Compare(_iterator->key(), _bounds.end()) > 0;
-  }
+  return _cmp->Compare(_iterator->key(), _bounds.end()) > 0;
 }
 
 bool RocksDBGenericIterator::reset() {
-  if (_reverse) {
-    return seek(_bounds.end());
-  } else {
-    return seek(_bounds.start());
-  }
+  return seek(_bounds.start());
 }
 
 bool RocksDBGenericIterator::skip(uint64_t count, uint64_t& skipped) {
@@ -340,11 +331,7 @@ bool RocksDBGenericIterator::skip(uint64_t count, uint64_t& skipped) {
 }
 
 bool RocksDBGenericIterator::seek(rocksdb::Slice const& key) {
-  if (_reverse) {
-    _iterator->SeekForPrev(key);
-  } else {
-    _iterator->Seek(key);
-  }
+  _iterator->Seek(key);
   return hasMore();
 }
 
@@ -371,11 +358,7 @@ bool RocksDBGenericIterator::next(GenericCallback const& cb, size_t limit) {
       return false;
     }
     --limit;
-    if (_reverse) {
-      _iterator->Prev();
-    } else {
-      _iterator->Next();
-    }
+    _iterator->Next();
     
     // validate that Iterator is in a good shape and hasn't failed
     arangodb::rocksutils::checkIteratorStatus(_iterator.get());

arangod/RocksDBEngine/RocksDBIterators.h

@@ -98,7 +98,7 @@ typedef std::function<bool(rocksdb::Slice const& key, rocksdb::Slice const& valu
 class RocksDBGenericIterator {
  public:
   RocksDBGenericIterator(rocksdb::ReadOptions& options,
-                         RocksDBKeyBounds const& bounds, bool reverse = false);
+                         RocksDBKeyBounds const& bounds);
   RocksDBGenericIterator(RocksDBGenericIterator&&) = default;
 
   ~RocksDBGenericIterator() {}
@@ -120,7 +120,6 @@ class RocksDBGenericIterator {
   bool outOfRange() const;
 
  private:
-  bool _reverse;
   RocksDBKeyBounds const _bounds;
   rocksdb::ReadOptions const _options;
   std::unique_ptr<rocksdb::Iterator> _iterator;

arangod/RocksDBEngine/RocksDBPrimaryIndex.cpp

@@ -275,6 +275,7 @@ class RocksDBPrimaryIndexInIterator final : public IndexIterator {
   bool const _allowCoveringIndexOptimization;
 };
 
+template<bool reverse>
 class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
  private:
   friend class RocksDBVPackIndex;
@@ -282,12 +283,11 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
  public:
   RocksDBPrimaryIndexRangeIterator(LogicalCollection* collection, transaction::Methods* trx,
                                    arangodb::RocksDBPrimaryIndex const* index,
-                                   bool reverse, RocksDBKeyBounds&& bounds,
+                                   RocksDBKeyBounds&& bounds,
                                    bool allowCoveringIndexOptimization)
       : IndexIterator(collection, trx),
         _index(index),
         _cmp(index->comparator()),
-        _reverse(reverse),
         _allowCoveringIndexOptimization(allowCoveringIndexOptimization),
         _bounds(std::move(bounds)) {
     TRI_ASSERT(index->columnFamily() == RocksDBColumnFamily::primary());
@@ -335,7 +335,7 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
       cb(RocksDBValue::documentId(_iterator->value()));
 
       --limit;
-      if (_reverse) {
+      if (reverse) {
         _iterator->Prev();
       } else {
         _iterator->Next();
@@ -370,7 +370,7 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
       cb(documentId, builder->slice());
 
       --limit;
-      if (_reverse) {
+      if (reverse) {
         _iterator->Prev();
       } else {
         _iterator->Next();
@@ -395,7 +395,7 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
 
       --count;
       ++skipped;
-      if (_reverse) {
+      if (reverse) {
         _iterator->Prev();
       } else {
         _iterator->Next();
@@ -411,7 +411,7 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
   void reset() override {
     TRI_ASSERT(_trx->state()->isRunning());
 
-    if (_reverse) {
+    if (reverse) {
       _iterator->SeekForPrev(_bounds.end());
     } else {
       _iterator->Seek(_bounds.start());
@@ -425,7 +425,7 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
  private:
   bool outOfRange() const {
     TRI_ASSERT(_trx->state()->isRunning());
-    if (_reverse) {
+    if (reverse) {
       return (_cmp->Compare(_iterator->key(), _bounds.start()) < 0);
     } else {
       return (_cmp->Compare(_iterator->key(), _bounds.end()) > 0);
@@ -435,7 +435,6 @@ class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
   arangodb::RocksDBPrimaryIndex const* _index;
   rocksdb::Comparator const* _cmp;
   std::unique_ptr<rocksdb::Iterator> _iterator;
-  bool const _reverse;
   bool const _allowCoveringIndexOptimization;
   RocksDBKeyBounds _bounds;
   // used for iterate_upper_bound iterate_lower_bound
@@ -691,8 +690,15 @@ std::unique_ptr<IndexIterator> RocksDBPrimaryIndex::iteratorForCondition(
   TRI_ASSERT(!isSorted() || opts.sorted);
   if (node == nullptr) {
     // full range scan
-    return std::make_unique<RocksDBPrimaryIndexRangeIterator>(
-        &_collection /*logical collection*/, trx, this, !opts.ascending /*reverse*/,
+    if (opts.ascending) {
+      // forward version
+      return std::make_unique<RocksDBPrimaryIndexRangeIterator<false>>(
+          &_collection /*logical collection*/, trx, this,
+          RocksDBKeyBounds::PrimaryIndex(_objectId, ::lowest, ::highest), opts.forceProjection);
+    }
+    // reverse version
+    return std::make_unique<RocksDBPrimaryIndexRangeIterator<true>>(
+        &_collection /*logical collection*/, trx, this,
         RocksDBKeyBounds::PrimaryIndex(_objectId, ::lowest, ::highest), opts.forceProjection);
   }
 
@@ -853,8 +859,15 @@ std::unique_ptr<IndexIterator> RocksDBPrimaryIndex::iteratorForCondition(
   }
 
   if (lowerFound && upperFound) {
-    return std::make_unique<RocksDBPrimaryIndexRangeIterator>(
-        &_collection /*logical collection*/, trx, this, !opts.ascending /*reverse*/,
+    if (opts.ascending) {
+      // forward version
+      return std::make_unique<RocksDBPrimaryIndexRangeIterator<false>>(
+          &_collection /*logical collection*/, trx, this,
+          RocksDBKeyBounds::PrimaryIndex(_objectId, lower, upper), opts.forceProjection);
+    }
+    // reverse version
+    return std::make_unique<RocksDBPrimaryIndexRangeIterator<true>>(
+        &_collection /*logical collection*/, trx, this,
         RocksDBKeyBounds::PrimaryIndex(_objectId, lower, upper), opts.forceProjection);
   }
 

arangod/RocksDBEngine/RocksDBPrimaryIndex.h

@@ -43,8 +43,8 @@ class Methods;
 
 class RocksDBPrimaryIndex final : public RocksDBIndex {
   friend class RocksDBPrimaryIndexEqIterator;
-  friend class RocksDBPrimaryIndexRangeIterator;
   friend class RocksDBPrimaryIndexInIterator;
+  template<bool reverse> friend class RocksDBPrimaryIndexRangeIterator;
   friend class RocksDBAllIndexIterator;
   friend class RocksDBAnyIndexIterator;
 

arangod/RocksDBEngine/RocksDBVPackIndex.cpp

@@ -168,6 +168,7 @@ class RocksDBVPackUniqueIndexIterator final : public IndexIterator {
 };
 
 /// @brief Iterator structure for RocksDB. We require a start and stop node
+template<bool reverse>
 class RocksDBVPackIndexIterator final : public IndexIterator {
  private:
   friend class RocksDBVPackIndex;
@@ -175,11 +176,10 @@ class RocksDBVPackIndexIterator final : public IndexIterator {
  public:
   RocksDBVPackIndexIterator(LogicalCollection* collection, transaction::Methods* trx,
                             arangodb::RocksDBVPackIndex const* index,
-                            bool reverse, RocksDBKeyBounds&& bounds)
+                            RocksDBKeyBounds&& bounds)
       : IndexIterator(collection, trx),
         _index(index),
         _cmp(static_cast<RocksDBVPackComparator const*>(index->comparator())),
-        _reverse(reverse),
         _bounds(std::move(bounds)) {
     TRI_ASSERT(index->columnFamily() == RocksDBColumnFamily::vpack());
 
@@ -295,7 +295,7 @@ class RocksDBVPackIndexIterator final : public IndexIterator {
   void reset() override {
     TRI_ASSERT(_trx->state()->isRunning());
 
-    if (_reverse) {
+    if (reverse) {
       _iterator->SeekForPrev(_bounds.end());
     } else {
       _iterator->Seek(_bounds.start());
@@ -317,7 +317,7 @@ class RocksDBVPackIndexIterator final : public IndexIterator {
     // so we really need to run the full-featured (read: expensive)
     // comparator
 
-    if (_reverse) {
+    if (reverse) {
       return (_cmp->Compare(_iterator->key(), _rangeBound) < 0);
     } else {
       return (_cmp->Compare(_iterator->key(), _rangeBound) > 0);
@@ -325,7 +325,7 @@ class RocksDBVPackIndexIterator final : public IndexIterator {
   }
 
   inline bool advance() {
-    if (_reverse) {
+    if (reverse) {
       _iterator->Prev();
     } else {
       _iterator->Next();
@@ -337,7 +337,6 @@ class RocksDBVPackIndexIterator final : public IndexIterator {
   arangodb::RocksDBVPackIndex const* _index;
   RocksDBVPackComparator const* _cmp;
   std::unique_ptr<rocksdb::Iterator> _iterator;
-  bool const _reverse;
   RocksDBKeyBounds _bounds;
   // used for iterate_upper_bound iterate_lower_bound
   rocksdb::Slice _rangeBound;
@@ -1008,7 +1007,12 @@ std::unique_ptr<IndexIterator> RocksDBVPackIndex::lookup(transaction::Methods* t
       _unique ? RocksDBKeyBounds::UniqueVPackIndex(_objectId, leftBorder, rightBorder)
               : RocksDBKeyBounds::VPackIndex(_objectId, leftBorder, rightBorder);
 
-  return std::make_unique<RocksDBVPackIndexIterator>(&_collection, trx, this, reverse, std::move(bounds));
+  if (reverse) {
+    // reverse version
+    return std::make_unique<RocksDBVPackIndexIterator<true>>(&_collection, trx, this, std::move(bounds));
+  }
+  // forward version
+  return std::make_unique<RocksDBVPackIndexIterator<false>>(&_collection, trx, this, std::move(bounds));
 }
 
 Index::FilterCosts RocksDBVPackIndex::supportsFilterCondition(

arangod/RocksDBEngine/RocksDBVPackIndex.h

@@ -58,7 +58,7 @@ class Methods;
 }
 
 class RocksDBVPackIndex : public RocksDBIndex {
-  friend class RocksDBVPackIndexIterator;
+  template<bool reverse> friend class RocksDBVPackIndexIterator;
 
  public:
   static uint64_t HashForKey(const rocksdb::Slice& key);