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

Major properties and storage refactoring in progress. UNSTABLE STATE

Browse Source
This commit is contained in:
Florijan Stamenkovic 2017-02-11 07:51:02 +01:00 committed by florijan
parent ee523d5080
commit 9adf5699d9
22 changed files with 479 additions and 539 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
cmake/copy_includes.cmake
View File

@ -52,9 +52,9 @@ FILE(COPY ${include_dir}/transactions/engine.hpp DESTINATION ${build_include_dir
FILE(COPY ${include_dir}/transactions/transaction_store.hpp DESTINATION ${build_include_dir}/transactions)
FILE(COPY ${include_dir}/transactions/transaction_read.hpp DESTINATION ${build_include_dir}/transactions)
FILE(COPY ${include_dir}/storage/typed_value.hpp DESTINATION ${build_include_dir}/storage/model)
FILE(COPY ${include_dir}/storage/typed_value_store.hpp DESTINATION ${build_include_dir}/storage/model)
FILE(COPY ${include_dir}/storage/typed_value_utils.hpp DESTINATION ${build_include_dir}/storage/model)
FILE(COPY ${include_dir}/storage/typed_value.hpp DESTINATION ${build_include_dir}/storage)
FILE(COPY ${include_dir}/storage/typed_value_store.hpp DESTINATION ${build_include_dir}/storage)
FILE(COPY ${include_dir}/storage/typed_value_utils.hpp DESTINATION ${build_include_dir}/storage)
FILE(COPY ${include_dir}/storage/garbage/delete_sensitive.hpp DESTINATION ${build_include_dir}/storage/garbage)
FILE(COPY ${include_dir}/storage/garbage/garbage.hpp DESTINATION ${build_include_dir}/storage/garbage)

115
include/communication/bolt/v1/serialization/bolt_serializer.hpp
View File

@ -6,25 +6,14 @@
#include "storage/edge_accessor.hpp"
#include "storage/vertex_accessor.hpp"
#include "storage/edge_type/edge_type.hpp"
#include "storage/label/label.hpp"
#include "storage/model/properties/all.hpp"
#include "storage/model/properties/properties.hpp"
#include "storage/vertex_record.hpp"
#include "storage/typed_value.hpp"
namespace bolt
{
namespace bolt {
template <class Stream>
class BoltSerializer
{
friend class Property;
template<class Stream>
class BoltSerializer {
// TODO: here shoud be friend but it doesn't work
// template <class Handler>
// friend void accept(const Property &property, Handler &h);
public:
public:
BoltSerializer(Stream &stream) : encoder(stream) {}
/** Serializes the vertex accessor into the packstream format
@ -36,38 +25,9 @@ public:
* }
*
*/
void write(const VertexAccessor &vertex)
{
// write signatures for the node struct and node data type
encoder.write_struct_header(3);
encoder.write(underlying_cast(pack::Node));
void write(const VertexAccessor &vertex);
// IMPORTANT: here we write a hardcorded 0 because we don't
// use internal IDs, but need to give something to Bolt
// note that OpenCypther has no id(x) function, so the client
// should not be able to do anything with this value anyway
encoder.write_integer(0);
// write the list of labels
auto labels = vertex.labels();
encoder.write_list_header(labels.size());
for (auto &label : labels)
encoder.write_string(label.get());
// write the property map
auto props = vertex.properties();
encoder.write_map_header(props.size());
for (auto &prop : props) {
write(prop.key.family_name());
prop.accept(*this);
}
}
/** Serializes the vertex accessor into the packstream format
/** Serializes the edge accessor into the packstream format
*
* struct[size = 5] Edge [signature = 0x52] {
* Integer edge_id; // IMPORTANT: always 0 since we don't do IDs
@ -80,57 +40,18 @@ public:
*/
void write(const EdgeAccessor &edge);
void write_null() { encoder.write_null(); }
// TODO document
void write_failure(const std::map<std::string, std::string> &data);
void write(const Null &) { encoder.write_null(); }
/**
* Writes a TypedValue (typically a property value in the edge or vertex).
*
* @param value The value to write.
*/
void write(const TypedValue& value);
void write(const Bool &prop) { encoder.write_bool(prop.value()); }
void write(const Float &prop) { encoder.write_double(prop.value()); }
void write(const Double &prop) { encoder.write_double(prop.value()); }
void write(const Int32 &prop) { encoder.write_integer(prop.value()); }
void write(const Int64 &prop) { encoder.write_integer(prop.value()); }
void write(const String &value) { encoder.write_string(value.value()); }
// Not yet implemented
void write(const ArrayBool &) { assert(false); }
// Not yet implemented
void write(const ArrayInt32 &) { assert(false); }
// Not yet implemented
void write(const ArrayInt64 &) { assert(false); }
// Not yet implemented
void write(const ArrayFloat &) { assert(false); }
// Not yet implemented
void write(const ArrayDouble &) { assert(false); }
// Not yet implemented
void write(const ArrayString &) { assert(false); }
void write_failure(const std::map<std::string, std::string> &data)
{
encoder.message_failure();
encoder.write_map_header(data.size());
for (auto const &kv : data) {
write(kv.first);
write(kv.second);
}
}
template <class T>
void handle(const T &prop)
{
write(prop);
}
protected:
protected:
Stream &encoder;
};
};
}

89
include/communication/bolt/v1/serialization/record_stream.hpp
View File

@ -7,45 +7,38 @@
#include "logging/default.hpp"
namespace bolt
{
namespace bolt {
/**
* compiled queries have to use this class in order to return results
* query code should not know about bolt protocol
*/
template <class Socket>
class RecordStream
{
public:
RecordStream(Socket &socket) : socket(socket)
{
template<class Socket>
class RecordStream {
public:
RecordStream(Socket &socket) : socket(socket) {
logger = logging::log->logger("Record Stream");
}
~RecordStream() = default;
// TODO: create apstract methods that are not bolt specific ---------------
void write_success()
{
void write_success() {
logger.trace("write_success");
bolt_encoder.message_success();
}
void write_success_empty()
{
void write_success_empty() {
logger.trace("write_success_empty");
bolt_encoder.message_success_empty();
}
void write_ignored()
{
void write_ignored() {
logger.trace("write_ignored");
bolt_encoder.message_ignored();
}
void write_empty_fields()
{
void write_empty_fields() {
bolt_encoder.message_success();
bolt_encoder.write_map_header(1);
bolt_encoder.write_string("fields");
@ -53,8 +46,7 @@ public:
chunk();
}
void write_fields(const std::vector<std::string> &fields)
{
void write_fields(const std::vector<std::string> &fields) {
// TODO: that should be one level below?
bolt_encoder.message_success();
@ -70,8 +62,7 @@ public:
send();
}
void write_field(const std::string &field)
{
void write_field(const std::string &field) {
bolt_encoder.message_success();
bolt_encoder.write_map_header(1);
bolt_encoder.write_string("fields");
@ -81,8 +72,7 @@ public:
send();
}
void write_list_header(size_t size)
{
void write_list_header(size_t size) {
bolt_encoder.write_list_header(size);
}
@ -92,8 +82,7 @@ public:
// TODO: write whole implementation (currently, only type is supported)
// { "stats": { "nodes created": 1, "properties set": 1},
// "type": "r" | "rw" | ...
void write_meta(const std::string &type)
{
void write_meta(const std::string &type) {
bolt_encoder.message_success();
bolt_encoder.write_map_header(1);
bolt_encoder.write_string("type");
@ -101,23 +90,21 @@ public:
chunk();
}
void write_failure(const std::map<std::string, std::string> &data)
{
void write_failure(const std::map<std::string, std::string> &data) {
serializer.write_failure(data);
chunk();
}
void write_count(const size_t count)
{
void write_count(const size_t count) {
write_record();
write_list_header(1);
write(Int64(count));
write(count);
chunk();
}
void write(const VertexAccessor &vertex) { serializer.write(vertex); }
void write_vertex_record(const VertexAccessor& va)
{
void write_vertex_record(const VertexAccessor &va) {
write_record();
write_list_header(1);
write(va);
@ -125,61 +112,43 @@ public:
}
void write(const EdgeAccessor &edge) { serializer.write(edge); }
void write_edge_record(const EdgeAccessor& ea)
{
void write_edge_record(const EdgeAccessor &ea) {
write_record();
write_list_header(1);
write(ea);
chunk();
}
void write(const StoredProperty<TypeGroupEdge> &prop)
{
prop.accept(serializer);
void write(const TypedValue& value) {
serializer.write(value);
}
void write(const StoredProperty<TypeGroupVertex> &prop)
{
prop.accept(serializer);
}
void write(const Null &prop) { serializer.write(prop); }
void write(const Bool &prop) { serializer.write(prop); }
void write(const Float &prop) { serializer.write(prop); }
void write(const Int32 &prop) { serializer.write(prop); }
void write(const Int64 &prop) { serializer.write(prop); }
void write(const Double &prop) { serializer.write(prop); }
void write(const String &prop) { serializer.write(prop); }
void write(const ArrayBool &prop) { serializer.write(prop); }
void write(const ArrayInt32 &prop) { serializer.write(prop); }
void write(const ArrayInt64 &prop) { serializer.write(prop); }
void write(const ArrayFloat &prop) { serializer.write(prop); }
void write(const ArrayDouble &prop) { serializer.write(prop); }
void write(const ArrayString &prop) { serializer.write(prop); }
void send() { chunked_buffer.flush(); }
void chunk() { chunked_encoder.write_chunk(); }
void _write_test()
{
// TODO WTF is this test doing here?
void _write_test() {
logger.trace("write_test");
write_fields({{"name"}});
write_record();
write_list_header(1);
write(String("max"));
bolt_encoder.write("max");
write_record();
write_list_header(1);
write(String("paul"));
bolt_encoder.write("paul");
write_success_empty();
}
protected:
protected:
Logger logger;
private:
private:
using socket_t = SocketStream<Socket>;
using buffer_t = ChunkedBuffer<socket_t>;
using chunked_encoder_t = ChunkedEncoder<buffer_t>;
@ -191,5 +160,5 @@ private:
chunked_encoder_t chunked_encoder{chunked_buffer};
bolt_encoder_t bolt_encoder{chunked_encoder};
bolt_serializer_t serializer{bolt_encoder};
};
};
}

16
include/communication/bolt/v1/session.hpp
View File

@ -12,12 +12,10 @@
#include "logging/default.hpp"
namespace bolt
{
namespace bolt {
class Session : public io::tcp::Stream<io::Socket>
{
public:
class Session : public io::tcp::Stream<io::Socket> {
public:
using Decoder = BoltDecoder;
using OutputStream = communication::OutputStream;
@ -26,10 +24,12 @@ public:
bool alive() const;
void execute(const byte *data, size_t len);
void close();
Bolt &bolt;
Db &active_db();
GraphDb &active_db();
Decoder decoder;
OutputStream output_stream{socket};
@ -37,7 +37,7 @@ public:
bool connected{false};
State *state;
protected:
protected:
Logger logger;
};
};
}

13
include/database/graph_db.hpp
View File

@ -4,6 +4,7 @@
#include "transactions/engine.hpp"
#include "mvcc/version_list.hpp"
#include "utils/pass_key.hpp"
#include "data_structures/concurrent/concurrent_set.hpp"
#include "storage/unique_object_store.hpp"
// forward declaring Edge and Vertex because they use
@ -26,9 +27,9 @@ class GraphDb {
public:
// definitions for what data types are used for a Label, Property, EdgeType
using Label = uint32_t;
using EdgeType = uint32_t;
using Property = uint32_t;
using Label = std::string*;
using EdgeType = std::string*;
using Property = std::string*;
/**
* This constructor will create a database with the name "default"
@ -83,7 +84,7 @@ public:
SkipList<mvcc::VersionList<Vertex>*> vertices_;
// unique object stores
UniqueObjectStore<std::string, Label> labels_;
UniqueObjectStore<std::string, EdgeType> edge_types_;
UniqueObjectStore<std::string, Property> properties_;
ConcurrentSet<std::string> labels_;
ConcurrentSet<std::string> edge_types_;
ConcurrentSet<std::string> properties_;
};

32
include/database/graph_db_accessor.hpp
View File

@ -3,14 +3,14 @@
// Created by Florijan Stamenkovic on 03.02.17.
//
#pragma
#pragma once
#include "graph_db.hpp"
#include "transactions/transaction.hpp"
class GraphDbAccessor {
GraphDbAccessor(GraphDb& db) : db_(db), transaction_(db.tx_engine.begin()) {}
GraphDbAccessor(GraphDb& db);
public:
/**
@ -36,24 +36,48 @@ public:
*/
GraphDb::Label label(const std::string& label_name);
/**
* Obtains the label name (a string) for the given label.
*
* @param label a Label.
* @return See above.
*/
std::string& label_name(const GraphDb::Label label) const;
/**
* Obtains the EdgeType for it's name.
* @return See above.
*/
GraphDb::EdgeType edge_type(const std::string& edge_type_name);
/**
* Obtains the edge type name (a string) for the given edge type.
*
* @param edge_type an EdgeType.
* @return See above.
*/
std::string& edge_type_name(const GraphDb::EdgeType edge_type) const;
/**
* Obtains the Property for it's name.
* @return See above.
*/
GraphDb::Property property(const std::string& property_name);
/**
* Obtains the property name (a string) for the given property.
*
* @param property a Property.
* @return See above.
*/
std::string& property_name(const GraphDb::Property property) const;
/** The current transaction */
tx::Transaction const transaction_;
tx::Transaction transaction_;
private:
GraphDb& db_;
// for privileged access to some RecordAccessor functionality (and similar)
const PassKey<GraphDb> pass_key;
const PassKey<GraphDbAccessor> pass_key;
};

15
include/query/i_plan_cpu.hpp
View File

@ -1,20 +1,19 @@
#pragma once
#include "communication/communication.hpp"
#include "database/graph_db.hpp"
#include "database/db_accessor.hpp"
#include "database/graph_db_accessor.hpp"
#include "query/strip/stripped.hpp"
template <typename Stream>
class IPlanCPU
{
template<typename Stream>
class IPlanCPU {
public:
virtual bool run(Db &db, plan_args_t &args, Stream &stream) = 0;
virtual bool run(GraphDbAccessor &db_accessor, TypedValueStore<>& args, Stream &stream) = 0;
virtual ~IPlanCPU() {}
};
template <typename Stream>
template<typename Stream>
using produce_t = IPlanCPU<Stream> *(*)();
template <typename Stream>
template<typename Stream>
using destruct_t = void (*)(IPlanCPU<Stream> *);

5
include/query/plan/program_loader.hpp
View File

@ -7,7 +7,6 @@
#include "config/config.hpp"
#include "logging/default.hpp"
#include "query/backend/cpp_old/cypher.hpp"
#include "query/dynamic_lib.hpp"
#include "query/frontend/cypher.hpp"
#include "query/plan/compiler.hpp"
@ -99,9 +98,7 @@ private:
QueryPreprocessor preprocessor;
// TODO: compile time switch between frontends and backends
using frontend_t = cypher::Frontend;
using backend_t = CypherBackend<Stream>;
PlanGenerator<frontend_t, backend_t> plan_generator;
PlanGenerator<cypher::Frontend, CypherBackend<Stream>> plan_generator;
PlanCompiler plan_compiler;

22
include/query/strip/stripped.hpp
View File

@ -2,12 +2,7 @@
#include <vector>
#include "storage/model/properties/property.hpp"
/*
* Query Plan Arguments Type
*/
using plan_args_t = std::vector<Property>;
#include "storage/typed_value_store.hpp"
/*
* QueryStripped contains:
@ -15,18 +10,17 @@ using plan_args_t = std::vector<Property>;
* * plan arguments stripped from query
* * hash of stripped query
*/
struct QueryStripped
{
QueryStripped(const std::string &&query, plan_args_t &&arguments,
uint64_t hash)
struct QueryStripped {
QueryStripped(const std::string &&query, const TypedValueStore<> &&arguments, uint64_t hash)
: query(std::forward<const std::string>(query)),
arguments(std::forward<plan_args_t>(arguments)), hash(hash)
{
}
arguments(std::forward<const TypedValueStore<>>(arguments)),
hash(hash) {}
QueryStripped(QueryStripped &other) = delete;
QueryStripped(QueryStripped &&other) = default;
std::string query;
plan_args_t arguments;
TypedValueStore<> arguments;
uint64_t hash;
};

68
include/query/strip/stripper.hpp
View File

@ -10,41 +10,31 @@
#include "logging/loggable.hpp"
#include "query/language/cypher/tokenizer/cypher_lexer.hpp"
#include "query/strip/stripped.hpp"
#include "storage/model/properties/all.hpp"
#include "storage/typed_value_store.hpp"
#include "utils/hashing/fnv.hpp"
#include "utils/string/transform.hpp"
#include "utils/variadic/variadic.hpp"
// TODO: Maybe std::move(v) is faster, but it must be cheked for validity.
template <class T, class V>
void store_query_param(plan_args_t &arguments, V &&v)
{
arguments.emplace_back(Property(T(std::move(v)), T::type));
}
template <typename... Ts>
class QueryStripper : public Loggable
{
template<typename... Ts>
class QueryStripper : public Loggable {
public:
QueryStripper(Ts &&... strip_types)
: Loggable("QueryStripper"),
strip_types(std::make_tuple(std::forward<Ts>(strip_types)...)),
lexer(std::make_unique<CypherLexer>())
{
lexer(std::make_unique<CypherLexer>()) {
}
QueryStripper(QueryStripper &other) = delete;
QueryStripper(QueryStripper &&other)
: Loggable("QueryStripper"), strip_types(std::move(other.strip_types)),
lexer(std::move(other.lexer))
{
lexer(std::move(other.lexer)) {
}
auto strip_space(const std::string &query) { return strip(query, " "); }
auto strip(const std::string &query, const std::string &separator = "")
{
auto strip(const std::string &query, const std::string &separator = "") {
// -------------------------------------------------------------------
// TODO: write speed tests and then optimize, because this
// function is called before every query execution !
@ -59,47 +49,38 @@ public:
// TMP size of supported token types
constexpr auto size = std::tuple_size<decltype(strip_types)>::value;
int counter = 0;
plan_args_t stripped_arguments;
TypedValueStore<> stripped_arguments;
std::string stripped_query;
stripped_query.reserve(query.size());
while (auto token = tokenizer.lookup())
{
if (_or(token.id, strip_types, std::make_index_sequence<size>{}))
{
auto index = counter++;
switch (token.id)
{
int counter = 0; // how many arguments have we processed so far
while (auto token = tokenizer.lookup()) {
if (_or(token.id, strip_types, std::make_index_sequence < size > {})) {
switch (token.id) {
case TK_LONG:
store_query_param<Int64>(stripped_arguments,
std::stol(token.value));
stripped_arguments.set(counter, std::stoi(token.value));
break;
case TK_STR:
// TODO: remove quotes view lexertl
token.value.erase(0, 1);
token.value.erase(token.value.length() - 1, 1);
// TODO: remove
store_query_param<String>(stripped_arguments, token.value);
stripped_arguments.set(counter, token.value);
break;
case TK_BOOL:
{
case TK_BOOL: {
bool value = token.value[0] == 'T' || token.value[0] == 't';
store_query_param<Bool>(stripped_arguments, value);
stripped_arguments.set(counter, value);
break;
}
case TK_FLOAT:
store_query_param<Float>(stripped_arguments,
std::stof(token.value));
stripped_arguments.set(counter, std::stof(token.value));
break;
default:
// TODO: other properties
assert(false);
}
stripped_query += std::to_string(index) + separator;
}
else
{
stripped_query += std::to_string(counter++) + separator;
} else {
// if token is keyword then lowercase because query hash
// should be the same
// TODO: probably we shoud do the lowercase before
@ -111,8 +92,7 @@ public:
token.id == TK_DELETE || token.id == TK_DETACH ||
token.id == TK_WHERE || token.id == TK_RETURN ||
token.id == TK_DISTINCT || token.id == TK_COUNT ||
token.id == TK_LABELS)
{
token.id == TK_LABELS) {
std::transform(token.value.begin(), token.value.end(),
token.value.begin(), ::tolower);
}
@ -130,16 +110,14 @@ private:
std::tuple<Ts...> strip_types;
CypherLexer::uptr lexer;
template <typename Value, typename Tuple, std::size_t... index>
bool _or(Value &&value, Tuple &&tuple, std::index_sequence<index...>)
{
template<typename Value, typename Tuple, std::size_t... index>
bool _or(Value &&value, Tuple &&tuple, std::index_sequence<index...>) {
return utils::or_vargs(std::forward<Value>(value),
std::get<index>(std::forward<Tuple>(tuple))...);
}
};
template <typename... Ts>
decltype(auto) make_query_stripper(Ts &&... ts)
{
template<typename... Ts>
decltype(auto) make_query_stripper(Ts &&... ts) {
return QueryStripper<Ts...>(std::forward<Ts>(ts)...);
}

58
include/query/util.hpp
View File

@ -7,10 +7,6 @@
#include "fmt/format.h"
#include "logging/default.hpp"
#include "storage/model/properties/properties.hpp"
#include "storage/model/properties/property.hpp"
#include "storage/model/properties/json_writer.hpp"
#include "utils/types/byte.hpp"
#include "utils/exceptions/basic_exception.hpp"
using std::cout;
@ -20,64 +16,42 @@ using std::endl;
// headers because it will create a unique namespace for each compilation unit
// http://stackoverflow.com/questions/2727582/multiple-definition-in-header-file
// but sometimes that might be a problem
namespace
{
namespace {
class CodeLineFormatException : public BasicException
{
public:
class CodeLineFormatException : public BasicException {
public:
using BasicException::BasicException;
};
};
template <typename... Args>
std::string format(const std::string &format_str, const Args &... args)
{
template<typename... Args>
std::string format(const std::string &format_str, const Args &... args) {
return fmt::format(format_str, args...);
}
}
template <typename... Args>
std::string code_line(const std::string &format_str, const Args &... args)
{
template<typename... Args>
std::string code_line(const std::string &format_str, const Args &... args) {
try {
return "\t" + format(format_str, args...) + "\n";
} catch (std::runtime_error &e) {
throw CodeLineFormatException(std::string(e.what()) + " " + format_str);
}
}
using name_properties_t = std::vector<std::pair<std::string, Property>>;
using indices_t = std::map<std::string, long>;
auto query_properties(indices_t &indices, properties_t &values)
{
name_properties_t properties;
for (auto &property_index : indices) {
properties.push_back(
std::make_pair(std::move(property_index.first),
std::move(values[property_index.second])));
}
return properties;
}
class CoutSocket
{
public:
class CoutSocket {
public:
CoutSocket() : logger(logging::log->logger("Cout Socket")) {}
int write(const std::string &str)
{
int write(const std::string &str) {
logger.info(str);
return str.size();
}
int write(const char *data, size_t len)
{
int write(const char *data, size_t len) {
logger.info(std::string(data, len));
return len;
}
int write(const byte *data, size_t len)
{
int write(const byte *data, size_t len) {
std::stringstream ss;
for (int i = 0; i < len; i++) {
ss << data[i];
@ -88,8 +62,8 @@ public:
return len;
}
private:
private:
Logger logger;
};
};
}

2
include/storage/edge.hpp
View File

@ -13,5 +13,5 @@ public:
mvcc::VersionList<Vertex>* from_;
mvcc::VersionList<Vertex>* to_;
GraphDb::EdgeType edge_type_;
TypedValueStore properties_;
TypedValueStore<GraphDb::Property> properties_;
};

92
include/storage/record_accessor.hpp
View File

@ -1,49 +1,37 @@
#pragma once
#include "mvcc/version_list.hpp"
#include "transactions/transaction.hpp"
#include "storage/typed_value.hpp"
#include "database/graph_db.hpp"
#include "database/graph_db_accessor.hpp"
#include "utils/pass_key.hpp"
template <typename TRecord, typename TDerived>
template<typename TRecord, typename TDerived>
class RecordAccessor {
public:
RecordAccessor(mvcc::VersionList<TRecord>* vlist, tx::Transaction &trans)
: vlist_(vlist), trans_(trans) {
record_ = vlist->find(trans_);
RecordAccessor(mvcc::VersionList<TRecord> *vlist, GraphDbAccessor *db_accessor)
: vlist_(vlist), db_accessor_(db_accessor) {
record_ = vlist->find(db_accessor->transaction_);
}
/**
* Indicates if this record is visible to the current transaction.
*
* @return
*/
bool is_visible() const {
return record_ != nullptr;
template<typename TValue>
void PropsSet(GraphDb::Property key, TValue value) {
update()->props_.set(key, value);
}
TypedValue at(GraphDb::Property key) const {
return record_->props_.at(key);
size_t PropsErase(GraphDb::Property key) {
return update()->props_.erase(key);
}
template <typename TValue>
void set(GraphDb::Property key, TValue value) {
update();
record_->props_.set(key, value);
const TypedValueStore<GraphDb::Property> &Properties() const {
return view().properties_;
}
size_t erase(GraphDb::Property key) {
update();
return record_->props_.erase(key);
}
void Accept(std::function<void(const GraphDb::Property key, const TypedValue& prop)> handler,
void PropertiesAccept(std::function<void(const GraphDb::Property key, const TypedValue &prop)> handler,
std::function<void()> finish = {}) const {
record_->props_.Accept(handler, finish);
view()->props_.Accept(handler, finish);
}
// Assumes same transaction
@ -64,29 +52,57 @@ public:
* @param pass_key Ignored.
* @return The version list of this accessor.
*/
mvcc::VersionList<TRecord>* vlist(PassKey<GraphDbAccessor> pass_key) const {
mvcc::VersionList<TRecord> *vlist(PassKey<GraphDbAccessor> pass_key) const {
return vlist_;
}
/**
* Returns a GraphDB accessor of this record accessor.
*
* @return
*/
const GraphDbAccessor &db_accessor() const {
return db_accessor_;
}
protected:
/**
* Ensures this accessor is fit for updating functions.
* Returns the update-ready version of the record.
*
* IMPORTANT: This function should be called from any
* method that will change the record (in terms of the
* property graph data).
* @return See above.
*/
void update() {
TRecord *update() {
// TODO consider renaming this to something more indicative
// of the underlying MVCC functionality (like "new_version" or so)
if (record_->is_visible_write(trans_))
return;
else
record_ = vlist_->update(trans_);
if (!record_->is_visible_write(db_accessor_->transaction_))
record_ = vlist_->update(db_accessor_->transaction_);
return record_;
}
mvcc::VersionList<TRecord>* vlist_;
tx::Transaction& trans_;
TRecord* record_;
/**
* Returns a version of the record that is only for viewing.
*
* @return See above.
*/
const TRecord *view() const {
return record_;
}
// The record (edge or vertex) this accessor provides access to.
mvcc::VersionList<TRecord> *vlist_;
// The database accessor for which this record accessor is created
// Provides means of getting to the transaction and database functions.
GraphDbAccessor *db_accessor_;
private:
/* The version of the record currently used in this transaction. Defaults to the
* latest viewable version (set in the constructor). After the first update done
* through this accessor a new, editable version, is created for this transaction,
* and set as the value of this variable.
*/
TRecord *record_;
};

8
include/storage/typed_value_store.hpp
View File

@ -10,14 +10,14 @@
* using a key of type Properties::TKey.
*
* The underlying implementation is not necessarily std::map.
*
* @tparam TKey The type of key used in this value store.
*/
template <typename TKey = uint32_t>
class TypedValueStore {
public:
using sptr = std::shared_ptr<TypedValueStore>;
/** The type of key used to get and set properties */
using TKey = uint32_t;
/**
* Returns a TypedValue (by reference) at the given key.
* If the key does not exist, the Null property is returned.
@ -65,7 +65,7 @@ public:
/**
* @return The number of Properties in this collection.
*/
size_t size();
size_t size() const;
/**

2
include/storage/vertex.hpp
View File

@ -14,5 +14,5 @@ public:
std::vector<mvcc::VersionList<Edge> *> out_;
std::vector<mvcc::VersionList<Edge> *> in_;
std::set<GraphDb::Label> labels_;
TypedValueStore properties_;
TypedValueStore<GraphDb::Property> properties_;
};

2
include/transactions/transaction.hpp
View File

@ -20,7 +20,7 @@ class Transaction : public TransactionRead
public:
Transaction(const Id &id, const Snapshot<Id> &snapshot, Engine &engine);
Transaction(const Transaction &) = delete;
Transaction(Transaction &&) = delete;
Transaction(Transaction &&) = default;
// Returns copy of transaction_read
TransactionRead transaction_read();

88
src/communication/bolt/v1/serialization/bolt_serializer.cpp
View File

@ -5,32 +5,92 @@
#include "communication/bolt/v1/transport/socket_stream.hpp"
#include "io/network/socket.hpp"
template <class Stream>
void bolt::BoltSerializer<Stream>::write(const EdgeAccessor &edge)
{
template<class Stream>
void bolt::BoltSerializer<Stream>::write(const VertexAccessor &vertex) {
// write signatures for the node struct and node data type
encoder.write_struct_header(3);
encoder.write(underlying_cast(pack::Node));
// IMPORTANT: here we write a hardcoded 0 because we don't
// use internal IDs, but need to give something to Bolt
// note that OpenCypher has no id(x) function, so the client
// should not be able to do anything with this value anyway
encoder.write_integer(0); // uID
// write the list of labels
auto labels = vertex.labels();
encoder.write_list_header(labels.size());
for (auto label : labels)
encoder.write_string(vertex.db_accessor().label_name(label));
// write the properties
const TypedValueStore &props = vertex.Properties();
encoder.write_map_header(props.size());
props.Accept([&vertex](const TypedValueStore::TKey &prop_name, const TypedValue &value) {
write(vertex.db_accessor().property_name(prop_name));
write(value);
});
}
template<class Stream>
void bolt::BoltSerializer<Stream>::write(const EdgeAccessor &edge) {
// write signatures for the edge struct and edge data type
encoder.write_struct_header(5);
encoder.write(underlying_cast(pack::Relationship));
// write the identifier for the node
encoder.write_integer(edge.id());
encoder.write_integer(edge.from().id());
encoder.write_integer(edge.to().id());
// IMPORTANT: here we write a hardcoded 0 because we don't
// use internal IDs, but need to give something to Bolt
// note that OpenCypher has no id(x) function, so the client
// should not be able to do anything with this value anyway
encoder.write_integer(0);
encoder.write_integer(0);
encoder.write_integer(0);
// write the type of the edge
encoder.write_string(edge.edge_type());
// write the property map
auto props = edge.properties();
const TypedValueStore& props = edge.Properties();
encoder.write_map_header(props.size());
props.Accept([&edge](const TypedValueStore::TKey &prop_name, const TypedValue &value) {
write(edge.db_accessor().property_name(prop_name));
write(value);
});
}
for (auto &prop : props) {
write(prop.key.family_name());
prop.accept(*this);
template<class Stream>
void bolt::BoltSerializer<Stream>::write(const TypedValue& value) {
switch (value.type_) {
case TypedValue::Type::Null:
encoder.write_null();
return;
case TypedValue::Type::Bool:
encoder.write_bool(value.Value<bool>());
return;
case TypedValue::Type::String:
encoder.write_string(value.Value<std::string>());
return;
case TypedValue::Type::Int:
encoder.write_integer(value.Value<int>());
return;
case TypedValue::Type::Float:
encoder.write_double(value.Value<float>());
return;
}
}
template class bolt::BoltSerializer<bolt::BoltEncoder<
template<class Stream>
void bolt::BoltSerializer<Stream>::write_failure(const std::map<std::string, std::string> &data) {
encoder.message_failure();
encoder.write_map_header(data.size());
for (auto const &kv : data) {
write(kv.first);
write(kv.second);
}
}
template
class bolt::BoltSerializer<bolt::BoltEncoder<
bolt::ChunkedEncoder<bolt::ChunkedBuffer<bolt::SocketStream<io::Socket>>>>>;

2
src/communication/bolt/v1/session.cpp
View File

@ -43,5 +43,5 @@ void Session::close()
bolt.close(this);
}
Db &Session::active_db() { return bolt.dbms.active(); }
GraphDb &Session::active_db() { return bolt.dbms.active(); }
}

23
src/database/graph_db_accessor.cpp
View File

@ -1,6 +1,9 @@
#include <database/creation_exception.hpp>
#include "database/graph_db_accessor.hpp"
GraphDbAccessor::GraphDbAccessor(GraphDb& db) : db_(db), transaction_(std::move(db.tx_engine.begin())) {}
VertexAccessor GraphDbAccessor::insert_vertex() {
auto vertex_vlist = new mvcc::VersionList<Vertex>();
vertex_vlist->insert(transaction_);
@ -32,8 +35,6 @@ EdgeAccessor GraphDbAccessor::insert_edge(
// TODO connect the vertices to edge
from.add_to_out(edge_vlist, pass_key);
to.add_to_in(edge_vlist, pass_key);
// from.vlist(pass_key).out_.emplace(edge_vlist);
// to.vlist(pass_key).in_.emplace(edge_vlist);
// TODO make this configurable
for (int i = 0; i < 5; ++i) {
@ -47,13 +48,25 @@ EdgeAccessor GraphDbAccessor::insert_edge(
}
GraphDb::Label GraphDbAccessor::label(const std::string& label_name) {
return db_.labels_.GetKey(label_name);
return &(*db_.labels_.access().insert(label_name).first);
}
std::string& GraphDbAccessor::label_name(const GraphDb::Label label) const {
return *label;
}
GraphDb::EdgeType GraphDbAccessor::edge_type(const std::string& edge_type_name){
return db_.edge_types_.GetKey(edge_type_name);
return &(*db_.edge_types_.access().insert(edge_type_name).first);
}
std::string& GraphDbAccessor::edge_type_name(const GraphDb::EdgeType edge_type) const {
return *edge_type;
}
GraphDb::Property GraphDbAccessor::property(const std::string& property_name) {
return db_.properties_.GetKey(property_name);
return &(*db_.properties_.access().insert(property_name).first);
}
std::string& GraphDbAccessor::property_name(const GraphDb::Property property) const {
return *property;
}

24
src/storage/edge_accessor.cpp
View File

@ -2,37 +2,35 @@
#include "storage/vertex_accessor.hpp"
void EdgeAccessor::set_edge_type(GraphDb::EdgeType edge_type) {
this->record_->edge_type_ = edge_type;
this->update()->edge_type_ = edge_type;
}
GraphDb::EdgeType EdgeAccessor::edge_type() const {
return this->record_->edge_type_;
return this->view()->edge_type_;
}
VertexAccessor EdgeAccessor::from() const {
return VertexAccessor(this->record_->from_, this->trans_);
return VertexAccessor(this->view()->from_, this->db_accessor_->transaction_);
}
VertexAccessor EdgeAccessor::to() const {
return VertexAccessor(this->record_->to_, this->trans_);
return VertexAccessor(this->view()->to_, this->db_accessor_->transaction_);
}
void EdgeAccessor::remove() {
// remove this edge's reference from the "from" vertex
auto vertex_from = from();
vertex_from.update();
std::remove(vertex_from.record_->out_.begin(),
vertex_from.record_->out_.end(),
auto vertex_from = from().update();
std::remove(vertex_from->out_.begin(),
vertex_from->out_.end(),
vlist_);
// remove this edge's reference from the "to" vertex
auto vertex_to = to();
vertex_to.update();
std::remove(vertex_to.record_->in_.begin(),
vertex_to.record_->in_.end(),
auto vertex_to = to().update();
std::remove(vertex_to->in_.begin(),
vertex_to->in_.end(),
vlist_);
// remove this record from the database via MVCC
vlist_->remove(record_, trans_);
vlist_->remove(update(), db_accessor_->transaction_);
}

2
src/storage/typed_value_store.cpp
View File

@ -43,7 +43,7 @@ void TypedValueStore::set(const TKey &key, const char *value) {
set(key, std::string(value));
}
size_t TypedValueStore::erase(const TKey &key) {
size_t TypedValueStore::erase(const TKey &key) const {
auto found = std::find_if(props_.begin(), props_.end(), [&key](std::pair<TKey, TypedValue> &kv){return kv.first == key;});
if (found != props_.end()) {
props_.erase(found);

34
src/storage/vertex_accessor.cpp
View File

@ -2,30 +2,28 @@
#include "storage/vertex_accessor.hpp"
size_t VertexAccessor::out_degree() const {
return this->record_->out_.size();
return this->view()->out_.size();
}
size_t VertexAccessor::in_degree() const {
return this->record_->in_.size();
return this->view()->in_.size();
}
bool VertexAccessor::add_label(GraphDb::Label label) {
update();
return this->record_->labels_.emplace(label).second;
return this->update()->labels_.emplace(label).second;
}
size_t VertexAccessor::remove_label(GraphDb::Label label) {
update();
return this->record_->labels_.erase(label);
return this->update()->labels_.erase(label);
}
bool VertexAccessor::has_label(GraphDb::Label label) const {
auto &label_set = this->record_->labels_;
auto &label_set = this->view()->labels_;
return label_set.find(label) != label_set.end();
}
const std::set<GraphDb::Label> &VertexAccessor::labels() const {
return this->record_->labels_;
const std::set<GraphDb::Label>& VertexAccessor::labels() const {
return this->view()->labels_;
}
bool VertexAccessor::remove() {
@ -33,7 +31,7 @@ bool VertexAccessor::remove() {
if (out_degree() > 0 || in_degree() > 0)
return false;
vlist_->remove(record_, trans_);
vlist_->remove(view(), db_accessor_->transaction_);
return true;
}
@ -41,21 +39,19 @@ void VertexAccessor::detach_remove() {
// removing edges via accessors is both safe
// and it should remove all the pointers in the relevant
// vertices (including this one)
for (auto edge_vlist : record_->out_)
EdgeAccessor(edge_vlist, trans_).remove();
for (auto edge_vlist : view()->out_)
EdgeAccessor(edge_vlist, db_accessor_->transaction_).remove();
for (auto edge_vlist : record_->in_)
EdgeAccessor(edge_vlist, trans_).remove();
for (auto edge_vlist : view()->in_)
EdgeAccessor(edge_vlist, db_accessor_->transaction_).remove();
vlist_->remove(record_, trans_);
vlist_->remove(view(), db_accessor_->transaction_);
}
void VertexAccessor::attach_in(mvcc::VersionList<Edge>* edge_vlist, PassKey<GraphDb>) {
update();
this->record_->in_.emplace_back(edge_vlist);
this->update()->in_.emplace_back(edge_vlist);
}
void VertexAccessor::attach_out(mvcc::VersionList<Edge>* edge_vlist, PassKey<GraphDb>) {
update();
this->record_->out_.emplace_back(edge_vlist);
this->update()->out_.emplace_back(edge_vlist);
}