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

[master < T1110] Add merge optimization to expand dynamically during runtime (#1110)

Browse Source
This commit is contained in:
Josipmrden
2023-09-08 17:12:25 +02:00
committed by GitHub
parent bd1852f407
commit 07dea328d8
29 changed files with 2164 additions and 1905 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/query/db_accessor.cpp
View File

@@ -129,10 +129,10 @@ query::Graph *SubgraphDbAccessor::getGraph() { return graph_; }
VertexAccessor SubgraphVertexAccessor::GetVertexAccessor() const { return impl_; }
auto SubgraphVertexAccessor::OutEdges(storage::View view) const -> decltype(impl_.OutEdges(view)) {
storage::Result<EdgeVertexAccessorResult> SubgraphVertexAccessor::OutEdges(storage::View view) const {
auto maybe_edges = impl_.impl_.OutEdges(view, {});
if (maybe_edges.HasError()) return maybe_edges.GetError();
auto edges = std::move(*maybe_edges);
auto edges = std::move(maybe_edges->edges);
const auto &graph_edges = graph_->edges();
std::vector<storage::EdgeAccessor> filteredOutEdges;
@@ -143,13 +143,18 @@ auto SubgraphVertexAccessor::OutEdges(storage::View view) const -> decltype(impl
}
}
return iter::imap(VertexAccessor::MakeEdgeAccessor, std::move(filteredOutEdges));
std::vector<EdgeAccessor> resulting_edges;
resulting_edges.reserve(filteredOutEdges.size());
std::ranges::transform(filteredOutEdges, std::back_inserter(resulting_edges),
[](auto const &edge) { return VertexAccessor::MakeEdgeAccessor(edge); });
return EdgeVertexAccessorResult{.edges = std::move(resulting_edges), .expanded_count = maybe_edges->expanded_count};
}
auto SubgraphVertexAccessor::InEdges(storage::View view) const -> decltype(impl_.InEdges(view)) {
storage::Result<EdgeVertexAccessorResult> SubgraphVertexAccessor::InEdges(storage::View view) const {
auto maybe_edges = impl_.impl_.InEdges(view, {});
if (maybe_edges.HasError()) return maybe_edges.GetError();
auto edges = std::move(*maybe_edges);
auto edges = std::move(maybe_edges->edges);
const auto &graph_edges = graph_->edges();
std::vector<storage::EdgeAccessor> filteredOutEdges;
@@ -160,7 +165,12 @@ auto SubgraphVertexAccessor::InEdges(storage::View view) const -> decltype(impl_
}
}
return iter::imap(VertexAccessor::MakeEdgeAccessor, std::move(filteredOutEdges));
std::vector<EdgeAccessor> resulting_edges;
resulting_edges.reserve(filteredOutEdges.size());
std::ranges::transform(filteredOutEdges, std::back_inserter(resulting_edges),
[](auto const &edge) { return VertexAccessor::MakeEdgeAccessor(edge); });
return EdgeVertexAccessorResult{.edges = std::move(resulting_edges), .expanded_count = maybe_edges->expanded_count};
}
} // namespace memgraph::query

77
src/query/db_accessor.hpp
View File

