RocksDBBounds now give info about their columnFamily. Adapted convenience methods that used those bounds to use correct column family now

arangod/RocksDBEngine/RocksDBCollection.cpp

@@ -1691,7 +1691,6 @@ uint64_t RocksDBCollection::recalculateCounts() {
   // count documents
   auto documentBounds = RocksDBKeyBounds::CollectionDocuments(_objectId);
   _numberDocuments = rocksutils::countKeyRange(globalRocksDB(), readOptions,
-                                               RocksDBColumnFamily::documents(),
                                                documentBounds);
 
   // update counter manager value

arangod/RocksDBEngine/RocksDBCommon.cpp

@@ -27,8 +27,8 @@
 #include "Basics/RocksDBUtils.h"
 #include "Basics/StringRef.h"
 #include "Logger/Logger.h"
-#include "RocksDBEngine/RocksDBComparator.h"
 #include "RocksDBEngine/RocksDBColumnFamily.h"
+#include "RocksDBEngine/RocksDBComparator.h"
 #include "RocksDBEngine/RocksDBEngine.h"
 #include "RocksDBEngine/RocksDBKey.h"
 #include "RocksDBEngine/RocksDBKeyBounds.h"
@@ -193,7 +193,8 @@ std::pair<TRI_voc_tick_t, TRI_voc_cid_t> mapObjectToCollection(
 }
 
 std::size_t countKeyRange(rocksdb::DB* db, rocksdb::ReadOptions const& opts,
-                          rocksdb::ColumnFamilyHandle* handle, RocksDBKeyBounds const& bounds) {
+                          RocksDBKeyBounds const& bounds) {
+  rocksdb::ColumnFamilyHandle* handle = bounds.columnFamily();
   rocksdb::Comparator const* cmp = db->GetOptions().comparator;
   std::unique_ptr<rocksdb::Iterator> it(db->NewIterator(opts, handle));
   std::size_t count = 0;
@@ -214,13 +215,13 @@ Result removeLargeRange(rocksdb::TransactionDB* db,
                         RocksDBKeyBounds const& bounds) {
   LOG_TOPIC(DEBUG, Logger::FIXME) << "removing large range: " << bounds;
   try {
+    rocksdb::ColumnFamilyHandle* handle = bounds.columnFamily();
     // delete files in range lower..upper
     rocksdb::Slice lower(bounds.start());
     rocksdb::Slice upper(bounds.end());
     {
-      rocksdb::Status status = rocksdb::DeleteFilesInRange(
-          db->GetBaseDB(), db->GetBaseDB()->DefaultColumnFamily(), &lower,
-          &upper);
+      rocksdb::Status status =
+          rocksdb::DeleteFilesInRange(db->GetBaseDB(), handle, &lower, &upper);
       if (!status.ok()) {
         // if file deletion failed, we will still iterate over the remaining
         // keys, so we don't need to abort and raise an error here
@@ -235,8 +236,7 @@ Result removeLargeRange(rocksdb::TransactionDB* db,
     rocksdb::WriteBatch batch;
     rocksdb::ReadOptions readOptions;
     readOptions.fill_cache = false;
-    std::unique_ptr<rocksdb::Iterator> it(
-        db->NewIterator(readOptions));
+    std::unique_ptr<rocksdb::Iterator> it(db->NewIterator(readOptions));
 
     // TODO: split this into multiple batches if batches get too big
     it->Seek(lower);
@@ -274,10 +274,13 @@ std::vector<std::pair<RocksDBKey, RocksDBValue>> collectionKVPairs(
     TRI_voc_tick_t databaseId) {
   std::vector<std::pair<RocksDBKey, RocksDBValue>> rv;
   RocksDBKeyBounds bounds = RocksDBKeyBounds::DatabaseCollections(databaseId);
-  iterateBounds(bounds, [&rv](rocksdb::Iterator* it) {
-    rv.emplace_back(RocksDBKey(it->key()),
+  iterateBounds(bounds,
+                [&rv](rocksdb::Iterator* it) {
+                  rv.emplace_back(
+                      RocksDBKey(it->key()),
                       RocksDBValue(RocksDBEntryType::Collection, it->value()));
-  }, arangodb::RocksDBColumnFamily::other());
+                },
+                arangodb::RocksDBColumnFamily::other());
   return rv;
 }
 
@@ -285,10 +288,13 @@ std::vector<std::pair<RocksDBKey, RocksDBValue>> viewKVPairs(
     TRI_voc_tick_t databaseId) {
   std::vector<std::pair<RocksDBKey, RocksDBValue>> rv;
   RocksDBKeyBounds bounds = RocksDBKeyBounds::DatabaseViews(databaseId);
-  iterateBounds(bounds, [&rv](rocksdb::Iterator* it) {
-    rv.emplace_back(RocksDBKey(it->key()),
+  iterateBounds(bounds,
+                [&rv](rocksdb::Iterator* it) {
+                  rv.emplace_back(
+                      RocksDBKey(it->key()),
                       RocksDBValue(RocksDBEntryType::View, it->value()));
-  }, arangodb::RocksDBColumnFamily::other());
+                },
+                arangodb::RocksDBColumnFamily::other());
   return rv;
 }
 

arangod/RocksDBEngine/RocksDBCommon.h

@@ -167,7 +167,7 @@ std::pair<TRI_voc_tick_t, TRI_voc_cid_t> mapObjectToCollection(uint64_t);
 
 /// Iterator over all keys in range and count them
 std::size_t countKeyRange(rocksdb::DB*, rocksdb::ReadOptions const&,
-                          rocksdb::ColumnFamilyHandle*, RocksDBKeyBounds const&);
+                          RocksDBKeyBounds const&);
 
 /// @brief helper method to remove large ranges of data
 /// Should mainly be used to implement the drop() call

arangod/RocksDBEngine/RocksDBKeyBounds.cpp

@@ -24,6 +24,7 @@
 
 #include "RocksDBKeyBounds.h"
 #include "Basics/Exceptions.h"
+#include "RocksDBEngine/RocksDBColumnFamily.h"
 #include "RocksDBEngine/RocksDBCommon.h"
 #include "RocksDBEngine/RocksDBTypes.h"
 
@@ -199,6 +200,28 @@ uint64_t RocksDBKeyBounds::objectId() const {
   }
 }
 
+rocksdb::ColumnFamilyHandle* RocksDBKeyBounds::columnFamily() const {
+  RocksDBEntryType type = static_cast<RocksDBEntryType>(_internals._buffer[0]);
+  switch (type) {
+    case RocksDBEntryType::Document:
+      return RocksDBColumnFamily::documents();
+    case RocksDBEntryType::PrimaryIndexValue:
+      return RocksDBColumnFamily::primary();
+    case RocksDBEntryType::EdgeIndexValue:
+      return RocksDBColumnFamily::edge();
+    case RocksDBEntryType::IndexValue:
+      return RocksDBColumnFamily::index();
+    case RocksDBEntryType::UniqueIndexValue:
+      return RocksDBColumnFamily::uniqueIndex();
+    case RocksDBEntryType::FulltextIndexValue:
+      return RocksDBColumnFamily::fulltext();
+    case RocksDBEntryType::GeoIndexValue:
+      return RocksDBColumnFamily::geo();
+    default:
+      return RocksDBColumnFamily::other();
+  }
+}
+
 // constructor for an empty bound. do not use for anything but to
 // default-construct a key bound!
 RocksDBKeyBounds::RocksDBKeyBounds()

arangod/RocksDBEngine/RocksDBKeyBounds.h

@@ -36,6 +36,10 @@
 
 #include <iosfwd>
 
+namespace rocksdb {
+class ColumnFamilyHandle;
+}
+
 namespace arangodb {
 
 class RocksDBKeyBounds {
@@ -170,6 +174,17 @@ class RocksDBKeyBounds {
     return _internals.end();
   }
 
+
+  //////////////////////////////////////////////////////////////////////////////
+  /// @brief Returns the column family from this Bound
+  ///
+  /// All bounds iterators need to iterate over the correct column families
+  /// with this helper function it is made sure that correct column family
+  /// for bound is used.
+  //////////////////////////////////////////////////////////////////////////////
+
+  rocksdb::ColumnFamilyHandle* columnFamily() const;
+
   //////////////////////////////////////////////////////////////////////////////
   /// @brief Returns the object ID for these bounds
   ///