Merge branch 'master' into T1312-MG-Cache-vertex-reads-on-disk

Read all the label-proprty indices from disk, independent of further filtering
General refactor
2023-10-18 08:44:18 +02:00 · 2023-10-17 11:37:10 +02:00 · 2023-10-17 10:04:52 +02:00 · 2023-10-17 08:01:05 +02:00 · 2023-10-16 13:56:21 +02:00 · 2023-10-16 13:22:22 +02:00
4 changed files with 220 additions and 54 deletions
--- a/src/storage/v2/disk/storage.cpp
+++ b/src/storage/v2/disk/storage.cpp
@@ -195,6 +195,11 @@ bool IsPropertyValueWithinInterval(const PropertyValue &value,
  return true;
 }

+bool disk_label_property_filter(const std::string &key, const std::string &label_property_prefix,
+                                const std::unordered_set<Gid> &gids, Gid curr_gid) {
+  return key.starts_with(label_property_prefix) && !utils::Contains(gids, curr_gid);
+}
+
 }  // namespace

 DiskStorage::DiskStorage(Config config)
@@ -479,20 +484,49 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, View view) {
    return VerticesIterable(disk_storage->edge_import_mode_cache_->Vertices(label, view, storage_, &transaction_));
  }

-  transaction_.index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = transaction_.index_storage_.back();
-  transaction_.index_deltas_storage_.emplace_back();
-  auto &index_deltas = transaction_.index_deltas_storage_.back();
+  utils::SkipList<memgraph::storage::Vertex> *indexed_vertices{nullptr};
+  std::list<storage::Delta> *index_deltas{nullptr};

-  auto gids = disk_storage->MergeVerticesFromMainCacheWithLabelIndexCache(&transaction_, label, view, index_deltas,
-                                                                          indexed_vertices.get());
-  disk_storage->LoadVerticesFromDiskLabelIndex(&transaction_, label, gids, index_deltas, indexed_vertices.get());
+  auto merge_with_main_cache = [&](auto &index, auto &index_delta_storage) -> std::unordered_set<Gid> {
+    indexed_vertices = &index[label];
+    index_deltas = &index_delta_storage.emplace_back();
+    return disk_storage->MergeVerticesFromMainCacheWithLabelIndexCache(&transaction_, label, view, *index_deltas,
+                                                                       indexed_vertices);
+  };

+  auto &cache = transaction_.label_index_cache_;
+  auto &cache_ci = transaction_.label_index_cache_ci_;
+  auto &index_delta_storage = transaction_.index_deltas_storage_;
+
+  SyncDeletedVertices(cache, label, index_delta_storage, view, merge_with_main_cache);
+
+  if (transaction_.command_id > cache_ci) {
+    cache_ci = transaction_.command_id;
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
+  }
+  if (cache.contains(label)) {
+    if (view == View::NEW) {
+      cache[label] = utils::SkipList<Vertex>();
+    }
+    // TODO
+    // we do not need this merge if we can make sure that
+    // the removing and re-adding the same label to a given
+    // vertex within one transaction is not permitted.
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(cache[label].access(), storage_, &transaction_, view));
+  }
+  cache[label] = utils::SkipList<Vertex>();
+  index_deltas = &index_delta_storage.emplace_back();
+  disk_storage->LoadVerticesFromDiskLabelIndex(&transaction_, label, {}, *index_deltas, &transaction_.vertices_);
+  merge_with_main_cache(cache, index_delta_storage);
  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
 }

 VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId property, View view) {
  auto *disk_storage = static_cast<DiskStorage *>(storage_);
+  const auto cache_key = std::make_pair(label, property);
+
  if (disk_storage->edge_import_status_ == EdgeImportMode::ACTIVE) {
    disk_storage->HandleLoadingLabelPropertyForEdgeImportCache(&transaction_, label, property);

@@ -500,34 +534,59 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
                                                                            view, storage_, &transaction_));
  }

-  transaction_.index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = transaction_.index_storage_.back();
-  transaction_.index_deltas_storage_.emplace_back();
-  auto &index_deltas = transaction_.index_deltas_storage_.back();
+  utils::SkipList<memgraph::storage::Vertex> *indexed_vertices{nullptr};
+  std::list<storage::Delta> *index_deltas{nullptr};

