diff --git a/src/query/interpreter.cpp b/src/query/interpreter.cpp index 2c5ff75ed..5aeaf0cf4 100644 --- a/src/query/interpreter.cpp +++ b/src/query/interpreter.cpp @@ -1401,6 +1401,54 @@ void Interpreter::Abort() { db_accessor_ = std::nullopt; } +namespace { +void RunTriggersIndividually(const utils::SkipList &triggers, InterpreterContext *interpreter_context) { + // Run the triggers + for (const auto &trigger : triggers.access()) { + spdlog::debug("Executing trigger '{}'", trigger.name()); + utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize}; + + // create a new transaction for each trigger + auto storage_acc = interpreter_context->db->Access(); + DbAccessor db_accessor{&storage_acc}; + + try { + trigger.Execute(&interpreter_context->plan_cache, &db_accessor, &execution_memory, + *interpreter_context->tsc_frequency, interpreter_context->execution_timeout_sec, + &interpreter_context->is_shutting_down); + } catch (const utils::BasicException &exception) { + spdlog::warn("Trigger {} failed with exception:\n{}", trigger.name(), exception.what()); + db_accessor.Abort(); + continue; + } + + auto maybe_constraint_violation = db_accessor.Commit(); + if (maybe_constraint_violation.HasError()) { + const auto &constraint_violation = maybe_constraint_violation.GetError(); + switch (constraint_violation.type) { + case storage::ConstraintViolation::Type::EXISTENCE: { + const auto &label_name = db_accessor.LabelToName(constraint_violation.label); + MG_ASSERT(constraint_violation.properties.size() == 1U); + const auto &property_name = db_accessor.PropertyToName(*constraint_violation.properties.begin()); + spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})", trigger.name(), + label_name, property_name); + break; + } + case storage::ConstraintViolation::Type::UNIQUE: { + const auto &label_name = db_accessor.LabelToName(constraint_violation.label); + std::stringstream property_names_stream; + utils::PrintIterable(property_names_stream, constraint_violation.properties, ", ", + [&](auto &stream, const auto &prop) { stream << db_accessor.PropertyToName(prop); }); + spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})", trigger.name(), + label_name, property_names_stream.str()); + break; + } + } + } + } +} +} // namespace + void Interpreter::Commit() { // It's possible that some queries did not finish because the user did // not pull all of the results from the query. @@ -1409,6 +1457,16 @@ void Interpreter::Commit() { // a query. if (!db_accessor_) return; + // Run the triggers + for (const auto &trigger : interpreter_context_->before_commit_triggers.access()) { + spdlog::debug("Executing trigger '{}'", trigger.name()); + utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize}; + trigger.Execute(&interpreter_context_->plan_cache, &*execution_db_accessor_, &execution_memory, + *interpreter_context_->tsc_frequency, interpreter_context_->execution_timeout_sec, + &interpreter_context_->is_shutting_down); + } + SPDLOG_DEBUG("Finished executing before commit triggers"); + auto maybe_constraint_violation = db_accessor_->Commit(); if (maybe_constraint_violation.HasError()) { const auto &constraint_violation = maybe_constraint_violation.GetError(); @@ -1438,16 +1496,15 @@ void Interpreter::Commit() { } } - // Run the triggers - for (const auto &trigger : interpreter_context_->triggers.access()) { - utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize}; - trigger.Execute(&interpreter_context_->plan_cache, &*execution_db_accessor_, &execution_memory, - *interpreter_context_->tsc_frequency, interpreter_context_->execution_timeout_sec, - &interpreter_context_->is_shutting_down); - } - execution_db_accessor_ = std::nullopt; db_accessor_ = std::nullopt; + + background_thread_.AddTask([interpreter_context = this->interpreter_context_] { + RunTriggersIndividually(interpreter_context->after_commit_triggers, interpreter_context); + SPDLOG_DEBUG("Finished executing after commit triggers"); // NOLINT(bugprone-lambda-function-name) + }); + + SPDLOG_DEBUG("Finished comitting the transaction"); } void Interpreter::AdvanceCommand() { diff --git a/src/query/interpreter.hpp b/src/query/interpreter.hpp index a3ed8e323..8200201d7 100644 --- a/src/query/interpreter.hpp +++ b/src/query/interpreter.hpp @@ -20,6 +20,7 @@ #include "utils/memory.hpp" #include "utils/skip_list.hpp" #include "utils/spin_lock.hpp" +#include "utils/thread_pool.hpp" #include "utils/timer.hpp" #include "utils/tsc.hpp" @@ -147,8 +148,14 @@ struct PreparedQuery { */ struct InterpreterContext { explicit InterpreterContext(storage::Storage *db) : db(db) { - // auto triggers_acc = triggers.access(); - // triggers_acc.insert(Trigger{"Creator", "CREATE (:CREATED)", &ast_cache, &antlr_lock}); + // { + // auto triggers_acc = before_commit_triggers.access(); + // triggers_acc.insert(Trigger{"BeforeCreator", "CREATE (:BEFORE)", &ast_cache, &antlr_lock}); + // } + // { + // auto triggers_acc = after_commit_triggers.access(); + // triggers_acc.insert(Trigger{"AfterCreator", "CREATE (:AFTER)", &ast_cache, &antlr_lock}); + // } } storage::Storage *db; @@ -171,7 +178,8 @@ struct InterpreterContext { utils::SkipList plan_cache; // use a thread safe container - utils::SkipList triggers; + utils::SkipList before_commit_triggers; + utils::SkipList after_commit_triggers; }; /// Function that is used to tell all active interpreters that they should stop @@ -304,6 +312,8 @@ class Interpreter final { bool in_explicit_transaction_{false}; bool expect_rollback_{false}; + utils::ThreadPool background_thread_{1}; + PreparedQuery PrepareTransactionQuery(std::string_view query_upper); void Commit(); void AdvanceCommand(); diff --git a/src/query/trigger.hpp b/src/query/trigger.hpp index 8e15b38cb..db59b3172 100644 --- a/src/query/trigger.hpp +++ b/src/query/trigger.hpp @@ -19,6 +19,8 @@ struct Trigger { // NOLINTNEXTLINE (modernize-use-nullptr) bool operator<(const std::string &other) const { return name_ < other; } + const auto &name() const { return name_; } + private: std::string name_; ParsedQuery parsed_statements_;