diff --git a/.clang-tidy b/.clang-tidy
index b0f274372..7a9d32294 100644
--- a/.clang-tidy
+++ b/.clang-tidy
@@ -52,7 +52,8 @@ Checks: '*,
          -readability-else-after-return,
          -readability-implicit-bool-conversion,
          -readability-magic-numbers,
-         -readability-named-parameter'
+         -readability-named-parameter,
+         -misc-no-recursion'
 WarningsAsErrors: ''
 HeaderFilterRegex: 'src/.*'
 AnalyzeTemporaryDtors: false
diff --git a/src/auth/models.cpp b/src/auth/models.cpp
index 33ebc72fe..be5f28b3e 100644
--- a/src/auth/models.cpp
+++ b/src/auth/models.cpp
@@ -47,6 +47,8 @@ std::string PermissionToString(Permission permission) {
       return "READ_FILE";
     case Permission::FREE_MEMORY:
       return "FREE_MEMORY";
+    case Permission::TRIGGER:
+      return "TRIGGER";
     case Permission::AUTH:
       return "AUTH";
   }
diff --git a/src/auth/models.hpp b/src/auth/models.hpp
index f1139e1ca..9e1b54977 100644
--- a/src/auth/models.hpp
+++ b/src/auth/models.hpp
@@ -25,16 +25,17 @@ enum class Permission : uint64_t {
   LOCK_PATH   = 1U << 11U,
   READ_FILE   = 1U << 12U,
   FREE_MEMORY = 1U << 13U,
+  TRIGGER     = 1U << 14U,
   AUTH        = 1U << 16U
 };
 // clang-format on
 
 // Constant list of all available permissions.
-const std::vector<Permission> 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};
+const std::vector<Permission> 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};
 
 // 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 2a9932595..a54894b80 100644
--- a/src/glue/auth.cpp
+++ b/src/glue/auth.cpp
@@ -32,6 +32,8 @@ auth::Permission PrivilegeToPermission(query::AuthQuery::Privilege privilege) {
       return auth::Permission::READ_FILE;
     case query::AuthQuery::Privilege::FREE_MEMORY:
       return auth::Permission::FREE_MEMORY;
+    case query::AuthQuery::Privilege::TRIGGER:
+      return auth::Permission::TRIGGER;
     case query::AuthQuery::Privilege::AUTH:
       return auth::Permission::AUTH;
   }
diff --git a/src/memgraph.cpp b/src/memgraph.cpp
index c9c17b334..e99ae8f43 100644
--- a/src/memgraph.cpp
+++ b/src/memgraph.cpp
@@ -980,7 +980,7 @@ int main(int argc, char **argv) {
     db_config.durability.snapshot_interval = std::chrono::seconds(FLAGS_storage_snapshot_interval_sec);
   }
   storage::Storage db(db_config);
-  query::InterpreterContext interpreter_context{&db};
+  query::InterpreterContext interpreter_context{&db, FLAGS_data_directory};
 
   query::SetExecutionTimeout(&interpreter_context, FLAGS_query_execution_timeout_sec);
 #ifdef MG_ENTERPRISE
diff --git a/src/query/CMakeLists.txt b/src/query/CMakeLists.txt
index 8ebe31425..5c692e44e 100644
--- a/src/query/CMakeLists.txt
+++ b/src/query/CMakeLists.txt
@@ -31,6 +31,7 @@ set(mg_query_sources
     procedure/mg_procedure_impl.cpp
     procedure/module.cpp
     procedure/py_module.cpp
+    serialization/property_value.cpp
     trigger.cpp
     typed_value.cpp)
 
@@ -38,7 +39,7 @@ add_library(mg-query STATIC ${mg_query_sources})
 add_dependencies(mg-query generate_lcp_query)
 target_include_directories(mg-query PUBLIC ${CMAKE_SOURCE_DIR}/include)
 target_link_libraries(mg-query dl cppitertools)
-target_link_libraries(mg-query mg-storage-v2 mg-utils)
+target_link_libraries(mg-query mg-storage-v2 mg-utils mg-kvstore)
 if("${MG_PYTHON_VERSION}" STREQUAL "")
     find_package(Python3 3.5 REQUIRED COMPONENTS Development)
 else()
diff --git a/src/query/exceptions.hpp b/src/query/exceptions.hpp
index cde5433d2..b8ddd7480 100644
--- a/src/query/exceptions.hpp
+++ b/src/query/exceptions.hpp
@@ -161,12 +161,18 @@ class ReplicationModificationInMulticommandTxException : public QueryException {
 class LockPathModificationInMulticommandTxException : public QueryException {
  public:
   LockPathModificationInMulticommandTxException()
-      : QueryException("Lock path clause not allowed in multicommand transactions.") {}
+      : QueryException("Lock path query not allowed in multicommand transactions.") {}
 };
 
 class FreeMemoryModificationInMulticommandTxException : public QueryException {
  public:
   FreeMemoryModificationInMulticommandTxException()
-      : QueryException("Lock path clause not allowed in multicommand transactions.") {}
+      : QueryException("Free memory query not allowed in multicommand transactions.") {}
+};
+
+class TriggerModificationInMulticommandTxException : public QueryException {
+ public:
+  TriggerModificationInMulticommandTxException()
+      : QueryException("Trigger queries not allowed in multicommand transactions.") {}
 };
 }  // namespace query
diff --git a/src/query/frontend/ast/ast.lcp b/src/query/frontend/ast/ast.lcp
index 4f9fdcf4c..c6e46f984 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)
+         dump replication lock_path read_file free_memory trigger)
       (:serialize))
     #>cpp
     AuthQuery() = default;
@@ -2232,7 +2232,7 @@ const std::vector<AuthQuery::Privilege> kPrivilegesAll = {
     AuthQuery::Privilege::REPLICATION,
     AuthQuery::Privilege::READ_FILE,
     AuthQuery::Privilege::LOCK_PATH,
-    AuthQuery::Privilege::FREE_MEMORY};
+    AuthQuery::Privilege::FREE_MEMORY, AuthQuery::Privilege::TRIGGER};
 cpp<#
 
 (lcp:define-class info-query (query)
@@ -2398,7 +2398,7 @@ cpp<#
   (:serialize (:slk))
   (:clone))
 
