Compose the cached indices instead of ptrs to them

During a full scan operation on the ON_DISK_TRANSACTIONAL mode, the
presence of the in-memory chache is checked, speeding up the lookup time
by eliminating the roundtrip to disk storage. The same is implemented
here when index-based scans are used, instead of full scans.

src/storage/v2/disk/storage.cpp

@@ -582,15 +582,21 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, View view) {
         disk_storage->edge_import_mode_cache_->Vertices(label, view, &transaction_, &disk_storage->constraints_));
   }
 
-  index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = index_storage_.back();
+  if (label_index_storage_.contains(label)) {
+    return VerticesIterable(AllVerticesIterable(label_index_storage_[label].access(), &transaction_, view,
+                                                &storage_->indices_, &storage_->constraints_, storage_->config_.items));
+  }
+
+  label_index_storage_.emplace(label, utils::SkipList<storage::Vertex>());
+  auto &indexed_vertices = label_index_storage_[label];
+
   index_deltas_storage_.emplace_back();
   auto &index_deltas = index_deltas_storage_.back();
 
-  auto gids = MergeVerticesFromMainCacheWithLabelIndexCache(label, view, index_deltas, indexed_vertices.get());
-  LoadVerticesFromDiskLabelIndex(label, gids, index_deltas, indexed_vertices.get());
+  auto gids = MergeVerticesFromMainCacheWithLabelIndexCache(label, view, index_deltas, indexed_vertices);
+  LoadVerticesFromDiskLabelIndex(label, gids, index_deltas, indexed_vertices);
 
-  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), &transaction_, view, &storage_->indices_,
+  return VerticesIterable(AllVerticesIterable(indexed_vertices.access(), &transaction_, view, &storage_->indices_,
                                               &storage_->constraints_, storage_->config_.items));
 }
 
@@ -603,8 +609,16 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
         label, property, std::nullopt, std::nullopt, view, &transaction_, &disk_storage->constraints_));
   }
 
-  index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = index_storage_.back();
+  if (label_property_index_storage_.contains(std::make_pair(label, property))) {
+    return VerticesIterable(AllVerticesIterable(label_property_index_storage_[std::make_pair(label, property)].access(),
+                                                &transaction_, view, &storage_->indices_, &storage_->constraints_,
+                                                storage_->config_.items));
+  }
+
+  label_property_index_storage_.emplace(
+      std::make_pair(std::make_pair(label, property), utils::SkipList<storage::Vertex>()));
+  auto &indexed_vertices = label_property_index_storage_[std::make_pair(label, property)];
+
   index_deltas_storage_.emplace_back();
   auto &index_deltas = index_deltas_storage_.back();
 
@@ -614,18 +628,18 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
            HasVertexProperty(vertex, property, &transaction_, view);
   };
 
-  const auto gids = MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
-      label, property, view, index_deltas, indexed_vertices.get(), label_property_filter);
+  const auto gids = MergeVerticesFromMainCacheWithLabelPropertyIndexCache(label, property, view, index_deltas,
+                                                                          indexed_vertices, label_property_filter);
 
   const auto disk_label_property_filter = [](const std::string &key, const std::string &label_property_prefix,
                                              const std::unordered_set<Gid> &gids, Gid curr_gid) -> bool {
     return key.starts_with(label_property_prefix) && !utils::Contains(gids, curr_gid);
   };
 
-  LoadVerticesFromDiskLabelPropertyIndex(label, property, gids, index_deltas, indexed_vertices.get(),
+  LoadVerticesFromDiskLabelPropertyIndex(label, property, gids, index_deltas, indexed_vertices,
                                          disk_label_property_filter);
 
-  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), &transaction_, view, &storage_->indices_,
+  return VerticesIterable(AllVerticesIterable(indexed_vertices.access(), &transaction_, view, &storage_->indices_,
                                               &storage_->constraints_, storage_->config_.items));
 }
 
@@ -640,8 +654,16 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
         &disk_storage->constraints_));
   }
 
