Compare commits

...

36 Commits

Author SHA1 Message Date
Marko Budiselic
bd5be78dfd Add dummy shard_rsm simulation test 2023-03-14 13:42:39 +00:00
Marko Budiselic
f94f72438d Add simple test and rr implementation 2023-03-14 12:49:41 +00:00
Marko Budiselic
491b651dcb Extend dummy test 2023-03-14 12:07:47 +00:00
Marko Budiselic
f42bd3237e A bit of cleanup 2023-03-14 10:22:10 +00:00
Marko Budiselic
ff07360b85 Add requests placeholders 2023-03-13 18:46:20 +00:00
Marko Budiselic
88724f18fa Add Path and Graph dummy msgs 2023-03-13 18:03:25 +00:00
János Benjamin Antal
27d99b620d Merge pull request #761 from memgraph/MG-access-control-benchmark-fixes
[🍍 < MG] Fix mgbench access control benchmark
2023-02-17 17:33:58 +01:00
János Benjamin Antal
a2ce9c4396 Trigger CI 2023-02-15 13:59:31 +01:00
János Benjamin Antal
3b0d531343 Supress clang-tidy warning 2023-02-15 09:07:48 +01:00
János Benjamin Antal
a17010ed16 Supress clang-tidy warnings 2023-02-15 08:37:45 +01:00
János Benjamin Antal
74f53369c0 Add two more queries to simple benchmark 2023-02-09 13:17:55 +01:00
János Benjamin Antal
2b3141879b Make the output table nicer for comparing results 2023-02-09 13:17:46 +01:00
János Benjamin Antal
53f95ed1a7 Format python file 2023-02-09 13:17:15 +01:00
János Benjamin Antal
b678e6a63b Handle bool flags properly in benchmark runner 2023-02-09 12:42:02 +01:00
János Benjamin Antal
563035645c Add command line flag to opt for using MultiFrame 2023-02-09 12:41:41 +01:00
János Benjamin Antal
12bc78ca2d Add command line flag to determine MultiFrame size 2023-02-09 12:41:17 +01:00
János Benjamin Antal
a9a388ce44 Use parametrized queries for vertex creation 2023-02-08 13:52:51 +01:00
János Benjamin Antal
6bc2e6d8b6 Merge remote-tracking branch 'origin/project-pineapples' into MG-access-control-benchmark-fixes 2023-02-08 13:48:50 +01:00
János Benjamin Antal
a02abc8f79 Merge pull request #763 from memgraph/T1235-MG-implement-EdgeUniquenessFilter-with-PullMultiple
Implement edge uniqueness filter with pull multiple
2023-02-08 13:40:15 +01:00
gvolfing
096d1ce5f4 Invert boolean logic when checking for unique edges 2023-02-08 12:57:21 +01:00
gvolfing
657279949a Fix compile error 2023-02-08 12:13:46 +01:00
gvolfing
25226cca92 Update src/query/v2/plan/operator.cpp
Co-authored-by: Jure Bajic <jure.bajic@memgraph.com>
2023-02-08 11:41:43 +01:00
János Benjamin Antal
292a55f4ff Add new line at the end of dataset file 2023-02-08 11:28:19 +01:00
János Benjamin Antal
b26c7d09ef Ignore not value equality property filters for ScanByPrimaryKey 2023-02-06 16:39:42 +01:00
János Benjamin Antal
4bad8c0d1e Filter edges on types 2023-02-06 16:14:39 +01:00
János Benjamin Antal
2b01f2280c Add TODO about failing query 2023-02-06 16:14:21 +01:00
János Benjamin Antal
7bf8550c86 Merge remote-tracking branch 'origin/project-pineapples' into MG-access-control-benchmark-fixes 2023-02-06 08:18:25 +01:00
János Benjamin Antal
41183b328b Invalidate consumed frames in ScanByPrimaryKey 2023-02-01 17:19:02 +01:00
János Benjamin Antal
c9a0c15c16 Make frames invalid on consumption 2023-02-01 16:17:06 +01:00
János Benjamin Antal
50327254e0 Make queries into a single line 2023-02-01 14:59:50 +01:00
János Benjamin Antal
24ae6069f0 Split edge creation into batches 2023-02-01 14:31:56 +01:00
János Benjamin Antal
7be66f0c54 Add unwind based dataset creator 2023-02-01 14:24:04 +01:00
János Benjamin Antal
b136cd71d2 Fix DistributedCreatedNodeCursor in case of UNWIND 2023-02-01 14:22:47 +01:00
János Benjamin Antal
a38401130e Set vertex id in Expand properly 2023-02-01 13:24:58 +01:00
János Benjamin Antal
9a805bff8b Merge remote-tracking branch 'origin/T1226-MG-Implement-scanbyprimarykey-with-multiframe' into MG-create-properties 2023-02-01 11:37:06 +01:00
János Benjamin Antal
da28a29c7f Pass properties when creating vertices 2023-01-31 17:09:41 +01:00
25 changed files with 454 additions and 158 deletions

View File

@@ -91,4 +91,12 @@ CheckOptions:
value: llvm
- key: modernize-use-nullptr.NullMacros
value: 'NULL'
- key: readability-identifier-length.MinimumVariableNameLength
value: '0'
- key: readability-identifier-length.MinimumParameterNameLength
value: '0'
- key: readability-identifier-length.MinimumLoopCounterNameLength
value: '0'
- key: readability-identifier-length.MinimumExceptionNameLength
value: '0'
...

View File

@@ -1,21 +1,21 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v2.3.0
rev: v4.4.0
hooks:
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/psf/black
rev: 22.10.0
rev: 23.1.0
hooks:
- id: black
- repo: https://github.com/pycqa/isort
rev: 5.10.1
rev: 5.12.0
hooks:
- id: isort
name: isort (python)
args: ["--profile", "black"]
- repo: https://github.com/pre-commit/mirrors-clang-format
rev: v13.0.0
rev: v15.0.7
hooks:
- id: clang-format

View File

