d4f0bb0e38
Add a report for the case where a sync replica does not confirm within a timeout: -Add a new exception: ReplicationException to be returned when one sync replica does not confirm the reception of messages (new data, new constraint/index, or for triggers) -Update the logic to throw the ReplicationException when needed for insertion of new data, triggers, or creation of new constraint/index -Add end-to-end tests to cover the loss of connection with sync/async replicas when adding new data, adding new constraint/indexes, and triggers Add end-to-end tests to cover the creation and drop of indexes, existence constraints, and uniqueness constraints Improved tooling function mg_sleep_and_assert to also show the last result when duration is exceeded
592 lines
28 KiB
C++
592 lines
28 KiB
C++
// Copyright 2022 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
|
|
// License, and you may not use this file except in compliance with the Business Source License.
|
|
//
|
|
// As of the Change Date specified in that file, in accordance with
|
|
// the Business Source License, use of this software will be governed
|
|
// by the Apache License, Version 2.0, included in the file
|
|
// licenses/APL.txt.
|
|
|
|
#include "storage/v2/replication/replication_server.hpp"
|
|
#include <atomic>
|
|
#include <filesystem>
|
|
|
|
#include "storage/v2/durability/durability.hpp"
|
|
#include "storage/v2/durability/paths.hpp"
|
|
#include "storage/v2/durability/serialization.hpp"
|
|
#include "storage/v2/durability/snapshot.hpp"
|
|
#include "storage/v2/durability/version.hpp"
|
|
#include "storage/v2/durability/wal.hpp"
|
|
#include "storage/v2/replication/config.hpp"
|
|
#include "storage/v2/transaction.hpp"
|
|
#include "utils/exceptions.hpp"
|
|
|
|
namespace memgraph::storage {
|
|
namespace {
|
|
std::pair<uint64_t, durability::WalDeltaData> ReadDelta(durability::BaseDecoder *decoder) {
|
|
try {
|
|
auto timestamp = ReadWalDeltaHeader(decoder);
|
|
SPDLOG_INFO(" Timestamp {}", timestamp);
|
|
auto delta = ReadWalDeltaData(decoder);
|
|
return {timestamp, delta};
|
|
} catch (const slk::SlkReaderException &) {
|
|
throw utils::BasicException("Missing data!");
|
|
} catch (const durability::RecoveryFailure &) {
|
|
throw utils::BasicException("Invalid data!");
|
|
}
|
|
};
|
|
} // namespace
|
|
|
|
Storage::ReplicationServer::ReplicationServer(Storage *storage, io::network::Endpoint endpoint,
|
|
const replication::ReplicationServerConfig &config)
|
|
: storage_(storage) {
|
|
// Create RPC server.
|
|
if (config.ssl) {
|
|
rpc_server_context_.emplace(config.ssl->key_file, config.ssl->cert_file, config.ssl->ca_file,
|
|
config.ssl->verify_peer);
|
|
} else {
|
|
rpc_server_context_.emplace();
|
|
}
|
|
// NOTE: The replication server must have a single thread for processing
|
|
// because there is no need for more processing threads - each replica can
|
|
// have only a single main server. Also, the single-threaded guarantee
|
|
// simplifies the rest of the implementation.
|
|
rpc_server_.emplace(std::move(endpoint), &*rpc_server_context_,
|
|
/* workers_count = */ 1);
|
|
|
|
rpc_server_->Register<replication::HeartbeatRpc>([this](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received HeartbeatRpc");
|
|
this->HeartbeatHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Register<replication::FrequentHeartbeatRpc>([](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received FrequentHeartbeatRpc");
|
|
FrequentHeartbeatHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Register<replication::AppendDeltasRpc>([this](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received AppendDeltasRpc");
|
|
this->AppendDeltasHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Register<replication::SnapshotRpc>([this](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received SnapshotRpc");
|
|
this->SnapshotHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Register<replication::WalFilesRpc>([this](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received WalFilesRpc");
|
|
this->WalFilesHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Register<replication::CurrentWalRpc>([this](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received CurrentWalRpc");
|
|
this->CurrentWalHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Register<replication::TimestampRpc>([this](auto *req_reader, auto *res_builder) {
|
|
spdlog::debug("Received TimestampRpc");
|
|
this->TimestampHandler(req_reader, res_builder);
|
|
});
|
|
rpc_server_->Start();
|
|
}
|
|
|
|
void Storage::ReplicationServer::HeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::HeartbeatReq req;
|
|
slk::Load(&req, req_reader);
|
|
replication::HeartbeatRes res{true, storage_->last_commit_timestamp_.load(), storage_->epoch_id_};
|
|
slk::Save(res, res_builder);
|
|
}
|
|
|
|
void Storage::ReplicationServer::FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::FrequentHeartbeatReq req;
|
|
slk::Load(&req, req_reader);
|
|
replication::FrequentHeartbeatRes res{true};
|
|
slk::Save(res, res_builder);
|
|
}
|
|
|
|
void Storage::ReplicationServer::AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::AppendDeltasReq req;
|
|
slk::Load(&req, req_reader);
|
|
|
|
replication::Decoder decoder(req_reader);
|
|
|
|
auto maybe_epoch_id = decoder.ReadString();
|
|
MG_ASSERT(maybe_epoch_id, "Invalid replication message");
|
|
|
|
if (*maybe_epoch_id != storage_->epoch_id_) {
|
|
storage_->epoch_history_.emplace_back(std::move(storage_->epoch_id_), storage_->last_commit_timestamp_);
|
|
storage_->epoch_id_ = std::move(*maybe_epoch_id);
|
|
}
|
|
|
|
if (storage_->wal_file_) {
|
|
if (req.seq_num > storage_->wal_file_->SequenceNumber() || *maybe_epoch_id != storage_->epoch_id_) {
|
|
storage_->wal_file_->FinalizeWal();
|
|
storage_->wal_file_.reset();
|
|
storage_->wal_seq_num_ = req.seq_num;
|
|
} else {
|
|
MG_ASSERT(storage_->wal_file_->SequenceNumber() == req.seq_num, "Invalid sequence number of current wal file");
|
|
storage_->wal_seq_num_ = req.seq_num + 1;
|
|
}
|
|
} else {
|
|
storage_->wal_seq_num_ = req.seq_num;
|
|
}
|
|
|
|
if (req.previous_commit_timestamp != storage_->last_commit_timestamp_.load()) {
|
|
// Empty the stream
|
|
bool transaction_complete = false;
|
|
while (!transaction_complete) {
|
|
SPDLOG_INFO("Skipping delta");
|
|
const auto [timestamp, delta] = ReadDelta(&decoder);
|
|
transaction_complete = durability::IsWalDeltaDataTypeTransactionEnd(delta.type);
|
|
}
|
|
|
|
replication::AppendDeltasRes res{false, storage_->last_commit_timestamp_.load()};
|
|
slk::Save(res, res_builder);
|
|
return;
|
|
}
|
|
|
|
ReadAndApplyDelta(&decoder);
|
|
|
|
replication::AppendDeltasRes res{true, storage_->last_commit_timestamp_.load()};
|
|
slk::Save(res, res_builder);
|
|
}
|
|
|
|
void Storage::ReplicationServer::SnapshotHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::SnapshotReq req;
|
|
slk::Load(&req, req_reader);
|
|
|
|
replication::Decoder decoder(req_reader);
|
|
|
|
utils::EnsureDirOrDie(storage_->snapshot_directory_);
|
|
|
|
const auto maybe_snapshot_path = decoder.ReadFile(storage_->snapshot_directory_);
|
|
MG_ASSERT(maybe_snapshot_path, "Failed to load snapshot!");
|
|
spdlog::info("Received snapshot saved to {}", *maybe_snapshot_path);
|
|
|
|
std::unique_lock<utils::RWLock> storage_guard(storage_->main_lock_);
|
|
// Clear the database
|
|
storage_->vertices_.clear();
|
|
storage_->edges_.clear();
|
|
|
|
storage_->constraints_ = Constraints();
|
|
storage_->indices_.label_index = LabelIndex(&storage_->indices_, &storage_->constraints_, storage_->config_.items);
|
|
storage_->indices_.label_property_index =
|
|
LabelPropertyIndex(&storage_->indices_, &storage_->constraints_, storage_->config_.items);
|
|
try {
|
|
spdlog::debug("Loading snapshot");
|
|
auto recovered_snapshot = durability::LoadSnapshot(*maybe_snapshot_path, &storage_->vertices_, &storage_->edges_,
|
|
&storage_->epoch_history_, &storage_->name_id_mapper_,
|
|
&storage_->edge_count_, storage_->config_.items);
|
|
spdlog::debug("Snapshot loaded successfully");
|
|
// If this step is present it should always be the first step of
|
|
// the recovery so we use the UUID we read from snasphost
|
|
storage_->uuid_ = std::move(recovered_snapshot.snapshot_info.uuid);
|
|
storage_->epoch_id_ = std::move(recovered_snapshot.snapshot_info.epoch_id);
|
|
const auto &recovery_info = recovered_snapshot.recovery_info;
|
|
storage_->vertex_id_ = recovery_info.next_vertex_id;
|
|
storage_->edge_id_ = recovery_info.next_edge_id;
|
|
storage_->timestamp_ = std::max(storage_->timestamp_, recovery_info.next_timestamp);
|
|
|
|
durability::RecoverIndicesAndConstraints(recovered_snapshot.indices_constraints, &storage_->indices_,
|
|
&storage_->constraints_, &storage_->vertices_);
|
|
} catch (const durability::RecoveryFailure &e) {
|
|
LOG_FATAL("Couldn't load the snapshot because of: {}", e.what());
|
|
}
|
|
storage_guard.unlock();
|
|
|
|
replication::SnapshotRes res{true, storage_->last_commit_timestamp_.load()};
|
|
slk::Save(res, res_builder);
|
|
|
|
// Delete other durability files
|
|
auto snapshot_files = durability::GetSnapshotFiles(storage_->snapshot_directory_, storage_->uuid_);
|
|
for (const auto &[path, uuid, _] : snapshot_files) {
|
|
if (path != *maybe_snapshot_path) {
|
|
storage_->file_retainer_.DeleteFile(path);
|
|
}
|
|
}
|
|
|
|
auto wal_files = durability::GetWalFiles(storage_->wal_directory_, storage_->uuid_);
|
|
if (wal_files) {
|
|
for (const auto &wal_file : *wal_files) {
|
|
storage_->file_retainer_.DeleteFile(wal_file.path);
|
|
}
|
|
|
|
storage_->wal_file_.reset();
|
|
}
|
|
}
|
|
|
|
void Storage::ReplicationServer::WalFilesHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::WalFilesReq req;
|
|
slk::Load(&req, req_reader);
|
|
|
|
const auto wal_file_number = req.file_number;
|
|
spdlog::debug("Received WAL files: {}", wal_file_number);
|
|
|
|
replication::Decoder decoder(req_reader);
|
|
|
|
utils::EnsureDirOrDie(storage_->wal_directory_);
|
|
|
|
for (auto i = 0; i < wal_file_number; ++i) {
|
|
LoadWal(&decoder);
|
|
}
|
|
|
|
replication::WalFilesRes res{true, storage_->last_commit_timestamp_.load()};
|
|
slk::Save(res, res_builder);
|
|
}
|
|
|
|
void Storage::ReplicationServer::CurrentWalHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::CurrentWalReq req;
|
|
slk::Load(&req, req_reader);
|
|
|
|
replication::Decoder decoder(req_reader);
|
|
|
|
utils::EnsureDirOrDie(storage_->wal_directory_);
|
|
|
|
LoadWal(&decoder);
|
|
|
|
replication::CurrentWalRes res{true, storage_->last_commit_timestamp_.load()};
|
|
slk::Save(res, res_builder);
|
|
}
|
|
|
|
void Storage::ReplicationServer::LoadWal(replication::Decoder *decoder) {
|
|
const auto temp_wal_directory = std::filesystem::temp_directory_path() / "memgraph" / durability::kWalDirectory;
|
|
utils::EnsureDir(temp_wal_directory);
|
|
auto maybe_wal_path = decoder->ReadFile(temp_wal_directory);
|
|
MG_ASSERT(maybe_wal_path, "Failed to load WAL!");
|
|
spdlog::trace("Received WAL saved to {}", *maybe_wal_path);
|
|
try {
|
|
auto wal_info = durability::ReadWalInfo(*maybe_wal_path);
|
|
if (wal_info.seq_num == 0) {
|
|
storage_->uuid_ = wal_info.uuid;
|
|
}
|
|
|
|
if (wal_info.epoch_id != storage_->epoch_id_) {
|
|
storage_->epoch_history_.emplace_back(wal_info.epoch_id, storage_->last_commit_timestamp_);
|
|
storage_->epoch_id_ = std::move(wal_info.epoch_id);
|
|
}
|
|
|
|
if (storage_->wal_file_) {
|
|
if (storage_->wal_file_->SequenceNumber() != wal_info.seq_num) {
|
|
storage_->wal_file_->FinalizeWal();
|
|
storage_->wal_seq_num_ = wal_info.seq_num;
|
|
storage_->wal_file_.reset();
|
|
}
|
|
} else {
|
|
storage_->wal_seq_num_ = wal_info.seq_num;
|
|
}
|
|
|
|
durability::Decoder wal;
|
|
const auto version = wal.Initialize(*maybe_wal_path, durability::kWalMagic);
|
|
if (!version) throw durability::RecoveryFailure("Couldn't read WAL magic and/or version!");
|
|
if (!durability::IsVersionSupported(*version)) throw durability::RecoveryFailure("Invalid WAL version!");
|
|
wal.SetPosition(wal_info.offset_deltas);
|
|
|
|
for (size_t i = 0; i < wal_info.num_deltas;) {
|
|
i += ReadAndApplyDelta(&wal);
|
|
}
|
|
|
|
spdlog::debug("{} loaded successfully", *maybe_wal_path);
|
|
} catch (const durability::RecoveryFailure &e) {
|
|
LOG_FATAL("Couldn't recover WAL deltas from {} because of: {}", *maybe_wal_path, e.what());
|
|
}
|
|
}
|
|
|
|
void Storage::ReplicationServer::TimestampHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
|
|
replication::TimestampReq req;
|
|
slk::Load(&req, req_reader);
|
|
|
|
replication::TimestampRes res{true, storage_->last_commit_timestamp_.load()};
|
|
slk::Save(res, res_builder);
|
|
}
|
|
|
|
Storage::ReplicationServer::~ReplicationServer() {
|
|
if (rpc_server_) {
|
|
rpc_server_->Shutdown();
|
|
rpc_server_->AwaitShutdown();
|
|
}
|
|
}
|
|
uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *decoder) {
|
|
auto edge_acc = storage_->edges_.access();
|
|
auto vertex_acc = storage_->vertices_.access();
|
|
|
|
std::optional<std::pair<uint64_t, storage::Storage::Accessor>> commit_timestamp_and_accessor;
|
|
auto get_transaction = [this, &commit_timestamp_and_accessor](uint64_t commit_timestamp) {
|
|
if (!commit_timestamp_and_accessor) {
|
|
commit_timestamp_and_accessor.emplace(commit_timestamp, storage_->Access());
|
|
} else if (commit_timestamp_and_accessor->first != commit_timestamp) {
|
|
throw utils::BasicException("Received more than one transaction!");
|
|
}
|
|
return &commit_timestamp_and_accessor->second;
|
|
};
|
|
|
|
uint64_t applied_deltas = 0;
|
|
auto max_commit_timestamp = storage_->last_commit_timestamp_.load();
|
|
|
|
for (bool transaction_complete = false; !transaction_complete; ++applied_deltas) {
|
|
const auto [timestamp, delta] = ReadDelta(decoder);
|
|
if (timestamp > max_commit_timestamp) {
|
|
max_commit_timestamp = timestamp;
|
|
}
|
|
|
|
transaction_complete = durability::IsWalDeltaDataTypeTransactionEnd(delta.type);
|
|
|
|
if (timestamp < storage_->timestamp_) {
|
|
continue;
|
|
}
|
|
|
|
SPDLOG_INFO(" Delta {}", applied_deltas);
|
|
switch (delta.type) {
|
|
case durability::WalDeltaData::Type::VERTEX_CREATE: {
|
|
spdlog::trace(" Create vertex {}", delta.vertex_create_delete.gid.AsUint());
|
|
auto transaction = get_transaction(timestamp);
|
|
transaction->CreateVertex(delta.vertex_create_delete.gid);
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::VERTEX_DELETE: {
|
|
spdlog::trace(" Delete vertex {}", delta.vertex_create_delete.gid.AsUint());
|
|
auto transaction = get_transaction(timestamp);
|
|
auto vertex = transaction->FindVertex(delta.vertex_create_delete.gid, storage::View::NEW);
|
|
if (!vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto ret = transaction->DeleteVertex(&*vertex);
|
|
if (ret.HasError() || !ret.GetValue()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::VERTEX_ADD_LABEL: {
|
|
spdlog::trace(" Vertex {} add label {}", delta.vertex_add_remove_label.gid.AsUint(),
|
|
delta.vertex_add_remove_label.label);
|
|
auto transaction = get_transaction(timestamp);
|
|
auto vertex = transaction->FindVertex(delta.vertex_add_remove_label.gid, storage::View::NEW);
|
|
if (!vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto ret = vertex->AddLabel(transaction->NameToLabel(delta.vertex_add_remove_label.label));
|
|
if (ret.HasError() || !ret.GetValue()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::VERTEX_REMOVE_LABEL: {
|
|
spdlog::trace(" Vertex {} remove label {}", delta.vertex_add_remove_label.gid.AsUint(),
|
|
delta.vertex_add_remove_label.label);
|
|
auto transaction = get_transaction(timestamp);
|
|
auto vertex = transaction->FindVertex(delta.vertex_add_remove_label.gid, storage::View::NEW);
|
|
if (!vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto ret = vertex->RemoveLabel(transaction->NameToLabel(delta.vertex_add_remove_label.label));
|
|
if (ret.HasError() || !ret.GetValue()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::VERTEX_SET_PROPERTY: {
|
|
spdlog::trace(" Vertex {} set property {} to {}", delta.vertex_edge_set_property.gid.AsUint(),
|
|
delta.vertex_edge_set_property.property, delta.vertex_edge_set_property.value);
|
|
auto transaction = get_transaction(timestamp);
|
|
auto vertex = transaction->FindVertex(delta.vertex_edge_set_property.gid, storage::View::NEW);
|
|
if (!vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto ret = vertex->SetProperty(transaction->NameToProperty(delta.vertex_edge_set_property.property),
|
|
delta.vertex_edge_set_property.value);
|
|
if (ret.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::EDGE_CREATE: {
|
|
spdlog::trace(" Create edge {} of type {} from vertex {} to vertex {}",
|
|
delta.edge_create_delete.gid.AsUint(), delta.edge_create_delete.edge_type,
|
|
delta.edge_create_delete.from_vertex.AsUint(), delta.edge_create_delete.to_vertex.AsUint());
|
|
auto transaction = get_transaction(timestamp);
|
|
auto from_vertex = transaction->FindVertex(delta.edge_create_delete.from_vertex, storage::View::NEW);
|
|
if (!from_vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto to_vertex = transaction->FindVertex(delta.edge_create_delete.to_vertex, storage::View::NEW);
|
|
if (!to_vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto edge = transaction->CreateEdge(&*from_vertex, &*to_vertex,
|
|
transaction->NameToEdgeType(delta.edge_create_delete.edge_type),
|
|
delta.edge_create_delete.gid);
|
|
if (edge.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::EDGE_DELETE: {
|
|
spdlog::trace(" Delete edge {} of type {} from vertex {} to vertex {}",
|
|
delta.edge_create_delete.gid.AsUint(), delta.edge_create_delete.edge_type,
|
|
delta.edge_create_delete.from_vertex.AsUint(), delta.edge_create_delete.to_vertex.AsUint());
|
|
auto transaction = get_transaction(timestamp);
|
|
auto from_vertex = transaction->FindVertex(delta.edge_create_delete.from_vertex, storage::View::NEW);
|
|
if (!from_vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto to_vertex = transaction->FindVertex(delta.edge_create_delete.to_vertex, storage::View::NEW);
|
|
if (!to_vertex) throw utils::BasicException("Invalid transaction!");
|
|
auto edges = from_vertex->OutEdges(
|
|
storage::View::NEW, {transaction->NameToEdgeType(delta.edge_create_delete.edge_type)}, &*to_vertex);
|
|
if (edges.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
if (edges->size() != 1) throw utils::BasicException("Invalid transaction!");
|
|
auto &edge = (*edges)[0];
|
|
auto ret = transaction->DeleteEdge(&edge);
|
|
if (ret.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::EDGE_SET_PROPERTY: {
|
|
spdlog::trace(" Edge {} set property {} to {}", delta.vertex_edge_set_property.gid.AsUint(),
|
|
delta.vertex_edge_set_property.property, delta.vertex_edge_set_property.value);
|
|
|
|
if (!storage_->config_.items.properties_on_edges)
|
|
throw utils::BasicException(
|
|
"Can't set properties on edges because properties on edges "
|
|
"are disabled!");
|
|
|
|
auto transaction = get_transaction(timestamp);
|
|
|
|
// The following block of code effectively implements `FindEdge` and
|
|
// yields an accessor that is only valid for managing the edge's
|
|
// properties.
|
|
auto edge = edge_acc.find(delta.vertex_edge_set_property.gid);
|
|
if (edge == edge_acc.end()) throw utils::BasicException("Invalid transaction!");
|
|
// The edge visibility check must be done here manually because we
|
|
// don't allow direct access to the edges through the public API.
|
|
{
|
|
bool is_visible = true;
|
|
Delta *delta = nullptr;
|
|
{
|
|
std::lock_guard<utils::SpinLock> guard(edge->lock);
|
|
is_visible = !edge->deleted;
|
|
delta = edge->delta;
|
|
}
|
|
ApplyDeltasForRead(&transaction->transaction_, delta, View::NEW, [&is_visible](const Delta &delta) {
|
|
switch (delta.action) {
|
|
case Delta::Action::ADD_LABEL:
|
|
case Delta::Action::REMOVE_LABEL:
|
|
case Delta::Action::SET_PROPERTY:
|
|
case Delta::Action::ADD_IN_EDGE:
|
|
case Delta::Action::ADD_OUT_EDGE:
|
|
case Delta::Action::REMOVE_IN_EDGE:
|
|
case Delta::Action::REMOVE_OUT_EDGE:
|
|
break;
|
|
case Delta::Action::RECREATE_OBJECT: {
|
|
is_visible = true;
|
|
break;
|
|
}
|
|
case Delta::Action::DELETE_OBJECT: {
|
|
is_visible = false;
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
if (!is_visible) throw utils::BasicException("Invalid transaction!");
|
|
}
|
|
EdgeRef edge_ref(&*edge);
|
|
// Here we create an edge accessor that we will use to get the
|
|
// properties of the edge. The accessor is created with an invalid
|
|
// type and invalid from/to pointers because we don't know them
|
|
// here, but that isn't an issue because we won't use that part of
|
|
// the API here.
|
|
auto ea = EdgeAccessor{edge_ref,
|
|
EdgeTypeId::FromUint(0UL),
|
|
nullptr,
|
|
nullptr,
|
|
&transaction->transaction_,
|
|
&storage_->indices_,
|
|
&storage_->constraints_,
|
|
storage_->config_.items};
|
|
|
|
auto ret = ea.SetProperty(transaction->NameToProperty(delta.vertex_edge_set_property.property),
|
|
delta.vertex_edge_set_property.value);
|
|
if (ret.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
|
|
case durability::WalDeltaData::Type::TRANSACTION_END: {
|
|
spdlog::trace(" Transaction end");
|
|
if (!commit_timestamp_and_accessor || commit_timestamp_and_accessor->first != timestamp)
|
|
throw utils::BasicException("Invalid data!");
|
|
auto ret = commit_timestamp_and_accessor->second.Commit(commit_timestamp_and_accessor->first);
|
|
if (ret.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
commit_timestamp_and_accessor = std::nullopt;
|
|
break;
|
|
}
|
|
|
|
case durability::WalDeltaData::Type::LABEL_INDEX_CREATE: {
|
|
spdlog::trace(" Create label index on :{}", delta.operation_label.label);
|
|
// Need to send the timestamp
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
if (storage_->CreateIndex(storage_->NameToLabel(delta.operation_label.label), timestamp).HasError())
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::LABEL_INDEX_DROP: {
|
|
spdlog::trace(" Drop label index on :{}", delta.operation_label.label);
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
if (storage_->DropIndex(storage_->NameToLabel(delta.operation_label.label), timestamp).HasError())
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::LABEL_PROPERTY_INDEX_CREATE: {
|
|
spdlog::trace(" Create label+property index on :{} ({})", delta.operation_label_property.label,
|
|
delta.operation_label_property.property);
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
if (storage_
|
|
->CreateIndex(storage_->NameToLabel(delta.operation_label_property.label),
|
|
storage_->NameToProperty(delta.operation_label_property.property), timestamp)
|
|
.HasError())
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::LABEL_PROPERTY_INDEX_DROP: {
|
|
spdlog::trace(" Drop label+property index on :{} ({})", delta.operation_label_property.label,
|
|
delta.operation_label_property.property);
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
if (storage_
|
|
->DropIndex(storage_->NameToLabel(delta.operation_label_property.label),
|
|
storage_->NameToProperty(delta.operation_label_property.property), timestamp)
|
|
.HasError())
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::EXISTENCE_CONSTRAINT_CREATE: {
|
|
spdlog::trace(" Create existence constraint on :{} ({})", delta.operation_label_property.label,
|
|
delta.operation_label_property.property);
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
auto ret = storage_->CreateExistenceConstraint(
|
|
storage_->NameToLabel(delta.operation_label_property.label),
|
|
storage_->NameToProperty(delta.operation_label_property.property), timestamp);
|
|
if (ret.HasError()) throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::EXISTENCE_CONSTRAINT_DROP: {
|
|
spdlog::trace(" Drop existence constraint on :{} ({})", delta.operation_label_property.label,
|
|
delta.operation_label_property.property);
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
if (storage_
|
|
->DropExistenceConstraint(storage_->NameToLabel(delta.operation_label_property.label),
|
|
storage_->NameToProperty(delta.operation_label_property.property), timestamp)
|
|
.HasError())
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::UNIQUE_CONSTRAINT_CREATE: {
|
|
std::stringstream ss;
|
|
utils::PrintIterable(ss, delta.operation_label_properties.properties);
|
|
spdlog::trace(" Create unique constraint on :{} ({})", delta.operation_label_properties.label, ss.str());
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
std::set<PropertyId> properties;
|
|
for (const auto &prop : delta.operation_label_properties.properties) {
|
|
properties.emplace(storage_->NameToProperty(prop));
|
|
}
|
|
auto ret = storage_->CreateUniqueConstraint(storage_->NameToLabel(delta.operation_label_properties.label),
|
|
properties, timestamp);
|
|
if (!ret.HasValue() || ret.GetValue() != UniqueConstraints::CreationStatus::SUCCESS)
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
case durability::WalDeltaData::Type::UNIQUE_CONSTRAINT_DROP: {
|
|
std::stringstream ss;
|
|
utils::PrintIterable(ss, delta.operation_label_properties.properties);
|
|
spdlog::trace(" Drop unique constraint on :{} ({})", delta.operation_label_properties.label, ss.str());
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid transaction!");
|
|
std::set<PropertyId> properties;
|
|
for (const auto &prop : delta.operation_label_properties.properties) {
|
|
properties.emplace(storage_->NameToProperty(prop));
|
|
}
|
|
auto ret = storage_->DropUniqueConstraint(storage_->NameToLabel(delta.operation_label_properties.label),
|
|
properties, timestamp);
|
|
if (ret.HasError() || ret.GetValue() != UniqueConstraints::DeletionStatus::SUCCESS)
|
|
throw utils::BasicException("Invalid transaction!");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (commit_timestamp_and_accessor) throw utils::BasicException("Invalid data!");
|
|
|
|
storage_->last_commit_timestamp_ = max_commit_timestamp;
|
|
|
|
return applied_deltas;
|
|
}
|
|
} // namespace memgraph::storage
|