@@ -105,6 +105,11 @@ class EdgeAccessor final {
bool operator!=(const EdgeAccessor &e) const noexcept { return !(*this == e); }
};
struct EdgeVertexAccessorResult {
std::vector<EdgeAccessor> edges;
int64_t expanded_count;
};
class VertexAccessor final {
public:
storage::VertexAccessor impl_;
@@ -153,37 +158,62 @@ class VertexAccessor final {
return impl_.ClearProperties();
}
auto InEdges(storage::View view, const std::vector<storage::EdgeTypeId> &edge_types) const
-> storage::Result<decltype(iter::imap(MakeEdgeAccessor, *impl_.InEdges(view)))> {
auto maybe_edges = impl_.InEdges(view, edge_types);
if (maybe_edges.HasError()) return maybe_edges.GetError();
return iter::imap(MakeEdgeAccessor, std::move(*maybe_edges));
storage::Result<EdgeVertexAccessorResult> InEdges(storage::View view,
const std::vector<storage::EdgeTypeId> &edge_types) const {
auto maybe_result = impl_.InEdges(view, edge_types);
if (maybe_result.HasError()) return maybe_result.GetError();
std::vector<EdgeAccessor> edges;
edges.reserve((*maybe_result).edges.size());
std::ranges::transform((*maybe_result).edges, std::back_inserter(edges),
[](auto const &edge) { return EdgeAccessor(edge); });
return EdgeVertexAccessorResult{.edges = edges, .expanded_count = (*maybe_result).expanded_count};
}
auto InEdges(storage::View view) const { return InEdges(view, {}); }
storage::Result<EdgeVertexAccessorResult> InEdges(storage::View view) const { return InEdges(view, {}); }
auto InEdges(storage::View view, const std::vector<storage::EdgeTypeId> &edge_types, const VertexAccessor &dest) const
-> storage::Result<decltype(iter::imap(MakeEdgeAccessor, *impl_.InEdges(view)))> {
auto maybe_edges = impl_.InEdges(view, edge_types, &dest.impl_);
if (maybe_edges.HasError()) return maybe_edges.GetError();
return iter::imap(MakeEdgeAccessor, std::move(*maybe_edges));
storage::Result<EdgeVertexAccessorResult> InEdges(storage::View view,
const std::vector<storage::EdgeTypeId> &edge_types,
const VertexAccessor &dest) const {
auto maybe_result = impl_.InEdges(view, edge_types, &dest.impl_);
if (maybe_result.HasError()) return maybe_result.GetError();
std::vector<EdgeAccessor> edges;
edges.reserve((*maybe_result).edges.size());
std::ranges::transform((*maybe_result).edges, std::back_inserter(edges),
[](auto const &edge) { return EdgeAccessor(edge); });
return EdgeVertexAccessorResult{.edges = edges, .expanded_count = (*maybe_result).expanded_count};
}
auto OutEdges(storage::View view, const std::vector<storage::EdgeTypeId> &edge_types) const
-> storage::Result<decltype(iter::imap(MakeEdgeAccessor, *impl_.OutEdges(view)))> {
auto maybe_edges = impl_.OutEdges(view, edge_types);
if (maybe_edges.HasError()) return maybe_edges.GetError();
return iter::imap(MakeEdgeAccessor, std::move(*maybe_edges));
storage::Result<EdgeVertexAccessorResult> OutEdges(storage::View view,
const std::vector<storage::EdgeTypeId> &edge_types) const {
auto maybe_result = impl_.OutEdges(view, edge_types);
if (maybe_result.HasError()) return maybe_result.GetError();
std::vector<EdgeAccessor> edges;
edges.reserve((*maybe_result).edges.size());
std::ranges::transform((*maybe_result).edges, std::back_inserter(edges),
[](auto const &edge) { return EdgeAccessor(edge); });
return EdgeVertexAccessorResult{.edges = edges, .expanded_count = (*maybe_result).expanded_count};
}
auto OutEdges(storage::View view) const { return OutEdges(view, {}); }
storage::Result<EdgeVertexAccessorResult> OutEdges(storage::View view) const { return OutEdges(view, {}); }
auto OutEdges(storage::View view, const std::vector<storage::EdgeTypeId> &edge_types,
const VertexAccessor &dest) const
-> storage::Result<decltype(iter::imap(MakeEdgeAccessor, *impl_.OutEdges(view)))> {
auto maybe_edges = impl_.OutEdges(view, edge_types, &dest.impl_);
if (maybe_edges.HasError()) return maybe_edges.GetError();
return iter::imap(MakeEdgeAccessor, std::move(*maybe_edges));
storage::Result<EdgeVertexAccessorResult> OutEdges(storage::View view,
const std::vector<storage::EdgeTypeId> &edge_types,
const VertexAccessor &dest) const {
auto maybe_result = impl_.OutEdges(view, edge_types, &dest.impl_);
if (maybe_result.HasError()) return maybe_result.GetError();
std::vector<EdgeAccessor> edges;
edges.reserve((*maybe_result).edges.size());
std::ranges::transform((*maybe_result).edges, std::back_inserter(edges),
[](auto const &edge) { return EdgeAccessor(edge); });
return EdgeVertexAccessorResult{.edges = edges, .expanded_count = (*maybe_result).expanded_count};
}
storage::Result<size_t> InDegree(storage::View view) const { return impl_.InDegree(view); }
@@ -260,7 +290,6 @@ class SubgraphVertexAccessor final {
} // namespace memgraph::query
namespace std {
template <>
struct hash<memgraph::query::VertexAccessor> {
size_t operator()(const memgraph::query::VertexAccessor &v) const { return std::hash<decltype(v.impl_)>{}(v.impl_); }

6
src/query/dump.cpp
View File

@@ -487,8 +487,8 @@ PullPlanDump::PullChunk PullPlanDump::CreateEdgePullChunk() {
}
auto &maybe_edges = *maybe_edge_iterable;
MG_ASSERT(maybe_edges.HasValue(), "Invalid database state!");
auto current_edge_iter = maybe_current_edge_iter ? *maybe_current_edge_iter : maybe_edges->begin();
for (; current_edge_iter != maybe_edges->end() && (!n || local_counter < *n); ++current_edge_iter) {
auto current_edge_iter = maybe_current_edge_iter ? *maybe_current_edge_iter : maybe_edges->edges.begin();
for (; current_edge_iter != maybe_edges->edges.end() && (!n || local_counter < *n); ++current_edge_iter) {
std::ostringstream os;
DumpEdge(&os, dba_, *current_edge_iter);
stream->Result({TypedValue(os.str())});
@@ -496,7 +496,7 @@ PullPlanDump::PullChunk PullPlanDump::CreateEdgePullChunk() {
++local_counter;
}
if (current_edge_iter != maybe_edges->end()) {
if (current_edge_iter != maybe_edges->edges.end()) {
maybe_current_edge_iter.emplace(current_edge_iter);
return std::nullopt;
}

4
src/query/dump.hpp
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
@@ -38,7 +38,7 @@ struct PullPlanDump {
using VertexAccessorIterableIterator = decltype(std::declval<VertexAccessorIterable>().begin());
using EdgeAccessorIterable = decltype(std::declval<VertexAccessor>().OutEdges(storage::View::OLD));
using EdgeAccessorIterableIterator = decltype(std::declval<EdgeAccessorIterable>().GetValue().begin());
using EdgeAccessorIterableIterator = decltype(std::declval<EdgeAccessorIterable>().GetValue().edges.begin());
VertexAccessorIterable vertices_iterable_;
bool internal_index_created_ = false;

151
src/query/plan/operator.cpp
View File

@@ -680,8 +680,8 @@ bool CheckExistingNode(const VertexAccessor &new_node, const Symbol &existing_no
return existing_node.ValueVertex() == new_node;
}
template <class TEdges>
auto UnwrapEdgesResult(storage::Result<TEdges> &&result) {
template <class TEdgesResult>
auto UnwrapEdgesResult(storage::Result<TEdgesResult> &&result) {
if (result.HasError()) {
switch (result.GetError()) {
case storage::Error::DELETED_OBJECT:
@@ -725,6 +725,13 @@ std::vector<Symbol> Expand::ModifiedSymbols(const SymbolTable &table) const {
Expand::ExpandCursor::ExpandCursor(const Expand &self, utils::MemoryResource *mem)
: self_(self), input_cursor_(self.input_->MakeCursor(mem)) {}
Expand::ExpandCursor::ExpandCursor(const Expand &self, int64_t input_degree, int64_t existing_node_degree,
utils::MemoryResource *mem)
: self_(self),
input_cursor_(self.input_->MakeCursor(mem)),
prev_input_degree_(input_degree),
prev_existing_degree_(existing_node_degree) {}
bool Expand::ExpandCursor::Pull(Frame &frame, ExecutionContext &context) {
SCOPED_PROFILE_OP("Expand");
@@ -800,58 +807,131 @@ void Expand::ExpandCursor::Reset() {
out_edges_it_ = std::nullopt;
}
ExpansionInfo Expand::ExpandCursor::GetExpansionInfo(Frame &frame) {
TypedValue &vertex_value = frame[self_.input_symbol_];
if (vertex_value.IsNull()) {
return ExpansionInfo{};
}
ExpectType(self_.input_symbol_, vertex_value, TypedValue::Type::Vertex);
auto &vertex = vertex_value.ValueVertex();
auto direction = self_.common_.direction;
if (!self_.common_.existing_node) {
return ExpansionInfo{.input_node = vertex, .direction = direction};
}
TypedValue &existing_node = frame[self_.common_.node_symbol];
if (existing_node.IsNull()) {
return ExpansionInfo{.input_node = vertex, .direction = direction};
}
ExpectType(self_.common_.node_symbol, existing_node, TypedValue::Type::Vertex);
auto &existing_vertex = existing_node.ValueVertex();
// -1 and -1 -> normal expansion
// -1 and expanded -> can't happen
// expanded and -1 -> reverse
// expanded and expanded -> see if can reverse
if ((prev_input_degree_ == -1 && prev_existing_degree_ == -1) || prev_input_degree_ < prev_existing_degree_) {
return ExpansionInfo{.input_node = vertex, .direction = direction, .existing_node = existing_vertex};
}
auto new_direction = direction;
switch (new_direction) {
case EdgeAtom::Direction::IN:
new_direction = EdgeAtom::Direction::OUT;
break;
case EdgeAtom::Direction::OUT:
new_direction = EdgeAtom::Direction::IN;
break;
default:
new_direction = EdgeAtom::Direction::BOTH;
break;
}
return ExpansionInfo{
.input_node = existing_vertex, .direction = new_direction, .existing_node = vertex, .reversed = true};
}
bool Expand::ExpandCursor::InitEdges(Frame &frame, ExecutionContext &context) {
// Input Vertex could be null if it is created by a failed optional match. In
// those cases we skip that input pull and continue with the next.
while (true) {
if (!input_cursor_->Pull(frame, context)) return false;
TypedValue &vertex_value = frame[self_.input_symbol_];
// Null check due to possible failed optional match.
if (vertex_value.IsNull()) continue;
expansion_info_ = GetExpansionInfo(frame);
ExpectType(self_.input_symbol_, vertex_value, TypedValue::Type::Vertex);
auto &vertex = vertex_value.ValueVertex();
if (!expansion_info_.input_node) {
continue;
}
auto direction = self_.common_.direction;
auto vertex = *expansion_info_.input_node;
auto direction = expansion_info_.direction;
int64_t num_expanded_first = -1;
if (direction == EdgeAtom::Direction::IN || direction == EdgeAtom::Direction::BOTH) {
if (self_.common_.existing_node) {
TypedValue &existing_node = frame[self_.common_.node_symbol];
// old_node_value may be Null when using optional matching
if (!existing_node.IsNull()) {
ExpectType(self_.common_.node_symbol, existing_node, TypedValue::Type::Vertex);
if (expansion_info_.existing_node) {
auto existing_node = *expansion_info_.existing_node;
context.db_accessor->PrefetchInEdges(vertex);
in_edges_.emplace(
UnwrapEdgesResult(vertex.InEdges(self_.view_, self_.common_.edge_types, existing_node.ValueVertex())));
auto edges_result = UnwrapEdgesResult(vertex.InEdges(self_.view_, self_.common_.edge_types, existing_node));
in_edges_.emplace(edges_result.edges);
num_expanded_first = edges_result.expanded_count;
}
} else {
context.db_accessor->PrefetchInEdges(vertex);
in_edges_.emplace(UnwrapEdgesResult(vertex.InEdges(self_.view_, self_.common_.edge_types)));
auto edges_result = UnwrapEdgesResult(vertex.InEdges(self_.view_, self_.common_.edge_types));
in_edges_.emplace(edges_result.edges);
num_expanded_first = edges_result.expanded_count;
}
if (in_edges_) {
in_edges_it_.emplace(in_edges_->begin());
}
}
int64_t num_expanded_second = -1;
if (direction == EdgeAtom::Direction::OUT || direction == EdgeAtom::Direction::BOTH) {
if (self_.common_.existing_node) {
TypedValue &existing_node = frame[self_.common_.node_symbol];
// old_node_value may be Null when using optional matching
if (!existing_node.IsNull()) {
ExpectType(self_.common_.node_symbol, existing_node, TypedValue::Type::Vertex);
if (expansion_info_.existing_node) {
auto existing_node = *expansion_info_.existing_node;
context.db_accessor->PrefetchOutEdges(vertex);
out_edges_.emplace(
UnwrapEdgesResult(vertex.OutEdges(self_.view_, self_.common_.edge_types, existing_node.ValueVertex())));
auto edges_result = UnwrapEdgesResult(vertex.OutEdges(self_.view_, self_.common_.edge_types, existing_node));
out_edges_.emplace(edges_result.edges);
num_expanded_second = edges_result.expanded_count;
}
} else {
context.db_accessor->PrefetchOutEdges(vertex);
out_edges_.emplace(UnwrapEdgesResult(vertex.OutEdges(self_.view_, self_.common_.edge_types)));
auto edges_result = UnwrapEdgesResult(vertex.OutEdges(self_.view_, self_.common_.edge_types));
out_edges_.emplace(edges_result.edges);
num_expanded_second = edges_result.expanded_count;
}
if (out_edges_) {
out_edges_it_.emplace(out_edges_->begin());
}
}
if (!expansion_info_.existing_node) {
return true;
}
num_expanded_first = num_expanded_first == -1 ? 0 : num_expanded_first;
num_expanded_second = num_expanded_second == -1 ? 0 : num_expanded_second;
int64_t total_expanded_edges = num_expanded_first + num_expanded_second;
if (!expansion_info_.reversed) {
prev_input_degree_ = total_expanded_edges;
} else {
prev_existing_degree_ = total_expanded_edges;
}
return true;
}
}
@@ -911,11 +991,12 @@ auto ExpandFromVertex(const VertexAccessor &vertex, EdgeAtom::Direction directio
};
storage::View view = storage::View::OLD;
utils::pmr::vector<decltype(wrapper(direction, *vertex.InEdges(view, edge_types)))> chain_elements(memory);
utils::pmr::vector<decltype(wrapper(direction, vertex.InEdges(view, edge_types).GetValue().edges))> chain_elements(
memory);
if (direction != EdgeAtom::Direction::OUT) {
db_accessor->PrefetchInEdges(vertex);
auto edges = UnwrapEdgesResult(vertex.InEdges(view, edge_types));
auto edges = UnwrapEdgesResult(vertex.InEdges(view, edge_types)).edges;
if (edges.begin() != edges.end()) {
chain_elements.emplace_back(wrapper(EdgeAtom::Direction::IN, std::move(edges)));
}
@@ -923,7 +1004,7 @@ auto ExpandFromVertex(const VertexAccessor &vertex, EdgeAtom::Direction directio
if (direction != EdgeAtom::Direction::IN) {
db_accessor->PrefetchOutEdges(vertex);
auto edges = UnwrapEdgesResult(vertex.OutEdges(view, edge_types));
auto edges = UnwrapEdgesResult(vertex.OutEdges(view, edge_types)).edges;
if (edges.begin() != edges.end()) {
chain_elements.emplace_back(wrapper(EdgeAtom::Direction::OUT, std::move(edges)));
}
@@ -1283,7 +1364,7 @@ class STShortestPathCursor : public query::plan::Cursor {
for (const auto &vertex : source_frontier) {
if (self_.common_.direction != EdgeAtom::Direction::IN) {
context.db_accessor->PrefetchOutEdges(vertex);
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types));
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : out_edges) {
#ifdef MG_ENTERPRISE
if (license::global_license_checker.IsEnterpriseValidFast() && context.auth_checker &&
@@ -1310,7 +1391,7 @@ class STShortestPathCursor : public query::plan::Cursor {
}
if (self_.common_.direction != EdgeAtom::Direction::OUT) {
dba.PrefetchInEdges(vertex);
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types));
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : in_edges) {
#ifdef MG_ENTERPRISE
if (license::global_license_checker.IsEnterpriseValidFast() && context.auth_checker &&
@@ -1351,7 +1432,7 @@ class STShortestPathCursor : public query::plan::Cursor {
for (const auto &vertex : sink_frontier) {
if (self_.common_.direction != EdgeAtom::Direction::OUT) {
context.db_accessor->PrefetchOutEdges(vertex);
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types));
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : out_edges) {
#ifdef MG_ENTERPRISE
if (license::global_license_checker.IsEnterpriseValidFast() && context.auth_checker &&
@@ -1377,7 +1458,7 @@ class STShortestPathCursor : public query::plan::Cursor {
}
if (self_.common_.direction != EdgeAtom::Direction::IN) {
dba.PrefetchInEdges(vertex);
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types));
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : in_edges) {
#ifdef MG_ENTERPRISE
if (license::global_license_checker.IsEnterpriseValidFast() && context.auth_checker &&
@@ -1469,12 +1550,12 @@ class SingleSourceShortestPathCursor : public query::plan::Cursor {
auto expand_from_vertex = [this, &expand_pair, &context](const auto &vertex) {
if (self_.common_.direction != EdgeAtom::Direction::IN) {
context.db_accessor->PrefetchOutEdges(vertex);
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types));
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : out_edges) expand_pair(edge, edge.To());
}
if (self_.common_.direction != EdgeAtom::Direction::OUT) {
context.db_accessor->PrefetchInEdges(vertex);
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types));
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : in_edges) expand_pair(edge, edge.From());
}
};
@@ -1672,14 +1753,14 @@ class ExpandWeightedShortestPathCursor : public query::plan::Cursor {
int64_t depth) {
if (self_.common_.direction != EdgeAtom::Direction::IN) {
context.db_accessor->PrefetchOutEdges(vertex);
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types));
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : out_edges) {
expand_pair(edge, edge.To(), weight, depth);
}
}
if (self_.common_.direction != EdgeAtom::Direction::OUT) {
context.db_accessor->PrefetchInEdges(vertex);
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types));
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : in_edges) {
expand_pair(edge, edge.From(), weight, depth);
}
@@ -1938,7 +2019,7 @@ class ExpandAllShortestPathsCursor : public query::plan::Cursor {
int64_t depth) {
if (self_.common_.direction != EdgeAtom::Direction::IN) {
context.db_accessor->PrefetchOutEdges(vertex);
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types));
auto out_edges = UnwrapEdgesResult(vertex.OutEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : out_edges) {
#ifdef MG_ENTERPRISE
if (license::global_license_checker.IsEnterpriseValidFast() && context.auth_checker &&
@@ -1953,7 +2034,7 @@ class ExpandAllShortestPathsCursor : public query::plan::Cursor {
}
if (self_.common_.direction != EdgeAtom::Direction::OUT) {
context.db_accessor->PrefetchInEdges(vertex);
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types));
auto in_edges = UnwrapEdgesResult(vertex.InEdges(storage::View::OLD, self_.common_.edge_types)).edges;
for (const auto &edge : in_edges) {
#ifdef MG_ENTERPRISE
if (license::global_license_checker.IsEnterpriseValidFast() && context.auth_checker &&

16
src/query/plan/operator.hpp
View File

@@ -764,6 +764,13 @@ struct ExpandCommon {
bool existing_node;
};
struct ExpansionInfo {
std::optional<VertexAccessor> input_node;
EdgeAtom::Direction direction;
std::optional<VertexAccessor> existing_node;
bool reversed{false};
};
/// Expansion operator. For a node existing in the frame it
/// expands one edge and one node and places them on the frame.
///
@@ -806,14 +813,16 @@ class Expand : public memgraph::query::plan::LogicalOperator {
class ExpandCursor : public Cursor {
public:
ExpandCursor(const Expand &, utils::MemoryResource *);
ExpandCursor(const Expand &, int64_t input_degree, int64_t existing_node_degree, utils::MemoryResource *);
bool Pull(Frame &, ExecutionContext &) override;
void Shutdown() override;
void Reset() override;
ExpansionInfo GetExpansionInfo(Frame &);
private:
using InEdgeT = std::remove_reference_t<decltype(*std::declval<VertexAccessor>().InEdges(storage::View::OLD))>;
using InEdgeT = std::vector<EdgeAccessor>;
using InEdgeIteratorT = decltype(std::declval<InEdgeT>().begin());
using OutEdgeT = std::remove_reference_t<decltype(*std::declval<VertexAccessor>().OutEdges(storage::View::OLD))>;
using OutEdgeT = std::vector<EdgeAccessor>;
using OutEdgeIteratorT = decltype(std::declval<OutEdgeT>().begin());
const Expand &self_;
@@ -826,6 +835,9 @@ class Expand : public memgraph::query::plan::LogicalOperator {
std::optional<InEdgeIteratorT> in_edges_it_;
std::optional<OutEdgeT> out_edges_;
std::optional<OutEdgeIteratorT> out_edges_it_;
ExpansionInfo expansion_info_;
int64_t prev_input_degree_{-1};
int64_t prev_existing_degree_{-1};
bool InitEdges(Frame &, ExecutionContext &);
};

4
src/query/procedure/mg_procedure_impl.cpp
View File

@@ -2120,7 +2120,7 @@ mgp_error mgp_vertex_iter_in_edges(mgp_vertex *v, mgp_memory *memory, mgp_edges_
LOG_FATAL("Unexpected error when getting the inbound edges of a vertex.");
}
}
it->in.emplace(std::move(*maybe_edges));
it->in.emplace(std::move(maybe_edges->edges));
it->in_it.emplace(it->in->begin());
#ifdef MG_ENTERPRISE
if (memgraph::license::global_license_checker.IsEnterpriseValidFast()) {
@@ -2180,7 +2180,7 @@ mgp_error mgp_vertex_iter_out_edges(mgp_vertex *v, mgp_memory *memory, mgp_edges
}
}
it->out.emplace(std::move(*maybe_edges));
it->out.emplace(std::move(maybe_edges->edges));
it->out_it.emplace(it->out->begin());
#ifdef MG_ENTERPRISE

10
src/query/procedure/mg_procedure_impl.hpp
View File

@@ -687,12 +687,14 @@ struct mgp_edges_iterator {
memgraph::utils::MemoryResource *memory;
mgp_vertex source_vertex;
std::optional<std::remove_reference_t<
decltype(*std::get<memgraph::query::VertexAccessor>(source_vertex.impl).InEdges(source_vertex.graph->view))>>
std::optional<std::remove_reference_t<decltype(std::get<memgraph::query::VertexAccessor>(source_vertex.impl)
.InEdges(source_vertex.graph->view)
->edges)>>
in;
std::optional<decltype(in->begin())> in_it;
std::optional<std::remove_reference_t<
decltype(*std::get<memgraph::query::VertexAccessor>(source_vertex.impl).OutEdges(source_vertex.graph->view))>>
std::optional<std::remove_reference_t<decltype(std::get<memgraph::query::VertexAccessor>(source_vertex.impl)
.OutEdges(source_vertex.graph->view)
->edges)>>
out;
std::optional<decltype(out->begin())> out_it;
std::optional<mgp_edge> current_e;

4
src/query/trigger_context.cpp
View File

@@ -306,7 +306,7 @@ void TriggerContext::AdaptForAccessor(DbAccessor *accessor) {
auto maybe_out_edges = maybe_from_vertex->OutEdges(storage::View::OLD);
MG_ASSERT(maybe_out_edges.HasValue());
const auto edge_gid = created_edge.object.Gid();
for (const auto &edge : *maybe_out_edges) {
for (const auto &edge : maybe_out_edges->edges) {
if (edge.Gid() == edge_gid) {
*it = detail::CreatedObject{edge};
++it;
@@ -327,7 +327,7 @@ void TriggerContext::AdaptForAccessor(DbAccessor *accessor) {
accessor->PrefetchOutEdges(*maybe_vertex);
auto maybe_out_edges = maybe_vertex->OutEdges(storage::View::OLD);
MG_ASSERT(maybe_out_edges.HasValue());
for (const auto &edge : *maybe_out_edges) {
for (const auto &edge : maybe_out_edges->edges) {
if (edge.Gid() == value.object.Gid()) {
*it = std::move(value);
it->object = edge;

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

@@ -1178,10 +1178,10 @@ bool DiskStorage::DiskAccessor::PrefetchEdgeFilter(const std::string_view disk_e
}
MG_ASSERT(edges_res.HasValue());
auto edges = edges_res.GetValue();
bool isEdgeAlreadyInMemory = std::any_of(edges.begin(), edges.end(), [edge_gid](const auto &edge_acc) {
return utils::SerializeIdType(edge_acc.Gid()) == edge_gid;
});
auto edges_result = edges_res.GetValue();
bool isEdgeAlreadyInMemory =
std::any_of(edges_result.edges.begin(), edges_result.edges.end(),
[edge_gid](const auto &edge_acc) { return utils::SerializeIdType(edge_acc.Gid()) == edge_gid; });
return !isEdgeAlreadyInMemory;
}

4
src/storage/v2/durability/snapshot.cpp
View File

@@ -1533,14 +1533,14 @@ void CreateSnapshot(Transaction *transaction, const std::filesystem::path &snaps
write_mapping(item.first);
snapshot.WritePropertyValue(item.second);
}
const auto &in_edges = maybe_in_edges.GetValue();
const auto &in_edges = maybe_in_edges.GetValue().edges;
snapshot.WriteUint(in_edges.size());
for (const auto &item : in_edges) {
snapshot.WriteUint(item.Gid().AsUint());
snapshot.WriteUint(item.FromVertex().Gid().AsUint());
write_mapping(item.EdgeType());
}
const auto &out_edges = maybe_out_edges.GetValue();
const auto &out_edges = maybe_out_edges.GetValue().edges;
snapshot.WriteUint(out_edges.size());
for (const auto &item : out_edges) {
snapshot.WriteUint(item.Gid().AsUint());

4
src/storage/v2/inmemory/replication/replication_server.cpp
View File

@@ -388,8 +388,8 @@ uint64_t InMemoryReplicationServer::ReadAndApplyDelta(InMemoryStorage *storage,
auto edges = from_vertex->OutEdges(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];
if (edges->edges.size() != 1) throw utils::BasicException("Invalid transaction!");
auto &edge = (*edges).edges[0];
auto ret = transaction->DeleteEdge(&edge);
if (ret.HasError()) throw utils::BasicException("Invalid transaction!");
break;

18
src/storage/v2/vertex_accessor.cpp
View File

@@ -423,7 +423,7 @@ Result<std::map<PropertyId, PropertyValue>> VertexAccessor::Properties(View view
return std::move(properties);
}
Result<std::vector<EdgeAccessor>> VertexAccessor::InEdges(View view, const std::vector<EdgeTypeId> &edge_types,
Result<EdgesVertexAccessorResult> VertexAccessor::InEdges(View view, const std::vector<EdgeTypeId> &edge_types,
const VertexAccessor *destination) const {
MG_ASSERT(!destination || destination->transaction_ == transaction_, "Invalid accessor!");
@@ -445,9 +445,11 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::InEdges(View view, const std::
bool deleted = false;
auto in_edges = edge_store{};
Delta *delta = nullptr;
int64_t expanded_count = 0;
{
auto guard = std::shared_lock{vertex_->lock};
deleted = vertex_->deleted;
expanded_count = static_cast<int64_t>(vertex_->in_edges.size());
// TODO: a better filter copy
if (edge_types.empty() && !destination) {
in_edges = vertex_->in_edges;
@@ -472,7 +474,7 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::InEdges(View view, const std::
auto const &cache = transaction_->manyDeltasCache;
if (auto resError = HasError(view, cache, vertex_, for_deleted_); resError) return *resError;
if (auto resInEdges = cache.GetInEdges(view, vertex_, destination_vertex, edge_types); resInEdges)
return {build_result(*resInEdges)};
return EdgesVertexAccessorResult{.edges = build_result(*resInEdges), .expanded_count = expanded_count};
}
auto const n_processed = ApplyDeltasForRead(
@@ -497,10 +499,11 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::InEdges(View view, const std::
if (!exists) return Error::NONEXISTENT_OBJECT;
if (deleted) return Error::DELETED_OBJECT;
return build_result(in_edges);
return EdgesVertexAccessorResult{.edges = build_result(in_edges), .expanded_count = expanded_count};
}
Result<std::vector<EdgeAccessor>> VertexAccessor::OutEdges(View view, const std::vector<EdgeTypeId> &edge_types,
Result<EdgesVertexAccessorResult> VertexAccessor::OutEdges(View view, const std::vector<EdgeTypeId> &edge_types,
const VertexAccessor *destination) const {
MG_ASSERT(!destination || destination->transaction_ == transaction_, "Invalid accessor!");
@@ -521,9 +524,11 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::OutEdges(View view, const std:
bool deleted = false;
auto out_edges = edge_store{};
Delta *delta = nullptr;
int64_t expanded_count = 0;
{
auto guard = std::shared_lock{vertex_->lock};
deleted = vertex_->deleted;
expanded_count = static_cast<int64_t>(vertex_->out_edges.size());
if (edge_types.empty() && !destination) {
out_edges = vertex_->out_edges;
} else {
@@ -547,7 +552,7 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::OutEdges(View view, const std:
auto const &cache = transaction_->manyDeltasCache;
if (auto resError = HasError(view, cache, vertex_, for_deleted_); resError) return *resError;
if (auto resOutEdges = cache.GetOutEdges(view, vertex_, dst_vertex, edge_types); resOutEdges)
return {build_result(*resOutEdges)};
return EdgesVertexAccessorResult{.edges = build_result(*resOutEdges), .expanded_count = expanded_count};
}
auto const n_processed = ApplyDeltasForRead(
@@ -571,7 +576,8 @@ Result<std::vector<EdgeAccessor>> VertexAccessor::OutEdges(View view, const std:
if (!exists) return Error::NONEXISTENT_OBJECT;
if (deleted) return Error::DELETED_OBJECT;
return build_result(out_edges);
return EdgesVertexAccessorResult{.edges = build_result(out_edges), .expanded_count = expanded_count};
}
Result<size_t> VertexAccessor::InDegree(View view) const {

10
src/storage/v2/vertex_accessor.hpp
View File

@@ -26,6 +26,7 @@ class EdgeAccessor;
class Storage;
struct Constraints;
struct Indices;
struct EdgesVertexAccessorResult;
class VertexAccessor final {
private:
@@ -89,13 +90,13 @@ class VertexAccessor final {
/// @throw std::bad_alloc
/// @throw std::length_error if the resulting vector exceeds
/// std::vector::max_size().
Result<std::vector<EdgeAccessor>> InEdges(View view, const std::vector<EdgeTypeId> &edge_types = {},
Result<EdgesVertexAccessorResult> InEdges(View view, const std::vector<EdgeTypeId> &edge_types = {},
const VertexAccessor *destination = nullptr) const;
/// @throw std::bad_alloc
/// @throw std::length_error if the resulting vector exceeds
/// std::vector::max_size().
Result<std::vector<EdgeAccessor>> OutEdges(View view, const std::vector<EdgeTypeId> &edge_types = {},
Result<EdgesVertexAccessorResult> OutEdges(View view, const std::vector<EdgeTypeId> &edge_types = {},
const VertexAccessor *destination = nullptr) const;
Result<size_t> InDegree(View view) const;
@@ -125,6 +126,11 @@ class VertexAccessor final {
bool for_deleted_{false};
};
struct EdgesVertexAccessorResult {
std::vector<EdgeAccessor> edges;
int64_t expanded_count;
};
} // namespace memgraph::storage
namespace std {