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

Add experimental/v1 of ON_DISK_TRANSACTIONAL storage (#850)

Browse Source 
Co-authored-by: Andi Skrgat <andi8647@gmail.com>
Co-authored-by: Aidar Samerkhanov <aidar.samerkhanov@memgraph.io>
This commit is contained in:
Marko Budiselić
2023-06-29 11:44:55 +02:00
committed by GitHub
parent aa4f68a37d
commit 9d056e7649
182 changed files with 26406 additions and 13468 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/mg_import_csv.cpp
View File

@@ -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
@@ -20,7 +20,8 @@
#include <unordered_map>
#include "helpers.hpp"
#include "storage/v2/storage.hpp"
#include "storage/v2/edge_accessor.hpp"
#include "storage/v2/inmemory/storage.hpp"
#include "utils/exceptions.hpp"
#include "utils/logging.hpp"
#include "utils/message.hpp"
@@ -421,7 +422,7 @@ void ProcessNodeRow(memgraph::storage::Storage *store, const std::vector<std::st
std::unordered_map<NodeId, memgraph::storage::Gid> *node_id_map) {
std::optional<NodeId> id;
auto acc = store->Access();
auto node = acc.CreateVertex();
auto node = acc->CreateVertex();
for (size_t i = 0; i < row.size(); ++i) {
const auto &field = fields[i];
const auto &value = row[i];
@@ -450,29 +451,29 @@ void ProcessNodeRow(memgraph::storage::Storage *store, const std::vector<std::st
} else {
pv_id = memgraph::storage::PropertyValue(node_id.id);
}
auto old_node_property = node.SetProperty(acc.NameToProperty(field.name), pv_id);
auto old_node_property = node.SetProperty(acc->NameToProperty(field.name), pv_id);
if (!old_node_property.HasValue()) throw LoadException("Couldn't add property '{}' to the node", field.name);
if (!old_node_property->IsNull()) throw LoadException("The property '{}' already exists", field.name);
}
id = node_id;
} else if (field.type == "LABEL") {
for (const auto &label : memgraph::utils::Split(value, FLAGS_array_delimiter)) {
auto node_label = node.AddLabel(acc.NameToLabel(label));
auto node_label = node.AddLabel(acc->NameToLabel(label));
if (!node_label.HasValue()) throw LoadException("Couldn't add label '{}' to the node", label);
if (!*node_label) throw LoadException("The label '{}' already exists", label);
}
} else if (field.type != "IGNORE") {
auto old_node_property = node.SetProperty(acc.NameToProperty(field.name), StringToValue(value, field.type));
auto old_node_property = node.SetProperty(acc->NameToProperty(field.name), StringToValue(value, field.type));
if (!old_node_property.HasValue()) throw LoadException("Couldn't add property '{}' to the node", field.name);
if (!old_node_property->IsNull()) throw LoadException("The property '{}' already exists", field.name);
}
}
for (const auto &label : additional_labels) {
auto node_label = node.AddLabel(acc.NameToLabel(label));
auto node_label = node.AddLabel(acc->NameToLabel(label));
if (!node_label.HasValue()) throw LoadException("Couldn't add label '{}' to the node", label);
if (!*node_label) throw LoadException("The label '{}' already exists", label);
}
if (acc.Commit().HasError()) throw LoadException("Couldn't store the node");
if (acc->Commit().HasError()) throw LoadException("Couldn't store the node");
}
void ProcessNodes(memgraph::storage::Storage *store, const std::string &nodes_path,
@@ -567,16 +568,16 @@ void ProcessRelationshipsRow(memgraph::storage::Storage *store, const std::vecto
if (!relationship_type) throw LoadException("Relationship TYPE must be set");
auto acc = store->Access();
auto from_node = acc.FindVertex(*start_id, memgraph::storage::View::NEW);
auto from_node = acc->FindVertex(*start_id, memgraph::storage::View::NEW);
if (!from_node) throw LoadException("From node must be in the storage");
auto to_node = acc.FindVertex(*end_id, memgraph::storage::View::NEW);
auto to_node = acc->FindVertex(*end_id, memgraph::storage::View::NEW);
if (!to_node) throw LoadException("To node must be in the storage");
auto relationship = acc.CreateEdge(&*from_node, &*to_node, acc.NameToEdgeType(*relationship_type));
auto relationship = acc->CreateEdge(&from_node.value(), &to_node.value(), acc->NameToEdgeType(*relationship_type));
if (!relationship.HasValue()) throw LoadException("Couldn't create the relationship");
for (const auto &property : properties) {
auto ret = relationship->SetProperty(acc.NameToProperty(property.first), property.second);
auto ret = relationship.GetValue().SetProperty(acc->NameToProperty(property.first), property.second);
if (!ret.HasValue()) {
if (ret.GetError() != memgraph::storage::Error::PROPERTIES_DISABLED) {
throw LoadException("Couldn't add property '{}' to the relationship", property.first);
@@ -589,7 +590,7 @@ void ProcessRelationshipsRow(memgraph::storage::Storage *store, const std::vecto
}
}
if (acc.Commit().HasError()) throw LoadException("Couldn't store the relationship");
if (acc->Commit().HasError()) throw LoadException("Couldn't store the relationship");
}
void ProcessRelationships(memgraph::storage::Storage *store, const std::string &relationships_path,
@@ -699,13 +700,13 @@ int main(int argc, char *argv[]) {
}
std::unordered_map<NodeId, memgraph::storage::Gid> node_id_map;
memgraph::storage::Storage store{{
std::unique_ptr<memgraph::storage::Storage> store{new memgraph::storage::InMemoryStorage{{
.items = {.properties_on_edges = FLAGS_storage_properties_on_edges},
.durability = {.storage_directory = FLAGS_data_directory,
.recover_on_startup = false,
.snapshot_wal_mode = memgraph::storage::Config::Durability::SnapshotWalMode::DISABLED,
.snapshot_on_exit = true},
}};
}}};
memgraph::utils::Timer load_timer;
@@ -715,7 +716,7 @@ int main(int argc, char *argv[]) {
std::optional<std::vector<Field>> header;
for (const auto &nodes_file : files) {
spdlog::info("Loading {}", nodes_file);
ProcessNodes(&store, nodes_file, &header, &node_id_map, additional_labels);
ProcessNodes(store.get(), nodes_file, &header, &node_id_map, additional_labels);
}
}
@@ -725,7 +726,7 @@ int main(int argc, char *argv[]) {
std::optional<std::vector<Field>> header;
for (const auto &relationships_file : files) {
spdlog::info("Loading {}", relationships_file);
ProcessRelationships(&store, relationships_file, type, &header, node_id_map);
ProcessRelationships(store.get(), relationships_file, type, &header, node_id_map);
}
}