From 6fc6a27288c6f44eefa033cefed79c61af5ce655 Mon Sep 17 00:00:00 2001 From: florijan Date: Fri, 12 Jan 2018 15:17:04 +0100 Subject: [PATCH] Refactor GraphDb Summary: GraphDb is refactored to become an API exposing different parts necessary for the database to function. These different parts can have different implementations in SingleNode or distributed Master/Server GraphDb implementations. Interally GraphDb is implemented using two class heirarchies. One contains all the members and correct wiring for each situation. The other takes care of initialization and shutdown. This architecture is practical because it can guarantee that the initialization of the object structure is complete, before initializing state. Reviewers: buda, mislav.bradac, dgleich, teon.banek Reviewed By: teon.banek Subscribers: pullbot Differential Revision: https://phabricator.memgraph.io/D1093 --- .../otto/parallel_connected_components.cpp | 34 +- src/CMakeLists.txt | 2 +- src/communication/bolt/v1/session.hpp | 11 +- .../bolt/v1/states/executing.hpp | 5 +- src/communication/messaging/distributed.hpp | 2 +- .../{graph_db_config.cpp => config.cpp} | 51 +- src/database/graph_db.cpp | 349 ++++++------ src/database/graph_db.hpp | 221 +++----- src/database/graph_db_accessor.cpp | 196 +++---- src/database/graph_db_accessor.hpp | 143 +++-- src/database/indexes/index_common.hpp | 4 +- src/database/indexes/key_index.hpp | 16 +- src/database/indexes/label_property_index.hpp | 56 +- src/database/storage.hpp | 57 ++ src/database/storage_gc.hpp | 124 +++++ .../{graph_db_datatypes.hpp => types.hpp} | 20 +- src/distributed/coordination_master.hpp | 2 +- src/distributed/serialization.hpp | 16 +- src/durability/recovery.cpp | 22 +- src/durability/recovery.hpp | 2 +- src/durability/snapshooter.cpp | 11 +- src/durability/snapshooter.hpp | 2 +- src/durability/wal.hpp | 7 +- src/memgraph_bolt.cpp | 71 +-- src/query/console.cpp | 4 +- src/query/console.hpp | 6 +- src/query/context.hpp | 5 +- src/query/frontend/ast/ast.hpp | 67 ++- .../frontend/ast/cypher_main_visitor.cpp | 45 +- .../frontend/ast/cypher_main_visitor.hpp | 8 +- .../interpret/awesome_memgraph_functions.cpp | 84 +-- .../interpret/awesome_memgraph_functions.hpp | 3 +- src/query/interpret/eval.hpp | 4 +- src/query/interpreter.cpp | 4 +- src/query/interpreter.hpp | 5 +- src/query/plan/operator.cpp | 206 ++++--- src/query/plan/operator.hpp | 219 ++++---- src/query/plan/preprocess.cpp | 6 +- src/query/plan/preprocess.hpp | 12 +- src/query/plan/rule_based_planner.hpp | 6 +- src/query/plan/vertex_count_cache.hpp | 27 +- src/storage/concurrent_id_mapper.hpp | 2 +- src/storage/concurrent_id_mapper_master.cpp | 4 +- src/storage/concurrent_id_mapper_master.hpp | 2 +- .../concurrent_id_mapper_rpc_messages.hpp | 6 +- src/storage/concurrent_id_mapper_worker.cpp | 4 +- src/storage/edge.hpp | 6 +- src/storage/edge_accessor.cpp | 11 +- src/storage/edge_accessor.hpp | 10 +- src/storage/edges.hpp | 13 +- src/storage/property_value_store.hpp | 4 +- src/storage/record_accessor.cpp | 14 +- src/storage/record_accessor.hpp | 27 +- src/storage/vertex.hpp | 4 +- src/storage/vertex_accessor.cpp | 8 +- src/storage/vertex_accessor.hpp | 30 +- src/utils/flag_validation.hpp | 1 + src/utils/random_graph_generator.hpp | 34 +- tests/benchmark/expansion.cpp | 16 +- tests/benchmark/query/planner.cpp | 35 +- tests/manual/console_test.cpp | 6 +- tests/manual/query_planner.cpp | 44 +- tests/manual/single_query.cpp | 4 +- tests/property_based/random_graph.cpp | 4 +- tests/unit/bolt_encoder.cpp | 4 +- tests/unit/bolt_session.cpp | 4 +- .../unit/concurrent_id_mapper_distributed.cpp | 5 +- .../unit/concurrent_id_mapper_single_node.cpp | 4 +- tests/unit/cypher_main_visitor.cpp | 8 +- tests/unit/database_key_index.cpp | 34 +- tests/unit/database_label_property_index.cpp | 21 +- tests/unit/database_transaction_timeout.cpp | 8 +- tests/unit/distributed_serialization.cpp | 3 +- tests/unit/durability.cpp | 107 ++-- tests/unit/graph_db.cpp | 12 +- tests/unit/graph_db_accessor.cpp | 34 +- tests/unit/graph_db_accessor_index_api.cpp | 33 +- tests/unit/interpreter.cpp | 12 +- tests/unit/network_timeouts.cpp | 4 +- tests/unit/property_value_store.cpp | 6 +- tests/unit/query_common.hpp | 58 +- tests/unit/query_cost_estimator.cpp | 31 +- tests/unit/query_expression_evaluator.cpp | 94 ++-- .../unit/query_plan_accumulate_aggregate.cpp | 49 +- tests/unit/query_plan_bag_semantics.cpp | 27 +- tests/unit/query_plan_common.hpp | 17 +- .../query_plan_create_set_remove_delete.cpp | 136 ++--- tests/unit/query_plan_edge_cases.cpp | 6 +- tests/unit/query_plan_match_filter_return.cpp | 170 +++--- tests/unit/query_planner.cpp | 276 ++++----- tests/unit/query_semantic.cpp | 524 ++++-------------- tests/unit/query_variable_start_planner.cpp | 41 +- tests/unit/record_edge_vertex_accessor.cpp | 68 +-- tests/unit/skiplist_suffix.cpp | 12 +- tests/unit/state_delta.cpp | 62 +-- tests/unit/typed_value.cpp | 4 +- tools/tests/mg_recovery_check.cpp | 8 +- 97 files changed, 2113 insertions(+), 2193 deletions(-) rename src/database/{graph_db_config.cpp => config.cpp} (51%) create mode 100644 src/database/storage.hpp create mode 100644 src/database/storage_gc.hpp rename src/database/{graph_db_datatypes.hpp => types.hpp} (82%) diff --git a/customers/otto/parallel_connected_components.cpp b/customers/otto/parallel_connected_components.cpp index c1f22de66..b01f6f1f4 100644 --- a/customers/otto/parallel_connected_components.cpp +++ b/customers/otto/parallel_connected_components.cpp @@ -25,9 +25,9 @@ DECLARE_int32(gc_cycle_sec); static const std::string kLabel{"kLabel"}; static const std::string kProperty{"kProperty"}; -void GenerateGraph(GraphDb &db) { +void GenerateGraph(database::GraphDb &db) { { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; dba.BuildIndex(dba.Label(kLabel), dba.Property(kProperty)); dba.Commit(); } @@ -56,7 +56,7 @@ void GenerateGraph(GraphDb &db) { SpinLock vertices_lock; for (int i = 0; i < FLAGS_thread_count; ++i) { threads.emplace_back([&db, &vertex_ids, &vertices, &vertices_lock, i]() { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; auto label = dba.Label(kLabel); auto property = dba.Property(kProperty); auto batch_size = FLAGS_vertex_count / FLAGS_thread_count; @@ -76,7 +76,7 @@ void GenerateGraph(GraphDb &db) { << timer.Elapsed().count() << " seconds."; } { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (int i = 0; i < FLAGS_vertex_count; ++i) vertices[i] = *dba.Transfer(vertices[i]); @@ -96,32 +96,32 @@ void GenerateGraph(GraphDb &db) { } } -auto EdgeIteration(GraphDb &db) { - GraphDbAccessor dba{db}; +auto EdgeIteration(database::GraphDb &db) { + database::GraphDbAccessor dba{db}; int64_t sum{0}; for (auto edge : dba.Edges(false)) sum += edge.from().gid() + edge.to().gid(); return sum; } -auto VertexIteration(GraphDb &db) { - GraphDbAccessor dba{db}; +auto VertexIteration(database::GraphDb &db) { + database::GraphDbAccessor dba{db}; int64_t sum{0}; for (auto v : dba.Vertices(false)) for (auto e : v.out()) sum += e.gid() + e.to().gid(); return sum; } -auto ConnectedComponentsEdges(GraphDb &db) { +auto ConnectedComponentsEdges(database::GraphDb &db) { UnionFind connectivity{FLAGS_vertex_count}; - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto edge : dba.Edges(false)) connectivity.Connect(edge.from().gid(), edge.to().gid()); return connectivity.Size(); } -auto ConnectedComponentsVertices(GraphDb &db) { +auto ConnectedComponentsVertices(database::GraphDb &db) { UnionFind connectivity{FLAGS_vertex_count}; - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto from : dba.Vertices(false)) { for (auto out_edge : from.out()) connectivity.Connect(from.gid(), out_edge.to().gid()); @@ -129,7 +129,7 @@ auto ConnectedComponentsVertices(GraphDb &db) { return connectivity.Size(); } -auto ConnectedComponentsVerticesParallel(GraphDb &db) { +auto ConnectedComponentsVerticesParallel(database::GraphDb &db) { UnionFind connectivity{FLAGS_vertex_count}; SpinLock connectivity_lock; @@ -143,7 +143,7 @@ auto ConnectedComponentsVerticesParallel(GraphDb &db) { for (int i = 0; i < FLAGS_thread_count; ++i) { threads.emplace_back( [&connectivity, &connectivity_lock, &bounds, &db, i]() { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto from : dba.Vertices(dba.Label(kLabel), dba.Property(kProperty), utils::MakeBoundInclusive(bounds[i]), @@ -159,11 +159,11 @@ auto ConnectedComponentsVerticesParallel(GraphDb &db) { return connectivity.Size(); } -auto Expansion(GraphDb &db) { +auto Expansion(database::GraphDb &db) { std::vector component_ids(FLAGS_vertex_count, -1); int next_component_id{0}; std::stack expansion_stack; - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto v : dba.Vertices(false)) { if (component_ids[v.gid()] != -1) continue; auto component_id = next_component_id++; @@ -186,7 +186,7 @@ int main(int argc, char **argv) { google::InitGoogleLogging(argv[0]); FLAGS_gc_cycle_sec = -1; - GraphDb db; + database::SingleNode db; GenerateGraph(db); auto timed_call = [&db](auto callable, const std::string &descr) { LOG(INFO) << "Running " << descr << "..."; diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 682ce1da6..aa53a46bc 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -9,10 +9,10 @@ set(memgraph_src_files communication/messaging/protocol.cpp communication/rpc/rpc.cpp data_structures/concurrent/skiplist_gc.cpp + database/config.cpp database/counters.cpp database/graph_db.cpp database/graph_db_accessor.cpp - database/graph_db_config.cpp database/state_delta.cpp distributed/coordination_master.cpp distributed/coordination_worker.cpp diff --git a/src/communication/bolt/v1/session.hpp b/src/communication/bolt/v1/session.hpp index a5268c5cb..6df216828 100644 --- a/src/communication/bolt/v1/session.hpp +++ b/src/communication/bolt/v1/session.hpp @@ -26,12 +26,7 @@ namespace communication::bolt { /** Encapsulates Dbms and Interpreter that are passed through the network server * and worker to the session. */ struct SessionData { - /** Constructs a SessionData object. - * @param args - Arguments forwarded to the GraphDb constructor. */ - template - SessionData(TArgs &&... args) : db(std::forward(args)...) {} - - GraphDb db; + database::MasterBase &db; query::Interpreter interpreter; }; @@ -202,7 +197,7 @@ class Session { // TODO: Rethink if there is a way to hide some members. At the momement all // of them are public. TSocket socket_; - GraphDb &db_; + database::MasterBase &db_; query::Interpreter &interpreter_; TimeoutSocket timeout_socket_{*this}; @@ -218,7 +213,7 @@ class Session { State state_{State::Handshake}; // GraphDbAccessor of active transaction in the session, can be null if // there is no associated transaction. - std::unique_ptr db_accessor_; + std::unique_ptr db_accessor_; // Time of the last event. std::chrono::time_point last_event_time_ = std::chrono::steady_clock::now(); diff --git a/src/communication/bolt/v1/states/executing.hpp b/src/communication/bolt/v1/states/executing.hpp index 686ccf342..306d324cc 100644 --- a/src/communication/bolt/v1/states/executing.hpp +++ b/src/communication/bolt/v1/states/executing.hpp @@ -75,13 +75,14 @@ State HandleRun(TSession &session, State state, Marker marker) { // TODO: Possible (but very unlikely) race condition, where we have alive // session during shutdown, but is_accepting_transactions_ isn't yet false. // We should probably create transactions under some locking mechanism. - if (!session.db_.is_accepting_transactions_) { + if (!session.db_.is_accepting_transactions()) { // Db is shutting down and doesn't accept new transactions so we should // close this session. return State::Close; } // Create new transaction. - session.db_accessor_ = std::make_unique(session.db_); + session.db_accessor_ = + std::make_unique(session.db_); } // If there was not explicitly started transaction before maybe we are diff --git a/src/communication/messaging/distributed.hpp b/src/communication/messaging/distributed.hpp index 49dd80b50..294d2f9a2 100644 --- a/src/communication/messaging/distributed.hpp +++ b/src/communication/messaging/distributed.hpp @@ -58,7 +58,7 @@ class System { friend class Writer; System(const std::string &address, uint16_t port); - System(const Endpoint &endpoint); + explicit System(const Endpoint &endpoint); System(const System &) = delete; System(System &&) = delete; System &operator=(const System &) = delete; diff --git a/src/database/graph_db_config.cpp b/src/database/config.cpp similarity index 51% rename from src/database/graph_db_config.cpp rename to src/database/config.cpp index 2ece07751..b41a17ceb 100644 --- a/src/database/graph_db_config.cpp +++ b/src/database/config.cpp @@ -1,19 +1,8 @@ -#include -#include #include -#include - #include "database/graph_db.hpp" #include "utils/flag_validation.hpp" -namespace fs = std::experimental::filesystem; - -// TODO review: tech docs say the default here is 'true', which it is in the -// community config. Should we set the default here to true? On some other -// points the tech docs are consistent with community config, and not with these -// defaults. - // Durability flags. DEFINE_bool(durability_enabled, false, "If durability (database persistence) should be enabled"); @@ -29,15 +18,37 @@ DEFINE_int32(snapshot_max_retained, -1, "Number of retained snapshots, -1 means without limit."); DEFINE_bool(snapshot_on_exit, false, "Snapshot on exiting the database."); -// Misc flags. -DEFINE_int32(gc_cycle_sec, 30, - "Amount of time between starts of two cleaning cycles in seconds. " - "-1 to turn off."); +// Misc flags DEFINE_int32(query_execution_time_sec, 180, "Maximum allowed query execution time. Queries exceeding this " "limit will be aborted. Value of -1 means no limit."); +DEFINE_int32(gc_cycle_sec, 30, + "Amount of time between starts of two cleaning cycles in seconds. " + "-1 to turn off."); -GraphDb::Config::Config() +// Distributed master/worker flags. +DEFINE_HIDDEN_int32(worker_id, 0, + "ID of a worker in a distributed system. Igored in " + "single-node and distributed-master."); +DEFINE_HIDDEN_string(master_host, "0.0.0.0", + "For master node indicates the host served on. For worker " + "node indicates the master location."); +DEFINE_VALIDATED_HIDDEN_int32( + master_port, 0, + "For master node the port on which to serve. For " + "worker node indicates the master's port.", + FLAG_IN_RANGE(0, std::numeric_limits::max())); +DEFINE_HIDDEN_string(worker_host, "0.0.0.0", + "For worker node indicates the host served on. For master " + "node this flag is not used."); +DEFINE_VALIDATED_HIDDEN_int32( + worker_port, 0, + "For master node it's unused. For worker node " + "indicates the port on which to serve. If zero (default value), a port is " + "chosen at random. Sent to the master when registring worker node.", + FLAG_IN_RANGE(0, std::numeric_limits::max())); + +database::Config::Config() // Durability flags. : durability_enabled{FLAGS_durability_enabled}, durability_directory{FLAGS_durability_directory}, @@ -47,4 +58,10 @@ GraphDb::Config::Config() snapshot_on_exit{FLAGS_snapshot_on_exit}, // Misc flags. gc_cycle_sec{FLAGS_gc_cycle_sec}, - query_execution_time_sec{FLAGS_query_execution_time_sec} {} + query_execution_time_sec{FLAGS_query_execution_time_sec}, + // Distributed flags. + worker_id{FLAGS_worker_id}, + master_endpoint{FLAGS_master_host, + static_cast(FLAGS_master_port)}, + worker_endpoint{FLAGS_worker_host, + static_cast(FLAGS_worker_port)} {} diff --git a/src/database/graph_db.cpp b/src/database/graph_db.cpp index 631ef10d0..05d457c2d 100644 --- a/src/database/graph_db.cpp +++ b/src/database/graph_db.cpp @@ -1,95 +1,191 @@ -#include -#include - -#include - #include "database/graph_db.hpp" -#include "database/graph_db_accessor.hpp" + +#include "communication/messaging/distributed.hpp" +#include "distributed/coordination_master.hpp" +#include "distributed/coordination_worker.hpp" #include "durability/paths.hpp" #include "durability/recovery.hpp" #include "durability/snapshooter.hpp" #include "storage/concurrent_id_mapper_master.hpp" +#include "storage/concurrent_id_mapper_single_node.hpp" #include "storage/concurrent_id_mapper_worker.hpp" #include "transactions/engine_master.hpp" #include "transactions/engine_single_node.hpp" #include "transactions/engine_worker.hpp" -#include "utils/timer.hpp" +#include "utils/flag_validation.hpp" -namespace fs = std::experimental::filesystem; +namespace database { +namespace impl { -#define INIT_MAPPERS(type, ...) \ - labels_ = std::make_unique>(__VA_ARGS__); \ - edge_types_ = std::make_unique>(__VA_ARGS__); \ - properties_ = std::make_unique>(__VA_ARGS__); +class Base { + public: + explicit Base(const Config &config) : config_(config) {} + virtual ~Base() {} -GraphDb::GraphDb(Config config) : GraphDb(config, 0) { - tx_engine_ = std::make_unique(&wal_); - counters_ = std::make_unique(); - INIT_MAPPERS(storage::SingleNodeConcurrentIdMapper); - Start(); -} + const Config config_; -GraphDb::GraphDb(communication::messaging::System &system, - distributed::MasterCoordination &master, Config config) - : GraphDb(config, 0) { - tx_engine_ = std::make_unique(system, &wal_); - auto counters = std::make_unique(system); - counters_ = std::move(counters); - INIT_MAPPERS(storage::MasterConcurrentIdMapper, system); - get_endpoint_ = [&master](int worker_id) { - return master.GetEndpoint(worker_id); - }; - Start(); -} + virtual Storage &storage() = 0; + virtual StorageGc &storage_gc() = 0; + virtual durability::WriteAheadLog &wal() = 0; + virtual tx::Engine &tx_engine() = 0; + virtual storage::ConcurrentIdMapper