From 8cd9f696cf33d525c2f36a1a72893a36d272df17 Mon Sep 17 00:00:00 2001 From: antonio2368 Date: Mon, 14 Jun 2021 15:47:57 +0200 Subject: [PATCH] Multiple isolation levels (#171) This PR introduces READ COMMITTED and READ UNCOMMITTED isolation levels. The isolation level can be set with a config or with a query for different scopes. --- config/flags.yaml | 4 + src/auth/models.cpp | 2 + src/auth/models.hpp | 12 +- src/glue/auth.cpp | 2 + src/memgraph.cpp | 175 +++++++++---- src/query/exceptions.hpp | 6 + src/query/frontend/ast/ast.lcp | 28 +- src/query/frontend/ast/ast_visitor.hpp | 3 +- .../frontend/ast/cypher_main_visitor.cpp | 28 ++ .../frontend/ast/cypher_main_visitor.hpp | 5 + .../opencypher/grammar/MemgraphCypher.g4 | 19 ++ .../opencypher/grammar/MemgraphCypherLexer.g4 | 11 + .../frontend/semantic/required_privileges.cpp | 2 + .../frontend/stripped_lexer_constants.hpp | 24 +- src/query/interpreter.cpp | 71 +++++- src/query/interpreter.hpp | 8 + src/storage/v2/config.hpp | 6 + src/storage/v2/isolation_level.hpp | 9 + src/storage/v2/mvcc.hpp | 14 +- src/storage/v2/storage.cpp | 16 +- src/storage/v2/storage.hpp | 12 +- src/storage/v2/transaction.hpp | 13 +- tests/benchmark/expansion.cpp | 1 + tests/e2e/CMakeLists.txt | 1 + tests/e2e/isolation_levels/CMakeLists.txt | 2 + .../e2e/isolation_levels/isolation_levels.cpp | 241 ++++++++++++++++++ tests/e2e/isolation_levels/workloads.yaml | 14 + tests/manual/single_query.cpp | 1 + tests/unit/CMakeLists.txt | 3 + tests/unit/cypher_main_visitor.cpp | 65 ++++- tests/unit/interpreter.cpp | 1 + tests/unit/query_required_privileges.cpp | 5 + tests/unit/storage_v2_isolation_level.cpp | 98 +++++++ tests/unit/storage_v2_wal_file.cpp | 4 +- 34 files changed, 810 insertions(+), 96 deletions(-) create mode 100644 src/storage/v2/isolation_level.hpp create mode 100644 tests/e2e/isolation_levels/CMakeLists.txt create mode 100644 tests/e2e/isolation_levels/isolation_levels.cpp create mode 100644 tests/e2e/isolation_levels/workloads.yaml create mode 100644 tests/unit/storage_v2_isolation_level.cpp diff --git a/config/flags.yaml b/config/flags.yaml index c610591d1..93f9c73ed 100644 --- a/config/flags.yaml +++ b/config/flags.yaml @@ -87,6 +87,10 @@ modifications: value: "0" override: true + - name: "isolation_level" + value: "SNAPSHOT_ISOLATION" + override: true + undocumented: - "flag_file" - "also_log_to_stderr" diff --git a/src/auth/models.cpp b/src/auth/models.cpp index be5f28b3e..cc34ca410 100644 --- a/src/auth/models.cpp +++ b/src/auth/models.cpp @@ -49,6 +49,8 @@ std::string PermissionToString(Permission permission) { return "FREE_MEMORY"; case Permission::TRIGGER: return "TRIGGER"; + case Permission::CONFIG: + return "CONFIG"; case Permission::AUTH: return "AUTH"; } diff --git a/src/auth/models.hpp b/src/auth/models.hpp index 9e1b54977..56cf897d2 100644 --- a/src/auth/models.hpp +++ b/src/auth/models.hpp @@ -26,16 +26,18 @@ enum class Permission : uint64_t { READ_FILE = 1U << 12U, FREE_MEMORY = 1U << 13U, TRIGGER = 1U << 14U, + CONFIG = 1U << 15U, AUTH = 1U << 16U }; // clang-format on // Constant list of all available permissions. -const std::vector kPermissionsAll = { - Permission::MATCH, Permission::CREATE, Permission::MERGE, Permission::DELETE, - Permission::SET, Permission::REMOVE, Permission::INDEX, Permission::STATS, - Permission::CONSTRAINT, Permission::DUMP, Permission::AUTH, Permission::REPLICATION, - Permission::LOCK_PATH, Permission::READ_FILE, Permission::FREE_MEMORY, Permission::TRIGGER}; +const std::vector kPermissionsAll = {Permission::MATCH, Permission::CREATE, Permission::MERGE, + Permission::DELETE, Permission::SET, Permission::REMOVE, + Permission::INDEX, Permission::STATS, Permission::CONSTRAINT, + Permission::DUMP, Permission::AUTH, Permission::REPLICATION, + Permission::LOCK_PATH, Permission::READ_FILE, Permission::FREE_MEMORY, + Permission::TRIGGER, Permission::CONFIG}; // Function that converts a permission to its string representation. std::string PermissionToString(Permission permission); diff --git a/src/glue/auth.cpp b/src/glue/auth.cpp index a54894b80..fe55c6dde 100644 --- a/src/glue/auth.cpp +++ b/src/glue/auth.cpp @@ -34,6 +34,8 @@ auth::Permission PrivilegeToPermission(query::AuthQuery::Privilege privilege) { return auth::Permission::FREE_MEMORY; case query::AuthQuery::Privilege::TRIGGER: return auth::Permission::TRIGGER; + case query::AuthQuery::Privilege::CONFIG: + return auth::Permission::CONFIG; case query::AuthQuery::Privilege::AUTH: return auth::Permission::AUTH; } diff --git a/src/memgraph.cpp b/src/memgraph.cpp index e99ae8f43..d128f1a2e 100644 --- a/src/memgraph.cpp +++ b/src/memgraph.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include @@ -28,6 +29,7 @@ #include "query/procedure/module.hpp" #include "query/procedure/py_module.hpp" #include "requests/requests.hpp" +#include "storage/v2/isolation_level.hpp" #include "storage/v2/storage.hpp" #include "storage/v2/view.hpp" #include "telemetry/telemetry.hpp" @@ -67,6 +69,42 @@ #include "glue/auth.hpp" #endif +namespace { +std::string GetAllowedEnumValuesString(const auto &mappings) { + std::vector allowed_values; + allowed_values.reserve(mappings.size()); + std::transform(mappings.begin(), mappings.end(), std::back_inserter(allowed_values), + [](const auto &mapping) { return std::string(mapping.first); }); + return utils::Join(allowed_values, ", "); +} + +enum class ValidationError : uint8_t { EmptyValue, InvalidValue }; + +utils::BasicResult IsValidEnumValueString(const auto &value, const auto &mappings) { + if (value.empty()) { + return ValidationError::EmptyValue; + } + + if (std::find_if(mappings.begin(), mappings.end(), [&](const auto &mapping) { return mapping.first == value; }) == + mappings.cend()) { + return ValidationError::InvalidValue; + } + + return {}; +} + +template +std::optional StringToEnum(const auto &value, const auto &mappings) { + const auto mapping_iter = + std::find_if(mappings.begin(), mappings.end(), [&](const auto &mapping) { return mapping.first == value; }); + if (mapping_iter == mappings.cend()) { + return std::nullopt; + } + + return mapping_iter->second; +} +} // namespace + // Bolt server flags. DEFINE_string(bolt_address, "0.0.0.0", "IP address on which the Bolt server should listen."); DEFINE_VALIDATED_int32(bolt_port, 7687, "Port on which the Bolt server should listen.", @@ -140,6 +178,72 @@ DEFINE_uint64(query_execution_timeout_sec, 180, "Maximum allowed query execution time. Queries exceeding this " "limit will be aborted. Value of 0 means no limit."); +// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables) +DEFINE_uint64( + memory_limit, 0, + "Total memory limit in MiB. Set to 0 to use the default values which are 100\% of the phyisical memory if the swap " + "is enabled and 90\% of the physical memory otherwise."); + +namespace { +using namespace std::literals; +constexpr std::array isolation_level_mappings{ + std::pair{"SNAPSHOT_ISOLATION"sv, storage::IsolationLevel::SNAPSHOT_ISOLATION}, + std::pair{"READ_COMMITTED"sv, storage::IsolationLevel::READ_COMMITTED}, + std::pair{"READ_UNCOMMITTED"sv, storage::IsolationLevel::READ_UNCOMMITTED}}; + +const std::string isolation_level_help_string = + fmt::format("Default isolation level used for the transactions. Allowed values: {}", + GetAllowedEnumValuesString(isolation_level_mappings)); +} // namespace + +// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables) +DEFINE_VALIDATED_string(isolation_level, "SNAPSHOT_ISOLATION", isolation_level_help_string.c_str(), { + if (const auto result = IsValidEnumValueString(value, isolation_level_mappings); result.HasError()) { + const auto error = result.GetError(); + switch (error) { + case ValidationError::EmptyValue: { + std::cout << "Isolation level cannot be empty." << std::endl; + break; + } + case ValidationError::InvalidValue: { + std::cout << "Invalid value for isolation level. Allowed values: " + << GetAllowedEnumValuesString(isolation_level_mappings) << std::endl; + break; + } + } + return false; + } + + return true; +}); + +namespace { +storage::IsolationLevel ParseIsolationLevel() { + const auto isolation_level = StringToEnum(FLAGS_isolation_level, isolation_level_mappings); + MG_ASSERT(isolation_level, "Invalid isolation level"); + return *isolation_level; +} + +int64_t GetMemoryLimit() { + if (FLAGS_memory_limit == 0) { + auto maybe_total_memory = utils::sysinfo::TotalMemory(); + MG_ASSERT(maybe_total_memory, "Failed to fetch the total physical memory"); + const auto maybe_swap_memory = utils::sysinfo::SwapTotalMemory(); + MG_ASSERT(maybe_swap_memory, "Failed to fetch the total swap memory"); + + if (*maybe_swap_memory == 0) { + // take only 90% of the total memory + *maybe_total_memory *= 9; + *maybe_total_memory /= 10; + } + return *maybe_total_memory * 1024; + } + + // We parse the memory as MiB every time + return FLAGS_memory_limit * 1024 * 1024; +} +} // namespace + namespace { std::vector query_modules_directories; } // namespace @@ -168,37 +272,30 @@ DEFINE_VALIDATED_string(query_modules_directory, "", DEFINE_bool(also_log_to_stderr, false, "Log messages go to stderr in addition to logfiles"); DEFINE_string(log_file, "", "Path to where the log should be stored."); -DEFINE_uint64( - memory_limit, 0, - "Total memory limit in MiB. Set to 0 to use the default values which are 100\% of the phyisical memory if the swap " - "is enabled and 90\% of the physical memory otherwise."); namespace { constexpr std::array log_level_mappings{ - std::pair{"TRACE", spdlog::level::trace}, std::pair{"DEBUG", spdlog::level::debug}, - std::pair{"INFO", spdlog::level::info}, std::pair{"WARNING", spdlog::level::warn}, - std::pair{"ERROR", spdlog::level::err}, std::pair{"CRITICAL", spdlog::level::critical}}; - -std::string GetAllowedLogLevelsString() { - std::vector allowed_log_levels; - allowed_log_levels.reserve(log_level_mappings.size()); - std::transform(log_level_mappings.cbegin(), log_level_mappings.cend(), std::back_inserter(allowed_log_levels), - [](const auto &mapping) { return mapping.first; }); - return utils::Join(allowed_log_levels, ", "); -} + std::pair{"TRACE"sv, spdlog::level::trace}, std::pair{"DEBUG"sv, spdlog::level::debug}, + std::pair{"INFO"sv, spdlog::level::info}, std::pair{"WARNING"sv, spdlog::level::warn}, + std::pair{"ERROR"sv, spdlog::level::err}, std::pair{"CRITICAL"sv, spdlog::level::critical}}; const std::string log_level_help_string = - fmt::format("Minimum log level. Allowed values: {}", GetAllowedLogLevelsString()); + fmt::format("Minimum log level. Allowed values: {}", GetAllowedEnumValuesString(log_level_mappings)); } // namespace DEFINE_VALIDATED_string(log_level, "WARNING", log_level_help_string.c_str(), { - if (value.empty()) { - std::cout << "Log level cannot be empty." << std::endl; - return false; - } - - if (std::find_if(log_level_mappings.cbegin(), log_level_mappings.cend(), - [&](const auto &mapping) { return mapping.first == value; }) == log_level_mappings.cend()) { - std::cout << "Invalid value for log level. Allowed values: " << GetAllowedLogLevelsString() << std::endl; + if (const auto result = IsValidEnumValueString(value, log_level_mappings); result.HasError()) { + const auto error = result.GetError(); + switch (error) { + case ValidationError::EmptyValue: { + std::cout << "Log level cannot be empty." << std::endl; + break; + } + case ValidationError::InvalidValue: { + std::cout << "Invalid value for log level. Allowed values: " << GetAllowedEnumValuesString(log_level_mappings) + << std::endl; + break; + } + } return false; } @@ -207,11 +304,9 @@ DEFINE_VALIDATED_string(log_level, "WARNING", log_level_help_string.c_str(), { namespace { void ParseLogLevel() { - const auto mapping_iter = std::find_if(log_level_mappings.cbegin(), log_level_mappings.cend(), - [](const auto &mapping) { return mapping.first == FLAGS_log_level; }); - MG_ASSERT(mapping_iter != log_level_mappings.cend(), "Invalid log level"); - - spdlog::set_level(mapping_iter->second); + const auto log_level = StringToEnum(FLAGS_log_level, log_level_mappings); + MG_ASSERT(log_level, "Invalid log level"); + spdlog::set_level(*log_level); } // 5 weeks * 7 days @@ -241,25 +336,6 @@ void ConfigureLogging() { spdlog::flush_on(spdlog::level::trace); ParseLogLevel(); } - -int64_t GetMemoryLimit() { - if (FLAGS_memory_limit == 0) { - auto maybe_total_memory = utils::sysinfo::TotalMemory(); - MG_ASSERT(maybe_total_memory, "Failed to fetch the total physical memory"); - const auto maybe_swap_memory = utils::sysinfo::SwapTotalMemory(); - MG_ASSERT(maybe_swap_memory, "Failed to fetch the total swap memory"); - - if (*maybe_swap_memory == 0) { - // take only 90% of the total memory - *maybe_total_memory *= 9; - *maybe_total_memory /= 10; - } - return *maybe_total_memory * 1024; - } - - // We parse the memory as MiB every time - return FLAGS_memory_limit * 1024 * 1024; -} } // namespace /// Encapsulates Dbms and Interpreter that are passed through the network server @@ -962,7 +1038,8 @@ int main(int argc, char **argv) { .snapshot_retention_count = FLAGS_storage_snapshot_retention_count, .wal_file_size_kibibytes = FLAGS_storage_wal_file_size_kib, .wal_file_flush_every_n_tx = FLAGS_storage_wal_file_flush_every_n_tx, - .snapshot_on_exit = FLAGS_storage_snapshot_on_exit}}; + .snapshot_on_exit = FLAGS_storage_snapshot_on_exit}, + .transaction = {.isolation_level = ParseIsolationLevel()}}; if (FLAGS_storage_snapshot_interval_sec == 0) { if (FLAGS_storage_wal_enabled) { LOG_FATAL( diff --git a/src/query/exceptions.hpp b/src/query/exceptions.hpp index b8ddd7480..6c757e91b 100644 --- a/src/query/exceptions.hpp +++ b/src/query/exceptions.hpp @@ -175,4 +175,10 @@ class TriggerModificationInMulticommandTxException : public QueryException { TriggerModificationInMulticommandTxException() : QueryException("Trigger queries not allowed in multicommand transactions.") {} }; + +class IsolationLevelModificationInMulticommandTxException : public QueryException { + public: + IsolationLevelModificationInMulticommandTxException() + : QueryException("Isolation level cannot be modified in multicommand transactions.") {} +}; } // namespace query diff --git a/src/query/frontend/ast/ast.lcp b/src/query/frontend/ast/ast.lcp index c6e46f984..81b6038e5 100644 --- a/src/query/frontend/ast/ast.lcp +++ b/src/query/frontend/ast/ast.lcp @@ -2193,7 +2193,7 @@ cpp<# (:serialize)) (lcp:define-enum privilege (create delete match merge set remove index stats auth constraint - dump replication lock_path read_file free_memory trigger) + dump replication lock_path read_file free_memory trigger config) (:serialize)) #>cpp AuthQuery() = default; @@ -2232,7 +2232,8 @@ const std::vector kPrivilegesAll = { AuthQuery::Privilege::REPLICATION, AuthQuery::Privilege::READ_FILE, AuthQuery::Privilege::LOCK_PATH, - AuthQuery::Privilege::FREE_MEMORY, AuthQuery::Privilege::TRIGGER}; + AuthQuery::Privilege::FREE_MEMORY, AuthQuery::Privilege::TRIGGER, + AuthQuery::Privilege::CONFIG}; cpp<# (lcp:define-class info-query (query) @@ -2432,4 +2433,27 @@ cpp<# (:serialize (:slk)) (:clone)) +(lcp:define-class isolation-level-query (query) + ((isolation_level "IsolationLevel" :scope :public) + (isolation_level_scope "IsolationLevelScope" :scope :public)) + + (:public + (lcp:define-enum isolation-level + (snapshot-isolation read-committed read-uncommitted) + (:serialize)) + (lcp:define-enum isolation-level-scope + (next session global) + (:serialize)) + #>cpp + IsolationLevelQuery() = default; + + DEFVISITABLE(QueryVisitor); + cpp<#) + (:private + #>cpp + friend class AstStorage; + cpp<#) + (:serialize (:slk)) + (:clone)) + (lcp:pop-namespace) ;; namespace query diff --git a/src/query/frontend/ast/ast_visitor.hpp b/src/query/frontend/ast/ast_visitor.hpp index 4523fd093..9d55f6b12 100644 --- a/src/query/frontend/ast/ast_visitor.hpp +++ b/src/query/frontend/ast/ast_visitor.hpp @@ -77,6 +77,7 @@ class LockPathQuery; class LoadCsv; class FreeMemoryQuery; class TriggerQuery; +class IsolationLevelQuery; using TreeCompositeVisitor = ::utils::CompositeVisitor< SingleQuery, CypherUnion, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator, AdditionOperator, @@ -110,6 +111,6 @@ class ExpressionVisitor template class QueryVisitor : public ::utils::Visitor {}; + FreeMemoryQuery, TriggerQuery, IsolationLevelQuery> {}; } // namespace query diff --git a/src/query/frontend/ast/cypher_main_visitor.cpp b/src/query/frontend/ast/cypher_main_visitor.cpp index 51cb93305..aaab87a4f 100644 --- a/src/query/frontend/ast/cypher_main_visitor.cpp +++ b/src/query/frontend/ast/cypher_main_visitor.cpp @@ -412,6 +412,33 @@ antlrcpp::Any CypherMainVisitor::visitShowTriggers(MemgraphCypher::ShowTriggersC return trigger_query; } +antlrcpp::Any CypherMainVisitor::visitIsolationLevelQuery(MemgraphCypher::IsolationLevelQueryContext *ctx) { + auto *isolation_level_query = storage_->Create(); + + isolation_level_query->isolation_level_scope_ = [scope = ctx->isolationLevelScope()]() { + if (scope->GLOBAL()) { + return IsolationLevelQuery::IsolationLevelScope::GLOBAL; + } + if (scope->SESSION()) { + return IsolationLevelQuery::IsolationLevelScope::SESSION; + } + return IsolationLevelQuery::IsolationLevelScope::NEXT; + }(); + + isolation_level_query->isolation_level_ = [level = ctx->isolationLevel()]() { + if (level->SNAPSHOT()) { + return IsolationLevelQuery::IsolationLevel::SNAPSHOT_ISOLATION; + } + if (level->COMMITTED()) { + return IsolationLevelQuery::IsolationLevel::READ_COMMITTED; + } + return IsolationLevelQuery::IsolationLevel::READ_UNCOMMITTED; + }(); + + query_ = isolation_level_query; + return isolation_level_query; +} + antlrcpp::Any CypherMainVisitor::visitCypherUnion(MemgraphCypher::CypherUnionContext *ctx) { bool distinct = !ctx->ALL(); auto *cypher_union = storage_->Create(distinct); @@ -844,6 +871,7 @@ antlrcpp::Any CypherMainVisitor::visitPrivilege(MemgraphCypher::PrivilegeContext if (ctx->READ_FILE()) return AuthQuery::Privilege::READ_FILE; if (ctx->FREE_MEMORY()) return AuthQuery::Privilege::FREE_MEMORY; if (ctx->TRIGGER()) return AuthQuery::Privilege::TRIGGER; + if (ctx->CONFIG()) return AuthQuery::Privilege::CONFIG; LOG_FATAL("Should not get here - unknown privilege!"); } diff --git a/src/query/frontend/ast/cypher_main_visitor.hpp b/src/query/frontend/ast/cypher_main_visitor.hpp index ca97ee02a..799f74986 100644 --- a/src/query/frontend/ast/cypher_main_visitor.hpp +++ b/src/query/frontend/ast/cypher_main_visitor.hpp @@ -238,6 +238,11 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor { */ antlrcpp::Any visitShowTriggers(MemgraphCypher::ShowTriggersContext *ctx) override; + /** + * @return IsolationLevelQuery* + */ + antlrcpp::Any visitIsolationLevelQuery(MemgraphCypher::IsolationLevelQueryContext *ctx) override; + /** * @return CypherUnion* */ diff --git a/src/query/frontend/opencypher/grammar/MemgraphCypher.g4 b/src/query/frontend/opencypher/grammar/MemgraphCypher.g4 index a3c578631..e90157503 100644 --- a/src/query/frontend/opencypher/grammar/MemgraphCypher.g4 +++ b/src/query/frontend/opencypher/grammar/MemgraphCypher.g4 @@ -14,8 +14,10 @@ memgraphCypherKeyword : cypherKeyword | BAD | BEFORE | CLEAR + | CONFIG | CSV | COMMIT + | COMMITTED | DATA | DELIMITER | DATABASE @@ -26,17 +28,22 @@ memgraphCypherKeyword : cypherKeyword | FOR | FREE | FROM + | GLOBAL | GRANT | HEADER | IDENTIFIED + | ISOLATION + | LEVEL | LOAD | LOCK | MAIN | MODE + | NEXT | NO | PASSWORD | PORT | PRIVILEGES + | READ | REGISTER | REPLICA | REPLICAS @@ -45,12 +52,16 @@ memgraphCypherKeyword : cypherKeyword | ROLE | ROLES | QUOTE + | SESSION + | SNAPSHOT | STATS | SYNC + | TRANSACTION | TRIGGER | TRIGGERS | TIMEOUT | TO + | UNCOMMITTED | UNLOCK | UPDATE | USER @@ -74,6 +85,7 @@ query : cypherQuery | lockPathQuery | freeMemoryQuery | triggerQuery + | isolationLevelQuery ; authQuery : createRole @@ -175,6 +187,7 @@ privilege : CREATE | READ_FILE | FREE_MEMORY | TRIGGER + | CONFIG ; privilegeList : privilege ( ',' privilege )* ; @@ -222,3 +235,9 @@ createTrigger : CREATE TRIGGER triggerName ( ON ( emptyVertex | emptyEdge ) ? ( dropTrigger : DROP TRIGGER triggerName ; showTriggers : SHOW TRIGGERS ; + +isolationLevel : SNAPSHOT ISOLATION | READ COMMITTED | READ UNCOMMITTED ; + +isolationLevelScope : GLOBAL | SESSION | NEXT ; + +isolationLevelQuery : SET isolationLevelScope TRANSACTION ISOLATION LEVEL isolationLevel ; diff --git a/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4 b/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4 index 5fdea9f31..37d8cafc1 100644 --- a/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4 +++ b/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4 @@ -20,6 +20,8 @@ BAD : B A D ; BEFORE : B E F O R E ; CLEAR : C L E A R ; COMMIT : C O M M I T ; +COMMITTED : C O M M I T T E D ; +CONFIG : C O N F I G ; CSV : C S V ; DATA : D A T A ; DELIMITER : D E L I M I T E R ; @@ -33,20 +35,25 @@ FOR : F O R ; FREE : F R E E ; FREE_MEMORY : F R E E UNDERSCORE M E M O R Y ; FROM : F R O M ; +GLOBAL : G L O B A L ; GRANT : G R A N T ; GRANTS : G R A N T S ; HEADER : H E A D E R ; IDENTIFIED : I D E N T I F I E D ; IGNORE : I G N O R E ; +ISOLATION : I S O L A T I O N ; +LEVEL : L E V E L ; LOAD : L O A D ; LOCK : L O C K ; LOCK_PATH : L O C K UNDERSCORE P A T H ; MAIN : M A I N ; MODE : M O D E ; +NEXT : N E X T ; NO : N O ; PASSWORD : P A S S W O R D ; PORT : P O R T ; PRIVILEGES : P R I V I L E G E S ; +READ : R E A D ; READ_FILE : R E A D UNDERSCORE F I L E ; REGISTER : R E G I S T E R ; REPLICA : R E P L I C A ; @@ -56,12 +63,16 @@ REVOKE : R E V O K E ; ROLE : R O L E ; ROLES : R O L E S ; QUOTE : Q U O T E ; +SESSION : S E S S I O N ; +SNAPSHOT : S N A P S H O T ; STATS : S T A T S ; SYNC : S Y N C ; TIMEOUT : T I M E O U T ; TO : T O ; +TRANSACTION : T R A N S A C T I O N ; TRIGGER : T R I G G E R ; TRIGGERS : T R I G G E R S ; +UNCOMMITTED : U N C O M M I T T E D ; UNLOCK : U N L O C K ; UPDATE : U P D A T E ; USER : U S E R ; diff --git a/src/query/frontend/semantic/required_privileges.cpp b/src/query/frontend/semantic/required_privileges.cpp index 0f16bff56..49e78c4bf 100644 --- a/src/query/frontend/semantic/required_privileges.cpp +++ b/src/query/frontend/semantic/required_privileges.cpp @@ -57,6 +57,8 @@ class PrivilegeExtractor : public QueryVisitor, public HierarchicalTreeVis void Visit(ReplicationQuery &replication_query) override { AddPrivilege(AuthQuery::Privilege::REPLICATION); } + void Visit(IsolationLevelQuery &isolation_level_query) override { AddPrivilege(AuthQuery::Privilege::CONFIG); } + bool PreVisit(Create & /*unused*/) override { AddPrivilege(AuthQuery::Privilege::CREATE); return false; diff --git a/src/query/frontend/stripped_lexer_constants.hpp b/src/query/frontend/stripped_lexer_constants.hpp index be388d708..5a03c9cf9 100644 --- a/src/query/frontend/stripped_lexer_constants.hpp +++ b/src/query/frontend/stripped_lexer_constants.hpp @@ -79,17 +79,19 @@ class Trie { const int kBitsetSize = 65536; const trie::Trie kKeywords = { - "union", "all", "optional", "match", "unwind", "as", "merge", "on", - "create", "set", "detach", "delete", "remove", "with", "distinct", "return", - "order", "by", "skip", "limit", "ascending", "asc", "descending", "desc", - "where", "or", "xor", "and", "not", "in", "starts", "ends", - "contains", "is", "null", "case", "when", "then", "else", "end", - "count", "filter", "extract", "any", "none", "single", "true", "false", - "reduce", "coalesce", "user", "password", "alter", "drop", "show", "stats", - "unique", "explain", "profile", "storage", "index", "info", "exists", "assert", - "constraint", "node", "key", "dump", "database", "call", "yield", "memory", - "mb", "kb", "unlimited", "free", "procedure", "query", "free_memory", "read_file", - "lock_path", "after", "before", "execute", "transaction", "trigger", "triggers", "update"}; + "union", "all", "optional", "match", "unwind", "as", "merge", "on", + "create", "set", "detach", "delete", "remove", "with", "distinct", "return", + "order", "by", "skip", "limit", "ascending", "asc", "descending", "desc", + "where", "or", "xor", "and", "not", "in", "starts", "ends", + "contains", "is", "null", "case", "when", "then", "else", "end", + "count", "filter", "extract", "any", "none", "single", "true", "false", + "reduce", "coalesce", "user", "password", "alter", "drop", "show", "stats", + "unique", "explain", "profile", "storage", "index", "info", "exists", "assert", + "constraint", "node", "key", "dump", "database", "call", "yield", "memory", + "mb", "kb", "unlimited", "free", "procedure", "query", "free_memory", "read_file", + "lock_path", "after", "before", "execute", "transaction", "trigger", "triggers", "update", + "comitted", "uncomitted", "global", "isolation", "level", "next", "read", "session", + "snapshot", "transaction"}; // Unicode codepoints that are allowed at the start of the unescaped name. const std::bitset kUnescapedNameAllowedStarts( diff --git a/src/query/interpreter.cpp b/src/query/interpreter.cpp index 39432f716..69bc4e3d6 100644 --- a/src/query/interpreter.cpp +++ b/src/query/interpreter.cpp @@ -622,7 +622,8 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper) in_explicit_transaction_ = true; expect_rollback_ = false; - db_accessor_ = std::make_unique(interpreter_context_->db->Access()); + db_accessor_ = + std::make_unique(interpreter_context_->db->Access(GetIsolationLevelOverride())); execution_db_accessor_.emplace(db_accessor_.get()); if (interpreter_context_->trigger_store->HasTriggers()) { @@ -1162,6 +1163,50 @@ PreparedQuery PrepareTriggerQuery(ParsedQuery parsed_query, const bool in_explic // NOLINTNEXTLINE(clang-analyzer-cplusplus.NewDeleteLeaks) } +constexpr auto ToStorageIsolationLevel(const IsolationLevelQuery::IsolationLevel isolation_level) noexcept { + switch (isolation_level) { + case IsolationLevelQuery::IsolationLevel::SNAPSHOT_ISOLATION: + return storage::IsolationLevel::SNAPSHOT_ISOLATION; + case IsolationLevelQuery::IsolationLevel::READ_COMMITTED: + return storage::IsolationLevel::READ_COMMITTED; + case IsolationLevelQuery::IsolationLevel::READ_UNCOMMITTED: + return storage::IsolationLevel::READ_UNCOMMITTED; + } +} + +PreparedQuery PrepareIsolationLevelQuery(ParsedQuery parsed_query, const bool in_explicit_transaction, + InterpreterContext *interpreter_context, Interpreter *interpreter) { + if (in_explicit_transaction) { + throw IsolationLevelModificationInMulticommandTxException(); + } + + auto *isolation_level_query = utils::Downcast(parsed_query.query); + MG_ASSERT(isolation_level_query); + + const auto isolation_level = ToStorageIsolationLevel(isolation_level_query->isolation_level_); + + auto callback = [isolation_level_query, isolation_level, interpreter_context, + interpreter]() -> std::function { + switch (isolation_level_query->isolation_level_scope_) { + case IsolationLevelQuery::IsolationLevelScope::GLOBAL: + return [interpreter_context, isolation_level] { interpreter_context->db->SetIsolationLevel(isolation_level); }; + case IsolationLevelQuery::IsolationLevelScope::SESSION: + return [interpreter, isolation_level] { interpreter->SetSessionIsolationLevel(isolation_level); }; + case IsolationLevelQuery::IsolationLevelScope::NEXT: + return [interpreter, isolation_level] { interpreter->SetNextTransactionIsolationLevel(isolation_level); }; + } + }(); + + return PreparedQuery{ + {}, + std::move(parsed_query.required_privileges), + [callback = std::move(callback)](AnyStream *stream, std::optional n) -> std::optional { + callback(); + return QueryHandlerResult::COMMIT; + }, + RWType::NONE}; +} + PreparedQuery PrepareInfoQuery(ParsedQuery parsed_query, bool in_explicit_transaction, std::map *summary, InterpreterContext *interpreter_context, storage::Storage *db, utils::MemoryResource *execution_memory) { @@ -1452,7 +1497,8 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string, (utils::Downcast(parsed_query.query) || utils::Downcast(parsed_query.query) || utils::Downcast(parsed_query.query) || utils::Downcast(parsed_query.query) || utils::Downcast(parsed_query.query))) { - db_accessor_ = std::make_unique(interpreter_context_->db->Access()); + db_accessor_ = + std::make_unique(interpreter_context_->db->Access(GetIsolationLevelOverride())); execution_db_accessor_.emplace(db_accessor_.get()); if (utils::Downcast(parsed_query.query) && interpreter_context_->trigger_store->HasTriggers()) { @@ -1503,6 +1549,9 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string, } else if (utils::Downcast(parsed_query.query)) { prepared_query = PrepareTriggerQuery(std::move(parsed_query), in_explicit_transaction_, interpreter_context_, &*execution_db_accessor_, params); + } else if (utils::Downcast(parsed_query.query)) { + prepared_query = + PrepareIsolationLevelQuery(std::move(parsed_query), in_explicit_transaction_, interpreter_context_, this); } else { LOG_FATAL("Should not get here -- unknown query type!"); } @@ -1690,4 +1739,22 @@ void Interpreter::AbortCommand(std::unique_ptr *query_execution) } } +std::optional Interpreter::GetIsolationLevelOverride() { + if (next_transaction_isolation_level) { + const auto isolation_level = *next_transaction_isolation_level; + next_transaction_isolation_level.reset(); + return isolation_level; + } + + return interpreter_isolation_level; +} + +void Interpreter::SetNextTransactionIsolationLevel(const storage::IsolationLevel isolation_level) { + next_transaction_isolation_level.emplace(isolation_level); +} + +void Interpreter::SetSessionIsolationLevel(const storage::IsolationLevel isolation_level) { + interpreter_isolation_level.emplace(isolation_level); +} + } // namespace query diff --git a/src/query/interpreter.hpp b/src/query/interpreter.hpp index 87bec040d..d68d9cee6 100644 --- a/src/query/interpreter.hpp +++ b/src/query/interpreter.hpp @@ -15,6 +15,7 @@ #include "query/stream.hpp" #include "query/trigger.hpp" #include "query/typed_value.hpp" +#include "storage/v2/isolation_level.hpp" #include "utils/event_counter.hpp" #include "utils/logging.hpp" #include "utils/memory.hpp" @@ -254,6 +255,9 @@ class Interpreter final { void RollbackTransaction(); + void SetNextTransactionIsolationLevel(storage::IsolationLevel isolation_level); + void SetSessionIsolationLevel(storage::IsolationLevel isolation_level); + /** * Abort the current multicommand transaction. */ @@ -306,10 +310,14 @@ class Interpreter final { bool in_explicit_transaction_{false}; bool expect_rollback_{false}; + std::optional interpreter_isolation_level; + std::optional next_transaction_isolation_level; + PreparedQuery PrepareTransactionQuery(std::string_view query_upper); void Commit(); void AdvanceCommand(); void AbortCommand(std::unique_ptr *query_execution); + std::optional GetIsolationLevelOverride(); size_t ActiveQueryExecutions() { return std::count_if(query_executions_.begin(), query_executions_.end(), diff --git a/src/storage/v2/config.hpp b/src/storage/v2/config.hpp index 42f9ebee1..5c3bab2ca 100644 --- a/src/storage/v2/config.hpp +++ b/src/storage/v2/config.hpp @@ -3,6 +3,8 @@ #include #include #include +#include "storage/v2/isolation_level.hpp" +#include "storage/v2/transaction.hpp" namespace storage { @@ -38,6 +40,10 @@ struct Config { bool snapshot_on_exit{false}; } durability; + + struct Transaction { + IsolationLevel isolation_level{IsolationLevel::SNAPSHOT_ISOLATION}; + } transaction; }; } // namespace storage diff --git a/src/storage/v2/isolation_level.hpp b/src/storage/v2/isolation_level.hpp new file mode 100644 index 000000000..42bf8f421 --- /dev/null +++ b/src/storage/v2/isolation_level.hpp @@ -0,0 +1,9 @@ +#pragma once + +#include + +namespace storage { + +enum class IsolationLevel : std::uint8_t { SNAPSHOT_ISOLATION, READ_COMMITTED, READ_UNCOMMITTED }; + +} // namespace storage diff --git a/src/storage/v2/mvcc.hpp b/src/storage/v2/mvcc.hpp index 211fc87af..1c449fa6b 100644 --- a/src/storage/v2/mvcc.hpp +++ b/src/storage/v2/mvcc.hpp @@ -24,8 +24,18 @@ inline void ApplyDeltasForRead(Transaction *transaction, const Delta *delta, Vie auto ts = delta->timestamp->load(std::memory_order_acquire); auto cid = delta->command_id; - // This is a committed change that we see so we shouldn't undo it. - if (ts < transaction->start_timestamp) { + // For SNAPSHOT ISOLATION -> we can only see the changes which were committed before the start of the current + // transaction + // + // For READ COMMITTED -> we can only see the changes which are committed. Commit timestamps of + // uncommitted changes are set to the transaction id of the transaction that made the change. Transaction id is + // always higher than start or commit timestamps so we know if the timestamp is lower than the initial transaction + // id value, that the change is committed. + // + // For READ UNCOMMITTED -> we accept any change. + if ((transaction->isolation_level == IsolationLevel::SNAPSHOT_ISOLATION && ts < transaction->start_timestamp) || + (transaction->isolation_level == IsolationLevel::READ_COMMITTED && ts < kTransactionInitialId) || + (transaction->isolation_level == IsolationLevel::READ_UNCOMMITTED)) { break; } diff --git a/src/storage/v2/storage.cpp b/src/storage/v2/storage.cpp index 8856d6628..fae4c710b 100644 --- a/src/storage/v2/storage.cpp +++ b/src/storage/v2/storage.cpp @@ -286,6 +286,7 @@ bool VerticesIterable::Iterator::operator==(const Iterator &other) const { Storage::Storage(Config config) : indices_(&constraints_, config.items), + isolation_level_(config.transaction.isolation_level), config_(config), snapshot_directory_(config_.durability.storage_directory / durability::kSnapshotDirectory), wal_directory_(config_.durability.storage_directory / durability::kWalDirectory), @@ -394,13 +395,13 @@ Storage::~Storage() { } } -Storage::Accessor::Accessor(Storage *storage) +Storage::Accessor::Accessor(Storage *storage, IsolationLevel isolation_level) : storage_(storage), // The lock must be acquired before creating the transaction object to // prevent freshly created transactions from dangling in an active state // during exclusive operations. storage_guard_(storage_->main_lock_), - transaction_(storage->CreateTransaction()), + transaction_(storage->CreateTransaction(isolation_level)), is_transaction_active_(true), config_(storage->config_.items) {} @@ -1227,7 +1228,7 @@ VerticesIterable Storage::Accessor::Vertices(LabelId label, PropertyId property, storage_->indices_.label_property_index.Vertices(label, property, lower_bound, upper_bound, view, &transaction_)); } -Transaction Storage::CreateTransaction() { +Transaction Storage::CreateTransaction(IsolationLevel isolation_level) { // We acquire the transaction engine lock here because we access (and // modify) the transaction engine variables (`transaction_id` and // `timestamp`) below. @@ -1248,7 +1249,7 @@ Transaction Storage::CreateTransaction() { start_timestamp = timestamp_++; } } - return {transaction_id, start_timestamp}; + return {transaction_id, start_timestamp, isolation_level}; } template @@ -1736,7 +1737,7 @@ void Storage::CreateSnapshot() { std::shared_lock storage_guard(main_lock_); // Create the transaction used to create the snapshot. - auto transaction = CreateTransaction(); + auto transaction = CreateTransaction(IsolationLevel::SNAPSHOT_ISOLATION); // Create snapshot. durability::CreateSnapshot(&transaction, snapshot_directory_, wal_directory_, @@ -1894,4 +1895,9 @@ std::vector Storage::ReplicasInfo() { }); } +void Storage::SetIsolationLevel(IsolationLevel isolation_level) { + std::unique_lock main_guard{main_lock_}; + isolation_level_ = isolation_level; +} + } // namespace storage diff --git a/src/storage/v2/storage.hpp b/src/storage/v2/storage.hpp index f61ff5546..fcb13c349 100644 --- a/src/storage/v2/storage.hpp +++ b/src/storage/v2/storage.hpp @@ -15,6 +15,7 @@ #include "storage/v2/edge.hpp" #include "storage/v2/edge_accessor.hpp" #include "storage/v2/indices.hpp" +#include "storage/v2/isolation_level.hpp" #include "storage/v2/mvcc.hpp" #include "storage/v2/name_id_mapper.hpp" #include "storage/v2/result.hpp" @@ -184,7 +185,7 @@ class Storage final { private: friend class Storage; - explicit Accessor(Storage *storage); + explicit Accessor(Storage *storage, IsolationLevel isolation_level); public: Accessor(const Accessor &) = delete; @@ -322,7 +323,9 @@ class Storage final { Config::Items config_; }; - Accessor Access() { return Accessor{this}; } + Accessor Access(std::optional override_isolation_level = {}) { + return Accessor{this, override_isolation_level.value_or(isolation_level_)}; + } const std::string &LabelToName(LabelId label) const; const std::string &PropertyToName(PropertyId property) const; @@ -423,8 +426,10 @@ class Storage final { void FreeMemory(); + void SetIsolationLevel(IsolationLevel isolation_level); + private: - Transaction CreateTransaction(); + Transaction CreateTransaction(IsolationLevel isolation_level); /// The force parameter determines the behaviour of the garbage collector. /// If it's set to true, it will behave as a global operation, i.e. it can't @@ -485,6 +490,7 @@ class Storage final { std::optional commit_log_; utils::Synchronized, utils::SpinLock> committed_transactions_; + IsolationLevel isolation_level_; Config config_; utils::Scheduler gc_runner_; diff --git a/src/storage/v2/transaction.hpp b/src/storage/v2/transaction.hpp index 979b26da8..dbf029dba 100644 --- a/src/storage/v2/transaction.hpp +++ b/src/storage/v2/transaction.hpp @@ -9,6 +9,7 @@ #include "storage/v2/delta.hpp" #include "storage/v2/edge.hpp" +#include "storage/v2/isolation_level.hpp" #include "storage/v2/property_value.hpp" #include "storage/v2/vertex.hpp" #include "storage/v2/view.hpp" @@ -19,8 +20,12 @@ const uint64_t kTimestampInitialId = 0; const uint64_t kTransactionInitialId = 1ULL << 63U; struct Transaction { - Transaction(uint64_t transaction_id, uint64_t start_timestamp) - : transaction_id(transaction_id), start_timestamp(start_timestamp), command_id(0), must_abort(false) {} + Transaction(uint64_t transaction_id, uint64_t start_timestamp, IsolationLevel isolation_level) + : transaction_id(transaction_id), + start_timestamp(start_timestamp), + command_id(0), + must_abort(false), + isolation_level(isolation_level) {} Transaction(Transaction &&other) noexcept : transaction_id(other.transaction_id), @@ -28,7 +33,8 @@ struct Transaction { commit_timestamp(std::move(other.commit_timestamp)), command_id(other.command_id), deltas(std::move(other.deltas)), - must_abort(other.must_abort) {} + must_abort(other.must_abort), + isolation_level(other.isolation_level) {} Transaction(const Transaction &) = delete; Transaction &operator=(const Transaction &) = delete; @@ -52,6 +58,7 @@ struct Transaction { uint64_t command_id; std::list deltas; bool must_abort; + IsolationLevel isolation_level; }; inline bool operator==(const Transaction &first, const Transaction &second) { diff --git a/tests/benchmark/expansion.cpp b/tests/benchmark/expansion.cpp index 3317a3158..5551f4829 100644 --- a/tests/benchmark/expansion.cpp +++ b/tests/benchmark/expansion.cpp @@ -4,6 +4,7 @@ #include "communication/result_stream_faker.hpp" #include "query/interpreter.hpp" #include "query/typed_value.hpp" +#include "storage/v2/isolation_level.hpp" #include "storage/v2/storage.hpp" class ExpansionBenchFixture : public benchmark::Fixture { diff --git a/tests/e2e/CMakeLists.txt b/tests/e2e/CMakeLists.txt index 6984f3530..1a14bd21a 100644 --- a/tests/e2e/CMakeLists.txt +++ b/tests/e2e/CMakeLists.txt @@ -1,3 +1,4 @@ add_subdirectory(replication) add_subdirectory(memory) add_subdirectory(triggers) +add_subdirectory(isolation_levels) diff --git a/tests/e2e/isolation_levels/CMakeLists.txt b/tests/e2e/isolation_levels/CMakeLists.txt new file mode 100644 index 000000000..d2441aced --- /dev/null +++ b/tests/e2e/isolation_levels/CMakeLists.txt @@ -0,0 +1,2 @@ +add_executable(memgraph__e2e__isolation_levels isolation_levels.cpp) +target_link_libraries(memgraph__e2e__isolation_levels gflags mgclient mg-utils mg-io Threads::Threads) diff --git a/tests/e2e/isolation_levels/isolation_levels.cpp b/tests/e2e/isolation_levels/isolation_levels.cpp new file mode 100644 index 000000000..df91c3e70 --- /dev/null +++ b/tests/e2e/isolation_levels/isolation_levels.cpp @@ -0,0 +1,241 @@ +#include +#include + +#include "utils/logging.hpp" +#include "utils/timer.hpp" + +DEFINE_uint64(bolt_port, 7687, "Bolt port"); +DEFINE_uint64(timeout, 120, "Timeout seconds"); + +namespace { + +auto GetClient() { + auto client = + mg::Client::Connect({.host = "127.0.0.1", .port = static_cast(FLAGS_bolt_port), .use_ssl = false}); + MG_ASSERT(client, "Failed to connect!"); + + return client; +} + +auto GetVertexCount(std::unique_ptr &client) { + MG_ASSERT(client->Execute("MATCH (n) RETURN count(n)")); + auto maybe_row = client->FetchOne(); + MG_ASSERT(maybe_row, "Failed to fetch vertex count"); + + const auto &row = *maybe_row; + MG_ASSERT(row.size() == 1, "Got invalid result for vertex count"); + + client->FetchOne(); + return row[0].ValueInt(); +} + +void CleanDatabase() { + auto client = GetClient(); + MG_ASSERT(client->Execute("MATCH (n) DETACH DELETE n;")); + client->DiscardAll(); +} + +void TestSnapshotIsolation(std::unique_ptr &client) { + spdlog::info("Verifying SNAPSHOT ISOLATION"); + + auto creator = GetClient(); + + MG_ASSERT(client->BeginTransaction()); + MG_ASSERT(creator->BeginTransaction()); + + constexpr auto vertex_count = 10; + for (size_t i = 0; i < vertex_count; ++i) { + MG_ASSERT(creator->Execute("CREATE ()")); + creator->DiscardAll(); + + auto current_vertex_count = GetVertexCount(client); + MG_ASSERT(current_vertex_count == 0, + "Invalid number of vertices found for SNAPSHOT ISOLATION (found {}, expected {}). Read vertices from a " + "transaction which started " + "at a later point.", + current_vertex_count, 0); + } + + MG_ASSERT(creator->CommitTransaction()); + + auto current_vertex_count = GetVertexCount(client); + MG_ASSERT(current_vertex_count == 0, + "Invalid number of vertices found for SNAPSHOT ISOLATION (found {}, expected {}). Read vertices from a " + "transaction which started " + "at a later point.", + current_vertex_count, 0); + MG_ASSERT(client->CommitTransaction()); + CleanDatabase(); +} + +void TestReadCommitted(std::unique_ptr &client) { + spdlog::info("Verifying READ COMMITTED"); + + auto creator = GetClient(); + + MG_ASSERT(client->BeginTransaction()); + MG_ASSERT(creator->BeginTransaction()); + + constexpr auto vertex_count = 10; + for (size_t i = 0; i < vertex_count; ++i) { + MG_ASSERT(creator->Execute("CREATE ()")); + creator->DiscardAll(); + + auto current_vertex_count = GetVertexCount(client); + MG_ASSERT(current_vertex_count == 0, + "Invalid number of vertices found for READ COMMITTED (found {}, expected {}. Read vertices from a " + "transaction which is not " + "committed.", + current_vertex_count, 0); + } + + MG_ASSERT(creator->CommitTransaction()); + + auto current_vertex_count = GetVertexCount(client); + MG_ASSERT(current_vertex_count == vertex_count, + "Invalid number of vertices found for READ COMMITTED (found {}, expected {}). Failed to read vertices " + "from a committed transaction", + current_vertex_count, vertex_count); + MG_ASSERT(client->CommitTransaction()); + CleanDatabase(); +} + +void TestReadUncommitted(std::unique_ptr &client) { + spdlog::info("Verifying READ UNCOMMITTED"); + + auto creator = GetClient(); + + MG_ASSERT(client->BeginTransaction()); + MG_ASSERT(creator->BeginTransaction()); + + constexpr auto vertex_count = 10; + for (size_t i = 1; i <= vertex_count; ++i) { + MG_ASSERT(creator->Execute("CREATE ()")); + creator->DiscardAll(); + + auto current_vertex_count = GetVertexCount(client); + MG_ASSERT(current_vertex_count == i, + "Invalid number of vertices found for READ UNCOMMITTED (found {}, expected {}). Failed to read vertices " + "from a different transaction.", + current_vertex_count, i); + } + + MG_ASSERT(creator->CommitTransaction()); + + auto current_vertex_count = GetVertexCount(client); + MG_ASSERT(current_vertex_count == vertex_count, + "Invalid number of vertices found for READ UNCOMMITTED (found {}, expected {}). Failed to read vertices " + "from a different transaction", + current_vertex_count, vertex_count); + MG_ASSERT(client->CommitTransaction()); + CleanDatabase(); +} + +constexpr std::array isolation_levels{std::pair{"SNAPSHOT ISOLATION", &TestSnapshotIsolation}, + std::pair{"READ COMMITTED", &TestReadCommitted}, + std::pair{"READ UNCOMMITTED", &TestReadUncommitted}}; + +void TestGlobalIsolationLevel() { + spdlog::info("\n\n----Test global isolation levels----\n"); + auto first_client = GetClient(); + auto second_client = GetClient(); + + for (const auto &[isolation_level, verification_function] : isolation_levels) { + spdlog::info("--------------------------"); + spdlog::info("Setting global isolation level to {}", isolation_level); + MG_ASSERT(first_client->Execute(fmt::format("SET GLOBAL TRANSACTION ISOLATION LEVEL {}", isolation_level))); + first_client->DiscardAll(); + + verification_function(first_client); + verification_function(second_client); + spdlog::info("--------------------------\n"); + } +} + +void TestSessionIsolationLevel() { + spdlog::info("\n\n----Test session isolation levels----\n"); + + auto global_client = GetClient(); + auto session_client = GetClient(); + for (const auto &[global_isolation_level, global_verification_function] : isolation_levels) { + spdlog::info("Setting global isolation level to {}", global_isolation_level); + MG_ASSERT(global_client->Execute(fmt::format("SET GLOBAL TRANSACTION ISOLATION LEVEL {}", global_isolation_level))); + global_client->DiscardAll(); + + for (const auto &[session_isolation_level, session_verification_function] : isolation_levels) { + spdlog::info("--------------------------"); + spdlog::info("Setting session isolation level to {}", session_isolation_level); + MG_ASSERT( + session_client->Execute(fmt::format("SET SESSION TRANSACTION ISOLATION LEVEL {}", session_isolation_level))); + session_client->DiscardAll(); + + spdlog::info("Verifying client which is using global isolation level"); + global_verification_function(global_client); + spdlog::info("Verifying client which is using session isolation level"); + session_verification_function(session_client); + spdlog::info("--------------------------\n"); + } + } +} + +// Priority of applying the isolation level from highest priority NEXT -> SESSION -> GLOBAL +void TestNextIsolationLevel() { + spdlog::info("\n\n----Test next isolation levels----\n"); + + auto global_client = GetClient(); + auto session_client = GetClient(); + for (const auto &[global_isolation_level, global_verification_function] : isolation_levels) { + spdlog::info("Setting global isolation level to {}", global_isolation_level); + MG_ASSERT(global_client->Execute(fmt::format("SET GLOBAL TRANSACTION ISOLATION LEVEL {}", global_isolation_level))); + global_client->DiscardAll(); + + for (const auto &[session_isolation_level, session_verification_function] : isolation_levels) { + spdlog::info("Setting session isolation level to {}", session_isolation_level); + MG_ASSERT( + session_client->Execute(fmt::format("SET SESSION TRANSACTION ISOLATION LEVEL {}", session_isolation_level))); + session_client->DiscardAll(); + + for (const auto &[next_isolation_level, next_verification_function] : isolation_levels) { + spdlog::info("--------------------------"); + spdlog::info("Verifying client which is using global isolation level"); + global_verification_function(global_client); + spdlog::info("Verifying client which is using session isolation level"); + session_verification_function(session_client); + + spdlog::info("Setting isolation level of the next transaction to {}", next_isolation_level); + MG_ASSERT(global_client->Execute(fmt::format("SET NEXT TRANSACTION ISOLATION LEVEL {}", next_isolation_level))); + global_client->DiscardAll(); + MG_ASSERT( + session_client->Execute(fmt::format("SET NEXT TRANSACTION ISOLATION LEVEL {}", next_isolation_level))); + session_client->DiscardAll(); + + spdlog::info("Verifying client which is using global isolation level while next isolation level is set"); + next_verification_function(global_client); + spdlog::info("Verifying client which is using session isolation level while next isolation level is set"); + next_verification_function(session_client); + + spdlog::info("Verifying client which is using global isolation level after the next isolation level was used"); + global_verification_function(global_client); + spdlog::info("Verifying client which is using session isolation level after the next isolation level was used"); + session_verification_function(session_client); + spdlog::info("--------------------------\n"); + } + } + } +} + +} // namespace + +int main(int argc, char **argv) { + google::SetUsageMessage("Memgraph E2E Isolation Levels"); + gflags::ParseCommandLineFlags(&argc, &argv, true); + logging::RedirectToStderr(); + + mg::Client::Init(); + + TestGlobalIsolationLevel(); + TestSessionIsolationLevel(); + TestNextIsolationLevel(); + + return 0; +} diff --git a/tests/e2e/isolation_levels/workloads.yaml b/tests/e2e/isolation_levels/workloads.yaml new file mode 100644 index 000000000..5d793396e --- /dev/null +++ b/tests/e2e/isolation_levels/workloads.yaml @@ -0,0 +1,14 @@ +bolt_port: &bolt_port "7687" +template_cluster: &template_cluster + cluster: + main: + args: ["--bolt-port", *bolt_port, "--log-level=TRACE"] + log_file: "isolation-levels-e2e.log" + setup_queries: [] + validation_queries: [] + +workloads: + - name: "Isolation levels" + binary: "tests/e2e/isolation_levels/memgraph__e2e__isolation_levels" + args: ["--bolt-port", *bolt_port] + <<: *template_cluster diff --git a/tests/manual/single_query.cpp b/tests/manual/single_query.cpp index ee60006ed..471b7779a 100644 --- a/tests/manual/single_query.cpp +++ b/tests/manual/single_query.cpp @@ -1,5 +1,6 @@ #include "communication/result_stream_faker.hpp" #include "query/interpreter.hpp" +#include "storage/v2/isolation_level.hpp" #include "storage/v2/storage.hpp" #include "utils/on_scope_exit.hpp" diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index 1adda7f54..5f61495a4 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -279,6 +279,9 @@ target_link_libraries(${test_prefix}storage_v2_wal_file mg-storage-v2 fmt) add_unit_test(storage_v2_replication.cpp) target_link_libraries(${test_prefix}storage_v2_replication mg-storage-v2 fmt) +add_unit_test(storage_v2_isolation_level.cpp) +target_link_libraries(${test_prefix}storage_v2_isolation_level mg-storage-v2) + # Test mg-auth if (MG_ENTERPRISE) diff --git a/tests/unit/cypher_main_visitor.cpp b/tests/unit/cypher_main_visitor.cpp index 9cb70fd99..fd673c3f4 100644 --- a/tests/unit/cypher_main_visitor.cpp +++ b/tests/unit/cypher_main_visitor.cpp @@ -2065,6 +2065,8 @@ TEST_P(CypherMainVisitorTest, GrantPrivilege) { {AuthQuery::Privilege::FREE_MEMORY}); check_auth_query(&ast_generator, "GRANT TRIGGER TO user", AuthQuery::Action::GRANT_PRIVILEGE, "", "", "user", {}, {AuthQuery::Privilege::TRIGGER}); + check_auth_query(&ast_generator, "GRANT CONFIG TO user", AuthQuery::Action::GRANT_PRIVILEGE, "", "", "user", {}, + {AuthQuery::Privilege::CONFIG}); } TEST_P(CypherMainVisitorTest, DenyPrivilege) { @@ -3129,20 +3131,20 @@ TEST_P(CypherMainVisitorTest, CreateTriggers) { const auto *query_template = "CREATE TRIGGER trigger {} {} COMMIT EXECUTE {}"; - std::array events{std::pair{"", query::TriggerQuery::EventType::ANY}, - std::pair{"ON CREATE", query::TriggerQuery::EventType::CREATE}, - std::pair{"ON () CREATE", query::TriggerQuery::EventType::VERTEX_CREATE}, - std::pair{"ON --> CREATE", query::TriggerQuery::EventType::EDGE_CREATE}, - std::pair{"ON DELETE", query::TriggerQuery::EventType::DELETE}, - std::pair{"ON () DELETE", query::TriggerQuery::EventType::VERTEX_DELETE}, - std::pair{"ON --> DELETE", query::TriggerQuery::EventType::EDGE_DELETE}, - std::pair{"ON UPDATE", query::TriggerQuery::EventType::UPDATE}, - std::pair{"ON () UPDATE", query::TriggerQuery::EventType::VERTEX_UPDATE}, - std::pair{"ON --> UPDATE", query::TriggerQuery::EventType::EDGE_UPDATE}}; + constexpr std::array events{std::pair{"", query::TriggerQuery::EventType::ANY}, + std::pair{"ON CREATE", query::TriggerQuery::EventType::CREATE}, + std::pair{"ON () CREATE", query::TriggerQuery::EventType::VERTEX_CREATE}, + std::pair{"ON --> CREATE", query::TriggerQuery::EventType::EDGE_CREATE}, + std::pair{"ON DELETE", query::TriggerQuery::EventType::DELETE}, + std::pair{"ON () DELETE", query::TriggerQuery::EventType::VERTEX_DELETE}, + std::pair{"ON --> DELETE", query::TriggerQuery::EventType::EDGE_DELETE}, + std::pair{"ON UPDATE", query::TriggerQuery::EventType::UPDATE}, + std::pair{"ON () UPDATE", query::TriggerQuery::EventType::VERTEX_UPDATE}, + std::pair{"ON --> UPDATE", query::TriggerQuery::EventType::EDGE_UPDATE}}; - std::array phases{"BEFORE", "AFTER"}; + constexpr std::array phases{"BEFORE", "AFTER"}; - std::array statements{ + constexpr std::array statements{ "", "SOME SUPER\nSTATEMENT", "Statement with 12312321 3 ", " Statement with 12312321 3 " }; @@ -3157,4 +3159,43 @@ TEST_P(CypherMainVisitorTest, CreateTriggers) { } } +namespace { +void ValidateSetIsolationLevelQuery(Base &ast_generator, const auto &query, const auto scope, + const auto isolation_level) { + auto *parsed_query = dynamic_cast(ast_generator.ParseQuery(query)); + EXPECT_EQ(parsed_query->isolation_level_scope_, scope); + EXPECT_EQ(parsed_query->isolation_level_, isolation_level); +} +} // namespace + +TEST_P(CypherMainVisitorTest, SetIsolationLevelQuery) { + auto &ast_generator = *GetParam(); + TestInvalidQuery("SET ISO", ast_generator); + TestInvalidQuery("SET TRANSACTION ISOLATION", ast_generator); + TestInvalidQuery("SET TRANSACTION ISOLATION LEVEL", ast_generator); + TestInvalidQuery("SET TRANSACTION ISOLATION LEVEL READ COMMITTED", ast_generator); + TestInvalidQuery("SET NEXT TRANSACTION ISOLATION LEVEL", ast_generator); + TestInvalidQuery("SET ISOLATION LEVEL READ COMMITTED", ast_generator); + TestInvalidQuery("SET GLOBAL ISOLATION LEVEL READ COMMITTED", ast_generator); + TestInvalidQuery("SET GLOBAL TRANSACTION ISOLATION LEVEL READ COMITTED", ast_generator); + TestInvalidQuery("SET GLOBAL TRANSACTION ISOLATION LEVEL READ_COMITTED", ast_generator); + TestInvalidQuery("SET SESSION TRANSACTION ISOLATION LEVEL READCOMITTED", ast_generator); + + constexpr std::array scopes{std::pair{"GLOBAL", query::IsolationLevelQuery::IsolationLevelScope::GLOBAL}, + std::pair{"SESSION", query::IsolationLevelQuery::IsolationLevelScope::SESSION}, + std::pair{"NEXT", query::IsolationLevelQuery::IsolationLevelScope::NEXT}}; + constexpr std::array isolation_levels{ + std::pair{"READ UNCOMMITTED", query::IsolationLevelQuery::IsolationLevel::READ_UNCOMMITTED}, + std::pair{"READ COMMITTED", query::IsolationLevelQuery::IsolationLevel::READ_COMMITTED}, + std::pair{"SNAPSHOT ISOLATION", query::IsolationLevelQuery::IsolationLevel::SNAPSHOT_ISOLATION}}; + + constexpr const auto *query_template = "SET {} TRANSACTION ISOLATION LEVEL {}"; + + for (const auto &[scope_string, scope] : scopes) { + for (const auto &[isolation_level_string, isolation_level] : isolation_levels) { + ValidateSetIsolationLevelQuery(ast_generator, fmt::format(query_template, scope_string, isolation_level_string), + scope, isolation_level); + } + } +} } // namespace diff --git a/tests/unit/interpreter.cpp b/tests/unit/interpreter.cpp index 194c4c815..70b92fbca 100644 --- a/tests/unit/interpreter.cpp +++ b/tests/unit/interpreter.cpp @@ -11,6 +11,7 @@ #include "query/stream.hpp" #include "query/typed_value.hpp" #include "query_common.hpp" +#include "storage/v2/isolation_level.hpp" #include "storage/v2/property_value.hpp" #include "utils/csv_parsing.hpp" #include "utils/logging.hpp" diff --git a/tests/unit/query_required_privileges.cpp b/tests/unit/query_required_privileges.cpp index 09fee4c0d..8ce38ee55 100644 --- a/tests/unit/query_required_privileges.cpp +++ b/tests/unit/query_required_privileges.cpp @@ -154,3 +154,8 @@ TEST_F(TestPrivilegeExtractor, TriggerQuery) { auto *query = storage.Create(); EXPECT_THAT(GetRequiredPrivileges(query), UnorderedElementsAre(AuthQuery::Privilege::TRIGGER)); } + +TEST_F(TestPrivilegeExtractor, SetIsolationLevelQuery) { + auto *query = storage.Create(); + EXPECT_THAT(GetRequiredPrivileges(query), UnorderedElementsAre(AuthQuery::Privilege::CONFIG)); +} diff --git a/tests/unit/storage_v2_isolation_level.cpp b/tests/unit/storage_v2_isolation_level.cpp new file mode 100644 index 000000000..f80bc2ef2 --- /dev/null +++ b/tests/unit/storage_v2_isolation_level.cpp @@ -0,0 +1,98 @@ +#include + +#include "storage/v2/isolation_level.hpp" +#include "storage/v2/storage.hpp" + +namespace { +int64_t VerticesCount(storage::Storage::Accessor &accessor) { + int64_t count{0}; + for ([[maybe_unused]] const auto &vertex : accessor.Vertices(storage::View::NEW)) { + ++count; + } + + return count; +} + +constexpr std::array isolation_levels{storage::IsolationLevel::SNAPSHOT_ISOLATION, + storage::IsolationLevel::READ_COMMITTED, + storage::IsolationLevel::READ_UNCOMMITTED}; + +std::string_view IsolationLevelToString(const storage::IsolationLevel isolation_level) { + switch (isolation_level) { + case storage::IsolationLevel::SNAPSHOT_ISOLATION: + return "SNAPSHOT_ISOLATION"; + case storage::IsolationLevel::READ_COMMITTED: + return "READ_COMMITTED"; + case storage::IsolationLevel::READ_UNCOMMITTED: + return "READ_UNCOMMITTED"; + } +} +} // namespace + +class StorageIsolationLevelTest : public ::testing::TestWithParam { + public: + struct PrintToStringParamName { + std::string operator()(const testing::TestParamInfo &info) { + return std::string(IsolationLevelToString(static_cast(info.param))); + } + }; +}; + +TEST_P(StorageIsolationLevelTest, Visibility) { + const auto default_isolation_level = GetParam(); + + for (const auto override_isolation_level : isolation_levels) { + storage::Storage storage{storage::Config{.transaction = {.isolation_level = default_isolation_level}}}; + auto creator = storage.Access(); + auto default_isolation_level_reader = storage.Access(); + auto override_isolation_level_reader = storage.Access(override_isolation_level); + + ASSERT_EQ(VerticesCount(default_isolation_level_reader), 0); + ASSERT_EQ(VerticesCount(override_isolation_level_reader), 0); + + constexpr auto iteration_count = 10; + { + SCOPED_TRACE(fmt::format( + "Visibility while the creator transaction is active " + "(default isolation level = {}, override isolation level = {})", + IsolationLevelToString(default_isolation_level), IsolationLevelToString(override_isolation_level))); + for (size_t i = 1; i <= iteration_count; ++i) { + creator.CreateVertex(); + + const auto check_vertices_count = [i](auto &accessor, const auto isolation_level) { + const auto expected_count = isolation_level == storage::IsolationLevel::READ_UNCOMMITTED ? i : 0; + EXPECT_EQ(VerticesCount(accessor), expected_count); + }; + check_vertices_count(default_isolation_level_reader, default_isolation_level); + check_vertices_count(override_isolation_level_reader, override_isolation_level); + } + } + + ASSERT_FALSE(creator.Commit().HasError()); + { + SCOPED_TRACE(fmt::format( + "Visibility after the creator transaction is committed " + "(default isolation level = {}, override isolation level = {})", + IsolationLevelToString(default_isolation_level), IsolationLevelToString(override_isolation_level))); + const auto check_vertices_count = [iteration_count](auto &accessor, const auto isolation_level) { + const auto expected_count = + isolation_level == storage::IsolationLevel::SNAPSHOT_ISOLATION ? 0 : iteration_count; + ASSERT_EQ(VerticesCount(accessor), expected_count); + }; + + check_vertices_count(default_isolation_level_reader, default_isolation_level); + check_vertices_count(override_isolation_level_reader, override_isolation_level); + } + + ASSERT_FALSE(default_isolation_level_reader.Commit().HasError()); + ASSERT_FALSE(override_isolation_level_reader.Commit().HasError()); + + SCOPED_TRACE("Visibility after a new transaction is started"); + auto verifier = storage.Access(); + ASSERT_EQ(VerticesCount(verifier), iteration_count); + ASSERT_FALSE(verifier.Commit().HasError()); + } +} + +INSTANTIATE_TEST_CASE_P(ParameterizedStorageIsolationLevelTests, StorageIsolationLevelTest, + ::testing::ValuesIn(isolation_levels), StorageIsolationLevelTest::PrintToStringParamName()); diff --git a/tests/unit/storage_v2_wal_file.cpp b/tests/unit/storage_v2_wal_file.cpp index 34b48b77e..79e03f351 100644 --- a/tests/unit/storage_v2_wal_file.cpp +++ b/tests/unit/storage_v2_wal_file.cpp @@ -45,7 +45,9 @@ class DeltaGenerator final { private: friend class DeltaGenerator; - explicit Transaction(DeltaGenerator *gen) : gen_(gen), transaction_(gen->transaction_id_++, gen->timestamp_++) {} + explicit Transaction(DeltaGenerator *gen) + : gen_(gen), + transaction_(gen->transaction_id_++, gen->timestamp_++, storage::IsolationLevel::SNAPSHOT_ISOLATION) {} public: storage::Vertex *CreateVertex() {