-  (lcp:define-class free-memory-query (query) ()
+(lcp:define-class free-memory-query (query) ()
   (:public
     #>cpp
     DEFVISITABLE(QueryVisitor<void>);
@@ -2406,4 +2406,30 @@ cpp<#
   (:serialize (:slk))
   (:clone))
 
+(lcp:define-class trigger-query (query)
+  ((action "Action" :scope :public)
+   (event_type "EventType" :scope :public)
+   (trigger_name "std::string" :scope :public)
+   (before_commit "bool" :scope :public)
+   (statement "std::string" :scope :public))
+
+  (:public
+    (lcp:define-enum action
+        (create-trigger drop-trigger show-triggers)
+      (:serialize))
+    (lcp:define-enum event-type
+        (any vertex_create edge_create create vertex_delete edge_delete delete vertex_update edge_update update)
+      (:serialize))
+    #>cpp
+    TriggerQuery() = default;
+
+    DEFVISITABLE(QueryVisitor<void>);
+    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 1539422e2..4523fd093 100644
--- a/src/query/frontend/ast/ast_visitor.hpp
+++ b/src/query/frontend/ast/ast_visitor.hpp
@@ -76,6 +76,7 @@ class ReplicationQuery;
 class LockPathQuery;
 class LoadCsv;
 class FreeMemoryQuery;
+class TriggerQuery;
 
 using TreeCompositeVisitor = ::utils::CompositeVisitor<
     SingleQuery, CypherUnion, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator, AdditionOperator,
@@ -107,8 +108,8 @@ class ExpressionVisitor
           None, ParameterLookup, Identifier, PrimitiveLiteral, RegexMatch> {};
 
 template <class TResult>
-class QueryVisitor
-    : public ::utils::Visitor<TResult, CypherQuery, ExplainQuery, ProfileQuery, IndexQuery, AuthQuery, InfoQuery,
-                              ConstraintQuery, DumpQuery, ReplicationQuery, LockPathQuery, FreeMemoryQuery> {};
+class QueryVisitor : public ::utils::Visitor<TResult, CypherQuery, ExplainQuery, ProfileQuery, IndexQuery, AuthQuery,
+                                             InfoQuery, ConstraintQuery, DumpQuery, ReplicationQuery, LockPathQuery,
+                                             FreeMemoryQuery, TriggerQuery> {};
 
 }  // namespace query
diff --git a/src/query/frontend/ast/cypher_main_visitor.cpp b/src/query/frontend/ast/cypher_main_visitor.cpp
index ea0aa3ccd..51cb93305 100644
--- a/src/query/frontend/ast/cypher_main_visitor.cpp
+++ b/src/query/frontend/ast/cypher_main_visitor.cpp
@@ -340,6 +340,78 @@ antlrcpp::Any CypherMainVisitor::visitFreeMemoryQuery(MemgraphCypher::FreeMemory
   return free_memory_query;
 }
 
+antlrcpp::Any CypherMainVisitor::visitTriggerQuery(MemgraphCypher::TriggerQueryContext *ctx) {
+  MG_ASSERT(ctx->children.size() == 1, "TriggerQuery should have exactly one child!");
+  auto *trigger_query = ctx->children[0]->accept(this).as<TriggerQuery *>();
+  query_ = trigger_query;
+  return trigger_query;
+}
+
+antlrcpp::Any CypherMainVisitor::visitCreateTrigger(MemgraphCypher::CreateTriggerContext *ctx) {
+  auto *trigger_query = storage_->Create<TriggerQuery>();
+  trigger_query->action_ = TriggerQuery::Action::CREATE_TRIGGER;
+  trigger_query->trigger_name_ = ctx->triggerName()->symbolicName()->accept(this).as<std::string>();
+
+  auto *statement = ctx->triggerStatement();
+  antlr4::misc::Interval interval{statement->start->getStartIndex(), statement->stop->getStopIndex()};
+  trigger_query->statement_ = ctx->start->getInputStream()->getText(interval);
+
+  trigger_query->event_type_ = [ctx] {
+    if (!ctx->ON()) {
+      return TriggerQuery::EventType::ANY;
+    }
+
+    if (ctx->CREATE(1)) {
+      if (ctx->emptyVertex()) {
+        return TriggerQuery::EventType::VERTEX_CREATE;
+      }
+      if (ctx->emptyEdge()) {
+        return TriggerQuery::EventType::EDGE_CREATE;
+      }
+      return TriggerQuery::EventType::CREATE;
+    }
+
+    if (ctx->DELETE()) {
+      if (ctx->emptyVertex()) {
+        return TriggerQuery::EventType::VERTEX_DELETE;
+      }
+      if (ctx->emptyEdge()) {
+        return TriggerQuery::EventType::EDGE_DELETE;
+      }
+      return TriggerQuery::EventType::DELETE;
+    }
+
+    if (ctx->UPDATE()) {
+      if (ctx->emptyVertex()) {
+        return TriggerQuery::EventType::VERTEX_UPDATE;
+      }
+      if (ctx->emptyEdge()) {
+        return TriggerQuery::EventType::EDGE_UPDATE;
+      }
+      return TriggerQuery::EventType::UPDATE;
+    }
+
+    LOG_FATAL("Invalid token allowed for the query");
+  }();
+
+  trigger_query->before_commit_ = ctx->BEFORE();
+
+  return trigger_query;
+}
+
+antlrcpp::Any CypherMainVisitor::visitDropTrigger(MemgraphCypher::DropTriggerContext *ctx) {
+  auto *trigger_query = storage_->Create<TriggerQuery>();
+  trigger_query->action_ = TriggerQuery::Action::DROP_TRIGGER;
+  trigger_query->trigger_name_ = ctx->triggerName()->symbolicName()->accept(this).as<std::string>();
+  return trigger_query;
+}
+
+antlrcpp::Any CypherMainVisitor::visitShowTriggers(MemgraphCypher::ShowTriggersContext *ctx) {
+  auto *trigger_query = storage_->Create<TriggerQuery>();
+  trigger_query->action_ = TriggerQuery::Action::SHOW_TRIGGERS;
+  return trigger_query;
+}
+
 antlrcpp::Any CypherMainVisitor::visitCypherUnion(MemgraphCypher::CypherUnionContext *ctx) {
   bool distinct = !ctx->ALL();
   auto *cypher_union = storage_->Create<CypherUnion>(distinct);
@@ -771,6 +843,7 @@ antlrcpp::Any CypherMainVisitor::visitPrivilege(MemgraphCypher::PrivilegeContext
   if (ctx->LOCK_PATH()) return AuthQuery::Privilege::LOCK_PATH;
   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;
   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 03fde1119..ca97ee02a 100644
--- a/src/query/frontend/ast/cypher_main_visitor.hpp
+++ b/src/query/frontend/ast/cypher_main_visitor.hpp
@@ -218,6 +218,26 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
    */
   antlrcpp::Any visitFreeMemoryQuery(MemgraphCypher::FreeMemoryQueryContext *ctx) override;
 
+  /**
+   * @return TriggerQuery*
+   */
+  antlrcpp::Any visitTriggerQuery(MemgraphCypher::TriggerQueryContext *ctx) override;
+
+  /**
+   * @return CreateTrigger*
+   */
+  antlrcpp::Any visitCreateTrigger(MemgraphCypher::CreateTriggerContext *ctx) override;
+
+  /**
+   * @return DropTrigger*
+   */
+  antlrcpp::Any visitDropTrigger(MemgraphCypher::DropTriggerContext *ctx) override;
+
+  /**
+   * @return ShowTriggers*
+   */
+  antlrcpp::Any visitShowTriggers(MemgraphCypher::ShowTriggersContext *ctx) override;
+
   /**
    * @return CypherUnion*
    */
diff --git a/src/query/frontend/opencypher/grammar/MemgraphCypher.g4 b/src/query/frontend/opencypher/grammar/MemgraphCypher.g4
index 8778baf01..a3c578631 100644
--- a/src/query/frontend/opencypher/grammar/MemgraphCypher.g4
+++ b/src/query/frontend/opencypher/grammar/MemgraphCypher.g4
@@ -7,18 +7,22 @@ options { tokenVocab=MemgraphCypherLexer; }
 import Cypher ;
 
 memgraphCypherKeyword : cypherKeyword
+                      | AFTER
                       | ALTER
                       | ASYNC
                       | AUTH
                       | BAD
+                      | BEFORE
                       | CLEAR
                       | CSV
+                      | COMMIT
                       | DATA
                       | DELIMITER
                       | DATABASE
                       | DENY
                       | DROP
                       | DUMP
+                      | EXECUTE
                       | FOR
                       | FREE
                       | FROM
@@ -43,9 +47,12 @@ memgraphCypherKeyword : cypherKeyword
                       | QUOTE
                       | STATS
                       | SYNC
+                      | TRIGGER
+                      | TRIGGERS
                       | TIMEOUT
                       | TO
                       | UNLOCK
+                      | UPDATE
                       | USER
                       | USERS
                       ;
@@ -66,6 +73,7 @@ query : cypherQuery
       | replicationQuery
       | lockPathQuery
       | freeMemoryQuery
+      | triggerQuery
       ;
 
 authQuery : createRole
@@ -92,6 +100,11 @@ replicationQuery : setReplicationRole
                  | showReplicas
                  ;
 
+triggerQuery : createTrigger
+             | dropTrigger
+             | showTriggers
+             ;
+
 clause : cypherMatch
        | unwind
        | merge
@@ -161,6 +174,7 @@ privilege : CREATE
           | LOCK_PATH
           | READ_FILE
           | FREE_MEMORY
+          | TRIGGER
           ;
 
 privilegeList : privilege ( ',' privilege )* ;
@@ -193,3 +207,18 @@ showReplicas  : SHOW REPLICAS ;
 lockPathQuery : ( LOCK | UNLOCK ) DATA DIRECTORY ;
 
 freeMemoryQuery : FREE MEMORY ;
+
+triggerName : symbolicName ;
+
+triggerStatement : .*? ;
+
+emptyVertex : '(' ')' ;
+
+emptyEdge : dash dash rightArrowHead ;
+
+createTrigger : CREATE TRIGGER triggerName ( ON ( emptyVertex | emptyEdge ) ? ( CREATE | UPDATE | DELETE ) ) ?
+              ( AFTER | BEFORE ) COMMIT EXECUTE triggerStatement ;
+
+dropTrigger : DROP TRIGGER triggerName ;
+
+showTriggers : SHOW TRIGGERS ;
diff --git a/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4 b/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4
index 9aeec8eec..5fdea9f31 100644
--- a/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4
+++ b/src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4
@@ -12,11 +12,14 @@ import CypherLexer ;
 
 UNDERSCORE : '_' ;
 
+AFTER          : A F T E R ;
 ALTER          : A L T E R ;
 ASYNC          : A S Y N C ;
 AUTH           : A U T H ;
 BAD            : B A D ;
+BEFORE         : B E F O R E ;
 CLEAR          : C L E A R ;
+COMMIT         : C O M M I T ;
 CSV            : C S V ;
 DATA           : D A T A ;
 DELIMITER      : D E L I M I T E R ;
@@ -25,9 +28,10 @@ DENY           : D E N Y ;
 DIRECTORY      : D I R E C T O R Y ;
 DROP           : D R O P ;
 DUMP           : D U M P ;
+EXECUTE        : E X E C U T E ;
 FOR            : F O R ;
 FREE           : F R E E ;
-FREE_MEMORY     : F R E E UNDERSCORE M E M O R Y ;
+FREE_MEMORY    : F R E E UNDERSCORE M E M O R Y ;
 FROM           : F R O M ;
 GRANT          : G R A N T ;
 GRANTS         : G R A N T S ;
@@ -56,6 +60,9 @@ STATS          : S T A T S ;
 SYNC           : S Y N C ;
 TIMEOUT        : T I M E O U T ;
 TO             : T O ;
+TRIGGER        : T R I G G E R ;
+TRIGGERS       : T R I G G E R S ;
 UNLOCK         : U N L O C K ;
+UPDATE         : U P D A T E ;
 USER           : U S E R ;
 USERS          : U S E R S ;
diff --git a/src/query/frontend/semantic/required_privileges.cpp b/src/query/frontend/semantic/required_privileges.cpp
index 4173c7fbc..0f16bff56 100644
--- a/src/query/frontend/semantic/required_privileges.cpp
+++ b/src/query/frontend/semantic/required_privileges.cpp
@@ -53,6 +53,8 @@ class PrivilegeExtractor : public QueryVisitor<void>, public HierarchicalTreeVis
 
   void Visit(FreeMemoryQuery &free_memory_query) override { AddPrivilege(AuthQuery::Privilege::FREE_MEMORY); }
 
+  void Visit(TriggerQuery &trigger_query) override { AddPrivilege(AuthQuery::Privilege::TRIGGER); }
+
   void Visit(ReplicationQuery &replication_query) override { AddPrivilege(AuthQuery::Privilege::REPLICATION); }
 
   bool PreVisit(Create & /*unused*/) override {
diff --git a/src/query/frontend/stripped.cpp b/src/query/frontend/stripped.cpp
index 8f2a3e67d..9a6c31959 100644
--- a/src/query/frontend/stripped.cpp
+++ b/src/query/frontend/stripped.cpp
@@ -35,6 +35,7 @@ StrippedQuery::StrippedQuery(const std::string &query) : original_(query) {
   };
 
   std::vector<std::pair<Token, std::string>> tokens;
+  std::string unstripped_chunk;
   for (int i = 0; i < static_cast<int>(original_.size());) {
     Token token = Token::UNMATCHED;
     int len = 0;
@@ -58,6 +59,13 @@ StrippedQuery::StrippedQuery(const std::string &query) : original_(query) {
     if (token == Token::UNMATCHED) throw LexingException("Invalid query.");
     tokens.emplace_back(token, original_.substr(i, len));
     i += len;
+
+    // if we notice execute, we create a trigger which has defined statements
+    // the statements will be parsed separately later on so we skip it for now
+    if (utils::IEquals(tokens.back().second, "execute")) {
+      unstripped_chunk = original_.substr(i);
+      break;
+    }
   }
 
   std::vector<std::string> token_strings;
@@ -79,6 +87,7 @@ StrippedQuery::StrippedQuery(const std::string &query) : original_(query) {
   // named expressions in return.
   for (int i = 0; i < static_cast<int>(tokens.size()); ++i) {
     auto &token = tokens[i];
+
     // We need to shift token index for every parameter since antlr's parser
     // thinks of parameter as two tokens.
     int token_index = token_strings.size() + parameters_.size();
@@ -123,6 +132,10 @@ StrippedQuery::StrippedQuery(const std::string &query) : original_(query) {
     }
   }
 
+  if (!unstripped_chunk.empty()) {
+    token_strings.push_back(std::move(unstripped_chunk));
+  }
+
   query_ = utils::Join(token_strings, " ");
   hash_ = utils::Fnv(query_);
 
@@ -156,6 +169,7 @@ StrippedQuery::StrippedQuery(const std::string &query) : original_(query) {
       }
       // There is only whitespace, nothing to do...
       if (it == tokens.end()) break;
+
       bool has_as = false;
       auto last_non_space = it;
       auto jt = it;
diff --git a/src/query/frontend/stripped_lexer_constants.hpp b/src/query/frontend/stripped_lexer_constants.hpp
index c08da89ad..be388d708 100644
--- a/src/query/frontend/stripped_lexer_constants.hpp
+++ b/src/query/frontend/stripped_lexer_constants.hpp
@@ -79,15 +79,17 @@ 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"};
+    "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"};
 
 // Unicode codepoints that are allowed at the start of the unescaped name.
 const std::bitset<kBitsetSize> kUnescapedNameAllowedStarts(
diff --git a/src/query/interpreter.cpp b/src/query/interpreter.cpp
index 4d8ca0221..dbd8c1828 100644
--- a/src/query/interpreter.cpp
+++ b/src/query/interpreter.cpp
@@ -20,6 +20,7 @@
 #include "query/plan/vertex_count_cache.hpp"
 #include "query/trigger.hpp"
 #include "query/typed_value.hpp"
+#include "storage/v2/property_value.hpp"
 #include "utils/algorithm.hpp"
 #include "utils/csv_parsing.hpp"
 #include "utils/event_counter.hpp"
@@ -439,7 +440,6 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, ReplQueryHandler *
       };
       return callback;
     }
-      return callback;
   }
 }
 
@@ -598,6 +598,12 @@ std::optional<ExecutionContext> PullPlan::Pull(AnyStream *stream, std::optional<
 using RWType = plan::ReadWriteTypeChecker::RWType;
 }  // namespace
 
+InterpreterContext::InterpreterContext(storage::Storage *db, const std::filesystem::path &data_directory) : db(db) {
+  auto storage_accessor = db->Access();
+  DbAccessor dba{&storage_accessor};
+  trigger_store.emplace(data_directory / "triggers", &ast_cache, &dba, &antlr_lock);
+}
+
 Interpreter::Interpreter(InterpreterContext *interpreter_context) : interpreter_context_(interpreter_context) {
   MG_ASSERT(interpreter_context_, "Interpreter context must not be NULL");
   // try {
@@ -668,8 +674,7 @@ PreparedQuery Interpreter::PrepareTransactionQuery(std::string_view query_upper)
       db_accessor_ = std::make_unique<storage::Storage::Accessor>(interpreter_context_->db->Access());
       execution_db_accessor_.emplace(db_accessor_.get());
 
-      if (interpreter_context_->before_commit_triggers.size() > 0 ||
-          interpreter_context_->after_commit_triggers.size() > 0) {
+      if (interpreter_context_->trigger_store->HasTriggers()) {
         trigger_context_collector_.emplace();
       }
     };
@@ -1095,6 +1100,116 @@ PreparedQuery PrepareFreeMemoryQuery(ParsedQuery parsed_query, const bool in_exp
                        RWType::NONE};
 }
 
+TriggerEventType ToTriggerEventType(const TriggerQuery::EventType event_type) {
+  switch (event_type) {
+    case TriggerQuery::EventType::ANY:
+      return TriggerEventType::ANY;
+
+    case TriggerQuery::EventType::CREATE:
+      return TriggerEventType::CREATE;
+
+    case TriggerQuery::EventType::VERTEX_CREATE:
+      return TriggerEventType::VERTEX_CREATE;
+
+    case TriggerQuery::EventType::EDGE_CREATE:
+      return TriggerEventType::EDGE_CREATE;
+
+    case TriggerQuery::EventType::DELETE:
+      return TriggerEventType::DELETE;
+
+    case TriggerQuery::EventType::VERTEX_DELETE:
+      return TriggerEventType::VERTEX_DELETE;
+
+    case TriggerQuery::EventType::EDGE_DELETE:
+      return TriggerEventType::EDGE_DELETE;
+
+    case TriggerQuery::EventType::UPDATE:
+      return TriggerEventType::UPDATE;
+
+    case TriggerQuery::EventType::VERTEX_UPDATE:
+      return TriggerEventType::VERTEX_UPDATE;
+
+    case TriggerQuery::EventType::EDGE_UPDATE:
+      return TriggerEventType::EDGE_UPDATE;
+  }
+}
+
+Callback CreateTrigger(TriggerQuery *trigger_query,
+                       const std::map<std::string, storage::PropertyValue> &user_parameters,
+                       InterpreterContext *interpreter_context, DbAccessor *dba) {
+  return {{}, [trigger_query, interpreter_context, dba, &user_parameters]() -> std::vector<std::vector<TypedValue>> {
+            interpreter_context->trigger_store->AddTrigger(
+                trigger_query->trigger_name_, trigger_query->statement_, user_parameters,
+                ToTriggerEventType(trigger_query->event_type_),
+                trigger_query->before_commit_ ? TriggerPhase::BEFORE_COMMIT : TriggerPhase::AFTER_COMMIT,
+                &interpreter_context->ast_cache, dba, &interpreter_context->antlr_lock);
+            return {};
+          }};
+}
+
+Callback DropTrigger(TriggerQuery *trigger_query, InterpreterContext *interpreter_context) {
+  return {{}, [trigger_query, interpreter_context]() -> std::vector<std::vector<TypedValue>> {
+            interpreter_context->trigger_store->DropTrigger(trigger_query->trigger_name_);
+            return {};
+          }};
+}
+
+Callback ShowTriggers(InterpreterContext *interpreter_context) {
+  return {{"trigger name", "statement", "event type", "phase"}, [interpreter_context] {
+            std::vector<std::vector<TypedValue>> results;
+            auto trigger_infos = interpreter_context->trigger_store->GetTriggerInfo();
+            results.reserve(trigger_infos.size());
+            for (auto &trigger_info : trigger_infos) {
+              std::vector<TypedValue> typed_trigger_info;
+              typed_trigger_info.reserve(4);
+              typed_trigger_info.emplace_back(std::move(trigger_info.name));
+              typed_trigger_info.emplace_back(std::move(trigger_info.statement));
+              typed_trigger_info.emplace_back(TriggerEventTypeToString(trigger_info.event_type));
+              typed_trigger_info.emplace_back(trigger_info.phase == TriggerPhase::BEFORE_COMMIT ? "BEFORE COMMIT"
+                                                                                                : "AFTER COMMIT");
+              results.push_back(std::move(typed_trigger_info));
+            }
+
+            return results;
+          }};
+}
+
+PreparedQuery PrepareTriggerQuery(ParsedQuery parsed_query, const bool in_explicit_transaction,
+                                  InterpreterContext *interpreter_context, DbAccessor *dba,
+                                  const std::map<std::string, storage::PropertyValue> &user_parameters) {
+  if (in_explicit_transaction) {
+    throw TriggerModificationInMulticommandTxException();
+  }
+
+  auto *trigger_query = utils::Downcast<TriggerQuery>(parsed_query.query);
+  MG_ASSERT(trigger_query);
+
+  auto callback = [trigger_query, interpreter_context, dba, &user_parameters] {
+    switch (trigger_query->action_) {
+      case TriggerQuery::Action::CREATE_TRIGGER:
+        return CreateTrigger(trigger_query, user_parameters, interpreter_context, dba);
+      case TriggerQuery::Action::DROP_TRIGGER:
+        return DropTrigger(trigger_query, interpreter_context);
+      case TriggerQuery::Action::SHOW_TRIGGERS:
+        return ShowTriggers(interpreter_context);
+    }
+  }();
+
+  auto results = callback.fn();
+
+  return PreparedQuery{std::move(callback.header), std::move(parsed_query.required_privileges),
+                       [pull_plan = std::make_shared<PullPlanVector>(std::move(results))](
+                           AnyStream *stream, std::optional<int> n) -> std::optional<QueryHandlerResult> {
+                         if (pull_plan->Pull(stream, n)) {
+                           return QueryHandlerResult::COMMIT;
+                         }
+                         return std::nullopt;
+                       },
+                       RWType::NONE};
+  // False positive report for the std::make_shared above
+  // NOLINTNEXTLINE(clang-analyzer-cplusplus.NewDeleteLeaks)
+}
+
 PreparedQuery PrepareInfoQuery(ParsedQuery parsed_query, bool in_explicit_transaction,
                                std::map<std::string, TypedValue> *summary, InterpreterContext *interpreter_context,
                                storage::Storage *db, utils::MemoryResource *execution_memory) {
@@ -1383,13 +1498,12 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
     // Some queries require an active transaction in order to be prepared.
     if (!in_explicit_transaction_ &&
         (utils::Downcast<CypherQuery>(parsed_query.query) || utils::Downcast<ExplainQuery>(parsed_query.query) ||
-         utils::Downcast<ProfileQuery>(parsed_query.query) || utils::Downcast<DumpQuery>(parsed_query.query))) {
+         utils::Downcast<ProfileQuery>(parsed_query.query) || utils::Downcast<DumpQuery>(parsed_query.query) ||
+         utils::Downcast<TriggerQuery>(parsed_query.query))) {
       db_accessor_ = std::make_unique<storage::Storage::Accessor>(interpreter_context_->db->Access());
       execution_db_accessor_.emplace(db_accessor_.get());
 
-      if (utils::Downcast<CypherQuery>(parsed_query.query) &&
-          (interpreter_context_->before_commit_triggers.size() > 0 ||
-           interpreter_context_->after_commit_triggers.size() > 0)) {
+      if (utils::Downcast<CypherQuery>(parsed_query.query) && interpreter_context_->trigger_store->HasTriggers()) {
         trigger_context_collector_.emplace();
       }
     }
@@ -1434,6 +1548,9 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
                                             &*execution_db_accessor_);
     } else if (utils::Downcast<FreeMemoryQuery>(parsed_query.query)) {
       prepared_query = PrepareFreeMemoryQuery(std::move(parsed_query), in_explicit_transaction_, interpreter_context_);
+    } else if (utils::Downcast<TriggerQuery>(parsed_query.query)) {
+      prepared_query = PrepareTriggerQuery(std::move(parsed_query), in_explicit_transaction_, interpreter_context_,
+                                           &*execution_db_accessor_, params);
     } else {
       LOG_FATAL("Should not get here -- unknown query type!");
     }
@@ -1488,7 +1605,7 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
                       interpreter_context->execution_timeout_sec, &interpreter_context->is_shutting_down,
                       trigger_context);
     } catch (const utils::BasicException &exception) {
-      spdlog::warn("Trigger '{}' failed with exception:\n{}", trigger.name(), exception.what());
+      spdlog::warn("Trigger '{}' failed with exception:\n{}", trigger.Name(), exception.what());
       db_accessor.Abort();
       continue;
     }
@@ -1501,7 +1618,7 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
           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(),
+          spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})", trigger.Name(),
                        label_name, property_name);
           break;
         }