-  index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = index_storage_.back();
+  if (label_property_index_storage_.contains(std::make_pair(label, property))) {
+    return VerticesIterable(AllVerticesIterable(label_property_index_storage_[std::make_pair(label, property)].access(),
+                                                &transaction_, view, &storage_->indices_, &storage_->constraints_,
+                                                storage_->config_.items));
+  }
+
+  label_property_index_storage_.emplace(
+      std::make_pair(std::make_pair(label, property), utils::SkipList<storage::Vertex>()));
+  auto &indexed_vertices = label_property_index_storage_[std::make_pair(label, property)];
+
   index_deltas_storage_.emplace_back();
   auto &index_deltas = index_deltas_storage_.back();
 
@@ -651,13 +673,13 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
            VertexHasEqualPropertyValue(vertex, property, value, &transaction_, view);
   };
 
-  const auto gids = MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
-      label, property, view, index_deltas, indexed_vertices.get(), label_property_filter);
+  const auto gids = MergeVerticesFromMainCacheWithLabelPropertyIndexCache(label, property, view, index_deltas,
+                                                                          indexed_vertices, label_property_filter);
 
   LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup(label, property, gids, value, index_deltas,
-                                                             indexed_vertices.get());
+                                                             indexed_vertices);
 
-  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), &transaction_, view, &storage_->indices_,
+  return VerticesIterable(AllVerticesIterable(indexed_vertices.access(), &transaction_, view, &storage_->indices_,
                                               &storage_->constraints_, storage_->config_.items));
 }
 
@@ -673,24 +695,32 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
         label, property, lower_bound, upper_bound, view, &transaction_, &disk_storage->constraints_));
   }
 
-  index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = index_storage_.back();
+  if (label_property_index_storage_.contains(std::make_pair(label, property))) {
+    return VerticesIterable(AllVerticesIterable(label_property_index_storage_[std::make_pair(label, property)].access(),
+                                                &transaction_, view, &storage_->indices_, &storage_->constraints_,
+                                                storage_->config_.items));
+  }
+
+  label_property_index_storage_.emplace(
+      std::make_pair(std::make_pair(label, property), utils::SkipList<storage::Vertex>()));
+  auto &indexed_vertices = label_property_index_storage_[std::make_pair(label, property)];
+
   index_deltas_storage_.emplace_back();
   auto &index_deltas = index_deltas_storage_.back();
 
   const auto gids = MergeVerticesFromMainCacheWithLabelPropertyIndexCacheForIntervalSearch(
-      label, property, view, lower_bound, upper_bound, index_deltas, indexed_vertices.get());
+      label, property, view, lower_bound, upper_bound, index_deltas, indexed_vertices);
 
   LoadVerticesFromDiskLabelPropertyIndexForIntervalSearch(label, property, gids, lower_bound, upper_bound, index_deltas,
-                                                          indexed_vertices.get());
+                                                          indexed_vertices);
 
-  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), &transaction_, view, &storage_->indices_,
+  return VerticesIterable(AllVerticesIterable(indexed_vertices.access(), &transaction_, view, &storage_->indices_,
                                               &storage_->constraints_, storage_->config_.items));
 }
 
 /// TODO: (andi) This should probably go into some other class not the storage. All utils methods
 std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWithLabelIndexCache(
-    LabelId label, View view, std::list<Delta> &index_deltas, utils::SkipList<Vertex> *indexed_vertices) {
+    LabelId label, View view, std::list<Delta> &index_deltas, utils::SkipList<Vertex> &indexed_vertices) {
   auto main_cache_acc = vertices_.access();
   std::unordered_set<Gid> gids;
   gids.reserve(main_cache_acc.size());
@@ -705,7 +735,7 @@ std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWit
       LoadVertexToLabelIndexCache(utils::SerializeVertexAsKeyForLabelIndex(label, vertex.gid),
                                   utils::SerializeVertexAsValueForLabelIndex(label, vertex.labels, vertex.properties),
                                   CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::nullopt, ts),
-                                  indexed_vertices->access());
+                                  indexed_vertices.access());
     }
   }
   return gids;
