Merge branch 'engine-api' of https://github.com/arangodb/arangodb into engine-api

arangod/RocksDBEngine/RocksDBCommon.cpp

@@ -30,6 +30,7 @@
 #include "StorageEngine/EngineSelectorFeature.h"
 #include "RocksDBEngine/RocksDBEngine.h"
 #include "RocksDBEngine/RocksDBKey.h"
+#include "RocksDBEngine/RocksDBComparator.h"
 
 #include <rocksdb/utilities/transaction_db.h>
 #include <rocksdb/utilities/transaction_db.h>
@@ -230,28 +231,25 @@ Result removeLargeRange(rocksdb::TransactionDB* db, RocksDBKeyBounds const& boun
 std::vector<std::pair<RocksDBKey,RocksDBValue>> collectionKVPairs(TRI_voc_tick_t databaseId){
   std::vector<std::pair<RocksDBKey,RocksDBValue>> rv;
   RocksDBKeyBounds bounds = RocksDBKeyBounds::DatabaseCollections(databaseId);
-  rocksdb::Iterator* it = globalRocksDB()->NewIterator(rocksdb::ReadOptions());
+  iterateBounds(bounds, [&rv](rocksdb::Iterator* it){
-  for (it->Seek(bounds.start()); it->Valid() && it->key() != bounds.end(); it->Next()) {
     rv.emplace_back(RocksDBKey(it->key()),RocksDBValue(RocksDBEntryType::Collection, it->value()));
-  }
+  });
   return rv;
 }
 std::vector<std::pair<RocksDBKey,RocksDBValue>> indexKVPairs(TRI_voc_tick_t databaseId){
   std::vector<std::pair<RocksDBKey,RocksDBValue>> rv;
   RocksDBKeyBounds bounds = RocksDBKeyBounds::DatabaseIndexes(databaseId);
-  rocksdb::Iterator* it = globalRocksDB()->NewIterator(rocksdb::ReadOptions());
+  iterateBounds(bounds, [&rv](rocksdb::Iterator* it){
-  for (it->Seek(bounds.start()); it->Valid() && it->key() != bounds.end(); it->Next()) {
     rv.emplace_back(RocksDBKey(it->key()),RocksDBValue(RocksDBEntryType::Index, it->value()));
-  }
+  });
   return rv;
 }
 std::vector<std::pair<RocksDBKey,RocksDBValue>> viewKVPairs(TRI_voc_tick_t databaseId){
   std::vector<std::pair<RocksDBKey,RocksDBValue>> rv;
   RocksDBKeyBounds bounds = RocksDBKeyBounds::DatabaseViews(databaseId);
-  rocksdb::Iterator* it = globalRocksDB()->NewIterator(rocksdb::ReadOptions());
+  iterateBounds(bounds, [&rv](rocksdb::Iterator* it){
-  for (it->Seek(bounds.start()); it->Valid() && it->key() != bounds.end(); it->Next()) {
     rv.emplace_back(RocksDBKey(it->key()),RocksDBValue(RocksDBEntryType::View, it->value()));
-  }
+  });
   return rv;
 }
 

arangod/RocksDBEngine/RocksDBCommon.h

@@ -28,9 +28,13 @@
 
 #include "Basics/Common.h"
 #include "Basics/Result.h"
-#include "RocksDBEngine/RocksDBValue.h"
+#include "RocksDBEngine/RocksDBComparator.h"
+#include "RocksDBEngine/RocksDBEngine.h"
 #include "RocksDBEngine/RocksDBKey.h"
+#include "RocksDBEngine/RocksDBKeyBounds.h"
+#include "RocksDBEngine/RocksDBValue.h"
 
+#include <rocksdb/iterator.h>
 #include <rocksdb/options.h>
 #include <rocksdb/status.h>
 
@@ -81,6 +85,17 @@ std::vector<std::pair<RocksDBKey,RocksDBValue>> collectionKVPairs(TRI_voc_tick_t
 std::vector<std::pair<RocksDBKey,RocksDBValue>> indexKVPairs(TRI_voc_tick_t databaseId);
 std::vector<std::pair<RocksDBKey,RocksDBValue>> viewKVPairs(TRI_voc_tick_t databaseId);
 
+// optional switch to std::function to reduce amount of includes and to avoid template
+// this helper is not meant for transactional usage!
+template<typename T> //T is a invokeable that takes a rocksdb::Iterator*
+void iterateBounds(RocksDBKeyBounds const& bounds, T callback, rocksdb::ReadOptions const& options = rocksdb::ReadOptions{}){
+  auto cmp = globalRocksEngine()->cmp();
+  std::unique_ptr<rocksdb::Iterator> it(globalRocksDB()->NewIterator(options));
+  for (it->Seek(bounds.start()); it->Valid() && cmp->Compare(it->key(), bounds.end()) < 0; it->Next()) {
+    callback(it.get());
+  }
+}
+
 }  // namespace rocksutils
 }  // namespace arangodb
 

arangod/RocksDBEngine/RocksDBComparator.h

@@ -42,6 +42,9 @@ class RocksDBComparator final : public rocksdb::Comparator {
 
   //////////////////////////////////////////////////////////////////////////////
   /// @brief Compares any two RocksDB keys.
+  /// returns -1 if lhs < rhs
+  ///          1 if lhs > rhs
+  ///          0 if lhs == rhs
   //////////////////////////////////////////////////////////////////////////////
   int Compare(rocksdb::Slice const& lhs, rocksdb::Slice const& rhs) const;