@@ -1510,7 +1627,7 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
           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(),
+          spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})", trigger.Name(),
                        label_name, property_names_stream.str());
           break;
         }
@@ -1535,7 +1652,7 @@ void Interpreter::Commit() {
 
   if (trigger_context) {
     // Run the triggers
-    for (const auto &trigger : interpreter_context_->before_commit_triggers.access()) {
+    for (const auto &trigger : interpreter_context_->trigger_store->BeforeCommitTriggers().access()) {
       utils::MonotonicBufferResource execution_memory{kExecutionMemoryBlockSize};
       AdvanceCommand();
       try {
@@ -1544,7 +1661,7 @@ void Interpreter::Commit() {
                         *trigger_context);
       } catch (const utils::BasicException &e) {
         throw utils::BasicException(
-            fmt::format("Trigger '{}' caused the transaction to fail.\nException: {}", trigger.name(), e.what()));
+            fmt::format("Trigger '{}' caused the transaction to fail.\nException: {}", trigger.Name(), e.what()));
       }
     }
     SPDLOG_DEBUG("Finished executing before commit triggers");
@@ -1581,11 +1698,11 @@ void Interpreter::Commit() {
     }
   }
 
-  if (trigger_context && interpreter_context_->after_commit_triggers.size() > 0) {
+  if (trigger_context && interpreter_context_->trigger_store->AfterCommitTriggers().size() > 0) {
     background_thread_.AddTask([trigger_context = std::move(*trigger_context),
                                 interpreter_context = this->interpreter_context_,
                                 user_transaction = std::shared_ptr(std::move(db_accessor_))]() mutable {
-      RunTriggersIndividually(interpreter_context->after_commit_triggers, interpreter_context,
+      RunTriggersIndividually(interpreter_context->trigger_store->AfterCommitTriggers(), interpreter_context,
                               std::move(trigger_context));
       user_transaction->FinalizeTransaction();
       SPDLOG_DEBUG("Finished executing after commit triggers");  // NOLINT(bugprone-lambda-function-name)
diff --git a/src/query/interpreter.hpp b/src/query/interpreter.hpp
index 8f42c430c..bd76731db 100644
--- a/src/query/interpreter.hpp
+++ b/src/query/interpreter.hpp
@@ -147,7 +147,7 @@ struct PreparedQuery {
  * been passed to an `Interpreter` instance.
  */
 struct InterpreterContext {
-  explicit InterpreterContext(storage::Storage *db) : db(db) {}
+  explicit InterpreterContext(storage::Storage *db, const std::filesystem::path &data_directory);
 
   storage::Storage *db;
 
@@ -168,9 +168,7 @@ struct InterpreterContext {
   utils::SkipList<QueryCacheEntry> ast_cache;
   utils::SkipList<PlanCacheEntry> plan_cache;
 
-  // use a thread safe container
-  utils::SkipList<Trigger> before_commit_triggers;
-  utils::SkipList<Trigger> after_commit_triggers;
+  std::optional<TriggerStore> trigger_store;
 };
 
 /// Function that is used to tell all active interpreters that they should stop
diff --git a/src/query/serialization/property_value.cpp b/src/query/serialization/property_value.cpp
new file mode 100644
index 000000000..262ff3cf0
--- /dev/null
+++ b/src/query/serialization/property_value.cpp
@@ -0,0 +1,94 @@
+#include "query/serialization/property_value.hpp"
+#include "storage/v2/property_value.hpp"
+#include "utils/logging.hpp"
+
+namespace query::serialization {
+
+nlohmann::json SerializePropertyValue(const storage::PropertyValue &property_value) {
+  using Type = storage::PropertyValue::Type;
+  switch (property_value.type()) {
+    case Type::Null:
+      return {};
+    case Type::Bool:
+      return property_value.ValueBool();
+    case Type::Int:
+      return property_value.ValueInt();
+    case Type::Double:
+      return property_value.ValueDouble();
+    case Type::String:
+      return property_value.ValueString();
+    case Type::List:
+      return SerializePropertyValueVector(property_value.ValueList());
+    case Type::Map:
+      return SerializePropertyValueMap(property_value.ValueMap());
+  }
+}
+
+nlohmann::json SerializePropertyValueVector(const std::vector<storage::PropertyValue> &values) {
+  nlohmann::json array = nlohmann::json::array();
+  for (const auto &value : values) {
+    array.push_back(SerializePropertyValue(value));
+  }
+  return array;
+}
+
+nlohmann::json SerializePropertyValueMap(const std::map<std::string, storage::PropertyValue> &parameters) {
+  nlohmann::json data = nlohmann::json::object();
+
+  for (const auto &[key, value] : parameters) {
+    data[key] = SerializePropertyValue(value);
+  }
+
+  return data;
+};
+
+storage::PropertyValue DeserializePropertyValue(const nlohmann::json &data) {
+  if (data.is_null()) {
+    return storage::PropertyValue();
+  }
+
+  if (data.is_boolean()) {
+    return storage::PropertyValue(data.get<bool>());
+  }
+
+  if (data.is_number_integer()) {
+    return storage::PropertyValue(data.get<int64_t>());
+  }
+
+  if (data.is_number_float()) {
+    return storage::PropertyValue(data.get<double>());
+  }
+
+  if (data.is_string()) {
+    return storage::PropertyValue(data.get<std::string>());
+  }
+
+  if (data.is_array()) {
+    return storage::PropertyValue(DeserializePropertyValueList(data));
+  }
+
+  MG_ASSERT(data.is_object(), "Unknown type found in the trigger storage");
+  return storage::PropertyValue(DeserializePropertyValueMap(data));
+}
+
+std::vector<storage::PropertyValue> DeserializePropertyValueList(const nlohmann::json::array_t &data) {
+  std::vector<storage::PropertyValue> property_values;
+  property_values.reserve(data.size());
+  for (const auto &value : data) {
+    property_values.emplace_back(DeserializePropertyValue(value));
+  }
+
+  return property_values;
+}
+
+std::map<std::string, storage::PropertyValue> DeserializePropertyValueMap(const nlohmann::json::object_t &data) {
+  std::map<std::string, storage::PropertyValue> property_values;
+
+  for (const auto &[key, value] : data) {
+    property_values.emplace(key, DeserializePropertyValue(value));
+  }
+
+  return property_values;
+}
+
+}  // namespace query::serialization
diff --git a/src/query/serialization/property_value.hpp b/src/query/serialization/property_value.hpp
new file mode 100644
index 000000000..2ea5892c2
--- /dev/null
+++ b/src/query/serialization/property_value.hpp
@@ -0,0 +1,21 @@
+#pragma once
+
+#include <json/json.hpp>
+
+#include "storage/v2/property_value.hpp"
+
+namespace query::serialization {
+
+nlohmann::json SerializePropertyValue(const storage::PropertyValue &property_value);
+
+nlohmann::json SerializePropertyValueVector(const std::vector<storage::PropertyValue> &values);
+
+nlohmann::json SerializePropertyValueMap(const std::map<std::string, storage::PropertyValue> &parameters);
+
+storage::PropertyValue DeserializePropertyValue(const nlohmann::json &data);
+
+std::vector<storage::PropertyValue> DeserializePropertyValueList(const nlohmann::json::array_t &data);
+
+std::map<std::string, storage::PropertyValue> DeserializePropertyValueMap(const nlohmann::json::object_t &data);
+
+}  // namespace query::serialization
diff --git a/src/query/trigger.cpp b/src/query/trigger.cpp
index 2a1ee4668..be46d19e8 100644
--- a/src/query/trigger.cpp
+++ b/src/query/trigger.cpp
@@ -1,3 +1,5 @@
+#include "query/trigger.hpp"
+
 #include <concepts>
 
 #include "query/context.hpp"
@@ -5,12 +7,12 @@
 #include "query/db_accessor.hpp"
 #include "query/frontend/ast/ast.hpp"
 #include "query/interpret/frame.hpp"
-#include "query/trigger.hpp"
+#include "query/serialization/property_value.hpp"
 #include "query/typed_value.hpp"
+#include "storage/v2/property_value.hpp"
 #include "utils/memory.hpp"
 
 namespace query {
-
 namespace {
 
 auto IdentifierString(const TriggerIdentifierTag tag) noexcept {
@@ -250,9 +252,7 @@ template <WithEmpty... TContainer>
 bool AnyContainsValue(const TContainer &...value_containers) {
   return (!value_containers.empty() || ...);
 }
-
 }  // namespace
-
 namespace detail {
 bool SetVertexLabel::IsValid() const { return object.IsVisible(storage::View::OLD); }
 
@@ -267,6 +267,40 @@ std::map<std::string, TypedValue> RemovedVertexLabel::ToMap(DbAccessor *dba) con
 }
 }  // namespace detail
 
+const char *TriggerEventTypeToString(const TriggerEventType event_type) {
+  switch (event_type) {
+    case TriggerEventType::ANY:
+      return "ANY";
+
+    case TriggerEventType::CREATE:
+      return "CREATE";
+
+    case TriggerEventType::VERTEX_CREATE:
+      return "() CREATE";
+
+    case TriggerEventType::EDGE_CREATE:
+      return "--> CREATE";
+
+    case TriggerEventType::DELETE:
+      return "DELETE";
+
+    case TriggerEventType::VERTEX_DELETE:
+      return "() DELETE";
+
+    case TriggerEventType::EDGE_DELETE:
+      return "--> DELETE";
+
+    case TriggerEventType::UPDATE:
+      return "UPDATE";
+
+    case TriggerEventType::VERTEX_UPDATE:
+      return "() UPDATE";
+
+    case TriggerEventType::EDGE_UPDATE:
+      return "--> UPDATE";
+  }
+}
+
 void TriggerContext::AdaptForAccessor(DbAccessor *accessor) {
   {
     // adapt created_vertices_
@@ -498,10 +532,12 @@ TriggerContextCollector::LabelChangesLists TriggerContextCollector::LabelMapToLi
   return {std::move(set_vertex_labels), std::move(removed_vertex_labels)};
 }
 
-Trigger::Trigger(std::string name, const std::string &query, utils::SkipList<QueryCacheEntry> *query_cache,
-                 DbAccessor *db_accessor, utils::SpinLock *antlr_lock, const TriggerEventType event_type)
-    : name_(std::move(name)),
-      parsed_statements_{ParseQuery(query, {}, query_cache, antlr_lock)},
+Trigger::Trigger(std::string name, const std::string &query,
+                 const std::map<std::string, storage::PropertyValue> &user_parameters,
+                 const TriggerEventType event_type, utils::SkipList<QueryCacheEntry> *query_cache,
+                 DbAccessor *db_accessor, utils::SpinLock *antlr_lock)
+    : name_{std::move(name)},
+      parsed_statements_{ParseQuery(query, user_parameters, query_cache, antlr_lock)},
       event_type_{event_type} {
   // We check immediately if the query is valid by trying to create a plan.
   GetPlan(db_accessor);
@@ -512,7 +548,7 @@ Trigger::TriggerPlan::TriggerPlan(std::unique_ptr<LogicalPlan> logical_plan, std
 
 std::shared_ptr<Trigger::TriggerPlan> Trigger::GetPlan(DbAccessor *db_accessor) const {
   std::lock_guard plan_guard{plan_lock_};
-  if (trigger_plan_ && !trigger_plan_->cached_plan.IsExpired()) {
+  if (parsed_statements_.is_cacheable && trigger_plan_ && !trigger_plan_->cached_plan.IsExpired()) {
     return trigger_plan_;
   }
 
@@ -592,4 +628,150 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
 
   cursor->Shutdown();
 }
+
+namespace {
+constexpr uint64_t kVersion{1};
+}  // namespace
+
+TriggerStore::TriggerStore(std::filesystem::path directory, utils::SkipList<QueryCacheEntry> *query_cache,
+                           DbAccessor *db_accessor, utils::SpinLock *antlr_lock)
+    : storage_{std::move(directory)} {
+  spdlog::info("Loading triggers...");
+
+  for (const auto &[trigger_name, trigger_data] : storage_) {
+    spdlog::debug("Loading trigger '{}'", trigger_name);
+    auto json_trigger_data = nlohmann::json::parse(trigger_data);
+
+    if (!json_trigger_data["version"].is_number_unsigned()) {
+      spdlog::debug("Invalid state of the trigger data.");
+      continue;
+    }
+    if (json_trigger_data["version"] != kVersion) {
+      spdlog::debug("Invalid version of the trigger data. Got {}");
+      continue;
+    }
+
+    if (!json_trigger_data["statement"].is_string()) {
+      spdlog::debug("Invalid state of the trigger data");
+      continue;
+    }
+    auto statement = json_trigger_data["statement"].get<std::string>();
+
+    if (!json_trigger_data["phase"].is_number_integer()) {
+      spdlog::debug("Invalid state of the trigger data");
+      continue;
+    }
+    const auto phase = json_trigger_data["phase"].get<TriggerPhase>();
+
+    if (!json_trigger_data["event_type"].is_number_integer()) {
+      spdlog::debug("Invalid state of the trigger data");
+      continue;
+    }
+    const auto event_type = json_trigger_data["event_type"].get<TriggerEventType>();
+
+    if (!json_trigger_data["user_parameters"].is_object()) {
+      spdlog::debug("Invalid state of the trigger data");
+      continue;
+    }
+    const auto user_parameters = serialization::DeserializePropertyValueMap(json_trigger_data["user_parameters"]);
+
+    std::optional<Trigger> trigger;
+    try {
+      trigger.emplace(trigger_name, statement, user_parameters, event_type, query_cache, db_accessor, antlr_lock);
+    } catch (const utils::BasicException &e) {
+      spdlog::debug("Failed to create a trigger '{}' because: {}", trigger_name, e.what());
+      continue;
+    }
+
+    auto triggers_acc =
+        phase == TriggerPhase::BEFORE_COMMIT ? before_commit_triggers_.access() : after_commit_triggers_.access();
+    triggers_acc.insert(std::move(*trigger));
+
+    spdlog::debug("Trigger loaded successfully!");
+  }
+}
+
+void TriggerStore::AddTrigger(const std::string &name, const std::string &query,
+                              const std::map<std::string, storage::PropertyValue> &user_parameters,
+                              TriggerEventType event_type, TriggerPhase phase,
+                              utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
+                              utils::SpinLock *antlr_lock) {
+  std::unique_lock store_guard{store_lock_};
+  if (storage_.Get(name)) {
+    throw utils::BasicException("Trigger with the same name already exists.");
+  }
+
+  std::optional<Trigger> trigger;
+  try {
+    trigger.emplace(name, query, user_parameters, event_type, query_cache, db_accessor, antlr_lock);
+  } catch (const utils::BasicException &e) {
+    const auto identifiers = GetPredefinedIdentifiers(event_type);
+    std::stringstream identifier_names_stream;
+    utils::PrintIterable(identifier_names_stream, identifiers, ", ",
+                         [](auto &stream, const auto &identifier) { stream << identifier.first.name_; });
+
+    throw utils::BasicException(
+        "Failed creating the trigger.\nError message: '{}'\nThe error was mostly likely generated because of the wrong "
+        "statement that this trigger executes.\nMake sure all predefined variables used are present for the specified "
+        "event.\nAllowed variables for event '{}' are: {}",
+        e.what(), TriggerEventTypeToString(event_type), identifier_names_stream.str());
+  }
+
+  nlohmann::json data = nlohmann::json::object();
+  data["statement"] = query;
+  data["user_parameters"] = serialization::SerializePropertyValueMap(user_parameters);
+  data["event_type"] = event_type;
+  data["phase"] = phase;
+  data["version"] = kVersion;
+  storage_.Put(name, data.dump());
+  store_guard.unlock();
+
+  auto triggers_acc =
+      phase == TriggerPhase::BEFORE_COMMIT ? before_commit_triggers_.access() : after_commit_triggers_.access();
+  triggers_acc.insert(std::move(*trigger));
+}
+
+void TriggerStore::DropTrigger(const std::string &name) {
+  std::unique_lock store_guard{store_lock_};
+  const auto maybe_trigger_data = storage_.Get(name);
+  if (!maybe_trigger_data) {
+    throw utils::BasicException("Trigger with name '{}' doesn't exist", name);
+  }
+
+  nlohmann::json data;
+  try {
+    data = nlohmann::json::parse(*maybe_trigger_data);
+  } catch (const nlohmann::json::parse_error &e) {
+    throw utils::BasicException("Couldn't load trigger data!");
+  }
+
+  if (!data.is_object()) {
+    throw utils::BasicException("Couldn't load trigger data!");
+  }
+
+  if (!data["phase"].is_number_integer()) {
+    throw utils::BasicException("Invalid type loaded inside the trigger data!");
+  }
+
+  auto triggers_acc =
+      data["phase"] == TriggerPhase::BEFORE_COMMIT ? before_commit_triggers_.access() : after_commit_triggers_.access();
+  triggers_acc.remove(name);
+  storage_.Delete(name);
+}
+
+std::vector<TriggerStore::TriggerInfo> TriggerStore::GetTriggerInfo() const {
+  std::vector<TriggerInfo> info;
+  info.reserve(before_commit_triggers_.size() + after_commit_triggers_.size());
+
+  const auto add_info = [&](const utils::SkipList<Trigger> &trigger_list, const TriggerPhase phase) {
+    for (const auto &trigger : trigger_list.access()) {
+      info.push_back({trigger.Name(), trigger.OriginalStatement(), trigger.EventType(), phase});
+    }
+  };
+
+  add_info(before_commit_triggers_, TriggerPhase::BEFORE_COMMIT);
+  add_info(after_commit_triggers_, TriggerPhase::AFTER_COMMIT);
+
+  return info;
+}
 }  // namespace query
diff --git a/src/query/trigger.hpp b/src/query/trigger.hpp
index 51119885e..a3e296aa0 100644
--- a/src/query/trigger.hpp
+++ b/src/query/trigger.hpp
@@ -6,9 +6,11 @@
 #include <type_traits>
 #include <utility>
 
+#include "kvstore/kvstore.hpp"
 #include "query/cypher_query_interpreter.hpp"
 #include "query/frontend/ast/ast.hpp"
 #include "query/typed_value.hpp"
+#include "storage/v2/property_value.hpp"
 #include "utils/concepts.hpp"
 #include "utils/fnv.hpp"
 
@@ -77,7 +79,7 @@ struct RemovedObjectProperty {
       : object{object}, key{key}, old_value{std::move(old_value)} {}
 
   std::map<std::string, TypedValue> ToMap(DbAccessor *dba) const {
-    return {{ObjectString<TAccessor>(), TypedValue{object}},
+    return {{detail::ObjectString<TAccessor>(), TypedValue{object}},
             {"key", TypedValue{dba->PropertyToName(key)}},
             {"old", old_value}};
   }
@@ -143,6 +145,8 @@ enum class TriggerEventType : uint8_t {
   UPDATE
 };
 
+const char *TriggerEventTypeToString(TriggerEventType event_type);
+
 static_assert(std::is_trivially_copy_constructible_v<VertexAccessor>,
               "VertexAccessor is not trivially copy constructible, move it where possible and remove this assert");
 static_assert(std::is_trivially_copy_constructible_v<EdgeAccessor>,
@@ -356,9 +360,9 @@ class TriggerContextCollector {
 };
 
 struct Trigger {
-  explicit Trigger(std::string name, const std::string &query, utils::SkipList<QueryCacheEntry> *query_cache,
-                   DbAccessor *db_accessor, utils::SpinLock *antlr_lock,
-                   TriggerEventType event_type = TriggerEventType::ANY);
+  explicit Trigger(std::string name, const std::string &query,
+                   const std::map<std::string, storage::PropertyValue> &user_parameters, TriggerEventType event_type,
+                   utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor, utils::SpinLock *antlr_lock);
 
   void Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution_memory, double tsc_frequency,
                double max_execution_time_sec, std::atomic<bool> *is_shutting_down, const TriggerContext &context) const;
@@ -370,7 +374,9 @@ struct Trigger {
   // NOLINTNEXTLINE (modernize-use-nullptr)
   bool operator<(const std::string &other) const { return name_ < other; }
 
-  const auto &name() const noexcept { return name_; }
+  const auto &Name() const noexcept { return name_; }
+  const auto &OriginalStatement() const noexcept { return parsed_statements_.query_string; }
+  auto EventType() const noexcept { return event_type_; }
 
  private:
   struct TriggerPlan {
@@ -391,4 +397,40 @@ struct Trigger {
   mutable utils::SpinLock plan_lock_;
   mutable std::shared_ptr<TriggerPlan> trigger_plan_;
 };
+
+enum class TriggerPhase : uint8_t { BEFORE_COMMIT, AFTER_COMMIT };
+
+struct TriggerStore {
+  explicit TriggerStore(std::filesystem::path directory, utils::SkipList<QueryCacheEntry> *query_cache,
+                        DbAccessor *db_accessor, utils::SpinLock *antlr_lock);
+
+  void AddTrigger(const std::string &name, const std::string &query,
+                  const std::map<std::string, storage::PropertyValue> &user_parameters, TriggerEventType event_type,
+                  TriggerPhase phase, utils::SkipList<QueryCacheEntry> *query_cache, DbAccessor *db_accessor,
+                  utils::SpinLock *antlr_lock);
+
+  void DropTrigger(const std::string &name);
+
+  struct TriggerInfo {
+    std::string name;
+    std::string statement;
+    TriggerEventType event_type;
+    TriggerPhase phase;
+  };
+
+  std::vector<TriggerInfo> GetTriggerInfo() const;
+
+  const auto &BeforeCommitTriggers() const noexcept { return before_commit_triggers_; }
+  const auto &AfterCommitTriggers() const noexcept { return after_commit_triggers_; }
+
+  bool HasTriggers() const noexcept { return before_commit_triggers_.size() > 0 || after_commit_triggers_.size() > 0; }
+
+ private:
+  utils::SpinLock store_lock_;
+  kvstore::KVStore storage_;
+
+  utils::SkipList<Trigger> before_commit_triggers_;
+  utils::SkipList<Trigger> after_commit_triggers_;
+};
+
 }  // namespace query
diff --git a/tests/benchmark/expansion.cpp b/tests/benchmark/expansion.cpp
index 9bfe2607b..3317a3158 100644
--- a/tests/benchmark/expansion.cpp
+++ b/tests/benchmark/expansion.cpp
@@ -11,6 +11,7 @@ class ExpansionBenchFixture : public benchmark::Fixture {
   std::optional<storage::Storage> db;
   std::optional<query::InterpreterContext> interpreter_context;
   std::optional<query::Interpreter> interpreter;
+  std::filesystem::path data_directory{std::filesystem::temp_directory_path() / "expansion-benchmark"};
 
   void SetUp(const benchmark::State &state) override {
     db.emplace();
@@ -34,7 +35,7 @@ class ExpansionBenchFixture : public benchmark::Fixture {
 
     MG_ASSERT(db->CreateIndex(label));
 
-    interpreter_context.emplace(&*db);
+    interpreter_context.emplace(&*db, data_directory);
     interpreter.emplace(&*interpreter_context);
   }
 
@@ -42,6 +43,7 @@ class ExpansionBenchFixture : public benchmark::Fixture {
     interpreter = std::nullopt;
     interpreter_context = std::nullopt;
     db = std::nullopt;
+    std::filesystem::remove_all(data_directory);
   }
 };
 
diff --git a/tests/manual/single_query.cpp b/tests/manual/single_query.cpp
index e0deff027..ee60006ed 100644
--- a/tests/manual/single_query.cpp
+++ b/tests/manual/single_query.cpp
@@ -1,6 +1,7 @@
 #include "communication/result_stream_faker.hpp"
 #include "query/interpreter.hpp"
 #include "storage/v2/storage.hpp"
+#include "utils/on_scope_exit.hpp"
 
 int main(int argc, char *argv[]) {
   gflags::ParseCommandLineFlags(&argc, &argv, true);
@@ -12,7 +13,9 @@ int main(int argc, char *argv[]) {
   }
 
   storage::Storage db;
-  query::InterpreterContext interpreter_context{&db};
+  auto data_directory = std::filesystem::temp_directory_path() / "single_query_test";
+  utils::OnScopeExit([&data_directory] { std::filesystem::remove_all(data_directory); });
+  query::InterpreterContext interpreter_context{&db, data_directory};
   query::Interpreter interpreter{&interpreter_context};
 
   ResultStreamFaker stream(&db);
diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt
index 8f7c2bb67..1adda7f54 100644
--- a/tests/unit/CMakeLists.txt
+++ b/tests/unit/CMakeLists.txt
@@ -98,6 +98,9 @@ target_link_libraries(${test_prefix}query_pretty_print mg-query)
 add_unit_test(query_trigger.cpp)
 target_link_libraries(${test_prefix}query_trigger mg-query)
 
+add_unit_test(query_serialization_property_value.cpp)
+target_link_libraries(${test_prefix}query_serialization_property_value mg-query)
+
 # Test query/procedure
 add_unit_test(query_procedure_mgp_type.cpp)
 target_link_libraries(${test_prefix}query_procedure_mgp_type mg-query)
diff --git a/tests/unit/cypher_main_visitor.cpp b/tests/unit/cypher_main_visitor.cpp
index e734d1cab..9cb70fd99 100644
--- a/tests/unit/cypher_main_visitor.cpp
+++ b/tests/unit/cypher_main_visitor.cpp
@@ -25,6 +25,8 @@
 #include "query/frontend/stripped.hpp"
 #include "query/typed_value.hpp"
 
+#include "utils/string.hpp"
+
 namespace {
 
 using namespace query;
@@ -2061,6 +2063,8 @@ TEST_P(CypherMainVisitorTest, GrantPrivilege) {
                    {AuthQuery::Privilege::READ_FILE});
   check_auth_query(&ast_generator, "GRANT FREE_MEMORY TO user", AuthQuery::Action::GRANT_PRIVILEGE, "", "", "user", {},
                    {AuthQuery::Privilege::FREE_MEMORY});
+  check_auth_query(&ast_generator, "GRANT TRIGGER TO user", AuthQuery::Action::GRANT_PRIVILEGE, "", "", "user", {},
+                   {AuthQuery::Privilege::TRIGGER});
 }
 
 TEST_P(CypherMainVisitorTest, DenyPrivilege) {
@@ -3054,4 +3058,103 @@ TEST_P(CypherMainVisitorTest, MemoryLimit) {
     CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
   }
 }
+
+namespace {
+void TestInvalidQuery(const auto &query, Base &ast_generator) {
+  ASSERT_THROW(ast_generator.ParseQuery(query), SyntaxException);
+}
+}  // namespace
+
+TEST_P(CypherMainVisitorTest, DropTrigger) {
+  auto &ast_generator = *GetParam();
+
+  TestInvalidQuery("DROP TR", ast_generator);
+  TestInvalidQuery("DROP TRIGGER", ast_generator);
+
+  auto *parsed_query = dynamic_cast<TriggerQuery *>(ast_generator.ParseQuery("DROP TRIGGER trigger"));
+  EXPECT_EQ(parsed_query->action_, TriggerQuery::Action::DROP_TRIGGER);
+  EXPECT_EQ(parsed_query->trigger_name_, "trigger");
+}
+
+TEST_P(CypherMainVisitorTest, ShowTriggers) {
+  auto &ast_generator = *GetParam();
+
+  TestInvalidQuery("SHOW TR", ast_generator);
+  TestInvalidQuery("SHOW TRIGGER", ast_generator);
+
+  auto *parsed_query = dynamic_cast<TriggerQuery *>(ast_generator.ParseQuery("SHOW TRIGGERS"));
+  EXPECT_EQ(parsed_query->action_, TriggerQuery::Action::SHOW_TRIGGERS);
+}
+
+namespace {
+void ValidateCreateQuery(Base &ast_generator, const auto &query, const auto &trigger_name,
+                         const query::TriggerQuery::EventType event_type, const auto &phase, const auto &statement) {
+  auto *parsed_query = dynamic_cast<TriggerQuery *>(ast_generator.ParseQuery(query));
+  EXPECT_EQ(parsed_query->action_, TriggerQuery::Action::CREATE_TRIGGER);
+  EXPECT_EQ(parsed_query->trigger_name_, trigger_name);
+  EXPECT_EQ(parsed_query->event_type_, event_type);
+  EXPECT_EQ(parsed_query->before_commit_, phase == "BEFORE");
+  EXPECT_EQ(parsed_query->statement_, statement);
+}
+}  // namespace
+
+TEST_P(CypherMainVisitorTest, CreateTriggers) {
+  auto &ast_generator = *GetParam();
+
+  TestInvalidQuery("CREATE TRIGGER", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON ", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON ()", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON -->", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON CREATE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON () CREATE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON --> CREATE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON DELETE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON () DELETE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON --> DELETE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON UPDATE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON () UPDATE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON --> UPDATE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON CREATE BEFORE", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON CREATE BEFORE COMMIT", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON CREATE AFTER", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON CREATE AFTER COMMIT", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON -> CREATE AFTER COMMIT EXECUTE a", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON ) CREATE AFTER COMMIT EXECUTE a", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON ( CREATE AFTER COMMIT EXECUTE a", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON CRETE AFTER COMMIT EXECUTE a", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON DELET AFTER COMMIT EXECUTE a", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON UPDTE AFTER COMMIT EXECUTE a", ast_generator);
+  TestInvalidQuery("CREATE TRIGGER trigger ON UPDATE COMMIT EXECUTE a", ast_generator);
+
+  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}};
+
+  std::array phases{"BEFORE", "AFTER"};
+
+  std::array statements{
+      "", "SOME SUPER\nSTATEMENT", "Statement with 12312321 3     ", "        Statement with 12312321 3     "
+
+  };
+
+  for (const auto &[event_string, event_type] : events) {
+    for (const auto &phase : phases) {
+      for (const auto &statement : statements) {
+        ValidateCreateQuery(ast_generator, fmt::format(query_template, event_string, phase, statement), "trigger",
+                            event_type, phase, utils::Trim(statement));
+      }
+    }
+  }
+}
+
 }  // namespace
diff --git a/tests/unit/interpreter.cpp b/tests/unit/interpreter.cpp
index 51444ed4f..194c4c815 100644
--- a/tests/unit/interpreter.cpp
+++ b/tests/unit/interpreter.cpp
@@ -1,4 +1,5 @@
 #include <cstdlib>
+#include <filesystem>
 
 #include "communication/bolt/v1/value.hpp"
 #include "communication/result_stream_faker.hpp"
@@ -32,7 +33,8 @@ auto ToEdgeList(const communication::bolt::Value &v) {
 class InterpreterTest : public ::testing::Test {
  protected:
   storage::Storage db_;
-  query::InterpreterContext interpreter_context_{&db_};
+  std::filesystem::path data_directory{std::filesystem::temp_directory_path() / "MG_tests_unit_interpreter"};
+  query::InterpreterContext interpreter_context_{&db_, data_directory};
   query::Interpreter interpreter_{&interpreter_context_};
 
   auto Prepare(const std::string &query, const std::map<std::string, storage::PropertyValue> &params = {}) {
diff --git a/tests/unit/main.cpp b/tests/unit/main.cpp
index 5b0fa17fb..cb05a4e3b 100644
--- a/tests/unit/main.cpp
+++ b/tests/unit/main.cpp
@@ -4,6 +4,6 @@
 int main(int argc, char **argv) {
   ::testing::InitGoogleTest(&argc, argv);
   logging::RedirectToStderr();
-  spdlog::set_level(spdlog::level::info);
+  spdlog::set_level(spdlog::level::warn);
   return RUN_ALL_TESTS();
 }
diff --git a/tests/unit/query_dump.cpp b/tests/unit/query_dump.cpp
index b565de5fe..ee19f8ddf 100644
--- a/tests/unit/query_dump.cpp
+++ b/tests/unit/query_dump.cpp
@@ -1,5 +1,6 @@
 #include <gtest/gtest.h>
 
+#include <filesystem>
 #include <map>
 #include <set>
 #include <vector>
@@ -187,7 +188,8 @@ DatabaseState GetState(storage::Storage *db) {
 }
 
 auto Execute(storage::Storage *db, const std::string &query) {
-  query::InterpreterContext context(db);
+  auto data_directory = std::filesystem::temp_directory_path() / "MG_tests_unit_query_dump";
+  query::InterpreterContext context(db, data_directory);
   query::Interpreter interpreter(&context);
   ResultStreamFaker stream(db);
 
@@ -700,7 +702,8 @@ TEST(DumpTest, ExecuteDumpDatabase) {
 
 class StatefulInterpreter {
  public:
-  explicit StatefulInterpreter(storage::Storage *db) : db_(db), context_(db_), interpreter_(&context_) {}
+  explicit StatefulInterpreter(storage::Storage *db)
+      : db_(db), context_(db_, data_directory_), interpreter_(&context_) {}
 
   auto Execute(const std::string &query) {
     ResultStreamFaker stream(db_);
@@ -714,11 +717,16 @@ class StatefulInterpreter {
   }
 
  private:
+  static const std::filesystem::path data_directory_;
+
   storage::Storage *db_;
   query::InterpreterContext context_;
   query::Interpreter interpreter_;
 };
 
+const std::filesystem::path StatefulInterpreter::data_directory_{std::filesystem::temp_directory_path() /
+                                                                 "MG_tests_unit_query_dump_stateful"};
+
 // NOLINTNEXTLINE(hicpp-special-member-functions)
 TEST(DumpTest, ExecuteDumpDatabaseInMulticommandTransaction) {
   storage::Storage db;
diff --git a/tests/unit/query_plan_edge_cases.cpp b/tests/unit/query_plan_edge_cases.cpp
index 71e558f17..fe3035240 100644
--- a/tests/unit/query_plan_edge_cases.cpp
+++ b/tests/unit/query_plan_edge_cases.cpp
@@ -2,6 +2,7 @@
 // that's not easily testable with single-phase testing. instead, for
 // easy testing and latter readability they are tested end-to-end.
 
+#include <filesystem>
 #include <optional>
 
 #include "gmock/gmock.h"
@@ -19,9 +20,11 @@ class QueryExecution : public testing::Test {
   std::optional<query::InterpreterContext> interpreter_context_;
   std::optional<query::Interpreter> interpreter_;
 
+  std::filesystem::path data_directory{std::filesystem::temp_directory_path() / "MG_tests_unit_query_plan_edge_cases"};
+
   void SetUp() {
     db_.emplace();
-    interpreter_context_.emplace(&*db_);
+    interpreter_context_.emplace(&*db_, data_directory);
     interpreter_.emplace(&*interpreter_context_);
   }
 
diff --git a/tests/unit/query_required_privileges.cpp b/tests/unit/query_required_privileges.cpp
index a5afc010c..09fee4c0d 100644
--- a/tests/unit/query_required_privileges.cpp
+++ b/tests/unit/query_required_privileges.cpp
@@ -149,3 +149,8 @@ TEST_F(TestPrivilegeExtractor, FreeMemoryQuery) {
   auto *query = storage.Create<FreeMemoryQuery>();
   EXPECT_THAT(GetRequiredPrivileges(query), UnorderedElementsAre(AuthQuery::Privilege::FREE_MEMORY));
 }
+
+TEST_F(TestPrivilegeExtractor, TriggerQuery) {
+  auto *query = storage.Create<TriggerQuery>();
+  EXPECT_THAT(GetRequiredPrivileges(query), UnorderedElementsAre(AuthQuery::Privilege::TRIGGER));
+}
diff --git a/tests/unit/query_serialization_property_value.cpp b/tests/unit/query_serialization_property_value.cpp
new file mode 100644
index 000000000..569d13ec2
--- /dev/null
+++ b/tests/unit/query_serialization_property_value.cpp
@@ -0,0 +1,85 @@
+#include <gtest/gtest.h>
+
+#include "query/serialization/property_value.hpp"
+#include "utils/logging.hpp"
+
+namespace {
+void ExpectPropEq(const storage::PropertyValue &a, const storage::PropertyValue &b) {
+  ASSERT_EQ(a.type(), b.type());
+  ASSERT_EQ(a, b);
+}
+
+void CheckJsonConversion(const storage::PropertyValue &property_value) {
+  const auto json_string = query::serialization::SerializePropertyValue(property_value).dump();
+  const auto json_object = nlohmann::json::parse(json_string);
+  ExpectPropEq(property_value, query::serialization::DeserializePropertyValue(json_object));
+}
+
+}  // namespace
+
+TEST(PropertyValueSerializationTest, Null) { CheckJsonConversion(storage::PropertyValue{}); }
+
+TEST(PropertyValueSerializationTest, Bool) {
+  CheckJsonConversion(storage::PropertyValue{true});
+  CheckJsonConversion(storage::PropertyValue{false});
+}
+
+TEST(PropertyValueSerializationTest, Int) {
+  CheckJsonConversion(storage::PropertyValue{1});
+  CheckJsonConversion(storage::PropertyValue{100});
+}
+
+TEST(PropertyValueSerializationTest, Double) {
+  CheckJsonConversion(storage::PropertyValue{1.0});
+  CheckJsonConversion(storage::PropertyValue{2.321});
+}
+
+TEST(PropertyValueSerializationTest, String) {
+  CheckJsonConversion(storage::PropertyValue{"TestString"});
+  CheckJsonConversion(storage::PropertyValue{""});
+}
+
+namespace {
+
+std::vector<storage::PropertyValue> GetPropertyValueListWithBasicTypes() {
+  return {storage::PropertyValue{}, storage::PropertyValue{true}, storage::PropertyValue{"string"},
+          storage::PropertyValue{1}, storage::PropertyValue{1.0}};
+}
+
+std::map<std::string, storage::PropertyValue> GetPropertyValueMapWithBasicTypes() {
+  return {{"null", storage::PropertyValue{}},
+          {"bool", storage::PropertyValue{true}},
+          {"int", storage::PropertyValue{1}},
+          {"double", storage::PropertyValue{1.0}},
+          {"string", storage::PropertyValue{"string"}}};
+}
+
+}  // namespace
+
+TEST(PropertyValueSerializationTest, List) {
+  storage::PropertyValue list = storage::PropertyValue{GetPropertyValueListWithBasicTypes()};
+
+  SPDLOG_DEBUG("Basic list");
+  CheckJsonConversion(list);
+
+  SPDLOG_DEBUG("Nested list");
+  CheckJsonConversion(storage::PropertyValue{std::vector<storage::PropertyValue>{list, list}});
+
+  SPDLOG_DEBUG("List with map");
+  list.ValueList().emplace_back(GetPropertyValueMapWithBasicTypes());
+  CheckJsonConversion(list);
+}
+
+TEST(PropertyValueSerializationTest, Map) {
+  auto map = GetPropertyValueMapWithBasicTypes();
+  SPDLOG_DEBUG("Basic map");
+  CheckJsonConversion(storage::PropertyValue{map});
+
+  SPDLOG_DEBUG("Nested map");
+  map.emplace("map", storage::PropertyValue{map});
+  CheckJsonConversion(storage::PropertyValue{map});
+
+  SPDLOG_DEBUG("Map with list");
+  map.emplace("list", storage::PropertyValue{GetPropertyValueListWithBasicTypes()});
+  CheckJsonConversion(storage::PropertyValue{map});
+}
diff --git a/tests/unit/query_trigger.cpp b/tests/unit/query_trigger.cpp
index 44936ef60..a5bb3ad91 100644
--- a/tests/unit/query_trigger.cpp
+++ b/tests/unit/query_trigger.cpp
@@ -551,3 +551,190 @@ TEST_F(TriggerContextTest, GlobalLabelChange) {
                                {"label", query::TypedValue{"LABEL"}}}});
   }
 }
+
+class TriggerStoreTest : public ::testing::Test {
+ protected:
+  const std::filesystem::path testing_directory{std::filesystem::temp_directory_path() / "MG_test_unit_query_trigger"};
+
+  void SetUp() override {
+    Clear();
+
+    storage_accessor.emplace(storage.Access());
+    dba.emplace(&*storage_accessor);
+  }
+
+  void TearDown() override {
+    Clear();
+
+    dba.reset();
+    storage_accessor.reset();
+  }
+
+  std::optional<query::DbAccessor> dba;
+
+  utils::SkipList<query::QueryCacheEntry> ast_cache;
+  utils::SpinLock antlr_lock;
+
+ private:
+  void Clear() {
+    if (!std::filesystem::exists(testing_directory)) return;
+    std::filesystem::remove_all(testing_directory);
+  }
+
+  storage::Storage storage;
+  std::optional<storage::Storage::Accessor> storage_accessor;
+};
+
+TEST_F(TriggerStoreTest, Load) {
+  std::optional<query::TriggerStore> store;
+
+  store.emplace(testing_directory, &ast_cache, &*dba, &antlr_lock);
+
+  const auto check_empty = [&] {
+    ASSERT_EQ(store->GetTriggerInfo().size(), 0);
+    ASSERT_EQ(store->BeforeCommitTriggers().size(), 0);
+    ASSERT_EQ(store->AfterCommitTriggers().size(), 0);
+  };
+
+  check_empty();
+
+  const auto *trigger_name_before = "trigger";
+  const auto *trigger_name_after = "trigger_after";
+  const auto *trigger_statement = "RETURN $parameter";
+  const auto event_type = query::TriggerEventType::VERTEX_CREATE;
+  store->AddTrigger(trigger_name_before, trigger_statement,
+                    std::map<std::string, storage::PropertyValue>{{"parameter", storage::PropertyValue{1}}}, event_type,
+                    query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock);
+  store->AddTrigger(trigger_name_after, trigger_statement,
+                    std::map<std::string, storage::PropertyValue>{{"parameter", storage::PropertyValue{"value"}}},
+                    event_type, query::TriggerPhase::AFTER_COMMIT, &ast_cache, &*dba, &antlr_lock);
+
+  const auto check_triggers = [&] {
+    ASSERT_EQ(store->GetTriggerInfo().size(), 2);
+
+    const auto verify_trigger = [&](const auto &trigger, const auto &name) {
+      ASSERT_EQ(trigger.Name(), name);
+      ASSERT_EQ(trigger.OriginalStatement(), trigger_statement);
+      ASSERT_EQ(trigger.EventType(), event_type);
+    };
+
+    const auto before_commit_triggers = store->BeforeCommitTriggers().access();
+    ASSERT_EQ(before_commit_triggers.size(), 1);
+    for (const auto &trigger : before_commit_triggers) {
+      verify_trigger(trigger, trigger_name_before);
+    }
+
+    const auto after_commit_triggers = store->AfterCommitTriggers().access();
+    ASSERT_EQ(after_commit_triggers.size(), 1);
+    for (const auto &trigger : after_commit_triggers) {
+      verify_trigger(trigger, trigger_name_after);
+    }
+  };
+
+  check_triggers();
+
+  // recreate trigger store, this should reload everything from the disk
+  store.emplace(testing_directory, &ast_cache, &*dba, &antlr_lock);
+  check_triggers();
+
+  ASSERT_NO_THROW(store->DropTrigger(trigger_name_after));
+  ASSERT_NO_THROW(store->DropTrigger(trigger_name_before));
+
+  check_empty();
+
+  store.emplace(testing_directory, &ast_cache, &*dba, &antlr_lock);
+
+  check_empty();
+}
+
+TEST_F(TriggerStoreTest, AddTrigger) {
+  query::TriggerStore store{testing_directory, &ast_cache, &*dba, &antlr_lock};
+
+  // Invalid query in statements
+  ASSERT_THROW(store.AddTrigger("trigger", "RETUR 1", {}, query::TriggerEventType::VERTEX_CREATE,
+                                query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock),
+               utils::BasicException);
+  ASSERT_THROW(store.AddTrigger("trigger", "RETURN createdEdges", {}, query::TriggerEventType::VERTEX_CREATE,
+                                query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock),
+               utils::BasicException);
+
+  ASSERT_THROW(store.AddTrigger("trigger", "RETURN $parameter", {}, query::TriggerEventType::VERTEX_CREATE,
+                                query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock),
+               utils::BasicException);
+
+  ASSERT_NO_THROW(store.AddTrigger(
+      "trigger", "RETURN $parameter",
+      std::map<std::string, storage::PropertyValue>{{"parameter", storage::PropertyValue{1}}},
+      query::TriggerEventType::VERTEX_CREATE, query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock));
+
+  // Inserting with the same name
+  ASSERT_THROW(store.AddTrigger("trigger", "RETURN 1", {}, query::TriggerEventType::VERTEX_CREATE,
+                                query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock),
+               utils::BasicException);
+  ASSERT_THROW(store.AddTrigger("trigger", "RETURN 1", {}, query::TriggerEventType::VERTEX_CREATE,
+                                query::TriggerPhase::AFTER_COMMIT, &ast_cache, &*dba, &antlr_lock),
+               utils::BasicException);
+
+  ASSERT_EQ(store.GetTriggerInfo().size(), 1);
+  ASSERT_EQ(store.BeforeCommitTriggers().size(), 1);
+  ASSERT_EQ(store.AfterCommitTriggers().size(), 0);
+}
+
+TEST_F(TriggerStoreTest, DropTrigger) {
+  query::TriggerStore store{testing_directory, &ast_cache, &*dba, &antlr_lock};
+
+  ASSERT_THROW(store.DropTrigger("Unknown"), utils::BasicException);
+
+  const auto *trigger_name = "trigger";
+  store.AddTrigger(trigger_name, "RETURN 1", {}, query::TriggerEventType::VERTEX_CREATE,
+                   query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock);
+
+  ASSERT_THROW(store.DropTrigger("Unknown"), utils::BasicException);
+  ASSERT_NO_THROW(store.DropTrigger(trigger_name));
+  ASSERT_EQ(store.GetTriggerInfo().size(), 0);
+}
+
+TEST_F(TriggerStoreTest, TriggerInfo) {
+  query::TriggerStore store{testing_directory, &ast_cache, &*dba, &antlr_lock};
+
+  std::vector<query::TriggerStore::TriggerInfo> expected_info;
+  store.AddTrigger("trigger", "RETURN 1", {}, query::TriggerEventType::VERTEX_CREATE,
+                   query::TriggerPhase::BEFORE_COMMIT, &ast_cache, &*dba, &antlr_lock);
+  expected_info.push_back(
+      {"trigger", "RETURN 1", query::TriggerEventType::VERTEX_CREATE, query::TriggerPhase::BEFORE_COMMIT});
+
+  const auto check_trigger_info = [&] {
+    const auto trigger_info = store.GetTriggerInfo();
+    ASSERT_EQ(expected_info.size(), trigger_info.size());
+    // ensure all of the expected trigger infos can be found in the retrieved infos
+    ASSERT_TRUE(std::all_of(expected_info.begin(), expected_info.end(), [&](const auto &info) {
+      return std::find_if(trigger_info.begin(), trigger_info.end(), [&](const auto &other) {
+               return info.name == other.name && info.statement == other.statement &&
+                      info.event_type == other.event_type && info.phase == other.phase;
+             }) != trigger_info.end();
+    }));
+  };
+
+  check_trigger_info();
+
+  store.AddTrigger("edge_update_trigger", "RETURN 1", {}, query::TriggerEventType::EDGE_UPDATE,
+                   query::TriggerPhase::AFTER_COMMIT, &ast_cache, &*dba, &antlr_lock);
+  expected_info.push_back(
+      {"edge_update_trigger", "RETURN 1", query::TriggerEventType::EDGE_UPDATE, query::TriggerPhase::AFTER_COMMIT});
+
+  check_trigger_info();
+
+  store.DropTrigger("edge_update_trigger");
+  const auto erase_from_expected = [&](const std::string_view name) {
+    const auto erase_count = std::erase_if(expected_info, [name](const auto &info) { return info.name == name; });
+    ASSERT_EQ(erase_count, 1);
+  };
+  erase_from_expected("edge_update_trigger");
+
+  check_trigger_info();
+
+  store.DropTrigger("trigger");
+  erase_from_expected("trigger");
+
+  check_trigger_info();
+}