@@ -714,7 +744,7 @@ std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWit
 void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelIndex(LabelId label,
                                                                const std::unordered_set<storage::Gid> &gids,
                                                                std::list<Delta> &index_deltas,
-                                                               utils::SkipList<Vertex> *indexed_vertices) {
+                                                               utils::SkipList<Vertex> &indexed_vertices) {
   auto *disk_label_index = static_cast<DiskLabelIndex *>(storage_->indices_.label_index_.get());
   auto disk_index_transaction = disk_label_index->CreateRocksDBTransaction();
   disk_index_transaction->SetReadTimestampForValidation(transaction_.start_timestamp);
@@ -736,14 +766,14 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelIndex(LabelId label,
       LoadVertexToLabelIndexCache(
           index_it->key().ToString(), index_it->value().ToString(),
           CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(key), deserializeTimestamp),
-          indexed_vertices->access());
+          indexed_vertices.access());
     }
   }
 }
 
 std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
     LabelId label, PropertyId property, View view, std::list<Delta> &index_deltas,
-    utils::SkipList<Vertex> *indexed_vertices, const auto &label_property_filter) {
+    utils::SkipList<Vertex> &indexed_vertices, const auto &label_property_filter) {
   auto main_cache_acc = vertices_.access();
   std::unordered_set<storage::Gid> gids;
   gids.reserve(main_cache_acc.size());
@@ -756,7 +786,7 @@ std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWit
       LoadVertexToLabelPropertyIndexCache(
           utils::SerializeVertexAsKeyForLabelPropertyIndex(label, property, vertex.gid),
           utils::SerializeVertexAsValueForLabelPropertyIndex(label, vertex.labels, vertex.properties),
-          CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::nullopt, ts), indexed_vertices->access());
+          CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::nullopt, ts), indexed_vertices.access());
     }
   }
 
@@ -766,7 +796,7 @@ std::unordered_set<Gid> DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWit
 void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndex(LabelId label, PropertyId property,
                                                                        const std::unordered_set<storage::Gid> &gids,
                                                                        std::list<Delta> &index_deltas,
-                                                                       utils::SkipList<Vertex> *indexed_vertices,
+                                                                       utils::SkipList<Vertex> &indexed_vertices,
                                                                        const auto &label_property_filter) {
   auto *disk_label_property_index =
       static_cast<DiskLabelPropertyIndex *>(storage_->indices_.label_property_index_.get());
@@ -790,14 +820,14 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndex(LabelId l
       LoadVertexToLabelPropertyIndexCache(
           index_it->key().ToString(), index_it->value().ToString(),
           CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(key), deserializeTimestamp),
-          indexed_vertices->access());
+          indexed_vertices.access());
     }
   }
 }
 
 void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup(
     LabelId label, PropertyId property, const std::unordered_set<storage::Gid> &gids, const PropertyValue &value,
-    std::list<Delta> &index_deltas, utils::SkipList<Vertex> *indexed_vertices) {
+    std::list<Delta> &index_deltas, utils::SkipList<Vertex> &indexed_vertices) {
   auto *disk_label_property_index =
       static_cast<DiskLabelPropertyIndex *>(storage_->indices_.label_property_index_.get());
   auto disk_index_transaction = disk_label_property_index->CreateRocksDBTransaction();
@@ -822,7 +852,7 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndexWithPointV
       LoadVertexToLabelPropertyIndexCache(
           index_it->key().ToString(), index_it->value().ToString(),
           CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(key), deserializeTimestamp),
-          indexed_vertices->access());
+          indexed_vertices.access());
     }
   }
 }
