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

Fix edge in vertex_Edge

Browse Source
This commit is contained in:
jbajic 2023-02-07 13:31:05 +01:00
parent 82c7c85428
commit dc48b4ae9b
2 changed files with 39 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/storage/v3/splitter.cpp
View File

@ -190,11 +190,12 @@ void PruneDeltas(Transaction &cloned_transaction, std::map<uint64_t, std::unique
void Splitter::AdjustClonedTransactions(std::map<uint64_t, std::unique_ptr<Transaction>> &cloned_transactions,
VertexContainer &cloned_vertices, EdgeContainer &cloned_edges,
const PrimaryKey &split_key) {
// Prune deltas whose delta chain points to vertex/edge that should not belong on that shard
for (auto &[commit_start, cloned_transaction] : cloned_transactions) {
AdjustClonedTransaction(*cloned_transaction, *start_logical_id_to_transaction_[commit_start], cloned_transactions,
cloned_vertices, cloned_edges, split_key);
}
// Prune deltas whose delta chain points to vertex/edge that should not belong on that shard
// Prune must be after ajdust, since next, and prev are not set and we cannot follow the chain
for (auto &[commit_start, cloned_transaction] : cloned_transactions) {
PruneDeltas(*cloned_transaction, cloned_transactions, split_key);
}
@ -226,6 +227,7 @@ void Splitter::AdjustClonedTransaction(Transaction &cloned_transaction, const Tr
const auto *delta = &*delta_it;
auto *cloned_delta = &*cloned_delta_it;
AdjustEdgeRef(*cloned_delta, cloned_edges);
while (delta->next != nullptr) {
// Align next ptr
AdjustDeltaNext(*delta, *cloned_delta, cloned_transactions);
@ -246,6 +248,36 @@ void Splitter::AdjustClonedTransaction(Transaction &cloned_transaction, const Tr
"Both iterators must be exhausted!");
}
void Splitter::AdjustEdgeRef(Delta &cloned_delta, EdgeContainer &cloned_edges) const {
switch (cloned_delta.action) {
case Delta::Action::ADD_IN_EDGE:
case Delta::Action::ADD_OUT_EDGE:
case Delta::Action::REMOVE_IN_EDGE:
case Delta::Action::REMOVE_OUT_EDGE: {
// Find edge
if (config_.items.properties_on_edges) {
// Only case when not finding is when the edge is not on splitted shard
// TODO Do this after prune an move condition into assert
if (const auto cloned_edge_it =
std::ranges::find_if(cloned_edges, [edge_ptr = cloned_delta.vertex_edge.edge.ptr](
const auto &elem) { return elem.second.gid == edge_ptr->gid; });
cloned_edge_it != cloned_edges.end()) {
cloned_delta.vertex_edge.edge = EdgeRef{&cloned_edge_it->second};
}
}
break;
}
case Delta::Action::DELETE_OBJECT:
case Delta::Action::RECREATE_OBJECT:
case Delta::Action::SET_PROPERTY:
case Delta::Action::ADD_LABEL:
case Delta::Action::REMOVE_LABEL: {
// noop
break;
}
}
}
void Splitter::AdjustDeltaNext(const Delta &original, Delta &cloned,
std::map<uint64_t, std::unique_ptr<Transaction>> &cloned_transactions) {
// Get cloned_delta->next transaction, using delta->next original transaction

10
src/storage/v3/splitter.hpp
View File

@ -76,15 +76,17 @@ class Splitter final {
static void ScanDeltas(std::set<uint64_t> &collected_transactions_start_id, Delta *delta);
static void AdjustClonedTransaction(Transaction &cloned_transaction, const Transaction &transaction,
std::map<uint64_t, std::unique_ptr<Transaction>> &cloned_transactions,
VertexContainer &cloned_vertices, EdgeContainer &cloned_edges,
const PrimaryKey &split_key);
void AdjustClonedTransaction(Transaction &cloned_transaction, const Transaction &transaction,
std::map<uint64_t, std::unique_ptr<Transaction>> &cloned_transactions,
VertexContainer &cloned_vertices, EdgeContainer &cloned_edges,
const PrimaryKey &split_key);
void AdjustClonedTransactions(std::map<uint64_t, std::unique_ptr<Transaction>> &cloned_transactions,
VertexContainer &cloned_vertices, EdgeContainer &cloned_edges,
const PrimaryKey &split_key);
void AdjustEdgeRef(Delta &cloned_delta, EdgeContainer &cloned_edges) const;
static void AdjustDeltaNext(const Delta &original, Delta &cloned,
std::map<uint64_t, std::unique_ptr<Transaction>> &cloned_transactions);