tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fixed imports and edge_properties in disk storage

Browse Source
This commit is contained in:
Andi Skrgat
2023-05-17 13:25:38 +02:00
parent c8d7bb743d
commit a015df3b93
7 changed files with 26 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/memgraph.cpp
View File

@@ -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<memgraph::storage::Storage>(new memgraph::storage::InMemoryStorage(db_config));
auto db = std::unique_ptr<memgraph::storage::Storage>(new memgraph::storage::DiskStorage(db_config));
auto db = std::unique_ptr<memgraph::storage::Storage>(new memgraph::storage::InMemoryStorage(db_config));
// auto db = std::unique_ptr<memgraph::storage::Storage>(new memgraph::storage::DiskStorage(db_config));
memgraph::query::InterpreterContext interpreter_context{
db.get(),

1
src/mg_import_csv.cpp
View File

@@ -20,6 +20,7 @@
#include <unordered_map>
#include "helpers.hpp"
#include "storage/v2/edge_accessor.hpp"
#include "storage/v2/inmemory/storage.hpp"
#include "utils/exceptions.hpp"
#include "utils/logging.hpp"

48
src/storage/v2/disk/storage.cpp
View File

@@ -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<EdgeAccessor> 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<DiskEdgeAccessor *>(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<DiskStorage *>(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<int64_t>(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<EdgeAccessor> 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<EdgeAccessor> 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<EdgeAccessor> 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<std::optional<EdgeAccessor>> 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<StorageDataManipulationError, void> 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

2
src/storage/v2/disk/storage.hpp
View File

@@ -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<EdgeAccessor> 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.

1
src/storage/v2/inmemory/storage.cpp
View File

@@ -393,7 +393,6 @@ Result<EdgeAccessor> 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!");

1
src/storage/v2/storage.hpp
View File

@@ -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"

2
tests/unit/storage_v2.cpp
View File

@@ -33,7 +33,7 @@ class StorageV2Test : public testing::Test {
};
// using StorageTypes = ::testing::Types<memgraph::storage::InMemoryStorage, memgraph::storage::DiskStorage>;
using StorageTypes = ::testing::Types<memgraph::storage::InMemoryStorage>;
using StorageTypes = ::testing::Types<memgraph::storage::DiskStorage>;
TYPED_TEST_CASE(StorageV2Test, StorageTypes);
// NOLINTNEXTLINE(hicpp-special-member-functions)