@@ -831,7 +861,7 @@ std::unordered_set<Gid>
 DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWithLabelPropertyIndexCacheForIntervalSearch(
     LabelId label, PropertyId property, View view, const std::optional<utils::Bound<PropertyValue>> &lower_bound,
     const std::optional<utils::Bound<PropertyValue>> &upper_bound, std::list<Delta> &index_deltas,
-    utils::SkipList<Vertex> *indexed_vertices) {
+    utils::SkipList<Vertex> &indexed_vertices) {
   auto main_cache_acc = vertices_.access();
   std::unordered_set<storage::Gid> gids;
   gids.reserve(main_cache_acc.size());
@@ -845,7 +875,7 @@ DiskStorage::DiskAccessor::MergeVerticesFromMainCacheWithLabelPropertyIndexCache
       LoadVertexToLabelPropertyIndexCache(
           utils::SerializeVertexAsKeyForLabelPropertyIndex(label, property, vertex.gid),
           utils::SerializeVertexAsValueForLabelPropertyIndex(label, vertex.labels, vertex.properties),
-          CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::nullopt, ts), indexed_vertices->access());
+          CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::nullopt, ts), indexed_vertices.access());
     }
   }
   return gids;
@@ -855,7 +885,7 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndexForInterva
     LabelId label, PropertyId property, const std::unordered_set<storage::Gid> &gids,
     const std::optional<utils::Bound<PropertyValue>> &lower_bound,
     const std::optional<utils::Bound<PropertyValue>> &upper_bound, std::list<Delta> &index_deltas,
-    utils::SkipList<Vertex> *indexed_vertices) {
+    utils::SkipList<Vertex> &indexed_vertices) {
   auto *disk_label_property_index =
       static_cast<DiskLabelPropertyIndex *>(storage_->indices_.label_property_index_.get());
 
@@ -885,7 +915,7 @@ void DiskStorage::DiskAccessor::LoadVerticesFromDiskLabelPropertyIndexForInterva
     LoadVertexToLabelPropertyIndexCache(
         index_it->key().ToString(), index_it->value().ToString(),
         CreateDeleteDeserializedIndexObjectDelta(index_deltas, std::move(key_str), deserializeTimestamp),
-        indexed_vertices->access());
+        indexed_vertices.access());
   }
 }
 
@@ -1061,13 +1091,25 @@ std::optional<VertexAccessor> DiskStorage::DiskAccessor::FindVertex(storage::Gid
     return VertexAccessor::Create(&*vertex_it, &transaction_, &storage_->indices_, &storage_->constraints_, config_,
                                   view);
   }