-  const auto label_property_filter = [this](const Vertex &vertex, LabelId label, PropertyId property,
-                                            View view) -> bool {
-    return VertexHasLabel(vertex, label, &transaction_, view) &&
-           HasVertexProperty(vertex, property, &transaction_, view);
+  auto merge_with_main_cache = [&](auto &index, auto &index_delta_storage) -> std::unordered_set<Gid> {
+    index[cache_key] = utils::SkipList<Vertex>();
+    indexed_vertices = &index[cache_key];
+    index_deltas = &index_delta_storage.emplace_back();
+
+    const auto label_property_filter = [this](const Vertex &vertex, LabelId label, PropertyId property,
+                                              View view) -> bool {
+      return VertexHasLabel(vertex, label, &transaction_, view) &&
+             HasVertexProperty(vertex, property, &transaction_, view);
+    };
+
+    return disk_storage->MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
+        &transaction_, label, property, view, *index_deltas, indexed_vertices, label_property_filter);
  };

-  const auto gids = disk_storage->MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
-      &transaction_, label, property, view, index_deltas, indexed_vertices.get(), label_property_filter);
+  auto &cache = transaction_.label_property_index_cache_;
+  auto &cache_ci = transaction_.label_property_index_cache_ci_;
+  auto &index_delta_storage = transaction_.index_deltas_storage_;

-  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);
-  };
-
-  disk_storage->LoadVerticesFromDiskLabelPropertyIndex(&transaction_, label, property, gids, index_deltas,
-                                                       indexed_vertices.get(), disk_label_property_filter);
+  SyncDeletedVertices(cache, cache_key, index_delta_storage, view, merge_with_main_cache);

+  if (transaction_.command_id > cache_ci) {
+    cache_ci = transaction_.command_id;
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
+  }
+  if (cache.contains(cache_key)) {
+    if (view == View::NEW) {
+      cache[cache_key] = utils::SkipList<Vertex>();
+    }
+    // TODO
+    // we do not need this merge if we can make sure that
+    // the removing and re-adding the same label to a given
+    // vertex within one transaction is not permitted.
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
+  }
+  cache[cache_key] = utils::SkipList<Vertex>();
+  index_deltas = &index_delta_storage.emplace_back();
+  disk_storage->LoadVerticesFromDiskLabelPropertyIndex(&transaction_, label, property, {}, *index_deltas,
+                                                       &transaction_.vertices_, disk_label_property_filter);
+  merge_with_main_cache(cache, index_delta_storage);
  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
 }

 VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId property, const PropertyValue &value,
                                                     View view) {
  auto *disk_storage = static_cast<DiskStorage *>(storage_);
+  const auto cache_key = std::make_pair(label, property);
+
  if (disk_storage->edge_import_status_ == EdgeImportMode::ACTIVE) {
    disk_storage->HandleLoadingLabelPropertyForEdgeImportCache(&transaction_, label, property);

@@ -536,23 +595,51 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
        &transaction_));
  }

-  transaction_.index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = transaction_.index_storage_.back();
-  transaction_.index_deltas_storage_.emplace_back();
-  auto &index_deltas = transaction_.index_deltas_storage_.back();
+  utils::SkipList<memgraph::storage::Vertex> *indexed_vertices{nullptr};
+  std::list<storage::Delta> *index_deltas{nullptr};

-  auto label_property_filter = [this, &value](const Vertex &vertex, LabelId label, PropertyId property,
-                                              View view) -> bool {
-    return VertexHasLabel(vertex, label, &transaction_, view) &&
-           VertexHasEqualPropertyValue(vertex, property, value, &transaction_, view);
+  auto merge_with_main_cache = [&](auto &index, auto &index_delta_storage) -> std::unordered_set<Gid> {
+    index[cache_key] = utils::SkipList<Vertex>();
+    indexed_vertices = &index[cache_key];
+    index_deltas = &index_delta_storage.emplace_back();
+
+    auto label_property_filter = [this, &value](const Vertex &vertex, LabelId label, PropertyId property,
+                                                View view) -> bool {
+      return VertexHasLabel(vertex, label, &transaction_, view) &&
+             VertexHasEqualPropertyValue(vertex, property, value, &transaction_, view);
+    };
+
+    return disk_storage->MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
+        &transaction_, label, property, view, *index_deltas, indexed_vertices, label_property_filter);
  };

