separate rocksdb directories

* Fix TriggerContext adaptation for accessors.
* Fix edge deserialization in case of the deleted vertex.

README.md

@@ -175,7 +175,6 @@ license](./licenses/BSL.txt).</br> Memgraph Enterprise is available under the
 
 - :purple_heart: [**Discord**](https://discord.gg/memgraph)
 - :ocean: [**Stack Overflow**](https://stackoverflow.com/questions/tagged/memgraphdb)
-- :busts_in_silhouette: [**Discourse forum**](https://discourse.memgraph.com/)
 - :bird: [**Twitter**](https://twitter.com/memgraphdb)
 - :movie_camera:
   [**YouTube**](https://www.youtube.com/channel/UCZ3HOJvHGxtQ_JHxOselBYg)

libs/setup.sh

@@ -210,7 +210,7 @@ pymgclient_tag="4f85c179e56302d46a1e3e2cf43509db65f062b3" # (2021-01-15)
 repo_clone_try_double "${primary_urls[pymgclient]}" "${secondary_urls[pymgclient]}" "pymgclient" "$pymgclient_tag"
 
 # mgconsole
-mgconsole_tag="v1.3.0" # (2022-11-20)
+mgconsole_tag="v1.4.0" # (2023-05-21)
 repo_clone_try_double "${primary_urls[mgconsole]}" "${secondary_urls[mgconsole]}" "mgconsole" "$mgconsole_tag" true
 
 spdlog_tag="v1.9.2" # (2021-08-12)

src/query/interpreter.cpp

@@ -3781,7 +3781,8 @@ void Interpreter::Abort() {
 
 namespace {
 void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, InterpreterContext *interpreter_context,
-                             TriggerContext trigger_context, std::atomic<TransactionStatus> *transaction_status) {
+                             TriggerContext original_trigger_context,
+                             std::atomic<TransactionStatus> *transaction_status) {
   // Run the triggers
   for (const auto &trigger : triggers.access()) {
     utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize};
@@ -3790,6 +3791,9 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
     auto storage_acc = interpreter_context->db->Access();
     DbAccessor db_accessor{storage_acc.get()};
 
+    // On-disk storage removes all Vertex/Edge Accessors because previous trigger tx finished.
+    // So we need to adapt TriggerContext based on user transaction which is still alive.
+    auto trigger_context = original_trigger_context;
     trigger_context.AdaptForAccessor(&db_accessor);
     try {
       trigger.Execute(&db_accessor, &execution_memory, interpreter_context->config.execution_timeout_sec,

src/storage/v2/config.hpp

@@ -69,6 +69,7 @@ struct Config {
 
   struct DiskConfig {
     std::filesystem::path main_storage_directory{"storage/rocksdb_main_storage"};
+    std::filesystem::path main_edge_directory{"storage/rocksdb_main_edge_storage"};
     std::filesystem::path label_index_directory{"storage/rocksdb_label_index"};
     std::filesystem::path label_property_index_directory{"storage/rocksdb_label_property_index"};
     std::filesystem::path unique_constraints_directory{"storage/rocksdb_unique_constraints"};
@@ -76,6 +77,7 @@ struct Config {
     std::filesystem::path id_name_mapper_directory{"storage/rocksdb_id_name_mapper"};
     std::filesystem::path durability_directory{"storage/rocksdb_durability"};
     std::filesystem::path wal_directory{"storage/rocksdb_wal"};
+    std::filesystem::path wal_edge_directory{"storage/rocksdb_wal_edge"};
   } disk;
 
   std::string name;

src/storage/v2/disk/label_property_index.hpp

@@ -15,7 +15,7 @@
 namespace memgraph::storage {
 
 /// TODO: andi. Too many copies, extract at one place
-using ParalellizedIndexCreationInfo =
+using ParallelizedIndexCreationInfo =
     std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
 
 class DiskLabelPropertyIndex : public storage::LabelPropertyIndex {

src/storage/v2/disk/rocksdb_storage.hpp

@@ -45,21 +45,6 @@ struct RocksDBStorage {
 
   rocksdb::Options options_;
   rocksdb::TransactionDB *db_;
-  rocksdb::ColumnFamilyHandle *vertex_chandle = nullptr;
-  rocksdb::ColumnFamilyHandle *edge_chandle = nullptr;
-  rocksdb::ColumnFamilyHandle *default_chandle = nullptr;
-
-  uint64_t ApproximateVertexCount() const {
-    uint64_t estimate_num_keys = 0;
-    db_->GetIntProperty(vertex_chandle, "rocksdb.estimate-num-keys", &estimate_num_keys);
-    return estimate_num_keys;
-  }
-
-  uint64_t ApproximateEdgeCount() const {
-    uint64_t estimate_num_keys = 0;
-    db_->GetIntProperty(edge_chandle, "rocksdb.estimate-num-keys", &estimate_num_keys);
-    return estimate_num_keys;
-  }
 };
 
 /// RocksDB comparator that compares keys with timestamps.

src/storage/v2/disk/storage.cpp

@@ -10,6 +10,7 @@
 // licenses/APL.txt.
 
 #include <limits>
+#include <memory>
 #include <optional>
 #include <stdexcept>
 #include <vector>
@@ -233,61 +234,52 @@ void DiskStorage::LoadUniqueConstraintInfoIfExists() const {
   }
 }
 
+void DiskStorage::PrepareRocksDBOptions() {
+  vertex_kvstore_->options_.create_if_missing = true;
+  vertex_kvstore_->options_.comparator = new ComparatorWithU64TsImpl();
+  vertex_kvstore_->options_.compression = rocksdb::kNoCompression;
+  vertex_kvstore_->options_.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;
+  vertex_kvstore_->options_.wal_dir = config_.disk.wal_directory;
+  vertex_kvstore_->options_.wal_compression = rocksdb::kNoCompression;
+  edge_kvstore_->options_.create_if_missing = true;
+  edge_kvstore_->options_.comparator = new ComparatorWithU64TsImpl();
+  edge_kvstore_->options_.compression = rocksdb::kNoCompression;
+  edge_kvstore_->options_.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;
+  edge_kvstore_->options_.wal_dir = config_.disk.wal_edge_directory;
+  edge_kvstore_->options_.wal_compression = rocksdb::kNoCompression;
+}
+
 DiskStorage::DiskStorage(Config config)
     : Storage(config, StorageMode::ON_DISK_TRANSACTIONAL),
-      kvstore_(std::make_unique<RocksDBStorage>()),
+      vertex_kvstore_(std::make_unique<RocksDBStorage>()),
+      edge_kvstore_(std::make_unique<RocksDBStorage>()),
       durability_kvstore_(std::make_unique<kvstore::KVStore>(config.disk.durability_directory)) {
   LoadTimestampIfExists();
   LoadIndexInfoIfExists();
   LoadConstraintsInfoIfExists();
-  kvstore_->options_.create_if_missing = true;
-  kvstore_->options_.comparator = new ComparatorWithU64TsImpl();
-  kvstore_->options_.compression = rocksdb::kNoCompression;
-  kvstore_->options_.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;
-  kvstore_->options_.wal_dir = config_.disk.wal_directory;
-  kvstore_->options_.wal_compression = rocksdb::kNoCompression;
-  std::vector<rocksdb::ColumnFamilyHandle *> column_handles;
-  std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
-  if (utils::DirExists(config.disk.main_storage_directory)) {
-    column_families.emplace_back(vertexHandle, kvstore_->options_);
-    column_families.emplace_back(edgeHandle, kvstore_->options_);
-    column_families.emplace_back(defaultHandle, kvstore_->options_);
-
-    logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(kvstore_->options_, rocksdb::TransactionDBOptions(),
-                                                              config.disk.main_storage_directory, column_families,
-                                                              &column_handles, &kvstore_->db_));
-    kvstore_->vertex_chandle = column_handles[0];
-    kvstore_->edge_chandle = column_handles[1];
-    kvstore_->default_chandle = column_handles[2];
-  } else {
-    logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(kvstore_->options_, rocksdb::TransactionDBOptions(),
-                                                              config.disk.main_storage_directory, &kvstore_->db_));
-    logging::AssertRocksDBStatus(
-        kvstore_->db_->CreateColumnFamily(kvstore_->options_, vertexHandle, &kvstore_->vertex_chandle));
-    logging::AssertRocksDBStatus(
-        kvstore_->db_->CreateColumnFamily(kvstore_->options_, edgeHandle, &kvstore_->edge_chandle));
-  }
+  PrepareRocksDBOptions();
+  logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(vertex_kvstore_->options_, rocksdb::TransactionDBOptions(),
+                                                            config.disk.main_storage_directory, &vertex_kvstore_->db_));
+  logging::AssertRocksDBStatus(rocksdb::TransactionDB::Open(edge_kvstore_->options_, rocksdb::TransactionDBOptions(),
+                                                            config.disk.main_edge_directory, &edge_kvstore_->db_));
 }
 
 DiskStorage::~DiskStorage() {
   durability_kvstore_->Put(lastTransactionStartTimeStamp, std::to_string(timestamp_));
-  logging::AssertRocksDBStatus(kvstore_->db_->DestroyColumnFamilyHandle(kvstore_->vertex_chandle));
-  logging::AssertRocksDBStatus(kvstore_->db_->DestroyColumnFamilyHandle(kvstore_->edge_chandle));
-  if (kvstore_->default_chandle) {
-    // We must destroy default column family handle only if it was read from existing database.
-    // https://github.com/facebook/rocksdb/issues/5006#issuecomment-1003154821
-    logging::AssertRocksDBStatus(kvstore_->db_->DestroyColumnFamilyHandle(kvstore_->default_chandle));
-  }
-  delete kvstore_->options_.comparator;
-  kvstore_->options_.comparator = nullptr;
+  delete vertex_kvstore_->options_.comparator;
+  vertex_kvstore_->options_.comparator = nullptr;
+  delete edge_kvstore_->options_.comparator;
+  edge_kvstore_->options_.comparator = nullptr;
 }
 
 DiskStorage::DiskAccessor::DiskAccessor(DiskStorage *storage, IsolationLevel isolation_level, StorageMode storage_mode)
     : Accessor(storage, isolation_level, storage_mode), config_(storage->config_.items) {
   rocksdb::WriteOptions write_options;
   auto txOptions = rocksdb::TransactionOptions{.set_snapshot = true};
-  disk_transaction_ = storage->kvstore_->db_->BeginTransaction(write_options, txOptions);
-  disk_transaction_->SetReadTimestampForValidation(transaction_.start_timestamp);
+  vertex_disk_transaction_ = storage->vertex_kvstore_->db_->BeginTransaction(write_options, txOptions);
+  vertex_disk_transaction_->SetReadTimestampForValidation(transaction_.start_timestamp);
+  edge_disk_transaction_ = storage->edge_kvstore_->db_->BeginTransaction(write_options, txOptions);
+  edge_disk_transaction_->SetReadTimestampForValidation(transaction_.start_timestamp);
 }
 
 DiskStorage::DiskAccessor::DiskAccessor(DiskAccessor &&other) noexcept
@@ -382,13 +374,11 @@ std::optional<EdgeAccessor> DiskStorage::DiskAccessor::DeserializeEdge(const roc
 }
 
 VerticesIterable DiskStorage::DiskAccessor::Vertices(View view) {
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
   rocksdb::ReadOptions ro;
   std::string strTs = utils::StringTimestamp(transaction_.start_timestamp);
   rocksdb::Slice ts(strTs);
   ro.timestamp = &ts;
-  auto it =
-      std::unique_ptr<rocksdb::Iterator>(disk_transaction_->GetIterator(ro, disk_storage->kvstore_->vertex_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_disk_transaction_->GetIterator(ro));
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
     LoadVertexToMainMemoryCache(it->key(), it->value());
   }
@@ -596,10 +586,7 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p
                                               &storage_->constraints_, storage_->config_.items));
 }
 
-uint64_t DiskStorage::DiskAccessor::ApproximateVertexCount() const {
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
-  return disk_storage->kvstore_->ApproximateVertexCount();
-}
+uint64_t DiskStorage::DiskAccessor::ApproximateVertexCount() const { return 0; }
 
 bool DiskStorage::PersistLabelIndexCreation(LabelId label) const {
   if (auto label_index_store = durability_kvstore_->Get(label_index_str); label_index_store.has_value()) {
@@ -693,8 +680,8 @@ uint64_t DiskStorage::GetDiskSpaceUsage() const {
 }
 
 StorageInfo DiskStorage::GetInfo() const {
-  auto vertex_count = kvstore_->ApproximateVertexCount();
-  auto edge_count = kvstore_->ApproximateEdgeCount();
+  auto vertex_count = 0U;
+  auto edge_count = 0U;
   double average_degree = 0.0;
   if (vertex_count) {
     // NOLINTNEXTLINE(bugprone-narrowing-conversions, cppcoreguidelines-narrowing-conversions)
@@ -761,9 +748,7 @@ std::optional<VertexAccessor> DiskStorage::DiskAccessor::FindVertex(storage::Gid
   auto strTs = utils::StringTimestamp(transaction_.start_timestamp);
   rocksdb::Slice ts(strTs);
   read_opts.timestamp = &ts;
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
-  auto it = std::unique_ptr<rocksdb::Iterator>(
-      disk_transaction_->GetIterator(read_opts, disk_storage->kvstore_->vertex_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_disk_transaction_->GetIterator(read_opts));
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
     const auto &key = it->key();
     if (Gid::FromUint(std::stoull(utils::ExtractGidFromKey(key.ToString()))) == gid) {
@@ -873,9 +858,7 @@ void DiskStorage::DiskAccessor::PrefetchEdges(const auto &prefetch_edge_filter)
   auto strTs = utils::StringTimestamp(transaction_.start_timestamp);
   rocksdb::Slice ts(strTs);
   read_opts.timestamp = &ts;
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
-  auto it = std::unique_ptr<rocksdb::Iterator>(
-      disk_transaction_->GetIterator(read_opts, disk_storage->kvstore_->edge_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_disk_transaction_->GetIterator(read_opts));
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
     const rocksdb::Slice &key = it->key();
     const auto edge_parts = utils::Split(key.ToStringView(), "|");
@@ -1134,9 +1117,8 @@ Result<std::optional<EdgeAccessor>> DiskStorage::DiskAccessor::DeleteEdge(EdgeAc
 /// TODO: at which storage naming
 /// TODO: this method should also delete the old key
 bool DiskStorage::DiskAccessor::WriteVertexToDisk(const Vertex &vertex) {
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
-  auto status = disk_transaction_->Put(disk_storage->kvstore_->vertex_chandle, utils::SerializeVertex(vertex),
-                                       utils::SerializeProperties(vertex.properties));
+  auto status =
+      vertex_disk_transaction_->Put(utils::SerializeVertex(vertex), utils::SerializeProperties(vertex.properties));
   if (status.ok()) {
     spdlog::debug("rocksdb: Saved vertex with key {} and ts {}", utils::SerializeVertex(vertex), *commit_timestamp_);
   } else if (status.IsBusy()) {
@@ -1153,13 +1135,11 @@ bool DiskStorage::DiskAccessor::WriteVertexToDisk(const Vertex &vertex) {
 
 /// TODO: at which storage naming
 bool DiskStorage::DiskAccessor::WriteEdgeToDisk(const EdgeRef edge, const std::string &serializedEdgeKey) {
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
   rocksdb::Status status;
   if (config_.properties_on_edges) {
-    status = disk_transaction_->Put(disk_storage->kvstore_->edge_chandle, serializedEdgeKey,
-                                    utils::SerializeProperties(edge.ptr->properties));
+    status = edge_disk_transaction_->Put(serializedEdgeKey, utils::SerializeProperties(edge.ptr->properties));
   } else {
-    status = disk_transaction_->Put(disk_storage->kvstore_->edge_chandle, serializedEdgeKey, "");
+    status = edge_disk_transaction_->Put(serializedEdgeKey, "");
   }
   if (status.ok()) {
     spdlog::debug("rocksdb: Saved edge with key {} and ts {}", serializedEdgeKey, *commit_timestamp_);
@@ -1175,8 +1155,7 @@ bool DiskStorage::DiskAccessor::WriteEdgeToDisk(const EdgeRef edge, const std::s
 }
 
 bool DiskStorage::DiskAccessor::DeleteVertexFromDisk(const std::string &vertex) {
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
-  auto status = disk_transaction_->Delete(disk_storage->kvstore_->vertex_chandle, vertex);
+  auto status = vertex_disk_transaction_->Delete(vertex);
   if (status.ok()) {
     spdlog::debug("rocksdb: Deleted vertex with key {}", vertex);
   } else if (status.IsBusy()) {
@@ -1190,8 +1169,7 @@ bool DiskStorage::DiskAccessor::DeleteVertexFromDisk(const std::string &vertex)
 }
 
 bool DiskStorage::DiskAccessor::DeleteEdgeFromDisk(const std::string &edge) {
-  auto *disk_storage = static_cast<DiskStorage *>(storage_);
-  auto status = disk_transaction_->Delete(disk_storage->kvstore_->edge_chandle, edge);
+  auto status = edge_disk_transaction_->Delete(edge);
   if (status.ok()) {
     spdlog::debug("rocksdb: Deleted edge with key {}", edge);
   } else if (status.IsBusy()) {
@@ -1366,7 +1344,7 @@ DiskStorage::DiskAccessor::CheckVertexConstraintsBeforeCommit(
   std::string strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
   rocksdb::Slice ts(strTs);
   ro.timestamp = &ts;
-  auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
     std::vector<LabelId> labels = utils::DeserializeLabelsFromMainDiskStorage(it->key().ToString());
     PropertyStore properties = utils::DeserializePropertiesFromMainDiskStorage(it->value().ToStringView());
@@ -1387,7 +1365,7 @@ DiskStorage::CheckExistingVerticesBeforeCreatingUniqueConstraint(LabelId label,
   std::string strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
   rocksdb::Slice ts(strTs);
   ro.timestamp = &ts;
-  auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
     const std::string key_str = it->key().ToString();
     std::vector<LabelId> labels = utils::DeserializeLabelsFromMainDiskStorage(key_str);
@@ -1435,13 +1413,21 @@ utils::BasicResult<StorageDataManipulationError, void> DiskStorage::DiskAccessor
 
   if (commit_timestamp_) {
     // commit_timestamp_ is set only if the transaction has writes.
-    logging::AssertRocksDBStatus(disk_transaction_->SetCommitTimestamp(*commit_timestamp_));
+    logging::AssertRocksDBStatus(vertex_disk_transaction_->SetCommitTimestamp(*commit_timestamp_));
+    logging::AssertRocksDBStatus(edge_disk_transaction_->SetCommitTimestamp(*commit_timestamp_));
   }
-  auto commitStatus = disk_transaction_->Commit();
-  delete disk_transaction_;
-  disk_transaction_ = nullptr;
-  if (!commitStatus.ok()) {
-    spdlog::error("rocksdb: Commit failed with status {}", commitStatus.ToString());
+  auto vertexCommitStatus = vertex_disk_transaction_->Commit();
+  delete vertex_disk_transaction_;
+  vertex_disk_transaction_ = nullptr;
+  if (!vertexCommitStatus.ok()) {
+    spdlog::error("rocksdb: Vertex commit failed with status {}", vertexCommitStatus.ToString());
+    return StorageDataManipulationError{SerializationError{}};
+  }
+  auto edgeCommitStatus = edge_disk_transaction_->Commit();
+  delete edge_disk_transaction_;
+  edge_disk_transaction_ = nullptr;
+  if (!edgeCommitStatus.ok()) {
+    spdlog::error("rocksdb: Commit failed with status {}", edgeCommitStatus.ToString());
     return StorageDataManipulationError{SerializationError{}};
   }
   spdlog::debug("rocksdb: Commit successful");
@@ -1458,7 +1444,7 @@ std::vector<std::pair<std::string, std::string>> DiskStorage::SerializeVerticesF
   auto strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
   rocksdb::Slice ts(strTs);
   ro.timestamp = &ts;
-  auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
 
   const std::string serialized_label = utils::SerializeIdType(label);
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
@@ -1484,7 +1470,7 @@ std::vector<std::pair<std::string, std::string>> DiskStorage::SerializeVerticesF
   auto strTs = utils::StringTimestamp(std::numeric_limits<uint64_t>::max());
   rocksdb::Slice ts(strTs);
   ro.timestamp = &ts;
-  auto it = std::unique_ptr<rocksdb::Iterator>(kvstore_->db_->NewIterator(ro, kvstore_->vertex_chandle));
+  auto it = std::unique_ptr<rocksdb::Iterator>(vertex_kvstore_->db_->NewIterator(ro));
 
   const std::string serialized_label = utils::SerializeIdType(label);
   for (it->SeekToFirst(); it->Valid(); it->Next()) {
@@ -1510,10 +1496,15 @@ void DiskStorage::DiskAccessor::Abort() {
   // disk_transaction correctly in destructor.
   // This happens in tests when we create and remove storage in one test. For example, in
   // query_plan_accumulate_aggregate.cpp
-  disk_transaction_->Rollback();
-  disk_transaction_->ClearSnapshot();
-  delete disk_transaction_;
-  disk_transaction_ = nullptr;
+  vertex_disk_transaction_->Rollback();
+  vertex_disk_transaction_->ClearSnapshot();
+  delete vertex_disk_transaction_;
+  vertex_disk_transaction_ = nullptr;
+
+  edge_disk_transaction_->Rollback();
+  edge_disk_transaction_->ClearSnapshot();
+  delete edge_disk_transaction_;
+  edge_disk_transaction_ = nullptr;
 
   is_transaction_active_ = false;
 }

src/storage/v2/disk/storage.hpp

@@ -208,7 +208,8 @@ class DiskStorage final : public Storage {
     Config::Items config_;
     std::vector<std::string> edges_to_delete_;
     std::vector<std::pair<std::string, std::string>> vertices_to_delete_;
-    rocksdb::Transaction *disk_transaction_;
+    rocksdb::Transaction *vertex_disk_transaction_;
+    rocksdb::Transaction *edge_disk_transaction_;
   };
 
   std::unique_ptr<Storage::Accessor> Access(std::optional<IsolationLevel> override_isolation_level) override {
@@ -219,7 +220,9 @@ class DiskStorage final : public Storage {
     return std::unique_ptr<DiskAccessor>(new DiskAccessor{this, isolation_level, storage_mode_});
   }
 
-  RocksDBStorage *GetRocksDBStorage() const { return kvstore_.get(); }
+  void PrepareRocksDBOptions();
+
+  RocksDBStorage *GetRocksDBStorage() const { return vertex_kvstore_.get(); }
 
   utils::BasicResult<StorageIndexDefinitionError, void> CreateIndex(
       LabelId label, std::optional<uint64_t> desired_commit_timestamp) override;
@@ -296,7 +299,8 @@ class DiskStorage final : public Storage {
 
   uint64_t CommitTimestamp(std::optional<uint64_t> desired_commit_timestamp = {});
 
-  std::unique_ptr<RocksDBStorage> kvstore_;
+  std::unique_ptr<RocksDBStorage> vertex_kvstore_;
+  std::unique_ptr<RocksDBStorage> edge_kvstore_;
   std::unique_ptr<kvstore::KVStore> durability_kvstore_;
 };
 

src/storage/v2/durability/durability.cpp

@@ -127,13 +127,13 @@ std::optional<std::vector<WalDurabilityInfo>> GetWalFiles(const std::filesystem:
 // recovery process.
 void RecoverIndicesAndConstraints(const RecoveredIndicesAndConstraints &indices_constraints, Indices *indices,
                                   Constraints *constraints, utils::SkipList<Vertex> *vertices,
-                                  const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info) {
+                                  const std::optional<ParallelizedIndexCreationInfo> &parallel_exec_info) {
   spdlog::info("Recreating indices from metadata.");
   // Recover label indices.
   spdlog::info("Recreating {} label indices from metadata.", indices_constraints.indices.label.size());
   for (const auto &item : indices_constraints.indices.label) {
     auto *mem_label_index = static_cast<InMemoryLabelIndex *>(indices->label_index_.get());
-    if (!mem_label_index->CreateIndex(item, vertices->access(), paralell_exec_info))
+    if (!mem_label_index->CreateIndex(item, vertices->access(), parallel_exec_info))
       throw RecoveryFailure("The label index must be created here!");
 
     spdlog::info("A label index is recreated from metadata.");
@@ -145,7 +145,7 @@ void RecoverIndicesAndConstraints(const RecoveredIndicesAndConstraints &indices_
                indices_constraints.indices.label_property.size());
   auto *mem_label_property_index = static_cast<InMemoryLabelPropertyIndex *>(indices->label_property_index_.get());
   for (const auto &item : indices_constraints.indices.label_property) {
-    if (!mem_label_property_index->CreateIndex(item.first, item.second, vertices->access(), std::nullopt))
+    if (!mem_label_property_index->CreateIndex(item.first, item.second, vertices->access(), parallel_exec_info))
       throw RecoveryFailure("The label+property index must be created here!");
     spdlog::info("A label+property index is recreated from metadata.");
   }

src/storage/v2/durability/durability.hpp

@@ -91,7 +91,7 @@ std::optional<std::vector<WalDurabilityInfo>> GetWalFiles(const std::filesystem:
                                                           std::string_view uuid = "",
                                                           std::optional<size_t> current_seq_num = {});
 
-using ParalellizedIndexCreationInfo =
+using ParallelizedIndexCreationInfo =
     std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
 
 // Helper function used to recover all discovered indices and constraints. The
@@ -102,7 +102,7 @@ using ParalellizedIndexCreationInfo =
 void RecoverIndicesAndConstraints(
     const RecoveredIndicesAndConstraints &indices_constraints, Indices *indices, Constraints *constraints,
     utils::SkipList<Vertex> *vertices,
-    const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info = std::nullopt);
+    const std::optional<ParallelizedIndexCreationInfo> &parallel_exec_info = std::nullopt);
 
 /// Recovers data either from a snapshot and/or WAL files.
 /// @throw RecoveryFailure

src/storage/v2/inmemory/indices_utils.hpp

@@ -19,7 +19,7 @@
 
 namespace memgraph::storage {
 
-using ParalellizedIndexCreationInfo =
+using ParallelizedIndexCreationInfo =
     std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
 
 /// Traverses deltas visible from transaction with start timestamp greater than
@@ -306,11 +306,11 @@ inline void CreateIndexOnSingleThread(utils::SkipList<Vertex>::Accessor &vertice
 template <typename TIndex, typename TIndexKey, typename TSKiplistIter, typename TFunc>
 inline void CreateIndexOnMultipleThreads(utils::SkipList<Vertex>::Accessor &vertices, TSKiplistIter skiplist_iter,
                                          TIndex &index, TIndexKey key,
-                                         const ParalellizedIndexCreationInfo &paralell_exec_info, const TFunc &func) {
+                                         const ParallelizedIndexCreationInfo &parallel_exec_info, const TFunc &func) {
   utils::MemoryTracker::OutOfMemoryExceptionEnabler oom_exception;
 
-  const auto &vertex_batches = paralell_exec_info.first;
-  const auto thread_count = std::min(paralell_exec_info.second, vertex_batches.size());
+  const auto &vertex_batches = parallel_exec_info.first;
+  const auto thread_count = std::min(parallel_exec_info.second, vertex_batches.size());
 
   MG_ASSERT(!vertex_batches.empty(),
             "The size of batches should always be greater than zero if you want to use the parallel version of index "

src/storage/v2/inmemory/label_index.cpp

@@ -25,7 +25,7 @@ void InMemoryLabelIndex::UpdateOnAddLabel(LabelId added_label, Vertex *vertex_af
 }
 
 bool InMemoryLabelIndex::CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices,
-                                     const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info) {
+                                     const std::optional<ParallelizedIndexCreationInfo> &parallel_exec_info) {
   const auto create_index_seq = [this](LabelId label, utils::SkipList<Vertex>::Accessor &vertices,
                                        std::map<LabelId, utils::SkipList<Entry>>::iterator it) {
     using IndexAccessor = decltype(it->second.access());
@@ -40,10 +40,10 @@ bool InMemoryLabelIndex::CreateIndex(LabelId label, utils::SkipList<Vertex>::Acc
 
   const auto create_index_par = [this](LabelId label, utils::SkipList<Vertex>::Accessor &vertices,
                                        std::map<LabelId, utils::SkipList<Entry>>::iterator label_it,
-                                       const ParalellizedIndexCreationInfo &paralell_exec_info) {
+                                       const ParallelizedIndexCreationInfo &parallel_exec_info) {
     using IndexAccessor = decltype(label_it->second.access());
 
-    CreateIndexOnMultipleThreads(vertices, label_it, index_, label, paralell_exec_info,
+    CreateIndexOnMultipleThreads(vertices, label_it, index_, label, parallel_exec_info,
                                  [](Vertex &vertex, LabelId label, IndexAccessor &index_accessor) {
                                    TryInsertLabelIndex(vertex, label, index_accessor);
                                  });
@@ -57,8 +57,8 @@ bool InMemoryLabelIndex::CreateIndex(LabelId label, utils::SkipList<Vertex>::Acc
     return false;
   }
 
-  if (paralell_exec_info) {
-    return create_index_par(label, vertices, it, *paralell_exec_info);
+  if (parallel_exec_info) {
+    return create_index_par(label, vertices, it, *parallel_exec_info);
   }
   return create_index_seq(label, vertices, it);
 }

src/storage/v2/inmemory/label_index.hpp

@@ -21,7 +21,7 @@ struct LabelIndexStats {
   double avg_degree;
 };
 
-using ParalellizedIndexCreationInfo =
+using ParallelizedIndexCreationInfo =
     std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
 
 class InMemoryLabelIndex : public storage::LabelIndex {
@@ -46,7 +46,7 @@ class InMemoryLabelIndex : public storage::LabelIndex {
 
   /// @throw std::bad_alloc
   bool CreateIndex(LabelId label, utils::SkipList<Vertex>::Accessor vertices,
-                   const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info);
+                   const std::optional<ParallelizedIndexCreationInfo> &parallel_exec_info);
 
   /// Returns false if there was no index to drop
   bool DropIndex(LabelId label) override;

src/storage/v2/inmemory/label_property_index.cpp

@@ -37,7 +37,7 @@ InMemoryLabelPropertyIndex::InMemoryLabelPropertyIndex(Indices *indices, Constra
 
 bool InMemoryLabelPropertyIndex::CreateIndex(LabelId label, PropertyId property,
                                              utils::SkipList<Vertex>::Accessor vertices,
-                                             const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info) {
+                                             const std::optional<ParallelizedIndexCreationInfo> &parallel_exec_info) {
   auto create_index_seq = [this](LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor &vertices,
                                  std::map<std::pair<LabelId, PropertyId>, utils::SkipList<Entry>>::iterator it) {
     using IndexAccessor = decltype(it->second.access());
@@ -53,11 +53,11 @@ bool InMemoryLabelPropertyIndex::CreateIndex(LabelId label, PropertyId property,
   auto create_index_par =
       [this](LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor &vertices,
              std::map<std::pair<LabelId, PropertyId>, utils::SkipList<Entry>>::iterator label_property_it,
-             const ParalellizedIndexCreationInfo &paralell_exec_info) {
+             const ParallelizedIndexCreationInfo &parallel_exec_info) {
         using IndexAccessor = decltype(label_property_it->second.access());
 
         CreateIndexOnMultipleThreads(
-            vertices, label_property_it, index_, std::make_pair(label, property), paralell_exec_info,
+            vertices, label_property_it, index_, std::make_pair(label, property), parallel_exec_info,
             [](Vertex &vertex, std::pair<LabelId, PropertyId> key, IndexAccessor &index_accessor) {
               TryInsertLabelPropertyIndex(vertex, key, index_accessor);
             });
@@ -72,8 +72,8 @@ bool InMemoryLabelPropertyIndex::CreateIndex(LabelId label, PropertyId property,
     return false;
   }
 
-  if (paralell_exec_info) {
-    return create_index_par(label, property, vertices, it, *paralell_exec_info);
+  if (parallel_exec_info) {
+    return create_index_par(label, property, vertices, it, *parallel_exec_info);
   }
   return create_index_seq(label, property, vertices, it);
 }

src/storage/v2/inmemory/label_property_index.hpp

@@ -21,7 +21,7 @@ struct LabelPropertyIndexStats {
 };
 
 /// TODO: andi. Too many copies, extract at one place
-using ParalellizedIndexCreationInfo =
+using ParallelizedIndexCreationInfo =
     std::pair<std::vector<std::pair<Gid, uint64_t>> /*vertex_recovery_info*/, uint64_t /*thread_count*/>;
 
 class InMemoryLabelPropertyIndex : public storage::LabelPropertyIndex {
@@ -43,7 +43,7 @@ class InMemoryLabelPropertyIndex : public storage::LabelPropertyIndex {
 
   /// @throw std::bad_alloc
   bool CreateIndex(LabelId label, PropertyId property, utils::SkipList<Vertex>::Accessor vertices,
-                   const std::optional<ParalellizedIndexCreationInfo> &paralell_exec_info);
+                   const std::optional<ParallelizedIndexCreationInfo> &parallel_exec_info);
 
   /// @throw std::bad_alloc
   void UpdateOnAddLabel(LabelId added_label, Vertex *vertex_after_update, const Transaction &tx) override;

tests/e2e/triggers/workloads.yaml

@@ -17,14 +17,14 @@ disk_template_cluster: &disk_template_cluster
   cluster:
     main:
       args: ["--bolt-port", *bolt_port, "--log-level=TRACE", "--storage-properties-on-edges=True"]
-      log_file: "triggers-e2e.log"
-      setup_queries: []
+      log_file: "triggers-e2e-disk.log"
+      setup_queries: ["storage mode on_disk_transactional"]
       validation_queries: []
 disk_storage_properties_edges_false: &disk_storage_properties_edges_false
   cluster:
     main:
       args: ["--bolt-port", *bolt_port, "--log-level=TRACE", "--also-log-to-stderr", "--storage-properties-on-edges=False"]
-      log_file: "triggers-e2e.log"
+      log_file: "triggers-e2e-disk.log"
       setup_queries: []
       validation_queries: []
 

tests/unit/disk_test_utils.hpp

@@ -20,17 +20,20 @@ namespace disk_test_utils {
 
 memgraph::storage::Config GenerateOnDiskConfig(const std::string &testName) {
   return {.disk = {.main_storage_directory = "rocksdb_" + testName + "_db",
+                   .main_edge_directory = "rocksdb_" + testName + "_edge_db",
                    .label_index_directory = "rocksdb_" + testName + "_label_index",
                    .label_property_index_directory = "rocksdb_" + testName + "_label_property_index",
                    .unique_constraints_directory = "rocksdb_" + testName + "_unique_constraints",
                    .name_id_mapper_directory = "rocksdb_" + testName + "_name_id_mapper",
                    .id_name_mapper_directory = "rocksdb_" + testName + "_id_name_mapper",
                    .durability_directory = "rocksdb_" + testName + "_durability",
-                   .wal_directory = "rocksdb_" + testName + "_wal"}};
+                   .wal_directory = "rocksdb_" + testName + "_wal",
+                   .wal_edge_directory = "rocksdb_" + testName + "_edge_wal"}};
 }
 
 void RemoveRocksDbDirs(const std::string &testName) {
   std::filesystem::remove_all("rocksdb_" + testName + "_db");
+  std::filesystem::remove_all("rocksdb_" + testName + "_edge_db");
   std::filesystem::remove_all("rocksdb_" + testName + "_label_index");
   std::filesystem::remove_all("rocksdb_" + testName + "_label_property_index");
   std::filesystem::remove_all("rocksdb_" + testName + "_unique_constraints");
@@ -38,6 +41,7 @@ void RemoveRocksDbDirs(const std::string &testName) {
   std::filesystem::remove_all("rocksdb_" + testName + "_id_name_mapper");
   std::filesystem::remove_all("rocksdb_" + testName + "_durability");
   std::filesystem::remove_all("rocksdb_" + testName + "_wal");
+  std::filesystem::remove_all("rocksdb_" + testName + "_edge_wal");
 }
 
 uint64_t GetRealNumberOfEntriesInRocksDB(rocksdb::TransactionDB *disk_storage) {