-  for (const auto &vec : index_storage_) {
-    acc = vec->access();
-    auto index_it = acc.find(gid);
-    if (index_it != acc.end()) {
-      return VertexAccessor::Create(&*index_it, &transaction_, &storage_->indices_, &storage_->constraints_, config_,
-                                    view);
+
+  auto find_in_indices = [&](auto &&index_storage) -> std::optional<VertexAccessor> {
+    for (auto &[key, skip_list] : index_storage) {
+      acc = skip_list.access();
+      auto index_it = acc.find(gid);
+      if (index_it != acc.end()) {
+        return VertexAccessor::Create(&*index_it, &transaction_, &storage_->indices_, &storage_->constraints_, config_,
+                                      view);
+      }
     }
+    return {};
+  };
+
+  if (auto vertex = find_in_indices(label_index_storage_)) {
+    return *vertex;
+  }
+
+  if (auto vertex = find_in_indices(label_property_index_storage_)) {
+    return *vertex;
   }
 
   rocksdb::ReadOptions read_opts;
@@ -1427,10 +1469,21 @@ DiskStorage::DiskAccessor::ClearDanglingVertices() {
 [[nodiscard]] utils::BasicResult<StorageDataManipulationError, void> DiskStorage::DiskAccessor::FlushIndexCache() {
   std::vector<std::vector<PropertyValue>> unique_storage;
 
-  for (const auto &vec : index_storage_) {
-    if (auto vertices_res = FlushVertices(vec->access(), unique_storage); vertices_res.HasError()) {
-      return vertices_res.GetError();
+  auto flush_index = [&](auto &&index_storage) -> utils::BasicResult<StorageDataManipulationError, void> {
+    for (const auto &[key, skip_list] : index_storage) {
+      if (auto vertices_res = FlushVertices(skip_list.access(), unique_storage); vertices_res.HasError()) {
+        return vertices_res.GetError();
+      }
     }
+    return {};
+  };
+
+  if (auto res = flush_index(label_index_storage_); res.HasError()) {
+    return res.GetError();
+  }
+
+  if (auto res = flush_index(label_property_index_storage_); res.HasError()) {
+    return res.GetError();
   }
 
   return {};

src/storage/v2/disk/storage.hpp

@@ -25,7 +25,9 @@
 
 #include <rocksdb/db.h>
 #include <rocksdb/slice.h>
+#include <map>
 #include <unordered_set>
+#include <utility>
 
 namespace memgraph::storage {
 
@@ -63,37 +65,37 @@ class DiskStorage final : public Storage {
 
     std::unordered_set<Gid> MergeVerticesFromMainCacheWithLabelIndexCache(LabelId label, View view,
                                                                           std::list<Delta> &index_deltas,
-                                                                          utils::SkipList<Vertex> *indexed_vertices);
+                                                                          utils::SkipList<Vertex> &indexed_vertices);
 
     void LoadVerticesFromDiskLabelIndex(LabelId label, const std::unordered_set<storage::Gid> &gids,
-                                        std::list<Delta> &index_deltas, utils::SkipList<Vertex> *indexed_vertices);
+                                        std::list<Delta> &index_deltas, utils::SkipList<Vertex> &indexed_vertices);
 
     std::unordered_set<Gid> MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
         LabelId label, PropertyId property, View view, std::list<Delta> &index_deltas,
-        utils::SkipList<Vertex> *indexed_vertices, const auto &label_property_filter);
+        utils::SkipList<Vertex> &indexed_vertices, const auto &label_property_filter);
 
     void LoadVerticesFromDiskLabelPropertyIndex(LabelId label, PropertyId property,
                                                 const std::unordered_set<storage::Gid> &gids,
                                                 std::list<Delta> &index_deltas,
-                                                utils::SkipList<Vertex> *indexed_vertices,
+                                                utils::SkipList<Vertex> &indexed_vertices,
                                                 const auto &label_property_filter);
 
     void LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup(LabelId label, PropertyId property,
                                                                     const std::unordered_set<storage::Gid> &gids,
                                                                     const PropertyValue &value,
                                                                     std::list<Delta> &index_deltas,
-                                                                    utils::SkipList<Vertex> *indexed_vertices);
+                                                                    utils::SkipList<Vertex> &indexed_vertices);
 
     std::unordered_set<Gid> MergeVerticesFromMainCacheWithLabelPropertyIndexCacheForIntervalSearch(
         LabelId label, PropertyId property, View view, const std::optional<utils::Bound<PropertyValue>> &lower_bound,
         const std::optional<utils::Bound<PropertyValue>> &upper_bound, std::list<Delta> &index_deltas,
-        utils::SkipList<Vertex> *indexed_vertices);
+        utils::SkipList<Vertex> &indexed_vertices);
 
     void LoadVerticesFromDiskLabelPropertyIndexForIntervalSearch(
         LabelId label, PropertyId property, const std::unordered_set<storage::Gid> &gids,
         const std::optional<utils::Bound<PropertyValue>> &lower_bound,
         const std::optional<utils::Bound<PropertyValue>> &upper_bound, std::list<Delta> &index_deltas,
-        utils::SkipList<Vertex> *indexed_vertices);
+        utils::SkipList<Vertex> &indexed_vertices);
 
    public:
     DiskAccessor(const DiskAccessor &) = delete;
@@ -266,7 +268,8 @@ class DiskStorage final : public Storage {
 
     /// Main storage
     utils::SkipList<Vertex> vertices_;
-    std::vector<std::unique_ptr<utils::SkipList<Vertex>>> index_storage_;
+    std::map<LabelId, utils::SkipList<Vertex>> label_index_storage_;
+    std::map<std::pair<LabelId, PropertyId>, utils::SkipList<Vertex>> label_property_index_storage_;
 
     /// We need them because query context for indexed reading is cleared after the query is done not after the
     /// transaction is done