-  const auto gids = disk_storage->MergeVerticesFromMainCacheWithLabelPropertyIndexCache(
-      &transaction_, label, property, view, index_deltas, indexed_vertices.get(), label_property_filter);
+  auto &cache = transaction_.label_property_index_cache_;
+  auto &cache_ci = transaction_.label_property_index_cache_ci_;
+  auto &index_delta_storage = transaction_.index_deltas_storage_;

-  disk_storage->LoadVerticesFromDiskLabelPropertyIndexWithPointValueLookup(&transaction_, label, property, gids, value,
-                                                                           index_deltas, indexed_vertices.get());
+  SyncDeletedVertices(cache, cache_key, index_delta_storage, view, merge_with_main_cache);

+  if (transaction_.command_id > cache_ci) {
+    cache_ci = transaction_.command_id;
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
+  }
+  if (cache.contains(cache_key)) {
+    if (view == View::NEW) {
+      cache[cache_key] = utils::SkipList<Vertex>();
+    }
+    // TODO
+    // we do not need this merge if we can make sure that
+    // the removing and re-adding the same label to a given
+    // vertex within one transaction is not permitted.
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
+  }
+  cache[cache_key] = utils::SkipList<Vertex>();
+  index_deltas = &index_delta_storage.emplace_back();
+  disk_storage->LoadVerticesFromDiskLabelPropertyIndex(&transaction_, label, property, {}, *index_deltas,
+                                                       &transaction_.vertices_, disk_label_property_filter);
+  merge_with_main_cache(cache, index_delta_storage);
  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
 }

@@ -561,6 +648,7 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
                                                     const std::optional<utils::Bound<PropertyValue>> &upper_bound,
                                                     View view) {
  auto *disk_storage = static_cast<DiskStorage *>(storage_);
+  const auto cache_key = std::make_pair(label, property);
  if (disk_storage->edge_import_status_ == EdgeImportMode::ACTIVE) {
    disk_storage->HandleLoadingLabelPropertyForEdgeImportCache(&transaction_, label, property);

@@ -568,17 +656,45 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
                                                                            view, storage_, &transaction_));
  }

-  transaction_.index_storage_.emplace_back(std::make_unique<utils::SkipList<storage::Vertex>>());
-  auto &indexed_vertices = transaction_.index_storage_.back();
-  transaction_.index_deltas_storage_.emplace_back();
-  auto &index_deltas = transaction_.index_deltas_storage_.back();
+  utils::SkipList<memgraph::storage::Vertex> *indexed_vertices{nullptr};
+  std::list<storage::Delta> *index_deltas{nullptr};

-  const auto gids = disk_storage->MergeVerticesFromMainCacheWithLabelPropertyIndexCacheForIntervalSearch(
-      &transaction_, label, property, view, lower_bound, upper_bound, index_deltas, indexed_vertices.get());
+  auto merge_with_main_cache = [&](auto &index, auto &index_delta_storage) -> std::unordered_set<Gid> {
+    index[cache_key] = utils::SkipList<Vertex>();
+    indexed_vertices = &index[cache_key];
+    index_deltas = &index_delta_storage.emplace_back();

-  disk_storage->LoadVerticesFromDiskLabelPropertyIndexForIntervalSearch(
-      &transaction_, label, property, gids, lower_bound, upper_bound, index_deltas, indexed_vertices.get());
+    return disk_storage->MergeVerticesFromMainCacheWithLabelPropertyIndexCacheForIntervalSearch(
+        &transaction_, label, property, view, lower_bound, upper_bound, *index_deltas, indexed_vertices);
+  };

