Implement message based actions (#538)
Create shard-side handlers for basic messages Implement the handlers for CreateVertices, CreateEdges and ScanAll. Use or modify the defined messages to interact with individual Shards and test their behavior. Shard is currently being owned by ShardRsm instances. The two top level dispatching functions Read() and Apply() are responsible for read- and write operations respectively. Currently there are a handful of messages that are defined but not utilized, these will be used in the near future, as well as a couple of handler functions with empty implementations.
This commit is contained in:
gvolfing
GitHub
parent
8e1f83acc9
commit
ecda71168c
src
io/simulator
query/v2
storage/v3
tests/simulation
@ -55,15 +55,17 @@ class SimulatorHandle {
|
||||
|
||||
void TimeoutPromisesPastDeadline() {
|
||||
const Time now = cluster_wide_time_microseconds_;
|
||||
|
||||
for (auto &[promise_key, dop] : promises_) {
|
||||
for (auto it = promises_.begin(); it != promises_.end();) {
|
||||
auto &[promise_key, dop] = *it;
|
||||
if (dop.deadline < now) {
|
||||
spdlog::debug("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(),
|
||||
promise_key.replier_address.ToString());
|
||||
spdlog::info("timing out request from requester {} to replier {}.", promise_key.requester_address.ToString(),
|
||||
promise_key.replier_address.ToString());
|
||||
std::move(dop).promise.TimeOut();
|
||||
promises_.erase(promise_key);
|
||||
it = promises_.erase(it);
|
||||
|
||||
stats_.timed_out_requests++;
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -16,52 +16,51 @@
|
||||
#include <map>
|
||||
#include <optional>
|
||||
#include <unordered_map>
|
||||
#include <utility>
|
||||
#include <variant>
|
||||
#include <vector>
|
||||
|
||||
#include "coordinator/hybrid_logical_clock.hpp"
|
||||
#include "storage/v3/id_types.hpp"
|
||||
#include "storage/v3/property_value.hpp"
|
||||
|
||||
/// Hybrid-logical clock
|
||||
struct Hlc {
|
||||
uint64_t logical_id;
|
||||
using Duration = std::chrono::microseconds;
|
||||
using Time = std::chrono::time_point<std::chrono::system_clock, Duration>;
|
||||
Time coordinator_wall_clock;
|
||||
namespace memgraph::msgs {
|
||||
|
||||
bool operator==(const Hlc &other) const = default;
|
||||
};
|
||||
using coordinator::Hlc;
|
||||
using storage::v3::LabelId;
|
||||
|
||||
struct Value;
|
||||
|
||||
struct Label {
|
||||
size_t id;
|
||||
LabelId id;
|
||||
};
|
||||
|
||||
// TODO(kostasrim) update this with CompoundKey, same for the rest of the file.
|
||||
using PrimaryKey = std::vector<memgraph::storage::v3::PropertyValue>;
|
||||
using PrimaryKey = std::vector<Value>;
|
||||
using VertexId = std::pair<Label, PrimaryKey>;
|
||||
using Gid = size_t;
|
||||
using PropertyId = memgraph::storage::v3::PropertyId;
|
||||
|
||||
struct EdgeType {
|
||||
std::string name;
|
||||
uint64_t id;
|
||||
};
|
||||
|
||||
struct EdgeId {
|
||||
VertexId id;
|
||||
Gid gid;
|
||||
};
|
||||
|
||||
struct Vertex {
|
||||
VertexId id;
|
||||
std::vector<Label> labels;
|
||||
};
|
||||
|
||||
struct Edge {
|
||||
VertexId src;
|
||||
VertexId dst;
|
||||
EdgeId id;
|
||||
EdgeType type;
|
||||
};
|
||||
|
||||
struct Vertex {
|
||||
VertexId id;
|
||||
std::vector<Label> labels;
|
||||
};
|
||||
|
||||
struct PathPart {
|
||||
Vertex dst;
|
||||
Gid edge;
|
||||
@ -75,11 +74,224 @@ struct Path {
|
||||
struct Null {};
|
||||
|
||||
struct Value {
|
||||
enum Type { NILL, BOOL, INT64, DOUBLE, STRING, LIST, MAP, VERTEX, EDGE, PATH };
|
||||
Value() : null_v{} {}
|
||||
|
||||
explicit Value(const bool val) : type(Type::Bool), bool_v(val) {}
|
||||
explicit Value(const int64_t val) : type(Type::Int64), int_v(val) {}
|
||||
explicit Value(const double val) : type(Type::Double), double_v(val) {}
|
||||
|
||||
explicit Value(const Vertex val) : type(Type::Vertex), vertex_v(val) {}
|
||||
|
||||
explicit Value(const std::string &val) : type(Type::String) { new (&string_v) std::string(val); }
|
||||
explicit Value(const char *val) : type(Type::String) { new (&string_v) std::string(val); }
|
||||
|
||||
explicit Value(const std::vector<Value> &val) : type(Type::List) { new (&list_v) std::vector<Value>(val); }
|
||||
|
||||
explicit Value(const std::map<std::string, Value> &val) : type(Type::Map) {
|
||||
new (&map_v) std::map<std::string, Value>(val);
|
||||
}
|
||||
|
||||
explicit Value(std::string &&val) noexcept : type(Type::String) { new (&string_v) std::string(std::move(val)); }
|
||||
|
||||
explicit Value(std::vector<Value> &&val) noexcept : type(Type::List) {
|
||||
new (&list_v) std::vector<Value>(std::move(val));
|
||||
}
|
||||
explicit Value(std::map<std::string, Value> &&val) noexcept : type(Type::Map) {
|
||||
new (&map_v) std::map<std::string, Value>(std::move(val));
|
||||
}
|
||||
|
||||
~Value() { DestroyValue(); }
|
||||
|
||||
void DestroyValue() noexcept {
|
||||
switch (type) {
|
||||
case Type::Null:
|
||||
case Type::Bool:
|
||||
case Type::Int64:
|
||||
case Type::Double:
|
||||
return;
|
||||
|
||||
case Type::String:
|
||||
std::destroy_at(&string_v);
|
||||
return;
|
||||
case Type::List:
|
||||
std::destroy_at(&list_v);
|
||||
return;
|
||||
case Type::Map:
|
||||
std::destroy_at(&map_v);
|
||||
return;
|
||||
|
||||
case Type::Vertex:
|
||||
std::destroy_at(&vertex_v);
|
||||
return;
|
||||
case Type::Path:
|
||||
std::destroy_at(&path_v);
|
||||
return;
|
||||
case Type::Edge:
|
||||
std::destroy_at(&edge_v);
|
||||
}
|
||||
}
|
||||
|
||||
Value(const Value &other) : type(other.type) {
|
||||
switch (other.type) {
|
||||
case Type::Null:
|
||||
return;
|
||||
case Type::Bool:
|
||||
this->bool_v = other.bool_v;
|
||||
return;
|
||||
case Type::Int64:
|
||||
this->int_v = other.int_v;
|
||||
return;
|
||||
case Type::Double:
|
||||
this->double_v = other.double_v;
|
||||
return;
|
||||
case Type::String:
|
||||
new (&string_v) std::string(other.string_v);
|
||||
return;
|
||||
case Type::List:
|
||||
new (&list_v) std::vector<Value>(other.list_v);
|
||||
return;
|
||||
case Type::Map:
|
||||
new (&map_v) std::map<std::string, Value>(other.map_v);
|
||||
return;
|
||||
case Type::Vertex:
|
||||
new (&vertex_v) Vertex(other.vertex_v);
|
||||
return;
|
||||
case Type::Edge:
|
||||
new (&edge_v) Edge(other.edge_v);
|
||||
return;
|
||||
case Type::Path:
|
||||
new (&path_v) Path(other.path_v);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
Value(Value &&other) noexcept : type(other.type) {
|
||||
switch (other.type) {
|
||||
case Type::Null:
|
||||
break;
|
||||
case Type::Bool:
|
||||
this->bool_v = other.bool_v;
|
||||
break;
|
||||
case Type::Int64:
|
||||
this->int_v = other.int_v;
|
||||
break;
|
||||
case Type::Double:
|
||||
this->double_v = other.double_v;
|
||||
break;
|
||||
case Type::String:
|
||||
new (&string_v) std::string(std::move(other.string_v));
|
||||
break;
|
||||
case Type::List:
|
||||
new (&list_v) std::vector<Value>(std::move(other.list_v));
|
||||
break;
|
||||
case Type::Map:
|
||||
new (&map_v) std::map<std::string, Value>(std::move(other.map_v));
|
||||
break;
|
||||
case Type::Vertex:
|
||||
new (&vertex_v) Vertex(std::move(other.vertex_v));
|
||||
break;
|
||||
case Type::Edge:
|
||||
new (&edge_v) Edge(std::move(other.edge_v));
|
||||
break;
|
||||
case Type::Path:
|
||||
new (&path_v) Path(std::move(other.path_v));
|
||||
break;
|
||||
}
|
||||
|
||||
other.DestroyValue();
|
||||
other.type = Type::Null;
|
||||
}
|
||||
|
||||
Value &operator=(const Value &other) {
|
||||
if (this == &other) return *this;
|
||||
|
||||
DestroyValue();
|
||||
type = other.type;
|
||||
|
||||
switch (other.type) {
|
||||
case Type::Null:
|
||||
break;
|
||||
case Type::Bool:
|
||||
this->bool_v = other.bool_v;
|
||||
break;
|
||||
case Type::Int64:
|
||||
this->int_v = other.int_v;
|
||||
break;
|
||||
case Type::Double:
|
||||
this->double_v = other.double_v;
|
||||
break;
|
||||
case Type::String:
|
||||
new (&string_v) std::string(other.string_v);
|
||||
break;
|
||||
case Type::List:
|
||||
new (&list_v) std::vector<Value>(other.list_v);
|
||||
break;
|
||||
case Type::Map:
|
||||
new (&map_v) std::map<std::string, Value>(other.map_v);
|
||||
break;
|
||||
case Type::Vertex:
|
||||
new (&vertex_v) Vertex(other.vertex_v);
|
||||
break;
|
||||
case Type::Edge:
|
||||
new (&edge_v) Edge(other.edge_v);
|
||||
break;
|
||||
case Type::Path:
|
||||
new (&path_v) Path(other.path_v);
|
||||
break;
|
||||
}
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
Value &operator=(Value &&other) noexcept {
|
||||
if (this == &other) return *this;
|
||||
|
||||
DestroyValue();
|
||||
type = other.type;
|
||||
|
||||
switch (other.type) {
|
||||
case Type::Null:
|
||||
break;
|
||||
case Type::Bool:
|
||||
this->bool_v = other.bool_v;
|
||||
break;
|
||||
case Type::Int64:
|
||||
this->int_v = other.int_v;
|
||||
break;
|
||||
case Type::Double:
|
||||
this->double_v = other.double_v;
|
||||
break;
|
||||
case Type::String:
|
||||
new (&string_v) std::string(std::move(other.string_v));
|
||||
break;
|
||||
case Type::List:
|
||||
new (&list_v) std::vector<Value>(std::move(other.list_v));
|
||||
break;
|
||||
case Type::Map:
|
||||
new (&map_v) std::map<std::string, Value>(std::move(other.map_v));
|
||||
break;
|
||||
case Type::Vertex:
|
||||
new (&vertex_v) Vertex(std::move(other.vertex_v));
|
||||
break;
|
||||
case Type::Edge:
|
||||
new (&edge_v) Edge(std::move(other.edge_v));
|
||||
break;
|
||||
case Type::Path:
|
||||
new (&path_v) Path(std::move(other.path_v));
|
||||
break;
|
||||
}
|
||||
|
||||
other.DestroyValue();
|
||||
other.type = Type::Null;
|
||||
|
||||
return *this;
|
||||
}
|
||||
enum class Type : uint8_t { Null, Bool, Int64, Double, String, List, Map, Vertex, Edge, Path };
|
||||
Type type{Type::Null};
|
||||
union {
|
||||
Null null_v;
|
||||
bool bool_v;
|
||||
uint64_t int_v;
|
||||
int64_t int_v;
|
||||
double double_v;
|
||||
std::string string_v;
|
||||
std::vector<Value> list_v;
|
||||
@ -88,8 +300,6 @@ struct Value {
|
||||
Edge edge_v;
|
||||
Path path_v;
|
||||
};
|
||||
|
||||
Type type;
|
||||
};
|
||||
|
||||
struct ValuesMap {
|
||||
@ -125,17 +335,23 @@ enum class StorageView { OLD = 0, NEW = 1 };
|
||||
|
||||
struct ScanVerticesRequest {
|
||||
Hlc transaction_id;
|
||||
size_t start_id;
|
||||
std::optional<std::vector<std::string>> props_to_return;
|
||||
VertexId start_id;
|
||||
std::optional<std::vector<PropertyId>> props_to_return;
|
||||
std::optional<std::vector<std::string>> filter_expressions;
|
||||
std::optional<size_t> batch_limit;
|
||||
StorageView storage_view;
|
||||
};
|
||||
|
||||
struct ScanResultRow {
|
||||
Value vertex;
|
||||
// empty() is no properties returned
|
||||
std::map<PropertyId, Value> props;
|
||||
};
|
||||
|
||||
struct ScanVerticesResponse {
|
||||
bool success;
|
||||
Values values;
|
||||
std::optional<VertexId> next_start_id;
|
||||
std::vector<ScanResultRow> results;
|
||||
};
|
||||
|
||||
using VertexOrEdgeIds = std::variant<VertexId, EdgeId>;
|
||||
@ -202,9 +418,23 @@ struct ExpandOneResponse {
|
||||
std::vector<ExpandOneResultRow> result;
|
||||
};
|
||||
|
||||
struct UpdateVertexProp {
|
||||
VertexId primary_key;
|
||||
std::vector<std::pair<PropertyId, Value>> property_updates;
|
||||
};
|
||||
|
||||
struct UpdateEdgeProp {
|
||||
Edge edge;
|
||||
std::vector<std::pair<PropertyId, Value>> property_updates;
|
||||
};
|
||||
|
||||
/*
|
||||
* Vertices
|
||||
*/
|
||||
struct NewVertex {
|
||||
std::vector<Label> label_ids;
|
||||
std::map<PropertyId, Value> properties;
|
||||
PrimaryKey primary_key;
|
||||
std::vector<std::pair<PropertyId, Value>> properties;
|
||||
};
|
||||
|
||||
struct CreateVerticesRequest {
|
||||
@ -216,8 +446,62 @@ struct CreateVerticesResponse {
|
||||
bool success;
|
||||
};
|
||||
|
||||
struct DeleteVerticesRequest {
|
||||
enum class DeletionType { DELETE, DETACH_DELETE };
|
||||
Hlc transaction_id;
|
||||
std::vector<std::vector<Value>> primary_keys;
|
||||
DeletionType deletion_type;
|
||||
};
|
||||
|
||||
struct DeleteVerticesResponse {
|
||||
bool success;
|
||||
};
|
||||
|
||||
struct UpdateVerticesRequest {
|
||||
Hlc transaction_id;
|
||||
std::vector<UpdateVertexProp> new_properties;
|
||||
};
|
||||
|
||||
struct UpdateVerticesResponse {
|
||||
bool success;
|
||||
};
|
||||
|
||||
/*
|
||||
* Edges
|
||||
*/
|
||||
struct CreateEdgesRequest {
|
||||
Hlc transaction_id;
|
||||
std::vector<Edge> edges;
|
||||
};
|
||||
|
||||
struct CreateEdgesResponse {
|
||||
bool success;
|
||||
};
|
||||
|
||||
struct DeleteEdgesRequest {
|
||||
Hlc transaction_id;
|
||||
std::vector<Edge> edges;
|
||||
};
|
||||
|
||||
struct DeleteEdgesResponse {
|
||||
bool success;
|
||||
};
|
||||
|
||||
struct UpdateEdgesRequest {
|
||||
Hlc transaction_id;
|
||||
std::vector<UpdateEdgeProp> new_properties;
|
||||
};
|
||||
|
||||
struct UpdateEdgesResponse {
|
||||
bool success;
|
||||
};
|
||||
|
||||
using ReadRequests = std::variant<ExpandOneRequest, GetPropertiesRequest, ScanVerticesRequest>;
|
||||
using ReadResponses = std::variant<ExpandOneResponse, GetPropertiesResponse, ScanVerticesResponse>;
|
||||
|
||||
using WriteRequests = CreateVerticesRequest;
|
||||
using WriteResponses = CreateVerticesResponse;
|
||||
using WriteRequests = std::variant<CreateVerticesRequest, DeleteVerticesRequest, UpdateVerticesRequest,
|
||||
CreateEdgesRequest, DeleteEdgesRequest, UpdateEdgesRequest>;
|
||||
using WriteResponses = std::variant<CreateVerticesResponse, DeleteVerticesResponse, UpdateVerticesResponse,
|
||||
CreateEdgesResponse, DeleteEdgesResponse, UpdateEdgesResponse>;
|
||||
|
||||
} // namespace memgraph::msgs
|
||||
|
||||
@ -15,7 +15,8 @@ set(storage_v3_src_files
|
||||
schemas.cpp
|
||||
schema_validator.cpp
|
||||
shard.cpp
|
||||
storage.cpp)
|
||||
storage.cpp
|
||||
shard_rsm.cpp)
|
||||
|
||||
# #### Replication #####
|
||||
define_add_lcp(add_lcp_storage lcp_storage_cpp_files generated_lcp_storage_files)
|
||||
|
||||
@ -521,6 +521,44 @@ ResultSchema<VertexAccessor> Shard::Accessor::CreateVertexAndValidate(
|
||||
return vertex_acc;
|
||||
}
|
||||
|
||||
ResultSchema<VertexAccessor> Shard::Accessor::CreateVertexAndValidate(
|
||||
LabelId primary_label, const std::vector<LabelId> &labels, const std::vector<PropertyValue> &primary_properties,
|
||||
const std::vector<std::pair<PropertyId, PropertyValue>> &properties) {
|
||||
if (primary_label != shard_->primary_label_) {
|
||||
throw utils::BasicException("Cannot add vertex to shard which does not hold the given primary label!");
|
||||
}
|
||||
auto maybe_schema_violation = GetSchemaValidator().ValidateVertexCreate(primary_label, labels, properties);
|
||||
if (maybe_schema_violation) {
|
||||
return {std::move(*maybe_schema_violation)};
|
||||
}
|
||||
OOMExceptionEnabler oom_exception;
|
||||
auto acc = shard_->vertices_.access();
|
||||
auto *delta = CreateDeleteObjectDelta(&transaction_);
|
||||
auto [it, inserted] = acc.insert({Vertex{delta, primary_properties}});
|
||||
|
||||
VertexAccessor vertex_acc{&it->vertex, &transaction_, &shard_->indices_,
|
||||
&shard_->constraints_, config_, shard_->vertex_validator_};
|
||||
MG_ASSERT(inserted, "The vertex must be inserted here!");
|
||||
MG_ASSERT(it != acc.end(), "Invalid Vertex accessor!");
|
||||
|
||||
// TODO(jbajic) Improve, maybe delay index update
|
||||
for (const auto &[property_id, property_value] : properties) {
|
||||
if (!shard_->schemas_.IsPropertyKey(primary_label, property_id)) {
|
||||
if (const auto err = vertex_acc.SetProperty(property_id, property_value); err.HasError()) {
|
||||
return {err.GetError()};
|
||||
}
|
||||
}
|
||||
}
|
||||
// Set secondary labels
|
||||
for (auto label : labels) {
|
||||
if (const auto err = vertex_acc.AddLabel(label); err.HasError()) {
|
||||
return {err.GetError()};
|
||||
}
|
||||
}
|
||||
delta->prev.Set(&it->vertex);
|
||||
return vertex_acc;
|
||||
}
|
||||
|
||||
std::optional<VertexAccessor> Shard::Accessor::FindVertex(std::vector<PropertyValue> primary_key, View view) {
|
||||
auto acc = shard_->vertices_.access();
|
||||
// Later on use label space
|
||||
|
||||
@ -237,11 +237,19 @@ class Shard final {
|
||||
|
||||
~Accessor();
|
||||
|
||||
// TODO(gvolfing) this is just a workaround for stitching remove this later.
|
||||
LabelId GetPrimaryLabel() const noexcept { return shard_->primary_label_; }
|
||||
|
||||
/// @throw std::bad_alloc
|
||||
ResultSchema<VertexAccessor> CreateVertexAndValidate(
|
||||
LabelId primary_label, const std::vector<LabelId> &labels,
|
||||
const std::vector<std::pair<PropertyId, PropertyValue>> &properties);
|
||||
|
||||
/// @throw std::bad_alloc
|
||||
ResultSchema<VertexAccessor> CreateVertexAndValidate(
|
||||
LabelId primary_label, const std::vector<LabelId> &labels, const std::vector<PropertyValue> &primary_properties,
|
||||
const std::vector<std::pair<PropertyId, PropertyValue>> &properties);
|
||||
|
||||
std::optional<VertexAccessor> FindVertex(std::vector<PropertyValue> primary_key, View view);
|
||||
|
||||
VerticesIterable Vertices(View view) {
|
||||
|
||||
439
src/storage/v3/shard_rsm.cpp
Normal file
439
src/storage/v3/shard_rsm.cpp
Normal file
@ -0,0 +1,439 @@
|
||||
// Copyright 2022 Memgraph Ltd.
|
||||
//
|
||||
// Use of this software is governed by the Business Source License
|
||||
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
|
||||
// License, and you may not use this file except in compliance with the Business Source License.
|
||||
//
|
||||
// As of the Change Date specified in that file, in accordance with
|
||||
// the Business Source License, use of this software will be governed
|
||||
// by the Apache License, Version 2.0, included in the file
|
||||
// licenses/APL.txt.
|
||||
|
||||
#include <iterator>
|
||||
#include <utility>
|
||||
|
||||
#include "storage/v3/shard_rsm.hpp"
|
||||
#include "storage/v3/vertex_accessor.hpp"
|
||||
|
||||
using memgraph::msgs::Label;
|
||||
using memgraph::msgs::PropertyId;
|
||||
using memgraph::msgs::Value;
|
||||
using memgraph::msgs::VertexId;
|
||||
|
||||
namespace {
|
||||
// TODO(gvolfing use come algorithm instead of explicit for loops)
|
||||
memgraph::storage::v3::PropertyValue ToPropertyValue(Value &&value) {
|
||||
using PV = memgraph::storage::v3::PropertyValue;
|
||||
PV ret;
|
||||
switch (value.type) {
|
||||
case Value::Type::Null:
|
||||
return PV{};
|
||||
case Value::Type::Bool:
|
||||
return PV(value.bool_v);
|
||||
case Value::Type::Int64:
|
||||
return PV(static_cast<int64_t>(value.int_v));
|
||||
case Value::Type::Double:
|
||||
return PV(value.double_v);
|
||||
case Value::Type::String:
|
||||
return PV(value.string_v);
|
||||
case Value::Type::List: {
|
||||
std::vector<PV> list;
|
||||
for (auto &elem : value.list_v) {
|
||||
list.emplace_back(ToPropertyValue(std::move(elem)));
|
||||
}
|
||||
return PV(list);
|
||||
}
|
||||
case Value::Type::Map: {
|
||||
std::map<std::string, PV> map;
|
||||
for (auto &[key, value] : value.map_v) {
|
||||
map.emplace(std::make_pair(key, ToPropertyValue(std::move(value))));
|
||||
}
|
||||
return PV(map);
|
||||
}
|
||||
// These are not PropertyValues
|
||||
case Value::Type::Vertex:
|
||||
case Value::Type::Edge:
|
||||
case Value::Type::Path:
|
||||
MG_ASSERT(false, "Not PropertyValue");
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
Value ToValue(const memgraph::storage::v3::PropertyValue &pv) {
|
||||
using memgraph::storage::v3::PropertyValue;
|
||||
|
||||
switch (pv.type()) {
|
||||
case PropertyValue::Type::Bool:
|
||||
return Value(pv.ValueBool());
|
||||
case PropertyValue::Type::Double:
|
||||
return Value(pv.ValueDouble());
|
||||
case PropertyValue::Type::Int:
|
||||
return Value(pv.ValueInt());
|
||||
case PropertyValue::Type::List: {
|
||||
std::vector<Value> list(pv.ValueList().size());
|
||||
for (const auto &elem : pv.ValueList()) {
|
||||
list.emplace_back(ToValue(elem));
|
||||
}
|
||||
|
||||
return Value(list);
|
||||
}
|
||||
case PropertyValue::Type::Map: {
|
||||
std::map<std::string, Value> map;
|
||||
for (const auto &[key, val] : pv.ValueMap()) {
|
||||
// maybe use std::make_pair once the && issue is resolved.
|
||||
map.emplace(key, ToValue(val));
|
||||
}
|
||||
|
||||
return Value(map);
|
||||
}
|
||||
case PropertyValue::Type::Null:
|
||||
return Value{};
|
||||
case PropertyValue::Type::String:
|
||||
return Value(pv.ValueString());
|
||||
case PropertyValue::Type::TemporalData: {
|
||||
// TBD -> we need to specify this in the messages, not a priority.
|
||||
MG_ASSERT(false, "Temporal datatypes are not yet implemented on Value!");
|
||||
return Value{};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::vector<std::pair<memgraph::storage::v3::PropertyId, memgraph::storage::v3::PropertyValue>> ConvertPropertyMap(
|
||||
std::vector<std::pair<PropertyId, Value>> &&properties) {
|
||||
std::vector<std::pair<memgraph::storage::v3::PropertyId, memgraph::storage::v3::PropertyValue>> ret;
|
||||
ret.reserve(properties.size());
|
||||
|
||||
for (auto &[key, value] : properties) {
|
||||
ret.emplace_back(std::make_pair(key, ToPropertyValue(std::move(value))));
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
std::vector<memgraph::storage::v3::PropertyValue> ConvertPropertyVector(std::vector<Value> &&vec) {
|
||||
std::vector<memgraph::storage::v3::PropertyValue> ret;
|
||||
ret.reserve(vec.size());
|
||||
|
||||
for (auto &elem : vec) {
|
||||
ret.push_back(ToPropertyValue(std::move(elem)));
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
std::vector<Value> ConvertValueVector(const std::vector<memgraph::storage::v3::PropertyValue> &vec) {
|
||||
std::vector<Value> ret;
|
||||
ret.reserve(vec.size());
|
||||
|
||||
for (const auto &elem : vec) {
|
||||
ret.push_back(ToValue(elem));
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
std::optional<std::map<PropertyId, Value>> CollectPropertiesFromAccessor(
|
||||
const memgraph::storage::v3::VertexAccessor &acc, const std::vector<memgraph::storage::v3::PropertyId> &props,
|
||||
memgraph::storage::v3::View view) {
|
||||
std::map<PropertyId, Value> ret;
|
||||
|
||||
for (const auto &prop : props) {
|
||||
auto result = acc.GetProperty(prop, view);
|
||||
if (result.HasError()) {
|
||||
spdlog::debug("Encountered an Error while trying to get a vertex property.");
|
||||
continue;
|
||||
}
|
||||
auto &value = result.GetValue();
|
||||
if (value.IsNull()) {
|
||||
spdlog::debug("The specified property does not exist but it should");
|
||||
continue;
|
||||
}
|
||||
ret.emplace(prop, ToValue(value));
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
std::optional<std::map<PropertyId, Value>> CollectAllPropertiesFromAccessor(
|
||||
const memgraph::storage::v3::VertexAccessor &acc, memgraph::storage::v3::View view) {
|
||||
std::map<PropertyId, Value> ret;
|
||||
auto iter = acc.Properties(view);
|
||||
if (iter.HasError()) {
|
||||
spdlog::debug("Encountered an error while trying to get vertex properties.");
|
||||
}
|
||||
|
||||
for (const auto &[prop_key, prop_val] : iter.GetValue()) {
|
||||
ret.emplace(prop_key, ToValue(prop_val));
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
Value ConstructValueVertex(const memgraph::storage::v3::VertexAccessor &acc, memgraph::storage::v3::View view) {
|
||||
// Get the vertex id
|
||||
auto prim_label = acc.PrimaryLabel(view).GetValue();
|
||||
Label value_label{.id = prim_label};
|
||||
|
||||
auto prim_key = ConvertValueVector(acc.PrimaryKey(view).GetValue());
|
||||
VertexId vertex_id = std::make_pair(value_label, prim_key);
|
||||
|
||||
// Get the labels
|
||||
auto vertex_labels = acc.Labels(view).GetValue();
|
||||
std::vector<Label> value_labels;
|
||||
for (const auto &label : vertex_labels) {
|
||||
Label l = {.id = label};
|
||||
value_labels.push_back(l);
|
||||
}
|
||||
|
||||
return Value({.id = vertex_id, .labels = value_labels});
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
namespace memgraph::storage::v3 {
|
||||
|
||||
msgs::WriteResponses ShardRsm::ApplyWrite(msgs::CreateVerticesRequest &&req) {
|
||||
auto acc = shard_->Access();
|
||||
|
||||
// Workaround untill we have access to CreateVertexAndValidate()
|
||||
// with the new signature that does not require the primary label.
|
||||
const auto prim_label = acc.GetPrimaryLabel();
|
||||
|
||||
bool action_successful = true;
|
||||
|
||||
for (auto &new_vertex : req.new_vertices) {
|
||||
/// TODO(gvolfing) Remove this. In the new implementation each shard
|
||||
/// should have a predetermined primary label, so there is no point in
|
||||
/// specifying it in the accessor functions. Their signature will
|
||||
/// change.
|
||||
/// TODO(gvolfing) Consider other methods than converting. Change either
|
||||
/// the way that the property map is stored in the messages, or the
|
||||
/// signature of CreateVertexAndValidate.
|
||||
auto converted_property_map = ConvertPropertyMap(std::move(new_vertex.properties));
|
||||
|
||||
// TODO(gvolfing) make sure if this conversion is actually needed.
|
||||
std::vector<memgraph::storage::v3::LabelId> converted_label_ids;
|
||||
converted_label_ids.reserve(new_vertex.label_ids.size());
|
||||
for (const auto &label_id : new_vertex.label_ids) {
|
||||
converted_label_ids.emplace_back(label_id.id);
|
||||
}
|
||||
|
||||
auto result_schema =
|
||||
acc.CreateVertexAndValidate(prim_label, converted_label_ids,
|
||||
ConvertPropertyVector(std::move(new_vertex.primary_key)), converted_property_map);
|
||||
|
||||
if (result_schema.HasError()) {
|
||||
auto &error = result_schema.GetError();
|
||||
|
||||
std::visit(
|
||||
[]<typename T>(T &&) {
|
||||
using ErrorType = std::remove_cvref_t<T>;
|
||||
if constexpr (std::is_same_v<ErrorType, SchemaViolation>) {
|
||||
spdlog::debug("Creating vertex failed with error: SchemaViolation");
|
||||
} else if constexpr (std::is_same_v<ErrorType, Error>) {
|
||||
spdlog::debug("Creating vertex failed with error: Error");
|
||||
} else {
|
||||
static_assert(kAlwaysFalse<T>, "Missing type from variant visitor");
|
||||
}
|
||||
},
|
||||
error);
|
||||
|
||||
action_successful = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
msgs::CreateVerticesResponse resp{};
|
||||
resp.success = action_successful;
|
||||
|
||||
if (action_successful) {
|
||||
auto result = acc.Commit(req.transaction_id.logical_id);
|
||||
if (result.HasError()) {
|
||||
resp.success = false;
|
||||
spdlog::debug(&"ConstraintViolation, commiting vertices was unsuccesfull with transaction id: "[req.transaction_id
|
||||
.logical_id]);
|
||||
}
|
||||
}
|
||||
return resp;
|
||||
}
|
||||
|
||||
msgs::WriteResponses ShardRsm::ApplyWrite(msgs::DeleteVerticesRequest &&req) {
|
||||
bool action_successful = true;
|
||||
auto acc = shard_->Access();
|
||||
|
||||
for (auto &propval : req.primary_keys) {
|
||||
if (!action_successful) {
|
||||
break;
|
||||
}
|
||||
|
||||
auto vertex_acc = acc.FindVertex(ConvertPropertyVector(std::move(propval)), View::OLD);
|
||||
|
||||
if (!vertex_acc) {
|
||||
spdlog::debug(
|
||||
&"Error while trying to delete vertex. Vertex to delete does not exist. Transaction id: "[req.transaction_id
|
||||
.logical_id]);
|
||||
action_successful = false;
|
||||
} else {
|
||||
// TODO(gvolfing)
|
||||
// Since we will not have different kinds of deletion types in one transaction,
|
||||
// we dont have to enter the switch statement on every iteration. Optimize this.
|
||||
switch (req.deletion_type) {
|
||||
case msgs::DeleteVerticesRequest::DeletionType::DELETE: {
|
||||
auto result = acc.DeleteVertex(&vertex_acc.value());
|
||||
if (result.HasError() || !(result.GetValue().has_value())) {
|
||||
action_successful = false;
|
||||
spdlog::debug(&"Error while trying to delete vertex. Transaction id: "[req.transaction_id.logical_id]);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
case msgs::DeleteVerticesRequest::DeletionType::DETACH_DELETE: {
|
||||
auto result = acc.DetachDeleteVertex(&vertex_acc.value());
|
||||
if (result.HasError() || !(result.GetValue().has_value())) {
|
||||
action_successful = false;
|
||||
spdlog::debug(
|
||||
&"Error while trying to detach and delete vertex. Transaction id: "[req.transaction_id.logical_id]);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
msgs::DeleteVerticesResponse resp{};
|
||||
resp.success = action_successful;
|
||||
|
||||
if (action_successful) {
|
||||
auto result = acc.Commit(req.transaction_id.logical_id);
|
||||
if (result.HasError()) {
|
||||
resp.success = false;
|
||||
spdlog::debug(&"ConstraintViolation, commiting vertices was unsuccesfull with transaction id: "[req.transaction_id
|
||||
.logical_id]);
|
||||
}
|
||||
}
|
||||
|
||||
return resp;
|
||||
}
|
||||
|
||||
msgs::WriteResponses ShardRsm::ApplyWrite(msgs::CreateEdgesRequest &&req) {
|
||||
auto acc = shard_->Access();
|
||||
bool action_successful = true;
|
||||
|
||||
for (auto &edge : req.edges) {
|
||||
auto vertex_acc_from_primary_key = edge.src.second;
|
||||
auto vertex_from_acc = acc.FindVertex(ConvertPropertyVector(std::move(vertex_acc_from_primary_key)), View::OLD);
|
||||
|
||||
auto vertex_acc_to_primary_key = edge.dst.second;
|
||||
auto vertex_to_acc = acc.FindVertex(ConvertPropertyVector(std::move(vertex_acc_to_primary_key)), View::OLD);
|
||||
|
||||
if (!vertex_from_acc || !vertex_to_acc) {
|
||||
action_successful = false;
|
||||
spdlog::debug(
|
||||
&"Error while trying to insert edge, vertex does not exist. Transaction id: "[req.transaction_id.logical_id]);
|
||||
break;
|
||||
}
|
||||
|
||||
auto from_vertex_id = VertexId(edge.src.first.id, ConvertPropertyVector(std::move(edge.src.second)));
|
||||
auto to_vertex_id = VertexId(edge.dst.first.id, ConvertPropertyVector(std::move(edge.dst.second)));
|
||||
auto edge_acc =
|
||||
acc.CreateEdge(from_vertex_id, to_vertex_id, EdgeTypeId::FromUint(edge.type.id), Gid::FromUint(edge.id.gid));
|
||||
|
||||
if (edge_acc.HasError()) {
|
||||
action_successful = false;
|
||||
spdlog::debug(&"Creating edge was not successful. Transaction id: "[req.transaction_id.logical_id]);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
msgs::CreateEdgesResponse resp{};
|
||||
|
||||
resp.success = action_successful;
|
||||
|
||||
if (action_successful) {
|
||||
auto result = acc.Commit(req.transaction_id.logical_id);
|
||||
if (result.HasError()) {
|
||||
resp.success = false;
|
||||
spdlog::debug(
|
||||
&"ConstraintViolation, commiting edge creation was unsuccesfull with transaction id: "[req.transaction_id
|
||||
.logical_id]);
|
||||
}
|
||||
}
|
||||
|
||||
return resp;
|
||||
}
|
||||
|
||||
msgs::ReadResponses ShardRsm::HandleRead(msgs::ScanVerticesRequest &&req) {
|
||||
auto acc = shard_->Access();
|
||||
bool action_successful = true;
|
||||
|
||||
std::vector<msgs::ScanResultRow> results;
|
||||
std::optional<msgs::VertexId> next_start_id;
|
||||
|
||||
const auto view = View(req.storage_view);
|
||||
auto vertex_iterable = acc.Vertices(view);
|
||||
bool did_reach_starting_point = false;
|
||||
uint64_t sample_counter = 0;
|
||||
|
||||
for (auto it = vertex_iterable.begin(); it != vertex_iterable.end(); ++it) {
|
||||
const auto &vertex = *it;
|
||||
|
||||
if (ConvertPropertyVector(std::move(req.start_id.second)) == vertex.PrimaryKey(View(req.storage_view)).GetValue()) {
|
||||
did_reach_starting_point = true;
|
||||
}
|
||||
|
||||
if (did_reach_starting_point) {
|
||||
std::optional<std::map<PropertyId, Value>> found_props;
|
||||
|
||||
if (req.props_to_return) {
|
||||
found_props = CollectPropertiesFromAccessor(vertex, req.props_to_return.value(), view);
|
||||
} else {
|
||||
found_props = CollectAllPropertiesFromAccessor(vertex, view);
|
||||
}
|
||||
|
||||
if (!found_props) {
|
||||
continue;
|
||||
}
|
||||
|
||||
results.emplace_back(
|
||||
msgs::ScanResultRow{.vertex = ConstructValueVertex(vertex, view), .props = found_props.value()});
|
||||
|
||||
++sample_counter;
|
||||
if (sample_counter == req.batch_limit) {
|
||||
// Reached the maximum specified batch size.
|
||||
// Get the next element before exiting.
|
||||
const auto &next_vertex = *(++it);
|
||||
next_start_id = ConstructValueVertex(next_vertex, view).vertex_v.id;
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
msgs::ScanVerticesResponse resp{};
|
||||
resp.success = action_successful;
|
||||
|
||||
if (action_successful) {
|
||||
resp.next_start_id = next_start_id;
|
||||
resp.results = std::move(results);
|
||||
}
|
||||
|
||||
return resp;
|
||||
}
|
||||
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::WriteResponses ShardRsm::ApplyWrite(msgs::UpdateVerticesRequest && /*req*/) {
|
||||
return msgs::UpdateVerticesResponse{};
|
||||
}
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::WriteResponses ShardRsm::ApplyWrite(msgs::DeleteEdgesRequest && /*req*/) { return msgs::DeleteEdgesResponse{}; }
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::WriteResponses ShardRsm::ApplyWrite(msgs::UpdateEdgesRequest && /*req*/) { return msgs::UpdateEdgesResponse{}; }
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::ReadResponses ShardRsm::HandleRead(msgs::ExpandOneRequest && /*req*/) { return msgs::ExpandOneResponse{}; }
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::ReadResponses ShardRsm::HandleRead(msgs::GetPropertiesRequest && /*req*/) {
|
||||
return msgs::GetPropertiesResponse{};
|
||||
}
|
||||
|
||||
} // namespace memgraph::storage::v3
|
||||
58
src/storage/v3/shard_rsm.hpp
Normal file
58
src/storage/v3/shard_rsm.hpp
Normal file
@ -0,0 +1,58 @@
|
||||
// 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.
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <memory>
|
||||
#include <variant>
|
||||
|
||||
#include <openssl/ec.h>
|
||||
#include "query/v2/requests.hpp"
|
||||
#include "storage/v3/shard.hpp"
|
||||
#include "storage/v3/vertex_accessor.hpp"
|
||||
|
||||
namespace memgraph::storage::v3 {
|
||||
|
||||
template <typename>
|
||||
constexpr auto kAlwaysFalse = false;
|
||||
|
||||
class ShardRsm {
|
||||
std::unique_ptr<Shard> shard_;
|
||||
|
||||
msgs::ReadResponses HandleRead(msgs::ExpandOneRequest &&req);
|
||||
msgs::ReadResponses HandleRead(msgs::GetPropertiesRequest &&req);
|
||||
msgs::ReadResponses HandleRead(msgs::ScanVerticesRequest &&req);
|
||||
|
||||
msgs::WriteResponses ApplyWrite(msgs::CreateVerticesRequest &&req);
|
||||
msgs::WriteResponses ApplyWrite(msgs::DeleteVerticesRequest &&req);
|
||||
msgs::WriteResponses ApplyWrite(msgs::UpdateVerticesRequest &&req);
|
||||
|
||||
msgs::WriteResponses ApplyWrite(msgs::CreateEdgesRequest &&req);
|
||||
msgs::WriteResponses ApplyWrite(msgs::DeleteEdgesRequest &&req);
|
||||
msgs::WriteResponses ApplyWrite(msgs::UpdateEdgesRequest &&req);
|
||||
|
||||
public:
|
||||
explicit ShardRsm(std::unique_ptr<Shard> &&shard) : shard_(std::move(shard)){};
|
||||
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::ReadResponses Read(msgs::ReadRequests requests) {
|
||||
return std::visit([&](auto &&request) mutable { return HandleRead(std::forward<decltype(request)>(request)); },
|
||||
std::move(requests));
|
||||
}
|
||||
|
||||
// NOLINTNEXTLINE(readability-convert-member-functions-to-static)
|
||||
msgs::WriteResponses Apply(msgs::WriteRequests requests) {
|
||||
return std::visit([&](auto &&request) mutable { return ApplyWrite(std::forward<decltype(request)>(request)); },
|
||||
std::move(requests));
|
||||
}
|
||||
};
|
||||
|
||||
} // namespace memgraph::storage::v3
|
||||
@ -1,34 +1,28 @@
|
||||
set(test_prefix memgraph__simulation__)
|
||||
|
||||
find_package(gflags)
|
||||
find_package(Boost REQUIRED)
|
||||
find_package(OpenSSL REQUIRED)
|
||||
|
||||
add_custom_target(memgraph__simulation)
|
||||
|
||||
function(add_simulation_test test_cpp san)
|
||||
function(add_simulation_test test_cpp)
|
||||
# get exec name (remove extension from the abs path)
|
||||
get_filename_component(exec_name ${test_cpp} NAME_WE)
|
||||
set(target_name ${test_prefix}${exec_name})
|
||||
add_executable(${target_name} ${test_cpp})
|
||||
|
||||
# OUTPUT_NAME sets the real name of a target when it is built and can be
|
||||
# used to help create two targets of the same name even though CMake
|
||||
# requires unique logical target names
|
||||
set_target_properties(${target_name} PROPERTIES OUTPUT_NAME ${exec_name})
|
||||
target_link_libraries(${target_name} gtest gmock mg-utils mg-io mg-io-simulator)
|
||||
|
||||
# sanitize
|
||||
target_compile_options(${target_name} PRIVATE -fsanitize=${san})
|
||||
target_link_options(${target_name} PRIVATE -fsanitize=${san})
|
||||
|
||||
target_link_libraries(${target_name} mg-storage-v3 mg-communication gtest gmock mg-utils mg-io mg-io-simulator Boost::headers)
|
||||
# register test
|
||||
add_test(${target_name} ${exec_name})
|
||||
add_dependencies(memgraph__simulation ${target_name})
|
||||
endfunction(add_simulation_test)
|
||||
|
||||
add_simulation_test(basic_request.cpp address)
|
||||
|
||||
add_simulation_test(raft.cpp address)
|
||||
|
||||
add_simulation_test(trial_query_storage/query_storage_test.cpp address)
|
||||
|
||||
add_simulation_test(sharded_map.cpp address)
|
||||
add_simulation_test(basic_request.cpp)
|
||||
add_simulation_test(raft.cpp)
|
||||
add_simulation_test(trial_query_storage/query_storage_test.cpp)
|
||||
add_simulation_test(sharded_map.cpp)
|
||||
add_simulation_test(shard_rsm.cpp)
|
||||
|
||||
366
tests/simulation/shard_rsm.cpp
Normal file
366
tests/simulation/shard_rsm.cpp
Normal file
@ -0,0 +1,366 @@
|
||||
// Copyright 2022 Memgraph Ltd.
|
||||
//
|
||||
// Use of this software is governed by the Business Source License
|
||||
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
|
||||
// License, and you may not use this file except in compliance with the Business Source License.
|
||||
//
|
||||
// As of the Change Date specified in that file, in accordance with
|
||||
// the Business Source License, use of this software will be governed
|
||||
// by the Apache License, Version 2.0, included in the file
|
||||
// licenses/APL.txt.
|
||||
|
||||
#include <iostream>
|
||||
#include <optional>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "io/address.hpp"
|
||||
#include "io/errors.hpp"
|
||||
#include "io/rsm/raft.hpp"
|
||||
#include "io/rsm/rsm_client.hpp"
|
||||
#include "io/simulator/simulator.hpp"
|
||||
#include "io/simulator/simulator_transport.hpp"
|
||||
#include "query/v2/requests.hpp"
|
||||
#include "storage/v3/property_value.hpp"
|
||||
#include "storage/v3/shard.hpp"
|
||||
#include "storage/v3/shard_rsm.hpp"
|
||||
#include "storage/v3/view.hpp"
|
||||
#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
|
||||
template <typename IoImpl>
|
||||
void RunShardRaft(Raft<IoImpl, ShardRsm, WriteRequests, WriteResponses, ReadRequests, ReadResponses> server) {
|
||||
server.Run();
|
||||
}
|
||||
|
||||
namespace {
|
||||
|
||||
uint64_t GetTransactionId() {
|
||||
static uint64_t transaction_id = 0;
|
||||
return transaction_id++;
|
||||
}
|
||||
|
||||
uint64_t GetUniqueInteger() {
|
||||
static uint64_t prop_val_val = 1001;
|
||||
return prop_val_val++;
|
||||
}
|
||||
|
||||
LabelId get_primary_label() { return LabelId::FromUint(0); }
|
||||
|
||||
SchemaProperty get_schema_property() {
|
||||
return {.property_id = PropertyId::FromUint(0), .type = common::SchemaType::INT};
|
||||
}
|
||||
|
||||
msgs::PrimaryKey GetPrimaryKey(int64_t value) {
|
||||
msgs::Value prop_val(static_cast<int64_t>(value));
|
||||
msgs::PrimaryKey primary_key = {prop_val};
|
||||
return primary_key;
|
||||
}
|
||||
|
||||
msgs::NewVertex get_new_vertex(int64_t value) {
|
||||
// Specify Labels.
|
||||
msgs::Label label1 = {.id = LabelId::FromUint(1)};
|
||||
std::vector<msgs::Label> label_ids = {label1};
|
||||
|
||||
// Specify primary key.
|
||||
msgs::PrimaryKey primary_key = GetPrimaryKey(value);
|
||||
|
||||
// Specify properties
|
||||
auto val1 = msgs::Value(static_cast<int64_t>(value));
|
||||
auto prop1 = std::make_pair(PropertyId::FromUint(0), val1);
|
||||
|
||||
auto val2 = msgs::Value(static_cast<int64_t>(value));
|
||||
auto prop2 = std::make_pair(PropertyId::FromUint(1), val1);
|
||||
|
||||
std::vector<std::pair<PropertyId, msgs::Value>> properties{prop1, prop2};
|
||||
|
||||
// NewVertex
|
||||
return {.label_ids = label_ids, .primary_key = primary_key, .properties = properties};
|
||||
}
|
||||
|
||||
// TODO(gvolfing) maybe rename that something that makes sense.
|
||||
std::vector<std::vector<msgs::Value>> GetValuePrimaryKeysWithValue(int64_t value) {
|
||||
msgs::Value val(static_cast<int64_t>(value));
|
||||
return {{val}};
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
// attempts to sending different requests
|
||||
namespace {
|
||||
|
||||
bool AttemtpToCreateVertex(ShardClient &client, int64_t value) {
|
||||
msgs::NewVertex vertex = get_new_vertex(value);
|
||||
|
||||
auto create_req = msgs::CreateVerticesRequest{};
|
||||
create_req.new_vertices = {vertex};
|
||||
create_req.transaction_id.logical_id = GetTransactionId();
|
||||
|
||||
while (true) {
|
||||
auto write_res = client.SendWriteRequest(create_req);
|
||||
if (write_res.HasError()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
auto write_response_result = write_res.GetValue();
|
||||
auto write_response = std::get<msgs::CreateVerticesResponse>(write_response_result);
|
||||
|
||||
return write_response.success;
|
||||
}
|
||||
}
|
||||
|
||||
bool AttemptToAddEdge(ShardClient &client, int64_t value_of_vertex_1, int64_t value_of_vertex_2, int64_t edge_gid,
|
||||
int64_t edge_type_id) {
|
||||
auto id = msgs::EdgeId{};
|
||||
msgs::Label label = {.id = get_primary_label()};
|
||||
|
||||
auto src = std::make_pair(label, GetPrimaryKey(value_of_vertex_1));
|
||||
auto dst = std::make_pair(label, GetPrimaryKey(value_of_vertex_2));
|
||||
id.gid = edge_gid;
|
||||
|
||||
auto type = msgs::EdgeType{};
|
||||
type.id = edge_type_id;
|
||||
|
||||
auto edge = msgs::Edge{};
|
||||
edge.id = id;
|
||||
edge.type = type;
|
||||
edge.src = src;
|
||||
edge.dst = dst;
|
||||
|
||||
msgs::CreateEdgesRequest create_req{};
|
||||
create_req.edges = {edge};
|
||||
create_req.transaction_id.logical_id = GetTransactionId();
|
||||
|
||||
while (true) {
|
||||
auto write_res = client.SendWriteRequest(create_req);
|
||||
if (write_res.HasError()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
auto write_response_result = write_res.GetValue();
|
||||
auto write_response = std::get<msgs::CreateEdgesResponse>(write_response_result);
|
||||
|
||||
return write_response.success;
|
||||
}
|
||||
}
|
||||
|
||||
std::tuple<size_t, std::optional<msgs::VertexId>> AttemptToScanAllWithBatchLimit(ShardClient &client,
|
||||
msgs::VertexId start_id,
|
||||
uint64_t batch_limit) {
|
||||
msgs::ScanVerticesRequest scan_req{};
|
||||
scan_req.batch_limit = batch_limit;
|
||||
scan_req.filter_expressions = std::nullopt;
|
||||
scan_req.props_to_return = std::nullopt;
|
||||
scan_req.start_id = start_id;
|
||||
scan_req.storage_view = msgs::StorageView::OLD;
|
||||
scan_req.transaction_id.logical_id = GetTransactionId();
|
||||
|
||||
while (true) {
|
||||
auto read_res = client.SendReadRequest(scan_req);
|
||||
if (read_res.HasError()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
auto write_response_result = read_res.GetValue();
|
||||
auto write_response = std::get<msgs::ScanVerticesResponse>(write_response_result);
|
||||
|
||||
MG_ASSERT(write_response.success);
|
||||
return {write_response.results.size(), write_response.next_start_id};
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
// tests
|
||||
namespace {
|
||||
|
||||
void TestCreateVertices(ShardClient &client) { MG_ASSERT(AttemtpToCreateVertex(client, GetUniqueInteger())); }
|
||||
|
||||
void TestAddEdge(ShardClient &client) {
|
||||
auto unique_prop_val_1 = GetUniqueInteger();
|
||||
auto unique_prop_val_2 = GetUniqueInteger();
|
||||
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_1));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_2));
|
||||
|
||||
auto edge_gid = GetUniqueInteger();
|
||||
auto edge_type_id = GetUniqueInteger();
|
||||
|
||||
MG_ASSERT(AttemptToAddEdge(client, unique_prop_val_1, unique_prop_val_2, edge_gid, edge_type_id));
|
||||
}
|
||||
|
||||
void TestScanAllOneGo(ShardClient &client) {
|
||||
auto unique_prop_val_1 = GetUniqueInteger();
|
||||
auto unique_prop_val_2 = GetUniqueInteger();
|
||||
auto unique_prop_val_3 = GetUniqueInteger();
|
||||
auto unique_prop_val_4 = GetUniqueInteger();
|
||||
auto unique_prop_val_5 = GetUniqueInteger();
|
||||
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_1));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_2));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_3));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_4));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_5));
|
||||
|
||||
msgs::Label prim_label = {.id = get_primary_label()};
|
||||
msgs::PrimaryKey prim_key = {msgs::Value(static_cast<int64_t>(unique_prop_val_1))};
|
||||
|
||||
msgs::VertexId v_id = {prim_label, prim_key};
|
||||
|
||||
auto [result_size, next_id] = AttemptToScanAllWithBatchLimit(client, v_id, 5);
|
||||
MG_ASSERT(result_size == 5);
|
||||
}
|
||||
|
||||
void TestScanAllWithSmallBatchSize(ShardClient &client) {
|
||||
auto unique_prop_val_1 = GetUniqueInteger();
|
||||
auto unique_prop_val_2 = GetUniqueInteger();
|
||||
auto unique_prop_val_3 = GetUniqueInteger();
|
||||
auto unique_prop_val_4 = GetUniqueInteger();
|
||||
auto unique_prop_val_5 = GetUniqueInteger();
|
||||
auto unique_prop_val_6 = GetUniqueInteger();
|
||||
auto unique_prop_val_7 = GetUniqueInteger();
|
||||
auto unique_prop_val_8 = GetUniqueInteger();
|
||||
auto unique_prop_val_9 = GetUniqueInteger();
|
||||
auto unique_prop_val_10 = GetUniqueInteger();
|
||||
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_1));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_2));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_3));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_4));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_5));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_6));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_7));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_8));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_9));
|
||||
MG_ASSERT(AttemtpToCreateVertex(client, unique_prop_val_10));
|
||||
|
||||
msgs::Label prim_label = {.id = get_primary_label()};
|
||||
msgs::PrimaryKey prim_key1 = {msgs::Value(static_cast<int64_t>(unique_prop_val_1))};
|
||||
|
||||
msgs::VertexId v_id_1 = {prim_label, prim_key1};
|
||||
|
||||
auto [result_size1, next_id1] = AttemptToScanAllWithBatchLimit(client, v_id_1, 3);
|
||||
MG_ASSERT(result_size1 == 3);
|
||||
|
||||
auto [result_size2, next_id2] = AttemptToScanAllWithBatchLimit(client, next_id1.value(), 3);
|
||||
MG_ASSERT(result_size2 == 3);
|
||||
|
||||
auto [result_size3, next_id3] = AttemptToScanAllWithBatchLimit(client, next_id2.value(), 3);
|
||||
MG_ASSERT(result_size3 == 3);
|
||||
|
||||
auto [result_size4, next_id4] = AttemptToScanAllWithBatchLimit(client, next_id3.value(), 3);
|
||||
MG_ASSERT(result_size4 == 1);
|
||||
MG_ASSERT(!next_id4);
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
int TestMessages() {
|
||||
SimulatorConfig config{
|
||||
.drop_percent = 0,
|
||||
.perform_timeouts = false,
|
||||
.scramble_messages = false,
|
||||
.rng_seed = 0,
|
||||
.start_time = Time::min() + std::chrono::microseconds{256 * 1024},
|
||||
.abort_time = Time::min() + std::chrono::microseconds{4 * 8 * 1024 * 1024},
|
||||
};
|
||||
|
||||
auto simulator = Simulator(config);
|
||||
|
||||
Io<SimulatorTransport> shard_server_io_1 = simulator.RegisterNew();
|
||||
const auto shard_server_1_address = shard_server_io_1.GetAddress();
|
||||
Io<SimulatorTransport> shard_server_io_2 = simulator.RegisterNew();
|
||||
const auto shard_server_2_address = shard_server_io_2.GetAddress();
|
||||
Io<SimulatorTransport> shard_server_io_3 = simulator.RegisterNew();
|
||||
const auto shard_server_3_address = shard_server_io_3.GetAddress();
|
||||
Io<SimulatorTransport> shard_client_io = simulator.RegisterNew();
|
||||
|
||||
PropertyValue min_pk(static_cast<int64_t>(0));
|
||||
std::vector<PropertyValue> min_prim_key = {min_pk};
|
||||
|
||||
PropertyValue max_pk(static_cast<int64_t>(10000000));
|
||||
std::vector<PropertyValue> max_prim_key = {max_pk};
|
||||
|
||||
auto shard_ptr1 = std::make_unique<Shard>(get_primary_label(), min_prim_key, max_prim_key);
|
||||
auto shard_ptr2 = std::make_unique<Shard>(get_primary_label(), min_prim_key, max_prim_key);
|
||||
auto shard_ptr3 = std::make_unique<Shard>(get_primary_label(), min_prim_key, max_prim_key);
|
||||
|
||||
shard_ptr1->CreateSchema(get_primary_label(), {get_schema_property()});
|
||||
shard_ptr2->CreateSchema(get_primary_label(), {get_schema_property()});
|
||||
shard_ptr3->CreateSchema(get_primary_label(), {get_schema_property()});
|
||||
|
||||
std::vector<Address> address_for_1{shard_server_2_address, shard_server_3_address};
|
||||
std::vector<Address> address_for_2{shard_server_1_address, shard_server_3_address};
|
||||
std::vector<Address> address_for_3{shard_server_1_address, shard_server_2_address};
|
||||
|
||||
ConcreteShardRsm shard_server1(std::move(shard_server_io_1), address_for_1, ShardRsm(std::move(shard_ptr1)));
|
||||
ConcreteShardRsm shard_server2(std::move(shard_server_io_2), address_for_2, ShardRsm(std::move(shard_ptr2)));
|
||||
ConcreteShardRsm shard_server3(std::move(shard_server_io_3), address_for_3, ShardRsm(std::move(shard_ptr3)));
|
||||
|
||||
auto server_thread1 = std::jthread([&shard_server1]() { shard_server1.Run(); });
|
||||
auto server_thread2 = std::jthread([&shard_server2]() { shard_server2.Run(); });
|
||||
auto server_thread3 = std::jthread([&shard_server3]() { shard_server3.Run(); });
|
||||
|
||||
simulator.IncrementServerCountAndWaitForQuiescentState(shard_server_1_address);
|
||||
simulator.IncrementServerCountAndWaitForQuiescentState(shard_server_2_address);
|
||||
simulator.IncrementServerCountAndWaitForQuiescentState(shard_server_3_address);
|
||||
|
||||
std::cout << "Beginning test after servers have become quiescent." << std::endl;
|
||||
|
||||
std::vector server_addrs = {shard_server_1_address, shard_server_2_address, shard_server_3_address};
|
||||
ShardClient client(shard_client_io, shard_server_1_address, server_addrs);
|
||||
|
||||
TestCreateVertices(client);
|
||||
TestAddEdge(client);
|
||||
TestScanAllOneGo(client);
|
||||
TestScanAllWithSmallBatchSize(client);
|
||||
|
||||
simulator.ShutDown();
|
||||
|
||||
SimulatorStats stats = simulator.Stats();
|
||||
|
||||
std::cout << "total messages: " << stats.total_messages << std::endl;
|
||||
std::cout << "dropped messages: " << stats.dropped_messages << std::endl;
|
||||
std::cout << "timed out requests: " << stats.timed_out_requests << std::endl;
|
||||
std::cout << "total requests: " << stats.total_requests << std::endl;
|
||||
std::cout << "total responses: " << stats.total_responses << std::endl;
|
||||
std::cout << "simulator ticks: " << stats.simulator_ticks << std::endl;
|
||||
|
||||
std::cout << "========================== SUCCESS :) ==========================" << std::endl;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
} // namespace memgraph::storage::v3::tests
|
||||
|
||||
int main() { return memgraph::storage::v3::tests::TestMessages(); }
|
||||
Loading…
Reference in New Issue
Block a user