From a015df3b933895991aaeb175711c5b435eb48ae1 Mon Sep 17 00:00:00 2001 From: Andi Skrgat Date: Wed, 17 May 2023 13:25:38 +0200 Subject: [PATCH] Fixed imports and edge_properties in disk storage --- src/memgraph.cpp | 4 +-- src/mg_import_csv.cpp | 1 + src/storage/v2/disk/storage.cpp | 48 ++++++++++++----------------- src/storage/v2/disk/storage.hpp | 2 +- src/storage/v2/inmemory/storage.cpp | 1 - src/storage/v2/storage.hpp | 1 + tests/unit/storage_v2.cpp | 2 +- 7 files changed, 26 insertions(+), 33 deletions(-) diff --git a/src/memgraph.cpp b/src/memgraph.cpp index f343927e1..3ac0287a2 100644 --- a/src/memgraph.cpp +++ b/src/memgraph.cpp @@ -893,8 +893,8 @@ int main(int argc, char **argv) { } db_config.durability.snapshot_interval = std::chrono::seconds(FLAGS_storage_snapshot_interval_sec); } - // auto db = std::unique_ptr(new memgraph::storage::InMemoryStorage(db_config)); - auto db = std::unique_ptr(new memgraph::storage::DiskStorage(db_config)); + auto db = std::unique_ptr(new memgraph::storage::InMemoryStorage(db_config)); + // auto db = std::unique_ptr(new memgraph::storage::DiskStorage(db_config)); memgraph::query::InterpreterContext interpreter_context{ db.get(), diff --git a/src/mg_import_csv.cpp b/src/mg_import_csv.cpp index 5929cb8de..63def9089 100644 --- a/src/mg_import_csv.cpp +++ b/src/mg_import_csv.cpp @@ -20,6 +20,7 @@ #include #include "helpers.hpp" +#include "storage/v2/edge_accessor.hpp" #include "storage/v2/inmemory/storage.hpp" #include "utils/exceptions.hpp" #include "utils/logging.hpp" diff --git a/src/storage/v2/disk/storage.cpp b/src/storage/v2/disk/storage.cpp index 7f90979aa..6396b9048 100644 --- a/src/storage/v2/disk/storage.cpp +++ b/src/storage/v2/disk/storage.cpp @@ -33,7 +33,6 @@ inline constexpr uint16_t kEpochHistoryRetention = 1000; } // namespace -/// TODO: indices should be initialized at some point after DiskStorage::DiskStorage(Config config) : Storage(config), indices_(&constraints_, config.items), @@ -250,11 +249,10 @@ std::optional DiskStorage::DiskAccessor::DeserializeEdge(const roc throw utils::BasicException("Non-existing vertices found during edge deserialization"); } const auto edge_type_id = storage::EdgeTypeId::FromUint(std::stoull(edge_parts[3])); - auto maybe_edge = - CreateEdge(&*from_acc, &*to_acc, edge_type_id, edge_gid, utils::ExtractTimestampFromDeserializedUserKey(key)); + auto maybe_edge = CreateEdge(&*from_acc, &*to_acc, edge_type_id, edge_gid, + utils::ExtractTimestampFromDeserializedUserKey(key), value.ToStringView()); MG_ASSERT(maybe_edge.HasValue()); - // TODO(andi): Initialize deserialized edge - // static_cast(maybe_edge->get())->InitializeDeserializedEdge(edge_type_id, value.ToStringView()); + return *maybe_edge; } @@ -300,7 +298,7 @@ VerticesIterable DiskStorage::DiskAccessor::Vertices(LabelId label, PropertyId p int64_t DiskStorage::DiskAccessor::ApproximateVertexCount() const { uint64_t estimate_num_keys = 0; auto *disk_storage = static_cast(storage_); - // TODO: This method should probably be organized. + // TODO: This method should probably be organized in a better way. disk_storage->kvstore_->db_->GetIntProperty(disk_storage->kvstore_->vertex_chandle, "rocksdb.estimate-num-keys", &estimate_num_keys); return static_cast(estimate_num_keys); @@ -523,7 +521,6 @@ void DiskStorage::DiskAccessor::PrefetchEdges(const auto &prefetch_edge_filter) } } -/// TOOD(andi): Add support for fetching in edges only for one vertex. Currently, all in edges are fetched. void DiskStorage::DiskAccessor::PrefetchInEdges(const VertexAccessor &vertex_acc) { PrefetchEdges([&vertex_acc](const std::string_view disk_edge_gid, const std::string_view disk_edge_direction) -> bool { @@ -531,7 +528,6 @@ void DiskStorage::DiskAccessor::PrefetchInEdges(const VertexAccessor &vertex_acc }); } -/// TODO(andi): Functionality of this method can probably be merged with the functionality of in-edges void DiskStorage::DiskAccessor::PrefetchOutEdges(const VertexAccessor &vertex_acc) { PrefetchEdges([&vertex_acc](const std::string_view disk_edge_gid, const std::string_view disk_edge_direction) -> bool { @@ -541,7 +537,7 @@ void DiskStorage::DiskAccessor::PrefetchOutEdges(const VertexAccessor &vertex_ac Result DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from, VertexAccessor *to, EdgeTypeId edge_type, storage::Gid gid, - uint64_t edge_commit_ts) { + uint64_t edge_commit_ts, std::string_view properties) { OOMExceptionEnabler oom_exception; MG_ASSERT(from->transaction_ == to->transaction_, "VertexAccessors must be from the same transaction when creating " @@ -584,17 +580,20 @@ Result DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from, storage_->edge_id_.store(std::max(storage_->edge_id_.load(std::memory_order_acquire), gid.AsUint() + 1), std::memory_order_release); - /// TODO(andi): Remove this once we add full support for edges. - MG_ASSERT(config_.properties_on_edges, "Properties on edges must be enabled currently for Disk version!"); - // if (config_.properties_on_edges) { - auto acc = storage_->edges_.access(); - auto delta = CreateDeleteDeserializedObjectDelta(&transaction_, edge_commit_ts); - auto [it, inserted] = acc.insert(Edge(gid, delta)); - MG_ASSERT(inserted, "The edge must be inserted here!"); - MG_ASSERT(it != acc.end(), "Invalid Edge accessor!"); - auto edge = EdgeRef(&*it); - delta->prev.Set(&*it); - // } + EdgeRef edge(gid); + if (config_.properties_on_edges) { + auto acc = storage_->edges_.access(); + auto *delta = CreateDeleteDeserializedObjectDelta(&transaction_, edge_commit_ts); + auto [it, inserted] = acc.insert(Edge(gid, delta)); + MG_ASSERT(inserted, "The edge must be inserted here!"); + MG_ASSERT(it != acc.end(), "Invalid Edge accessor!"); + edge = EdgeRef(&*it); + if (delta) { + delta->prev.Set(&*it); + } + edge.ptr->properties.SetBuffer(properties); + } + from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge); to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge); @@ -653,7 +652,7 @@ Result DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from, EdgeRef edge(gid); if (config_.properties_on_edges) { auto acc = storage_->edges_.access(); - auto delta = CreateDeleteObjectDelta(&transaction_); + auto *delta = CreateDeleteObjectDelta(&transaction_); auto [it, inserted] = acc.insert(Edge(gid, delta)); MG_ASSERT(inserted, "The edge must be inserted here!"); MG_ASSERT(it != acc.end(), "Invalid Edge accessor!"); @@ -812,7 +811,6 @@ Result> DiskStorage::DiskAccessor::DeleteEdge(EdgeAc CreateAndLinkDelta(&transaction_, edge_ptr, Delta::RecreateObjectTag()); edge_ptr->deleted = true; } - // TODO(andi): How to know that the edge is deleted if properties_on_edges are disabled? CreateAndLinkDelta(&transaction_, from_vertex, Delta::AddOutEdgeTag(), edge_type, to_vertex, edge_ref); CreateAndLinkDelta(&transaction_, to_vertex, Delta::AddInEdgeTag(), edge_type, from_vertex, edge_ref); @@ -1000,7 +998,6 @@ utils::BasicResult DiskStorage::DiskAccessor return {}; } -/// TODO(andi): I think we should have one Abort method per storage. void DiskStorage::DiskAccessor::Abort() { MG_ASSERT(is_transaction_active_, "The transaction is already terminated!"); @@ -1543,11 +1540,6 @@ void DiskStorage::CollectGarbage() { MG_ASSERT(edge_acc.remove(edge), "Invalid database state!"); } } - - // TODO(andi): Remove this after we are assured that deserialization works - edges_.clear(); - vertices_.clear(); - spdlog::debug("Cleared caches"); } // tell the linker he can find the CollectGarbage definitions here diff --git a/src/storage/v2/disk/storage.hpp b/src/storage/v2/disk/storage.hpp index 92ff269cb..ca74d85de 100644 --- a/src/storage/v2/disk/storage.hpp +++ b/src/storage/v2/disk/storage.hpp @@ -176,7 +176,7 @@ class DiskStorage final : public Storage { /// TODO(andi): Consolidate this vertex creation methods and find from in-memory version where are they used. Result CreateEdge(VertexAccessor *from, VertexAccessor *to, EdgeTypeId edge_type, storage::Gid gid, - uint64_t edge_commit_ts); + uint64_t edge_commit_ts, std::string_view properties); /// Flushes vertices and edges to the disk with the commit timestamp. /// At the time of calling, the commit_timestamp_ must already exist. diff --git a/src/storage/v2/inmemory/storage.cpp b/src/storage/v2/inmemory/storage.cpp index 9fdfef3f8..230341191 100644 --- a/src/storage/v2/inmemory/storage.cpp +++ b/src/storage/v2/inmemory/storage.cpp @@ -393,7 +393,6 @@ Result InMemoryStorage::InMemoryAccessor::CreateEdge(VertexAccesso EdgeRef edge(gid); if (config_.properties_on_edges) { auto acc = storage_->edges_.access(); - auto *delta = CreateDeleteObjectDelta(&transaction_); auto [it, inserted] = acc.insert(Edge(gid, delta)); MG_ASSERT(inserted, "The edge must be inserted here!"); diff --git a/src/storage/v2/storage.hpp b/src/storage/v2/storage.hpp index d9ab19018..31e1dd688 100644 --- a/src/storage/v2/storage.hpp +++ b/src/storage/v2/storage.hpp @@ -19,6 +19,7 @@ #include "storage/v2/config.hpp" #include "storage/v2/durability/paths.hpp" #include "storage/v2/durability/wal.hpp" +#include "storage/v2/edge_accessor.hpp" #include "storage/v2/indices.hpp" #include "storage/v2/mvcc.hpp" #include "storage/v2/replication/config.hpp" diff --git a/tests/unit/storage_v2.cpp b/tests/unit/storage_v2.cpp index d91b7e82c..f26eff948 100644 --- a/tests/unit/storage_v2.cpp +++ b/tests/unit/storage_v2.cpp @@ -33,7 +33,7 @@ class StorageV2Test : public testing::Test { }; // using StorageTypes = ::testing::Types; -using StorageTypes = ::testing::Types; +using StorageTypes = ::testing::Types; TYPED_TEST_CASE(StorageV2Test, StorageTypes); // NOLINTNEXTLINE(hicpp-special-member-functions)