+  auto &cache = transaction_.label_property_index_cache_;
+  auto &cache_ci = transaction_.label_property_index_cache_ci_;
+  auto &index_delta_storage = transaction_.index_deltas_storage_;
+
+  SyncDeletedVertices(cache, cache_key, index_delta_storage, view, merge_with_main_cache);
+
+  if (transaction_.command_id > cache_ci) {
+    cache_ci = transaction_.command_id;
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
+  }
+  if (cache.contains(cache_key)) {
+    if (view == View::NEW) {
+      cache[cache_key] = utils::SkipList<Vertex>();
+    }
+    // TODO
+    // we do not need this merge if we can make sure that
+    // the removing and re-adding the same label to a given
+    // vertex within one transaction is not permitted.
+    merge_with_main_cache(cache, index_delta_storage);
+    return VerticesIterable(AllVerticesIterable(cache[cache_key].access(), storage_, &transaction_, view));
+  }
+  cache[cache_key] = utils::SkipList<Vertex>();
+  index_deltas = &index_delta_storage.emplace_back();
+  disk_storage->LoadVerticesFromDiskLabelPropertyIndex(&transaction_, label, property, {}, *index_deltas,
+                                                       &transaction_.vertices_, disk_label_property_filter);
+  merge_with_main_cache(cache, index_delta_storage);
  return VerticesIterable(AllVerticesIterable(indexed_vertices->access(), storage_, &transaction_, view));
 }

