Compare commits

...

6 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
12 changed files with 136 additions and 47 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

@@ -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;
@@ -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

@@ -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, ());

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());
}