@@ -64,7 +64,11 @@
#include "utils/tsc.hpp"
#include "utils/variant_helpers.hpp"
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_bool(use_multi_frame, false, "Whether to use MultiFrame or not");
namespace EventCounter {
extern Event ReadQuery;
extern Event WriteQuery;
extern Event ReadWriteQuery;
@@ -74,6 +78,7 @@ extern const Event LabelPropertyIndexCreated;
extern const Event StreamsCreated;
extern const Event TriggersCreated;
} // namespace EventCounter
namespace memgraph::query::v2 {
@@ -688,7 +693,7 @@ PullPlan::PullPlan(const std::shared_ptr<CachedPlan> plan, const Parameters &par
: plan_(plan),
cursor_(plan->plan().MakeCursor(execution_memory)),
frame_(plan->symbol_table().max_position(), execution_memory),
multi_frame_(plan->symbol_table().max_position(), kNumberOfFramesInMultiframe, execution_memory),
multi_frame_(plan->symbol_table().max_position(), FLAGS_default_multi_frame_size, execution_memory),
memory_limit_(memory_limit) {
ctx_.db_accessor = dba;
ctx_.symbol_table = plan->symbol_table();
@@ -812,8 +817,7 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::PullMultiple(AnyStrea
std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *stream, std::optional<int> n,
const std::vector<Symbol> &output_symbols,
std::map<std::string, TypedValue> *summary) {
auto should_pull_multiple = false; // TODO on the long term, we will only use PullMultiple
if (should_pull_multiple) {
if (FLAGS_use_multi_frame) {
return PullMultiple(stream, n, output_symbols, summary);
}
// Set up temporary memory for a single Pull. Initial memory comes from the

View File

@@ -17,6 +17,9 @@
#include "query/v2/bindings/frame.hpp"
#include "utils/pmr/vector.hpp"
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DEFINE_uint64(default_multi_frame_size, 100, "Default size of MultiFrame");
namespace memgraph::query::v2 {
static_assert(std::forward_iterator<ValidFramesReader::Iterator>);

View File

@@ -13,10 +13,14 @@
#include <iterator>
#include <gflags/gflags.h>
#include "query/v2/bindings/frame.hpp"
// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
DECLARE_uint64(default_multi_frame_size);
namespace memgraph::query::v2 {
constexpr uint64_t kNumberOfFramesInMultiframe = 1000; // TODO have it configurable
class ValidFramesConsumer;
class ValidFramesModifier;

View File

@@ -218,6 +218,7 @@ class DistributedCreateNodeCursor : public Cursor {
}
std::vector<msgs::NewVertex> NodeCreationInfoToRequest(ExecutionContext &context, Frame &frame) {
primary_keys_.clear();
std::vector<msgs::NewVertex> requests;
msgs::PrimaryKey pk;
msgs::NewVertex rqst;
@@ -227,22 +228,27 @@ class DistributedCreateNodeCursor : public Cursor {
ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, nullptr,
storage::v3::View::NEW);
if (const auto *node_info_properties = std::get_if<PropertiesMapList>(&node_info_.properties)) {
for (const auto &[key, value_expression] : *node_info_properties) {
for (const auto &[property, value_expression] : *node_info_properties) {
TypedValue val = value_expression->Accept(evaluator);
if (context.request_router->IsPrimaryKey(primary_label, key)) {
rqst.primary_key.push_back(TypedValueToValue(val));
pk.push_back(TypedValueToValue(val));
auto msgs_value = TypedValueToValue(val);
if (context.request_router->IsPrimaryProperty(primary_label, property)) {
rqst.primary_key.push_back(msgs_value);
pk.push_back(std::move(msgs_value));
} else {
rqst.properties.emplace_back(property, std::move(msgs_value));
}
}
} else {
auto property_map = evaluator.Visit(*std::get<ParameterLookup *>(node_info_.properties)).ValueMap();
for (const auto &[key, value] : property_map) {
auto key_str = std::string(key);
auto property_id = context.request_router->NameToProperty(key_str);
if (context.request_router->IsPrimaryKey(primary_label, property_id)) {
rqst.primary_key.push_back(TypedValueToValue(value));
pk.push_back(TypedValueToValue(value));
}
for (const auto &[property, typed_value] : property_map) {
auto property_str = std::string(property);
auto property_id = context.request_router->NameToProperty(property_str);
auto msgs_value = TypedValueToValue(typed_value);
if (context.request_router->IsPrimaryProperty(primary_label, property_id)) {
rqst.primary_key.push_back(msgs_value);
pk.push_back(std::move(msgs_value));
} else
rqst.properties.emplace_back(property_id, std::move(msgs_value));
}
}
@@ -268,6 +274,7 @@ class DistributedCreateNodeCursor : public Cursor {
}
std::vector<msgs::NewVertex> NodeCreationInfoToRequests(ExecutionContext &context, MultiFrame &multi_frame) {
primary_keys_.clear();
std::vector<msgs::NewVertex> requests;
auto multi_frame_modifier = multi_frame.GetValidFramesModifier();
for (auto &frame : multi_frame_modifier) {
@@ -280,22 +287,27 @@ class DistributedCreateNodeCursor : public Cursor {
ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, nullptr,
storage::v3::View::NEW);
if (const auto *node_info_properties = std::get_if<PropertiesMapList>(&node_info_.properties)) {
for (const auto &[key, value_expression] : *node_info_properties) {
for (const auto &[property, value_expression] : *node_info_properties) {
TypedValue val = value_expression->Accept(evaluator);
if (context.request_router->IsPrimaryKey(primary_label, key)) {
rqst.primary_key.push_back(TypedValueToValue(val));
pk.push_back(TypedValueToValue(val));
auto msgs_value = TypedValueToValue(val);
if (context.request_router->IsPrimaryProperty(primary_label, property)) {
rqst.primary_key.push_back(msgs_value);
pk.push_back(std::move(msgs_value));
} else {
rqst.properties.emplace_back(property, std::move(msgs_value));
}
}
} else {
auto property_map = evaluator.Visit(*std::get<ParameterLookup *>(node_info_.properties)).ValueMap();
for (const auto &[key, value] : property_map) {
auto key_str = std::string(key);
auto property_id = context.request_router->NameToProperty(key_str);
if (context.request_router->IsPrimaryKey(primary_label, property_id)) {
rqst.primary_key.push_back(TypedValueToValue(value));
pk.push_back(TypedValueToValue(value));
}
for (const auto &[property, typed_value] : property_map) {
auto property_str = std::string(property);
auto property_id = context.request_router->NameToProperty(property_str);
auto msgs_value = TypedValueToValue(typed_value);
if (context.request_router->IsPrimaryProperty(primary_label, property_id)) {
rqst.primary_key.push_back(msgs_value);
pk.push_back(std::move(msgs_value));
} else
rqst.properties.emplace_back(property_id, std::move(msgs_value));
}
}
@@ -497,7 +509,7 @@ class DistributedScanAllAndFilterCursor : public Cursor {
if (!own_multi_frame_.has_value()) {
own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
own_frames_it_ = own_frames_consumer_->begin();
}
@@ -693,7 +705,7 @@ class DistributedScanByPrimaryKeyCursor : public Cursor {
void EnsureOwnMultiFrameIsGood(MultiFrame &output_multi_frame) {
if (!own_multi_frame_.has_value()) {
own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
own_frames_it_ = own_frames_consumer_->begin();
}
@@ -1335,15 +1347,14 @@ bool ContainsSameEdge(const TypedValue &a, const TypedValue &b) {
}
bool IsExpansionOk(Frame &frame, const Symbol &expand_symbol, const std::vector<Symbol> &previous_symbols) {
const auto &expand_value = frame[expand_symbol];
for (const auto &previous_symbol : previous_symbols) {
const auto &previous_value = frame[previous_symbol];
// This shouldn't raise a TypedValueException, because the planner
// makes sure these are all of the expected type. In case they are not
// an error should be raised long before this code is executed.
if (ContainsSameEdge(previous_value, expand_value)) return false;
}
return true;
// This shouldn't raise a TypedValueException, because the planner
// makes sure these are all of the expected type. In case they are not
// an error should be raised long before this code is executed.
return std::ranges::all_of(previous_symbols,
[&frame, &expand_value = frame[expand_symbol]](const auto &previous_symbol) {
const auto &previous_value = frame[previous_symbol];
return !ContainsSameEdge(previous_value, expand_value);
});
}
} // namespace
@@ -1595,6 +1606,7 @@ class AggregateCursor : public Cursor {
ExpressionEvaluator evaluator(&frame, context->symbol_table, context->evaluation_context,
context->request_router, storage::v3::View::NEW);
ProcessOne(frame, &evaluator);
frame.MakeInvalid();
}
}
@@ -2201,7 +2213,7 @@ class UnwindCursor : public Cursor {
if (!own_multi_frame_.has_value()) {
own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
own_frames_it_ = own_frames_consumer_->begin();
}
@@ -2960,27 +2972,16 @@ class DistributedCreateExpandCursor : public Cursor {
const auto &v1 = v1_value.ValueVertex();
const auto &v2 = OtherVertex(frame);
// Set src and dest vertices
// TODO(jbajic) Currently we are only handling scenario where vertices
// are matched
const auto set_vertex = [&context](const auto &vertex, auto &vertex_id) {
vertex_id.first = vertex.PrimaryLabel();
for (const auto &[key, val] : vertex.Properties()) {
if (context.request_router->IsPrimaryKey(vertex_id.first.id, key)) {
vertex_id.second.push_back(val);
}
}
};
std::invoke([&]() {
switch (edge_info.direction) {
case EdgeAtom::Direction::IN: {
set_vertex(v2, request.src_vertex);
set_vertex(v1, request.dest_vertex);
request.src_vertex = v2.Id();
request.dest_vertex = v1.Id();
break;
}
case EdgeAtom::Direction::OUT: {
set_vertex(v1, request.src_vertex);
set_vertex(v2, request.dest_vertex);
request.src_vertex = v1.Id();
request.dest_vertex = v2.Id();
break;
}
case EdgeAtom::Direction::BOTH:
@@ -3122,6 +3123,9 @@ class DistributedExpandCursor : public Cursor {
auto &vertex = vertex_value.ValueVertex();
msgs::ExpandOneRequest request;
request.direction = DirectionToMsgsDirection(self_.common_.direction);
std::transform(self_.common_.edge_types.begin(), self_.common_.edge_types.end(),
std::back_inserter(request.edge_types),
[](const storage::v3::EdgeTypeId edge_type_id) { return msgs::EdgeType{edge_type_id}; });
// to not fetch any properties of the edges
request.edge_properties.emplace();
request.src_vertices.push_back(vertex.Id());
@@ -3262,6 +3266,9 @@ class DistributedExpandCursor : public Cursor {
msgs::ExpandOneRequest request;
request.direction = DirectionToMsgsDirection(self_.common_.direction);
std::transform(self_.common_.edge_types.begin(), self_.common_.edge_types.end(),
std::back_inserter(request.edge_types),
[](const storage::v3::EdgeTypeId edge_type_id) { return msgs::EdgeType{edge_type_id}; });
// to not fetch any properties of the edges
request.edge_properties.emplace();
for (const auto &frame : own_multi_frame_->GetValidFramesReader()) {
@@ -3375,7 +3382,7 @@ class DistributedExpandCursor : public Cursor {
void EnsureOwnMultiFrameIsGood(MultiFrame &output_multi_frame) {
if (!own_multi_frame_.has_value()) {
own_multi_frame_.emplace(MultiFrame(output_multi_frame.GetFirstFrame().elems().size(),
kNumberOfFramesInMultiframe, output_multi_frame.GetMemoryResource()));
FLAGS_default_multi_frame_size, output_multi_frame.GetMemoryResource()));
own_frames_consumer_.emplace(own_multi_frame_->GetValidFramesConsumer());
own_frames_it_ = own_frames_consumer_->begin();
}

View File

@@ -597,6 +597,9 @@ class IndexLookupRewriter final : public HierarchicalLogicalOperatorVisitor {
[](const auto &schema_elem) { return schema_elem.property_id; });
for (const auto &property_filter : property_filters) {
if (property_filter.property_filter->type_ != PropertyFilter::Type::EQUAL) {
continue;
}
const auto &property_id = db_->NameToProperty(property_filter.property_filter->property_.name);
if (std::find(schema_properties.begin(), schema_properties.end(), property_id) != schema_properties.end()) {
pk_temp.emplace_back(std::make_pair(property_filter.expression, property_filter));

View File

@@ -106,6 +106,7 @@ class RequestRouterInterface {
virtual std::vector<msgs::ExpandOneResultRow> ExpandOne(msgs::ExpandOneRequest request) = 0;
virtual std::vector<msgs::CreateExpandResponse> CreateExpand(std::vector<msgs::NewExpand> new_edges) = 0;
virtual std::vector<msgs::GetPropertiesResultRow> GetProperties(msgs::GetPropertiesRequest request) = 0;
virtual std::vector<msgs::GraphResponse> GetGraph(msgs::GraphRequest req) = 0;
virtual storage::v3::EdgeTypeId NameToEdgeType(const std::string &name) const = 0;
virtual storage::v3::PropertyId NameToProperty(const std::string &name) const = 0;
@@ -117,7 +118,7 @@ class RequestRouterInterface {
virtual std::optional<storage::v3::EdgeTypeId> MaybeNameToEdgeType(const std::string &name) const = 0;
virtual std::optional<storage::v3::LabelId> MaybeNameToLabel(const std::string &name) const = 0;
virtual bool IsPrimaryLabel(storage::v3::LabelId label) const = 0;
virtual bool IsPrimaryKey(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const = 0;
virtual bool IsPrimaryProperty(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const = 0;
virtual std::optional<std::pair<uint64_t, uint64_t>> AllocateInitialEdgeIds(io::Address coordinator_address) = 0;
virtual void InstallSimulatorTicker(std::function<bool()> tick_simulator) = 0;
@@ -231,7 +232,7 @@ class RequestRouter : public RequestRouterInterface {
return edge_types_.IdToName(id.AsUint());
}
bool IsPrimaryKey(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const override {
bool IsPrimaryProperty(storage::v3::LabelId primary_label, storage::v3::PropertyId property) const override {
const auto schema_it = shards_map_.schemas.find(primary_label);
MG_ASSERT(schema_it != shards_map_.schemas.end(), "Invalid primary label id: {}", primary_label.AsUint());
@@ -403,6 +404,44 @@ class RequestRouter : public RequestRouterInterface {
return result_rows;
}
std::vector<msgs::GraphResponse> GetGraph(msgs::GraphRequest req) override {
SPDLOG_WARN("RequestRouter::GetGraph(GraphRequest) not fully implemented");
// InitializeRequests
// TODO(gitbuda): This seems quite expensive because potentially a lot of requests has to be initialized.
auto multi_shards = shards_map_.GetAllShards();
std::vector<ShardRequestState<msgs::GraphRequest>> requests = {};
for (auto &shards : multi_shards) {
for (auto &[key, shard] : shards) {
MG_ASSERT(!shard.peers.empty());
msgs::GraphRequest request;
request.transaction_id = transaction_id_;
ShardRequestState<msgs::GraphRequest> shard_request_state{
.shard = shard,
.request = std::move(request),
};
requests.emplace_back(std::move(shard_request_state));
}
}
spdlog::trace("created {} Graph requests", requests.size());
// SendRequests and CollectResponses
RunningRequests<msgs::GraphRequest> running_requests = {};
running_requests.reserve(requests.size());
for (size_t i = 0; i < requests.size(); i++) {
auto &request = requests[i];
io::ReadinessToken readiness_token{i};
auto &storage_client = GetStorageClientForShard(request.shard);
storage_client.SendAsyncReadRequest(request.request, notifier_, readiness_token);
running_requests.emplace(readiness_token.GetId(), request);
}
spdlog::trace("sent {} Graph requests in parallel", running_requests.size());
auto responses = DriveReadResponses<msgs::GraphRequest, msgs::GraphResponse>(running_requests);
spdlog::trace("got back {} Graph responses after driving to completion", responses.size());
return responses;
}
std::optional<storage::v3::PropertyId> MaybeNameToProperty(const std::string &name) const override {
return shards_map_.GetPropertyId(name);
}

View File

@@ -79,6 +79,18 @@ struct Vertex {
friend bool operator==(const Vertex &lhs, const Vertex &rhs) { return lhs.id == rhs.id; }
};
struct Path {
// TODO(gitbuda): Define how a path should look like.
};
struct Graph {
std::vector<Vertex> vertices;
std::vector<Edge> edges;
friend bool operator==(const Graph &lhs, const Graph &rhs) {
LOG_FATAL("Implement memgraph::msgs::Graph::operator==");
}
};
// TODO(gitbuda): Figure out how to serialize memgraph::msgs::Graph/Path.
struct Null {};
struct Value {
@@ -562,6 +574,21 @@ struct UpdateEdgesResponse {
std::optional<ShardError> error;
};
// TODO(gitbuda): Add more filtering options.
struct GraphRequest {
Hlc transaction_id;
std::optional<VertexId> maybe_start_id;
// The empty optional means return all of the properties, while an empty list means do not return any properties
std::optional<std::vector<PropertyId>> props_to_return;
std::optional<size_t> batch_limit;
StorageView storage_view{StorageView::NEW};
};
struct GraphResponse {
std::optional<ShardError> error;
Graph data;
};
struct CommitRequest {
Hlc transaction_id;
Hlc commit_timestamp;
@@ -571,8 +598,8 @@ struct CommitResponse {
std::optional<ShardError> error;
};
using ReadRequests = std::variant<ExpandOneRequest, GetPropertiesRequest, ScanVerticesRequest>;
using ReadResponses = std::variant<ExpandOneResponse, GetPropertiesResponse, ScanVerticesResponse>;
using ReadRequests = std::variant<ExpandOneRequest, GetPropertiesRequest, ScanVerticesRequest, GraphRequest>;
using ReadResponses = std::variant<ExpandOneResponse, GetPropertiesResponse, ScanVerticesResponse, GraphResponse>;
using WriteRequests = std::variant<CreateVerticesRequest, DeleteVerticesRequest, UpdateVerticesRequest,
CreateExpandRequest, DeleteEdgesRequest, UpdateEdgesRequest, CommitRequest>;
@@ -582,7 +609,6 @@ using WriteResponses = std::variant<CreateVerticesResponse, DeleteVerticesRespon
} // namespace memgraph::msgs
namespace std {
template <>
struct hash<memgraph::msgs::Value>;

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
@@ -25,10 +25,12 @@
#include "utils/template_utils.hpp"
namespace memgraph::storage::v3 {
using EdgeAccessors = std::vector<storage::v3::EdgeAccessor>;
using EdgeUniquenessFunction = std::function<EdgeAccessors(EdgeAccessors &&, msgs::EdgeDirection)>;
using EdgeFiller =
std::function<ShardResult<void>(const EdgeAccessor &edge, bool is_in_edge, msgs::ExpandOneResultRow &result_row)>;
using conversions::FromPropertyValueToValue;
using msgs::Value;
template <typename T>

View File

@@ -50,17 +50,13 @@
#include "utils/logging.hpp"
namespace memgraph::storage::v3 {
using msgs::Label;
using msgs::PropertyId;
using msgs::Value;
using conversions::ConvertPropertyMap;
using conversions::ConvertPropertyVector;
using conversions::ConvertValueVector;
using conversions::FromMap;
using conversions::FromPropertyValueToValue;
using conversions::ToMsgsVertexId;
using conversions::ToPropertyValue;
using msgs::PropertyId;
using msgs::Value;
auto CreateErrorResponse(const ShardError &shard_error, const auto transaction_id, const std::string_view action) {
msgs::ShardError message_shard_error{shard_error.code, shard_error.message};
@@ -530,7 +526,7 @@ msgs::ReadResponses ShardRsm::HandleRead(msgs::GetPropertiesRequest &&req) {
std::vector<std::pair<PropertyId, Value>> result;
result.reserve(value.size());
for (auto &[id, val] : value) {
result.emplace_back(std::make_pair(id, std::move(val)));
result.emplace_back(id, std::move(val));
}
return result;
};
@@ -706,4 +702,12 @@ msgs::ReadResponses ShardRsm::HandleRead(msgs::GetPropertiesRequest &&req) {
});
}
msgs::ReadResponses ShardRsm::HandleRead(msgs::GraphRequest &&req) {
shard_->Access(req.transaction_id);
SPDLOG_WARN("ShardRsm::HandleRead(GraphRequest) not fully implemented");
msgs::GraphResponse response;
response.data = msgs::Graph{.vertices = {}, .edges = {}};
return response;
}
} // namespace memgraph::storage::v3

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
@@ -27,6 +27,7 @@ class ShardRsm {
msgs::ReadResponses HandleRead(msgs::ExpandOneRequest &&req);
msgs::ReadResponses HandleRead(msgs::GetPropertiesRequest &&req);
msgs::ReadResponses HandleRead(msgs::ScanVerticesRequest &&req);
msgs::ReadResponses HandleRead(msgs::GraphRequest &&req);
msgs::WriteResponses ApplyWrite(msgs::CreateVerticesRequest &&req);
msgs::WriteResponses ApplyWrite(msgs::DeleteVerticesRequest &&req);

View File

@@ -14,7 +14,6 @@
import argparse
import json
FIELDS = [
{
"name": "throughput",
@@ -85,39 +84,32 @@ def compare_results(results_from, results_to, fields):
if group == "__import__":
continue
for scenario, summary_to in scenarios.items():
summary_from = recursive_get(
results_from, dataset, variant, group, scenario,
value={})
if len(summary_from) > 0 and \
summary_to["count"] != summary_from["count"] or \
summary_to["num_workers"] != \
summary_from["num_workers"]:
summary_from = recursive_get(results_from, dataset, variant, group, scenario, value={})
if (
len(summary_from) > 0
and summary_to["count"] != summary_from["count"]
or summary_to["num_workers"] != summary_from["num_workers"]
):
raise Exception("Incompatible results!")
testcode = "/".join([dataset, variant, group, scenario,
"{:02d}".format(
summary_to["num_workers"])])
testcode = "/".join([dataset, variant, group, scenario, "{:02d}".format(summary_to["num_workers"])])
row = {}
performance_changed = False
for field in fields:
key = field["name"]
if key in summary_to:
row[key] = compute_diff(
summary_from.get(key, None),
summary_to[key])
row[key] = compute_diff(summary_from.get(key, None), summary_to[key])
elif key in summary_to["database"]:
row[key] = compute_diff(
recursive_get(summary_from, "database", key,
value=None),
summary_to["database"][key])
recursive_get(summary_from, "database", key, value=None), summary_to["database"][key]
)
else:
row[key] = compute_diff(
recursive_get(summary_from, "metadata", key,
"average", value=None),
summary_to["metadata"][key]["average"])
if "diff" not in row[key] or \
("diff_treshold" in field and
abs(row[key]["diff"]) >=
field["diff_treshold"]):
recursive_get(summary_from, "metadata", key, "average", value=None),
summary_to["metadata"][key]["average"],
)
if "diff" not in row[key] or (
"diff_treshold" in field and abs(row[key]["diff"]) >= field["diff_treshold"]
):
performance_changed = True
if performance_changed:
ret[testcode] = row
@@ -130,29 +122,36 @@ def generate_remarkup(fields, data):
ret += "<table>\n"
ret += " <tr>\n"
ret += " <th>Testcode</th>\n"
ret += "\n".join(map(lambda x: " <th>{}</th>".format(
x["name"].replace("_", " ").capitalize()), fields)) + "\n"
ret += (
"\n".join(
map(
lambda x: " <th>{}</th>".format(x["name"].replace("_", " ").capitalize()),
fields,
)
)
+ "\n"
)
ret += " </tr>\n"
for testcode in sorted(data.keys()):
ret += " <tr>\n"
ret += " <td>{}</td>\n".format(testcode)
for field in fields:
result = data[testcode][field["name"]]
value = result["value"] * field["scaling"]
if "diff" in result:
diff = result["diff"]
arrow = "arrow-up" if diff >= 0 else "arrow-down"
if not (field["positive_diff_better"] ^ (diff >= 0)):
color = "green"
result = data[testcode].get(field["name"])
if result != None:
value = result["value"] * field["scaling"]
if "diff" in result:
diff = result["diff"]
arrow = "arrow-up" if diff >= 0 else "arrow-down"
if not (field["positive_diff_better"] ^ (diff >= 0)):
color = "green"
else:
color = "red"
sign = "{{icon {} color={}}}".format(arrow, color)
ret += ' <td bgcolor="{}">{:.3f}{} ({:+.2%})</td>\n'.format(
color, value, field["unit"], diff
)
else:
color = "red"
sign = "{{icon {} color={}}}".format(arrow, color)
ret += " <td>{:.3f}{} //({:+.2%})// {}</td>\n".format(
value, field["unit"], diff, sign)
else:
ret += " <td>{:.3f}{} //(new)// " \
"{{icon plus color=blue}}</td>\n".format(
value, field["unit"])
ret += '<td bgcolor="blue">{:.3f}{} //(new)// </td>\n'.format(value, field["unit"])
ret += " </tr>\n"
ret += "</table>\n"
else:
@@ -161,11 +160,14 @@ def generate_remarkup(fields, data):
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Compare results of multiple benchmark runs.")
parser.add_argument("--compare", action="append", nargs=2,
metavar=("from", "to"),
help="compare results between `from` and `to` files")
parser = argparse.ArgumentParser(description="Compare results of multiple benchmark runs.")
parser.add_argument(
"--compare",
action="append",
nargs=2,
metavar=("from", "to"),
help="compare results between `from` and `to` files",
)
parser.add_argument("--output", default="", help="output file name")
args = parser.parse_args()

View File

@@ -0,0 +1,139 @@
# 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.
import argparse
import random
import helpers
# Explaination of datasets:
# - empty_only_index: contains index; contains no data
# - small: contains index; contains data (small dataset)
#
# Datamodel is as follow:
#
# ┌──────────────┐
# │ Permission │
# ┌────────────────┐ │ Schema:uuid │ ┌────────────┐
# │:IS_FOR_IDENTITY├────┤ Index:name ├───┤:IS_FOR_FILE│
# └┬───────────────┘ └──────────────┘ └────────────┤
# │ │
# ┌──────▼──────────────┐ ┌──▼────────────────┐
# │ Identity │ │ File │
# │ Schema:uuid │ │ Schema:uuid │
# │ Index:email │ │ Index:name │
# └─────────────────────┘ │ Index:platformId │
# └───────────────────┘
#
# - File: attributes: ["uuid", "name", "platformId"]
# - Permission: attributes: ["uuid", "name"]
# - Identity: attributes: ["uuid", "email"]
#
# Indexes:
# - File: [File(uuid), File(platformId), File(name)]
# - Permission: [Permission(uuid), Permission(name)]
# - Identity: [Identity(uuid), Identity(email)]
#
# Edges:
# - (:Permission)-[:IS_FOR_FILE]->(:File)
# - (:Permission)-[:IS_FOR_IDENTITYR]->(:Identity)
#
# AccessControl specific: uuid is the schema
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--number_of_identities", type=int, default=10)
parser.add_argument("--number_of_files", type=int, default=10)
parser.add_argument("--percentage_of_permissions", type=float, default=1.0)
parser.add_argument("--filename", default="dataset.cypher")
args = parser.parse_args()
number_of_identities = args.number_of_identities
number_of_files = args.number_of_files
percentage_of_permissions = args.percentage_of_permissions
filename = args.filename
assert number_of_identities >= 0
assert number_of_files >= 0
assert percentage_of_permissions > 0.0 and percentage_of_permissions <= 1.0
assert filename != ""
with open(filename, "w") as f:
f.write("MATCH (n) DETACH DELETE n;\n")
# Create the indexes
f.write("CREATE INDEX ON :File;\n")
f.write("CREATE INDEX ON :Permission;\n")
f.write("CREATE INDEX ON :Identity;\n")
f.write("CREATE INDEX ON :File(platformId);\n")
f.write("CREATE INDEX ON :File(name);\n")
f.write("CREATE INDEX ON :Permission(name);\n")
f.write("CREATE INDEX ON :Identity(email);\n")
# Create extra index: in distributed, this will be the schema
f.write("CREATE INDEX ON :File(uuid);\n")
f.write("CREATE INDEX ON :Permission(uuid);\n")
f.write("CREATE INDEX ON :Identity(uuid);\n")
uuid = 1
# Create the nodes File
f.write("UNWIND [")
for index in range(0, number_of_files):
if index != 0:
f.write(",")
f.write(f' {{uuid: {uuid}, platformId: "platform_id", name: "name_file_{uuid}"}}')
uuid += 1
f.write("] AS props CREATE (:File {uuid: props.uuid, platformId: props.platformId, name: props.name});\n")
identities = []
f.write("UNWIND [")
# Create the nodes Identity
for index in range(0, number_of_identities):
if index != 0:
f.write(",")
f.write(f' {{uuid: {uuid}, name: "mail_{uuid}@something.com"}}')
uuid += 1
f.write("] AS props CREATE (:Identity {uuid: props.uuid, name: props.name});\n")
f.write("UNWIND [")
created = 0
for outer_index in range(0, number_of_files):
for inner_index in range(0, number_of_identities):
file_uuid = outer_index + 1
identity_uuid = number_of_files + inner_index + 1
if random.random() <= percentage_of_permissions:
if created > 0:
f.write(",")
f.write(
f' {{permUuid: {uuid}, permName: "name_permission_{uuid}", fileUuid: {file_uuid}, identityUuid: {identity_uuid}}}'
)
created += 1
uuid += 1
if created == 5000:
f.write(
"] AS props MATCH (file:File {uuid:props.fileUuid}), (identity:Identity {uuid: props.identityUuid}) CREATE (permission:Permission {uuid: props.permUuid, name: props.permName}) CREATE (permission)-[: IS_FOR_FILE]->(file) CREATE (permission)-[: IS_FOR_IDENTITY]->(identity);\nUNWIND ["
)
created = 0
f.write(
"] AS props MATCH (file:File {uuid:props.fileUuid}), (identity:Identity {uuid: props.identityUuid}) CREATE (permission:Permission {uuid: props.permUuid, name: props.permName}) CREATE (permission)-[: IS_FOR_FILE]->(file) CREATE (permission)-[: IS_FOR_IDENTITY]->(identity);\n"
)
if __name__ == "__main__":
main()

View File

@@ -353,7 +353,7 @@ class AccessControl(Dataset):
def benchmark__create__vertex(self):
self.next_value_idx += 1
query = (f"CREATE (:File {{uuid: {self.next_value_idx}}});", {})
query = ("CREATE (:File {uuid: $uuid})", {"uuid": self.next_value_idx})
return query
def benchmark__create__edges(self):
@@ -379,6 +379,24 @@ class AccessControl(Dataset):
return query
def benchmark__match__match_all_vertices_with_edges(self):
self.next_value_idx += 1
query = ("MATCH (permission:Permission)-[e:IS_FOR_FILE]->(file:File) RETURN *", {})
return query
def benchmark__match__match_users_with_permission_for_files(self):
file_uuid_1 = self._get_random_uuid("File")
file_uuid_2 = self._get_random_uuid("File")
min_file_uuid = min(file_uuid_1, file_uuid_2)
max_file_uuid = max(file_uuid_1, file_uuid_2)
query = (
"MATCH (f:File)<-[ff:IS_FOR_FILE]-(p:Permission)-[fi:IS_FOR_IDENTITY]->(i:Identity) WHERE f.uuid >= $min_file_uuid AND f.uuid <= $max_file_uuid RETURN *",
{"min_file_uuid": min_file_uuid, "max_file_uuid": max_file_uuid},
)
return query
def benchmark__match__match_users_with_permission_for_specific_file(self):
file_uuid = self._get_random_uuid("File")
query = (
"MATCH (f:File {uuid: $file_uuid})<-[ff:IS_FOR_FILE]-(p:Permission)-[fi:IS_FOR_IDENTITY]->(i:Identity) RETURN *",
{"file_uuid": file_uuid},
)
return query

View File

@@ -68,6 +68,15 @@ class Memgraph:
self._cleanup()
atexit.unregister(self._cleanup)
# Returns None if string_value is not true or false, casing doesn't matter
def _get_bool_value(self, string_value):
lower_string_value = string_value.lower()
if lower_string_value == "true":
return True
if lower_string_value == "false":
return False
return None
def _get_args(self, **kwargs):
data_directory = os.path.join(self._directory.name, "memgraph")
if self._memgraph_version >= (0, 50, 0):
@@ -83,7 +92,13 @@ class Memgraph:
args_list = self._extra_args.split(" ")
assert len(args_list) % 2 == 0
for i in range(0, len(args_list), 2):
kwargs[args_list[i]] = args_list[i + 1]
key = args_list[i]
value = args_list[i + 1]
maybe_bool_value = self._get_bool_value(value)
if maybe_bool_value is not None:
kwargs[key] = maybe_bool_value
else:
kwargs[key] = value
return _convert_args_to_flags(self._memgraph_binary, **kwargs)

View File

@@ -1,8 +1,12 @@
4
8
uuid
email
name
platformId
permUuid
permName
fileUuid
identityUuid
2
IS_FOR_IDENTITY
IS_FOR_FILE

View File

@@ -1,8 +1,12 @@
4
8
uuid
email
name
platformId
permUuid
permName
fileUuid
identityUuid
2
IS_FOR_IDENTITY
IS_FOR_FILE

View File

@@ -1,8 +1,12 @@
4
8
uuid
email
name
platformId
permUuid
permName
fileUuid
identityUuid
2
IS_FOR_IDENTITY
IS_FOR_FILE

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
@@ -118,6 +118,12 @@ class MockedShardRsm {
return resp;
}
msgs::GraphResponse ReadImpl(msgs::GraphRequest rqst) {
msgs::GraphResponse resp;
SPDLOG_INFO("MockedShardRsm::ReadImpl");
return resp;
}
ReadResponses Read(ReadRequests read_requests) {
return {std::visit([this]<typename T>(T &&request) { return ReadResponses{ReadImpl(std::forward<T>(request))}; },
std::move(read_requests))};

View File

@@ -44,40 +44,23 @@ using CompoundKey = coordinator::PrimaryKey;
using coordinator::Coordinator;
using coordinator::CoordinatorClient;
using coordinator::CoordinatorRsm;
using coordinator::HlcRequest;
using coordinator::HlcResponse;
using coordinator::ShardMap;
using coordinator::ShardMetadata;
using coordinator::Shards;
using coordinator::Status;
using io::Address;
using io::Io;
using io::ResponseEnvelope;
using io::ResponseFuture;
using io::Time;
using io::TimedOut;
using io::rsm::Raft;
using io::rsm::ReadRequest;
using io::rsm::ReadResponse;
using io::rsm::StorageReadRequest;
using io::rsm::StorageReadResponse;
using io::rsm::StorageWriteRequest;
using io::rsm::StorageWriteResponse;
using io::rsm::WriteRequest;
using io::rsm::WriteResponse;
using io::simulator::Simulator;
using io::simulator::SimulatorConfig;
using io::simulator::SimulatorStats;
using io::simulator::SimulatorTransport;
using msgs::CreateVerticesRequest;
using msgs::CreateVerticesResponse;
using msgs::ScanVerticesRequest;
using msgs::ScanVerticesResponse;
using msgs::VertexId;
using storage::v3::LabelId;
using storage::v3::SchemaProperty;
using storage::v3::tests::MockedShardRsm;
using utils::BasicResult;
namespace {
@@ -230,10 +213,19 @@ void TestGetProperties(query::v2::RequestRouterInterface &request_router) {
auto result = request_router.GetProperties(std::move(request));
MG_ASSERT(result.size() == 3);
}
template <typename RequestRouter>
void TestAggregate(RequestRouter &request_router) {}
void TestGetGraph(query::v2::RequestRouterInterface &rr) {
msgs::GraphRequest req;
auto graphs = rr.GetGraph(req);
MG_ASSERT(graphs.size() == 2);
for (const auto &graph : graphs) {
MG_ASSERT(graph.data.vertices.size() == 0);
MG_ASSERT(graph.data.edges.size() == 0);
}
}
void DoTest() {
SimulatorConfig config{
.drop_percent = 0,
@@ -356,6 +348,7 @@ void DoTest() {
TestCreateVertices(request_router);
TestCreateExpand(request_router);
TestGetProperties(request_router);
TestGetGraph(request_router);
simulator.ShutDown();

View File

@@ -33,32 +33,21 @@
#include "utils/result.hpp"
namespace memgraph::storage::v3::tests {
using io::Address;
using io::Io;
using io::ResponseEnvelope;
using io::ResponseFuture;
using io::Time;
using io::TimedOut;
using io::rsm::Raft;
using io::rsm::ReadRequest;
using io::rsm::ReadResponse;
using io::rsm::RsmClient;
using io::rsm::WriteRequest;
using io::rsm::WriteResponse;
using io::simulator::Simulator;
using io::simulator::SimulatorConfig;
using io::simulator::SimulatorStats;
using io::simulator::SimulatorTransport;
using utils::BasicResult;
using msgs::ReadRequests;
using msgs::ReadResponses;
using msgs::WriteRequests;
using msgs::WriteResponses;
using ShardClient = RsmClient<SimulatorTransport, WriteRequests, WriteResponses, ReadRequests, ReadResponses>;
using ConcreteShardRsm = Raft<SimulatorTransport, ShardRsm, WriteRequests, WriteResponses, ReadRequests, ReadResponses>;
// TODO(gvolfing) test vertex deletion with DETACH_DELETE as well
@@ -1462,6 +1451,21 @@ void TestGetProperties(ShardClient &client) {
}
}
void TestGetGraph(ShardClient &client) {
SPDLOG_WARN("shard_rsm.cpp:TestGetGraph(ShardClient) not yet implemented");
msgs::GraphRequest req{};
while (true) {
auto read_res = client.SendReadRequest(req);
if (read_res.HasError()) {
continue;
}
auto res = read_res.GetValue();
auto graph_res = std::get<msgs::GraphResponse>(res);
MG_ASSERT(graph_res.error == std::nullopt);
return;
}
}
} // namespace
int TestMessages() {
@@ -1549,8 +1553,11 @@ int TestMessages() {
// GetProperties tests
TestGetProperties(client);
simulator.ShutDown();
// GetGraph tests
TestGetGraph(client);
simulator.ShutDown();
SimulatorStats stats = simulator.Stats();
std::cout << "total messages: " << stats.total_messages << std::endl;

View File

@@ -27,6 +27,7 @@ class MockedRequestRouter : public RequestRouterInterface {
MOCK_METHOD(std::vector<msgs::ExpandOneResultRow>, ExpandOne, (msgs::ExpandOneRequest));
MOCK_METHOD(std::vector<msgs::CreateExpandResponse>, CreateExpand, (std::vector<msgs::NewExpand>));
MOCK_METHOD(std::vector<msgs::GetPropertiesResultRow>, GetProperties, (msgs::GetPropertiesRequest));
MOCK_METHOD(std::vector<msgs::GraphResponse>, GetGraph, (msgs::GraphRequest));
MOCK_METHOD(void, StartTransaction, ());
MOCK_METHOD(void, Commit, ());
@@ -41,7 +42,7 @@ class MockedRequestRouter : public RequestRouterInterface {
MOCK_METHOD(std::optional<storage::v3::EdgeTypeId>, MaybeNameToEdgeType, (const std::string &), (const));
MOCK_METHOD(std::optional<storage::v3::LabelId>, MaybeNameToLabel, (const std::string &), (const));
MOCK_METHOD(bool, IsPrimaryLabel, (storage::v3::LabelId), (const));
MOCK_METHOD(bool, IsPrimaryKey, (storage::v3::LabelId, storage::v3::PropertyId), (const));
MOCK_METHOD(bool, IsPrimaryProperty, (storage::v3::LabelId, storage::v3::PropertyId), (const));
MOCK_METHOD((std::optional<std::pair<uint64_t, uint64_t>>), AllocateInitialEdgeIds, (io::Address));
MOCK_METHOD(void, InstallSimulatorTicker, (std::function<bool()>));
MOCK_METHOD(const std::vector<coordinator::SchemaProperty> &, GetSchemaForLabel, (storage::v3::LabelId), (const));

View File

@@ -58,7 +58,7 @@ TEST(CreateNodeTest, CreateNodeCursor) {
MockedRequestRouter router;
EXPECT_CALL(router, CreateVertices(_)).Times(1).WillOnce(Return(std::vector<msgs::CreateVerticesResponse>{}));
EXPECT_CALL(router, IsPrimaryLabel(_)).WillRepeatedly(Return(true));
EXPECT_CALL(router, IsPrimaryKey(_, _)).WillRepeatedly(Return(true));
EXPECT_CALL(router, IsPrimaryProperty(_, _)).WillRepeatedly(Return(true));
auto context = MakeContext(ast, symbol_table, &router, &id_alloc);
auto multi_frame = CreateMultiFrame(context.symbol_table.max_position());
cursor->PullMultiple(multi_frame, context);

View File

@@ -97,6 +97,8 @@ class MockedRequestRouter : public RequestRouterInterface {
std::vector<GetPropertiesResultRow> GetProperties(GetPropertiesRequest rqst) override { return {}; }
std::vector<msgs::GraphResponse> GetGraph(msgs::GraphRequest rqst) override { return {}; }
const std::string &PropertyToName(memgraph::storage::v3::PropertyId id) const override {
return properties_.IdToName(id.AsUint());
}
@@ -123,7 +125,7 @@ class MockedRequestRouter : public RequestRouterInterface {
bool IsPrimaryLabel(LabelId label) const override { return true; }
bool IsPrimaryKey(LabelId primary_label, PropertyId property) const override { return true; }
bool IsPrimaryProperty(LabelId primary_label, PropertyId property) const override { return true; }
std::optional<std::pair<uint64_t, uint64_t>> AllocateInitialEdgeIds(io::Address coordinator_address) override {
return {};