@@ -1181,10 +1297,22 @@ bool DiskStorage::DeleteEdgeFromConnectivityIndex(Transaction *transaction, cons
    Transaction *transaction) {
  std::vector<std::vector<PropertyValue>> unique_storage;

-  for (const auto &vec : transaction->index_storage_) {
-    if (auto vertices_res = FlushVertices(transaction, vec->access(), unique_storage); vertices_res.HasError()) {
-      return vertices_res.GetError();
+  auto flush_index = [&](auto &index_storage) -> utils::BasicResult<StorageManipulationError, void> {
+    for (const auto &[k, v] : index_storage) {
+      if (auto vertices_res = FlushVertices(transaction, v.access(), unique_storage); vertices_res.HasError()) {
+        return vertices_res.GetError();
+      }
    }
+
+    return {};
+  };
+
+  if (auto res = flush_index(transaction->label_index_cache_); res.HasError()) {
+    return res.GetError();
+  }
+
+  if (auto res = flush_index(transaction->label_property_index_cache_); res.HasError()) {
+    return res.GetError();
  }

  return {};
@@ -1309,12 +1437,24 @@ std::optional<VertexAccessor> DiskStorage::FindVertex(storage::Gid gid, Transact
  if (vertex_it != acc.end()) {
    return VertexAccessor::Create(&*vertex_it, this, transaction, view);
  }
-  for (const auto &vec : transaction->index_storage_) {
-    acc = vec->access();
-    auto index_it = acc.find(gid);
-    if (index_it != acc.end()) {
-      return VertexAccessor::Create(&*index_it, this, transaction, view);
+
+  auto find_in_indices = [&](auto &index_storage) -> std::optional<VertexAccessor> {
+    for (auto &[k, v] : index_storage) {
+      acc = v.access();
+      auto index_it = acc.find(gid);
+      if (index_it != acc.end()) {
+        return VertexAccessor::Create(&*index_it, this, transaction, view);
+      }
    }
+    return {};
+  };
+
+  if (auto vertex = find_in_indices(transaction->label_index_cache_)) {
+    return *vertex;
+  }
+
+  if (auto vertex = find_in_indices(transaction->label_property_index_cache_)) {
+    return *vertex;
  }

  rocksdb::ReadOptions read_opts;
--- a/src/storage/v2/disk/storage.hpp
+++ b/src/storage/v2/disk/storage.hpp
@@ -47,6 +47,18 @@ class DiskStorage final : public Storage {

    explicit DiskAccessor(auto tag, DiskStorage *storage, IsolationLevel isolation_level, StorageMode storage_mode);

+    template <typename TIndex, typename TIndexCacheKey, typename TMergeFunc>
+    void SyncDeletedVertices(TIndex &index, TIndexCacheKey &cache_key,
+                             std::vector<std::list<Delta>> &index_delta_storage, View view, TMergeFunc &merge_func) {
+      if (transaction_.vertices_to_delete_.empty()) {
+        return;
+      }
+      index[cache_key] = utils::SkipList<Vertex>();
+      if (view == View::OLD) {
+        merge_func(index, index_delta_storage);
+      }
+    }
+
   public:
    DiskAccessor(const DiskAccessor &) = delete;
    DiskAccessor &operator=(const DiskAccessor &) = delete;
--- a/src/storage/v2/transaction.hpp
+++ b/src/storage/v2/transaction.hpp
@@ -113,7 +113,13 @@ struct Transaction {
  rocksdb::Transaction *disk_transaction_;
  /// Main storage
  utils::SkipList<Vertex> vertices_;
-  std::vector<std::unique_ptr<utils::SkipList<Vertex>>> index_storage_;
+
+  using LabelIndex = LabelId;
+  using LabelPropertyIndex = std::pair<LabelId, PropertyId>;
+  std::map<LabelId, utils::SkipList<Vertex>> label_index_cache_;
+  uint64_t label_index_cache_ci_{command_id};
+  std::map<LabelPropertyIndex, utils::SkipList<Vertex>> label_property_index_cache_;
+  uint64_t label_property_index_cache_ci_{command_id};

  /// We need them because query context for indexed reading is cleared after the query is done not after the
  /// transaction is done
--- a/src/utils/bound.hpp
+++ b/src/utils/bound.hpp
@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -33,6 +33,14 @@ class Bound {
  Bound &operator=(const Bound &other) = default;
  Bound &operator=(Bound &&other) = default;

+  // bool operator<(const Bound &other) const {
+  //   return std::make_tuple(value_, type_) < std::make_tuple(other.value_, other.type_);
+  // }
+
+  // bool operator==(const Bound &other) const {
+  //   return std::make_tuple(value_, type_) == std::make_tuple(other.value_, other.type_);
+  // }
+
  /** Value for the bound. */
  const auto &value() const { return value_; }
  /** Whether the bound is inclusive or exclusive. */
Author	SHA1	Message	Date
gvolfing	3f560fe602	Merge branch 'master' into T1312-MG-Cache-vertex-reads-on-disk	2023-10-18 08:44:18 +02:00
gvolfing	c3067d3640	Read all the label-proprty indices from disk, independent of further filtering	2023-10-17 11:37:10 +02:00
gvolfing	a40b18b8aa	General refactor	2023-10-17 10:04:52 +02:00
gvolfing	422f9199bd	Apply suggestions from code review Co-authored-by: Andi <andi8647@gmail.com>	2023-10-17 08:01:05 +02:00
gvolfing	89585fad1c	Remove unnecessary local variable	2023-10-16 13:56:21 +02:00
gvolfing	8feefc20b7	Change caching mechanism. Instead of trying to read into the index caches directly, read into the main chache, and merge the index and main caches when appropriate. This is needed because of several index modifying calls within one transaction makes the indexing process convoluted, and should be propertly addressed when the On-disk LRU cache will be properly implemented.	2023-10-16 13:22:22 +02:00
gvolfing	646818a0b4	Merge branch 'master' into T1312-MG-Cache-vertex-reads-on-disk	2023-10-11 13:36:04 +02:00
gvolfing	2f4156c2c9	Make lambda take another argument instead of capturing	2023-10-11 13:34:57 +02:00
gvolfing	61d301310a	Remove merge-conflict related comments	2023-10-11 13:14:29 +02:00
gvolfing	73a554a617	Merge branch 'master' into T1312-MG-Cache-vertex-reads-on-disk	2023-10-11 13:09:50 +02:00
gvolfing	bc34a0add6	Merge branch 'master' into T1312-MG-Cache-vertex-reads-on-disk	2023-10-11 10:19:14 +02:00
gvolfing	af4abd4db0	Remove comments	2023-10-10 16:57:00 +02:00
gvolfing	f806419b1d	Remove comments	2023-10-10 16:14:53 +02:00
gvolfing	8e3be813ac	General clean-up	2023-10-09 19:43:10 +02:00
gvolfing	b850a66afd	General clean-up	2023-10-09 19:13:03 +02:00
gvolfing	96dcfae63e	Remove unnecessary metadata placeholder	2023-10-05 12:47:28 +02:00
gvolfing	1b6a73a021	Cache the indices upon reading them from disk Create and utilize caches for indices, one per index access type. This is needed so different access methods within one transaction will not override the otherwise would-be common cache. On top of the caches themselves a counter has also been added per access methos type to make sure that calls to AdvanceMethod are handled correctly from the caching point of view.	2023-10-05 08:37:28 +02:00