Update markers

* Add initial schema implementation

* Add index to schema

* List schemas and enable multiple properties

* Implement SchemaTypes

* Apply suggestions from code review

Co-authored-by: Jeremy B <97525434+42jeremy@users.noreply.github.com>
Co-authored-by: János Benjamin Antal <antaljanosbenjamin@users.noreply.github.com>

* Address review comments

* Remove Map and List

* Apply suggestions from code review

Co-authored-by: Kostas Kyrimis  <kostaskyrim@gmail.com>

Co-authored-by: Jeremy B <97525434+42jeremy@users.noreply.github.com>
Co-authored-by: János Benjamin Antal <antaljanosbenjamin@users.noreply.github.com>
Co-authored-by: Kostas Kyrimis  <kostaskyrim@gmail.com>

.clang-tidy

@@ -88,3 +88,4 @@ CheckOptions:
   - key:             modernize-use-nullptr.NullMacros
     value:           'NULL'
 ...
+

src/auth/models.cpp

@@ -37,7 +37,7 @@ const std::vector<Permission> kPermissionsAll = {
     Permission::CONSTRAINT, Permission::DUMP,      Permission::AUTH,        Permission::REPLICATION,
     Permission::DURABILITY, Permission::READ_FILE, Permission::FREE_MEMORY, Permission::TRIGGER,
     Permission::CONFIG,     Permission::STREAM,    Permission::MODULE_READ, Permission::MODULE_WRITE,
-    Permission::WEBSOCKET};
+    Permission::WEBSOCKET,  Permission::SCHEMA};
 }  // namespace
 
 std::string PermissionToString(Permission permission) {
@@ -84,6 +84,8 @@ std::string PermissionToString(Permission permission) {
       return "MODULE_WRITE";
     case Permission::WEBSOCKET:
       return "WEBSOCKET";
+    case Permission::SCHEMA:
+      return "SCHEMA";
   }
 }
 

src/auth/models.hpp

@@ -38,7 +38,8 @@ enum class Permission : uint64_t {
   STREAM       = 1U << 17U,
   MODULE_READ  = 1U << 18U,
   MODULE_WRITE = 1U << 19U,
-  WEBSOCKET    = 1U << 20U
+  WEBSOCKET    = 1U << 20U,
+  SCHEMA       = 1U << 21U
 };
 // clang-format on
 

src/common/types.hpp

@@ -9,12 +9,11 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
-#include "storage/v2/commit_error.hpp"
+#pragma once
 
-namespace memgraph::storage {
+#include <cstdint>
 
-bool operator==(const CommitError &lhs, const CommitError &rhs) {
-  return lhs.type == rhs.type && lhs.maybe_constraint_violation == rhs.maybe_constraint_violation;
-}
+namespace memgraph::common {
+enum class SchemaType : uint8_t { BOOL, INT, STRING, DATE, LOCALTIME, LOCALDATETIME, DURATION };
 
-}  // namespace memgraph::storage
+}  // namespace memgraph::common

src/glue/auth.cpp

@@ -57,6 +57,8 @@ auth::Permission PrivilegeToPermission(query::AuthQuery::Privilege privilege) {
       return auth::Permission::MODULE_WRITE;
     case query::AuthQuery::Privilege::WEBSOCKET:
       return auth::Permission::WEBSOCKET;
+    case query::AuthQuery::Privilege::SCHEMA:
+      return auth::Permission::SCHEMA;
   }
 }
 }  // namespace memgraph::glue

src/memgraph.cpp

@@ -252,11 +252,6 @@ DEFINE_double(query_execution_timeout_sec, 600,
               "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(replication_replica_check_frequency_sec, 1,
-              "The time duration between two replica checks/pings. If < 1, replicas will NOT be checked at all. NOTE: "
-              "The MAIN instance allocates a new thread for each REPLICA.");
-
 // NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_uint64(
     memory_limit, 0,
@@ -1221,7 +1216,6 @@ int main(int argc, char **argv) {
       &db,
       {.query = {.allow_load_csv = FLAGS_allow_load_csv},
        .execution_timeout_sec = FLAGS_query_execution_timeout_sec,
-       .replication_replica_check_frequency = std::chrono::seconds(FLAGS_replication_replica_check_frequency_sec),
        .default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
        .default_pulsar_service_url = FLAGS_pulsar_service_url,
        .stream_transaction_conflict_retries = FLAGS_stream_transaction_conflict_retries,

src/query/config.hpp

@@ -21,8 +21,6 @@ struct InterpreterConfig {
 
   // The default execution timeout is 10 minutes.
   double execution_timeout_sec{600.0};
-  // The same as \ref memgraph::storage::replication::ReplicationClientConfig
-  std::chrono::seconds replication_replica_check_frequency{1};
 
   std::string default_kafka_bootstrap_servers;
   std::string default_pulsar_service_url;

src/query/db_accessor.hpp

@@ -324,7 +324,7 @@ class DbAccessor final {
 
   void AdvanceCommand() { accessor_->AdvanceCommand(); }
 
-  utils::BasicResult<storage::CommitError, void> Commit() { return accessor_->Commit(); }
+  utils::BasicResult<storage::ConstraintViolation, void> Commit() { return accessor_->Commit(); }
 
   void Abort() { accessor_->Abort(); }
 

src/query/frontend/ast/ast.lcp

@@ -17,6 +17,7 @@
 #include <variant>
 #include <vector>
 
+#include "common/types.hpp"
 #include "query/frontend/ast/ast_visitor.hpp"
 #include "query/frontend/semantic/symbol.hpp"
 #include "query/interpret/awesome_memgraph_functions.hpp"
@@ -2253,7 +2254,7 @@ cpp<#
     (lcp:define-enum privilege
         (create delete match merge set remove index stats auth constraint
          dump replication durability read_file free_memory trigger config stream module_read module_write
-         websocket)
+         websocket schema)
       (:serialize))
     #>cpp
     AuthQuery() = default;
@@ -2295,7 +2296,7 @@ const std::vector<AuthQuery::Privilege> kPrivilegesAll = {
     AuthQuery::Privilege::FREE_MEMORY, AuthQuery::Privilege::TRIGGER,
     AuthQuery::Privilege::CONFIG,      AuthQuery::Privilege::STREAM,
     AuthQuery::Privilege::MODULE_READ, AuthQuery::Privilege::MODULE_WRITE,
-    AuthQuery::Privilege::WEBSOCKET};
+    AuthQuery::Privilege::WEBSOCKET,   AuthQuery::Privilege::SCHEMA};
 cpp<#
 
 (lcp:define-class info-query (query)
@@ -2665,5 +2666,37 @@ cpp<#
   (:serialize (:slk))
   (:clone))
 
+(lcp:define-class schema-query (query)
+  ((action "Action" :scope :public)
+   (label "LabelIx" :scope :public
+          :slk-load (lambda (member)
+                     #>cpp
+                     slk::Load(&self->${member}, reader, storage);
+                     cpp<#)
+          :clone (lambda (source dest)
+                   #>cpp
+                   ${dest} = storage->GetLabelIx(${source}.name);
+                   cpp<#))
+    (schema_type_map "std::unordered_map<PropertyIx, common::SchemaType>"
+      :slk-save #'slk-save-property-map
+      :slk-load #'slk-load-property-map
+      :scope :public))
+
+  (:public
+    (lcp:define-enum action
+        (create-schema drop-schema show-schema show-schemas)
+      (:serialize))
+    #>cpp
+    SchemaQuery() = default;
+
+    DEFVISITABLE(QueryVisitor<void>);
+    cpp<#)
+  (:private
+    #>cpp
+    friend class AstStorage;
+    cpp<#)
+  (:serialize (:slk))
+  (:clone))
+
 (lcp:pop-namespace) ;; namespace query
 (lcp:pop-namespace) ;; namespace memgraph

src/query/frontend/ast/ast_visitor.hpp

@@ -94,6 +94,7 @@ class StreamQuery;
 class SettingQuery;
 class VersionQuery;
 class Foreach;
+class SchemaQuery;
 
 using TreeCompositeVisitor = utils::CompositeVisitor<
     SingleQuery, CypherUnion, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator, AdditionOperator,
@@ -125,9 +126,9 @@ 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, TriggerQuery,
-                            IsolationLevelQuery, CreateSnapshotQuery, StreamQuery, SettingQuery, VersionQuery> {};
+class QueryVisitor : public utils::Visitor<TResult, CypherQuery, ExplainQuery, ProfileQuery, IndexQuery, AuthQuery,
+                                           InfoQuery, ConstraintQuery, DumpQuery, ReplicationQuery, LockPathQuery,
+                                           FreeMemoryQuery, TriggerQuery, IsolationLevelQuery, CreateSnapshotQuery,
+                                           StreamQuery, SettingQuery, VersionQuery, SchemaQuery> {};
 
 }  // namespace memgraph::query

src/query/frontend/ast/cypher_main_visitor.cpp

@@ -27,6 +27,7 @@
 
 #include <boost/preprocessor/cat.hpp>
 
+#include "common/types.hpp"
 #include "query/exceptions.hpp"
 #include "query/frontend/ast/ast.hpp"
 #include "query/frontend/ast/ast_visitor.hpp"
@@ -1338,6 +1339,7 @@ antlrcpp::Any CypherMainVisitor::visitPrivilege(MemgraphCypher::PrivilegeContext
   if (ctx->MODULE_READ()) return AuthQuery::Privilege::MODULE_READ;
   if (ctx->MODULE_WRITE()) return AuthQuery::Privilege::MODULE_WRITE;
   if (ctx->WEBSOCKET()) return AuthQuery::Privilege::WEBSOCKET;
+  if (ctx->SCHEMA()) return AuthQuery::Privilege::SCHEMA;
   LOG_FATAL("Should not get here - unknown privilege!");
 }
 
@@ -2336,6 +2338,94 @@ antlrcpp::Any CypherMainVisitor::visitForeach(MemgraphCypher::ForeachContext *ct
   return for_each;
 }
 
+antlrcpp::Any CypherMainVisitor::visitSchemaQuery(MemgraphCypher::SchemaQueryContext *ctx) {
+  MG_ASSERT(ctx->children.size() == 1, "SchemaQuery should have exactly one child!");
+  auto *schema_query = ctx->children[0]->accept(this).as<SchemaQuery *>();
+  query_ = schema_query;
+  return schema_query;
+}
+
+antlrcpp::Any CypherMainVisitor::visitShowSchema(MemgraphCypher::ShowSchemaContext *ctx) {
+  auto *schema_query = storage_->Create<SchemaQuery>();
+  schema_query->action_ = SchemaQuery::Action::SHOW_SCHEMA;
+  schema_query->label_ = AddLabel(ctx->labelName()->accept(this));
+  query_ = schema_query;
+  return schema_query;
+}
+
+antlrcpp::Any CypherMainVisitor::visitShowSchemas(MemgraphCypher::ShowSchemasContext * /*ctx*/) {
+  auto *schema_query = storage_->Create<SchemaQuery>();
+  schema_query->action_ = SchemaQuery::Action::SHOW_SCHEMAS;
+  query_ = schema_query;
+  return schema_query;
+}
+
+antlrcpp::Any CypherMainVisitor::visitPropertyType(MemgraphCypher::PropertyTypeContext *ctx) {
+  MG_ASSERT(ctx->symbolicName());
+  const auto property_type = utils::ToLowerCase(ctx->symbolicName()->accept(this).as<std::string>());
+  if (property_type == "bool") {
+    return common::SchemaType::BOOL;
+  }
+  if (property_type == "string") {
+    return common::SchemaType::STRING;
+  }
+  if (property_type == "integer") {
+    return common::SchemaType::INT;
+  }
+  if (property_type == "date") {
+    return common::SchemaType::DATE;
+  }
+  if (property_type == "duration") {
+    return common::SchemaType::DURATION;
+  }
+  if (property_type == "localdatetime") {
+    return common::SchemaType::LOCALDATETIME;
+  }
+  if (property_type == "localtime") {
+    return common::SchemaType::LOCALTIME;
+  }
+  throw SyntaxException("Property type must be one of the supported types!");
+}
+
+/**
+ * @return Schema*
+ */
+antlrcpp::Any CypherMainVisitor::visitSchemaTypeMap(MemgraphCypher::SchemaTypeMapContext *ctx) {
+  std::unordered_map<PropertyIx, common::SchemaType> map;
+  for (auto *property_key_pair : ctx->propertyKeyTypePair()) {
+    PropertyIx key = property_key_pair->propertyKeyName()->accept(this);
+    common::SchemaType type = property_key_pair->propertyType()->accept(this);
+    if (!map.insert({key, type}).second) {
+      throw SemanticException("Same property name can't appear twice in a schema map.");
+    }
+  }
+  return map;
+}
+
+antlrcpp::Any CypherMainVisitor::visitCreateSchema(MemgraphCypher::CreateSchemaContext *ctx) {
+  auto *schema_query = storage_->Create<SchemaQuery>();
+  schema_query->action_ = SchemaQuery::Action::CREATE_SCHEMA;
+  schema_query->label_ = AddLabel(ctx->labelName()->accept(this));
+  if (!ctx->schemaTypeMap()) {
+    throw SemanticException("Schema property map must exist!");
+  }
+  schema_query->schema_type_map_ =
+      ctx->schemaTypeMap()->accept(this).as<std::unordered_map<PropertyIx, common::SchemaType>>();
+  query_ = schema_query;
+  return schema_query;
+}
+
+/**
+ * @return Schema*
+ */
+antlrcpp::Any CypherMainVisitor::visitDropSchema(MemgraphCypher::DropSchemaContext *ctx) {
+  auto *schema_query = storage_->Create<SchemaQuery>();
+  schema_query->action_ = SchemaQuery::Action::DROP_SCHEMA;
+  schema_query->label_ = AddLabel(ctx->labelName()->accept(this));
+  query_ = schema_query;
+  return schema_query;
+}
+
 LabelIx CypherMainVisitor::AddLabel(const std::string &name) { return storage_->GetLabelIx(name); }
 
 PropertyIx CypherMainVisitor::AddProperty(const std::string &name) { return storage_->GetPropertyIx(name); }

src/query/frontend/ast/cypher_main_visitor.hpp

@@ -849,6 +849,41 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
    */
   antlrcpp::Any visitForeach(MemgraphCypher::ForeachContext *ctx) override;
 
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitPropertyType(MemgraphCypher::PropertyTypeContext *ctx) override;
+
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitSchemaTypeMap(MemgraphCypher::SchemaTypeMapContext *ctx) override;
+
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitSchemaQuery(MemgraphCypher::SchemaQueryContext *ctx) override;
+
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitShowSchema(MemgraphCypher::ShowSchemaContext *ctx) override;
+
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitShowSchemas(MemgraphCypher::ShowSchemasContext *ctx) override;
+
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitCreateSchema(MemgraphCypher::CreateSchemaContext *ctx) override;
+
+  /**
+   * @return Schema*
+   */
+  antlrcpp::Any visitDropSchema(MemgraphCypher::DropSchemaContext *ctx) override;
+
  public:
   Query *query() { return query_; }
   const static std::string kAnonPrefix;

src/query/frontend/opencypher/grammar/MemgraphCypher.g4

@@ -46,10 +46,10 @@ memgraphCypherKeyword : cypherKeyword
                       | DROP
                       | DUMP
                       | EXECUTE
-                      | FOR
-                      | FOREACH
                       | FREE
                       | FROM
+                      | FOR
+                      | FOREACH
                       | GLOBAL
                       | GRANT
                       | HEADER
@@ -76,6 +76,8 @@ memgraphCypherKeyword : cypherKeyword
                       | ROLE
                       | ROLES
                       | QUOTE
+                      | SCHEMA
+                      | SCHEMAS
                       | SESSION
                       | SETTING
                       | SETTINGS
@@ -122,6 +124,7 @@ query : cypherQuery
       | streamQuery
       | settingQuery
       | versionQuery
+      | schemaQuery
       ;
 
 authQuery : createRole
@@ -192,6 +195,12 @@ settingQuery : setSetting
              | showSettings
              ;
 
+schemaQuery : showSchema
+            | showSchemas
+            | createSchema
+            | dropSchema
+            ;
+
 loadCsv : LOAD CSV FROM csvFile ( WITH | NO ) HEADER
          ( IGNORE BAD ) ?
          ( DELIMITER delimiter ) ?
@@ -254,6 +263,7 @@ privilege : CREATE
           | MODULE_READ
           | MODULE_WRITE
           | WEBSOCKET
+          | SCHEMA
           ;
 
 privilegeList : privilege ( ',' privilege )* ;
@@ -374,3 +384,17 @@ showSetting : SHOW DATABASE SETTING settingName ;
 showSettings : SHOW DATABASE SETTINGS ;
 
 versionQuery : SHOW VERSION ;
+
+showSchema : SHOW SCHEMA ON ':' labelName ;
+
+showSchemas : SHOW SCHEMAS ;
+
+propertyType : symbolicName ;
+
+propertyKeyTypePair : propertyKeyName propertyType ;
+
+schemaTypeMap : '(' propertyKeyTypePair ( ',' propertyKeyTypePair )* ')' ;
+
+createSchema : CREATE SCHEMA ON ':' labelName schemaTypeMap ;
+
+dropSchema : DROP SCHEMA ON ':' labelName ;

src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4

@@ -89,6 +89,8 @@ REVOKE              : R E V O K E ;
 ROLE                : R O L E ;
 ROLES               : R O L E S ;
 QUOTE               : Q U O T E ;
+SCHEMA              : S C H E M A ;
+SCHEMAS             : S C H E M A S ;
 SERVICE_URL         : S E R V I C E UNDERSCORE U R L ;
 SESSION             : S E S S I O N ;
 SETTING             : S E T T I N G ;

src/query/frontend/semantic/required_privileges.cpp

@@ -80,6 +80,8 @@ class PrivilegeExtractor : public QueryVisitor<void>, public HierarchicalTreeVis
 
   void Visit(VersionQuery & /*version_query*/) override { AddPrivilege(AuthQuery::Privilege::STATS); }
 
+  void Visit(SchemaQuery & /*schema_query*/) override { AddPrivilege(AuthQuery::Privilege::SCHEMA); }
+
   bool PreVisit(Create & /*unused*/) override {
     AddPrivilege(AuthQuery::Privilege::CREATE);
     return false;

src/query/frontend/stripped_lexer_constants.hpp

@@ -204,8 +204,9 @@ const trie::Trie kKeywords = {"union",
                               "pulsar",
                               "service_url",
                               "version",
-                              "websocket"
-                              "foreach"};
+                              "websocket",
+                              "foreach",
+                              "schema"};
 
 // Unicode codepoints that are allowed at the start of the unescaped name.
 const std::bitset<kBitsetSize> kUnescapedNameAllowedStarts(

src/query/interpreter.cpp

@@ -44,6 +44,7 @@
 #include "query/trigger.hpp"
 #include "query/typed_value.hpp"
 #include "storage/v2/property_value.hpp"
+#include "storage/v2/schemas.hpp"
 #include "utils/algorithm.hpp"
 #include "utils/csv_parsing.hpp"
 #include "utils/event_counter.hpp"
@@ -160,8 +161,7 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
 
   /// @throw QueryRuntimeException if an error ocurred.
   void RegisterReplica(const std::string &name, const std::string &socket_address,
-                       const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
-                       const std::chrono::seconds replica_check_frequency) override {
+                       const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) override {
     if (db_->GetReplicationRole() == storage::ReplicationRole::REPLICA) {
       // replica can't register another replica
       throw QueryRuntimeException("Replica can't register another replica!");
@@ -183,9 +183,8 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
         io::network::Endpoint::ParseSocketOrIpAddress(socket_address, query::kDefaultReplicationPort);
     if (maybe_ip_and_port) {
       auto [ip, port] = *maybe_ip_and_port;
-      auto ret = db_->RegisterReplica(
-          name, {std::move(ip), port}, repl_mode,
-          {.timeout = timeout, .replica_check_frequency = replica_check_frequency, .ssl = std::nullopt});
+      auto ret =
+          db_->RegisterReplica(name, {std::move(ip), port}, repl_mode, {.timeout = timeout, .ssl = std::nullopt});
       if (ret.HasError()) {
         throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
       }
@@ -450,7 +449,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
       return callback;
     }
     case ReplicationQuery::Action::SHOW_REPLICATION_ROLE: {
-      callback.header = {"replication role"};
+      callback.header = {"replication mode"};
       callback.fn = [handler = ReplQueryHandler{interpreter_context->db}] {
         auto mode = handler.ShowReplicationRole();
         switch (mode) {
@@ -469,7 +468,6 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
       const auto &sync_mode = repl_query->sync_mode_;
       auto socket_address = repl_query->socket_address_->Accept(evaluator);
       auto timeout = EvaluateOptionalExpression(repl_query->timeout_, &evaluator);
-      const auto replica_check_frequency = interpreter_context->config.replication_replica_check_frequency;
       std::optional<double> maybe_timeout;
       if (timeout.IsDouble()) {
         maybe_timeout = timeout.ValueDouble();
@@ -477,9 +475,8 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
         maybe_timeout = static_cast<double>(timeout.ValueInt());
       }
       callback.fn = [handler = ReplQueryHandler{interpreter_context->db}, name, socket_address, sync_mode,
-                     maybe_timeout, replica_check_frequency]() mutable {
-        handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout,
-                                replica_check_frequency);
+                     maybe_timeout]() mutable {
+        handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout);
         return std::vector<std::vector<TypedValue>>();
       };
       notifications->emplace_back(SeverityLevel::INFO, NotificationCode::REGISTER_REPLICA,
@@ -516,6 +513,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
               typed_replica.emplace_back(TypedValue("async"));
               break;
           }
+          typed_replica.emplace_back(TypedValue(static_cast<int64_t>(replica.sync_mode)));
           if (replica.timeout) {
             typed_replica.emplace_back(TypedValue(*replica.timeout));
           } else {
@@ -823,6 +821,108 @@ Callback HandleSettingQuery(SettingQuery *setting_query, const Parameters &param
   }
 }
 
+Callback HandleSchemaQuery(SchemaQuery *schema_query, InterpreterContext *interpreter_context,
+                           std::vector<Notification> *notifications) {
+  Callback callback;
+  switch (schema_query->action_) {
+    case SchemaQuery::Action::SHOW_SCHEMAS: {
+      callback.header = {"label", "primary_key", "primary_key_type"};
+      callback.fn = [interpreter_context]() {
+        auto *db = interpreter_context->db;
+        auto schemas_info = db->ListAllSchemas();
+        std::vector<std::vector<TypedValue>> results;
+        results.reserve(schemas_info.schemas.size());
+
+        for (const auto &[label_id, schema_types] : schemas_info.schemas) {
+          std::vector<TypedValue> schema_info_row;
+          schema_info_row.reserve(3);
+
+          schema_info_row.emplace_back(db->LabelToName(label_id));
+          std::vector<std::string> primary_key_properties;
+          primary_key_properties.reserve(schema_types.size());
+          std::transform(schema_types.begin(), schema_types.end(), std::back_inserter(primary_key_properties),
+                         [&db](const auto &schema_type) {
+                           return db->PropertyToName(schema_type.property_id) +
+                                  "::" + storage::SchemaTypeToString(schema_type.type);
+                         });
+
+          schema_info_row.emplace_back(utils::Join(primary_key_properties, ", "));
+          schema_info_row.emplace_back(schema_types.size() == 1 ? "Single" : "Composite");
+
+          results.push_back(std::move(schema_info_row));
+        }
+        return results;
+      };
+      return callback;
+    }
+    case SchemaQuery::Action::SHOW_SCHEMA: {
+      callback.header = {"property_name", "property_type"};
+      callback.fn = [interpreter_context, primary_label = schema_query->label_]() {
+        auto *db = interpreter_context->db;
+        const auto label = db->NameToLabel(primary_label.name);
+        const auto schemas_info = db->GetSchema(label);
+        MG_ASSERT(schemas_info.schemas.size() < 2, "There can be only one schema under single label!");
+        std::vector<std::vector<TypedValue>> results;
+        if (!schemas_info.schemas.empty()) {
+          const auto schema = schemas_info.schemas[0];
+
+          for (const auto &schema_property : schema.second) {
+            std::vector<TypedValue> schema_info_row;
+            schema_info_row.reserve(2);
+
+            schema_info_row.emplace_back(db->PropertyToName(schema_property.property_id));
+            schema_info_row.emplace_back(storage::SchemaTypeToString(schema_property.type));
+
+            results.push_back(std::move(schema_info_row));
+          }
+        }
+        return results;
+      };
+      return callback;
+    }
+    case SchemaQuery::Action::CREATE_SCHEMA: {
+      auto schema_type_map = schema_query->schema_type_map_;
+      if (schema_query->schema_type_map_.empty()) {
+        throw SyntaxException("One or more types have to be defined in schema definition.");
+      }
+      callback.fn = [interpreter_context, primary_label = schema_query->label_,
+                     schema_type_map = std::move(schema_type_map)]() {
+        auto *db = interpreter_context->db;
+        const auto label = db->NameToLabel(primary_label.name);
+        std::vector<storage::SchemaPropertyType> schemas_types;
+        schemas_types.reserve(schema_type_map.size());
+        for (const auto &schema_type : schema_type_map) {
+          auto property_id = db->NameToProperty(schema_type.first.name);
+          schemas_types.push_back({schema_type.second, property_id});
+        }
+        if (!db->CreateSchema(label, schemas_types)) {
+          throw QueryException(fmt::format("Schema on label :{} already exists!", primary_label.name));
+        }
+        return std::vector<std::vector<TypedValue>>{};
+      };
+      notifications->emplace_back(SeverityLevel::INFO, NotificationCode::CREATE_SCHEMA,
+                                  fmt::format("Create schema on label :{}", schema_query->label_.name));
+      return callback;
+    }
+    case SchemaQuery::Action::DROP_SCHEMA: {
+      callback.fn = [interpreter_context, primary_label = schema_query->label_]() {
+        auto *db = interpreter_context->db;
+        const auto label = db->NameToLabel(primary_label.name);
+
+        if (!db->DropSchema(label)) {
+          throw QueryException(fmt::format("Schema on label :{} does not exist!", primary_label.name));
+        }
+
+        return std::vector<std::vector<TypedValue>>{};
+      };
+      notifications->emplace_back(SeverityLevel::INFO, NotificationCode::DROP_SCHEMA,
+                                  fmt::format("Dropped schema on label :{}", schema_query->label_.name));
+      return callback;
+    }
+  }
+  return callback;
+}
+
 // Struct for lazy pulling from a vector
 struct PullPlanVector {
   explicit PullPlanVector(std::vector<std::vector<TypedValue>> values) : values_(std::move(values)) {}
@@ -2018,6 +2118,32 @@ PreparedQuery PrepareConstraintQuery(ParsedQuery parsed_query, bool in_explicit_
                        RWType::NONE};
 }
 
+PreparedQuery PrepareSchemaQuery(ParsedQuery parsed_query, bool in_explicit_transaction,
+                                 InterpreterContext *interpreter_context, std::vector<Notification> *notifications) {
+  if (in_explicit_transaction) {
+    throw ConstraintInMulticommandTxException();
+  }
+  auto *schema_query = utils::Downcast<SchemaQuery>(parsed_query.query);
+  MG_ASSERT(schema_query);
+  auto callback = HandleSchemaQuery(schema_query, interpreter_context, notifications);
+
+  return PreparedQuery{std::move(callback.header), std::move(parsed_query.required_privileges),
+                       [handler = std::move(callback.fn), action = QueryHandlerResult::NOTHING,
+                        pull_plan = std::shared_ptr<PullPlanVector>(nullptr)](
+                           AnyStream *stream, std::optional<int> n) mutable -> std::optional<QueryHandlerResult> {
+                         if (!pull_plan) {
+                           auto results = handler();
+                           pull_plan = std::make_shared<PullPlanVector>(std::move(results));
+                         }
+
+                         if (pull_plan->Pull(stream, n)) {
+                           return action;
+                         }
+                         return std::nullopt;
+                       },
+                       RWType::NONE};
+}
+
 void Interpreter::BeginTransaction() {
   const auto prepared_query = PrepareTransactionQuery("BEGIN");
   prepared_query.query_handler(nullptr, {});
@@ -2151,6 +2277,9 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
       prepared_query = PrepareSettingQuery(std::move(parsed_query), in_explicit_transaction_, &*execution_db_accessor_);
     } else if (utils::Downcast<VersionQuery>(parsed_query.query)) {
       prepared_query = PrepareVersionQuery(std::move(parsed_query), in_explicit_transaction_);
+    } else if (utils::Downcast<SchemaQuery>(parsed_query.query)) {
+      prepared_query = PrepareSchemaQuery(std::move(parsed_query), in_explicit_transaction_, interpreter_context_,
+                                          &query_execution->notifications);
     } else {
       LOG_FATAL("Should not get here -- unknown query type!");
     }
@@ -2209,39 +2338,25 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
       continue;
     }
 
-    auto maybe_commit_error = db_accessor.Commit();
-    if (maybe_commit_error.HasError()) {
-      const auto &commit_error = maybe_commit_error.GetError();
-      switch (commit_error.type) {
-        case storage::CommitError::Type::UNABLE_TO_SYNC_REPLICATE: {
-          // TODO(gitbuda): This is tricky because this is an internal
-          // operation. Consider stopping main Memgraph instance here.
-          spdlog::warn("Trigger '{}' failed to commit due to inability to replicate data to SYNC replica",
-                       trigger.Name());
+    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::CommitError::Type::CONSTRAINT_VIOLATION: {
-          MG_ASSERT(commit_error.maybe_constraint_violation.has_value());
-          const auto &constraint_violation = *commit_error.maybe_constraint_violation;
-          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;
-            }
-          }
+        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;
         }
       }
@@ -2286,40 +2401,28 @@ void Interpreter::Commit() {
     trigger_context_collector_.reset();
   };
 
-  auto maybe_commit_error = db_accessor_->Commit();
-  if (maybe_commit_error.HasError()) {
-    const auto &commit_error = maybe_commit_error.GetError();
-    switch (commit_error.type) {
-      case storage::CommitError::Type::UNABLE_TO_SYNC_REPLICATE: {
+  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: {
+        auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
+        MG_ASSERT(constraint_violation.properties.size() == 1U);
+        auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
         reset_necessary_members();
-        throw QueryException("Unable to commit due to inability to replicate to SYNC replica");
+        throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
+                             property_name);
         break;
       }
-      case storage::CommitError::Type::CONSTRAINT_VIOLATION: {
-        MG_ASSERT(commit_error.maybe_constraint_violation.has_value());
-        const auto &constraint_violation = *commit_error.maybe_constraint_violation;
-        switch (constraint_violation.type) {
-          case storage::ConstraintViolation::Type::EXISTENCE: {
-            auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
-            MG_ASSERT(constraint_violation.properties.size() == 1U);
-            auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
-            reset_necessary_members();
-            throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
-                                 property_name);
-            break;
-          }
-          case storage::ConstraintViolation::Type::UNIQUE: {
-            auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
-            std::stringstream property_names_stream;
-            utils::PrintIterable(
-                property_names_stream, constraint_violation.properties, ", ",
-                [this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
-            reset_necessary_members();
-            throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
-                                 property_names_stream.str());
-            break;
-          }
-        }
+      case storage::ConstraintViolation::Type::UNIQUE: {
+        auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
+        std::stringstream property_names_stream;
+        utils::PrintIterable(
+            property_names_stream, constraint_violation.properties, ", ",
+            [this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
+        reset_necessary_members();
+        throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
+                             property_names_stream.str());
         break;
       }
     }

src/query/interpreter.hpp

@@ -137,8 +137,7 @@ class ReplicationQueryHandler {
 
   /// @throw QueryRuntimeException if an error ocurred.
   virtual void RegisterReplica(const std::string &name, const std::string &socket_address,
-                               const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
-                               const std::chrono::seconds replica_check_frequency) = 0;
+                               const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) = 0;
 
   /// @throw QueryRuntimeException if an error ocurred.
   virtual void DropReplica(const std::string &replica_name) = 0;

src/query/metadata.cpp

@@ -38,6 +38,8 @@ constexpr std::string_view GetCodeString(const NotificationCode code) {
       return "CreateIndex"sv;
     case NotificationCode::CREATE_STREAM:
       return "CreateStream"sv;
+    case NotificationCode::CREATE_SCHEMA:
+      return "CreateSchema"sv;
     case NotificationCode::CHECK_STREAM:
       return "CheckStream"sv;
     case NotificationCode::CREATE_TRIGGER:
@@ -48,6 +50,8 @@ constexpr std::string_view GetCodeString(const NotificationCode code) {
       return "DropReplica"sv;
     case NotificationCode::DROP_INDEX:
       return "DropIndex"sv;
+    case NotificationCode::DROP_SCHEMA:
+      return "DropSchema"sv;
     case NotificationCode::DROP_STREAM:
       return "DropStream"sv;
     case NotificationCode::DROP_TRIGGER:
@@ -68,6 +72,10 @@ constexpr std::string_view GetCodeString(const NotificationCode code) {
       return "ReplicaPortWarning"sv;
     case NotificationCode::SET_REPLICA:
       return "SetReplica"sv;
+    case NotificationCode::SHOW_SCHEMA:
+      return "ShowSchema"sv;
+    case NotificationCode::SHOW_SCHEMAS:
+      return "ShowSchemas"sv;
     case NotificationCode::START_STREAM:
       return "StartStream"sv;
     case NotificationCode::START_ALL_STREAMS:
@@ -114,4 +122,4 @@ std::string ExecutionStatsKeyToString(const ExecutionStats::Key key) {
   }
 }
 
-}  // namespace memgraph::query
+}  // namespace memgraph::query

src/query/metadata.hpp

@@ -26,12 +26,14 @@ enum class SeverityLevel : uint8_t { INFO, WARNING };
 enum class NotificationCode : uint8_t {
   CREATE_CONSTRAINT,
   CREATE_INDEX,
+  CREATE_SCHEMA,
   CHECK_STREAM,
   CREATE_STREAM,
   CREATE_TRIGGER,
   DROP_CONSTRAINT,
   DROP_INDEX,
   DROP_REPLICA,
+  DROP_SCHEMA,
   DROP_STREAM,
   DROP_TRIGGER,
   EXISTANT_INDEX,
@@ -42,6 +44,8 @@ enum class NotificationCode : uint8_t {
   REPLICA_PORT_WARNING,
   REGISTER_REPLICA,
   SET_REPLICA,
+  SHOW_SCHEMA,
+  SHOW_SCHEMAS,
   START_STREAM,
   START_ALL_STREAMS,
   STOP_STREAM,

src/storage/v2/CMakeLists.txt

@@ -1,6 +1,5 @@
 set(storage_v2_src_files
     commit_log.cpp
-    commit_error.cpp
     constraints.cpp
     temporal.cpp
     durability/durability.cpp
@@ -11,6 +10,7 @@ set(storage_v2_src_files
     indices.cpp
     property_store.cpp
     vertex_accessor.cpp
+    schemas.cpp
     storage.cpp)
 
 ##### Replication #####

src/storage/v2/commit_error.hpp

@@ -1,32 +0,0 @@
-// Copyright 2022 Memgraph Ltd.
-//
-// Use of this software is governed by the Business Source License
-// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
-// License, and you may not use this file except in compliance with the Business Source License.
-//
-// As of the Change Date specified in that file, in accordance with
-// the Business Source License, use of this software will be governed
-// by the Apache License, Version 2.0, included in the file
-// licenses/APL.txt.
-
-#pragma once
-
-#include <optional>
-
-#include "storage/v2/constraints.hpp"
-
-namespace memgraph::storage {
-
-struct CommitError {
-  enum class Type {
-    CONSTRAINT_VIOLATION,
-    UNABLE_TO_SYNC_REPLICATE,
-  };
-  Type type;
-
-  std::optional<ConstraintViolation> maybe_constraint_violation;
-};
-
-bool operator==(const CommitError &lhs, const CommitError &rhs);
-
-}  // namespace memgraph::storage

src/storage/v2/durability/durability.cpp

@@ -178,6 +178,7 @@ std::optional<RecoveryInfo> RecoverData(const std::filesystem::path &snapshot_di
 
   RecoveryInfo recovery_info;
   RecoveredIndicesAndConstraints indices_constraints;
+  memgraph::storage::SchemasMap recovered_schemas;
   std::optional<uint64_t> snapshot_timestamp;
   if (!snapshot_files.empty()) {
     spdlog::info("Try recovering from snapshot directory {}.", snapshot_directory);

src/storage/v2/durability/marker.hpp

@@ -38,6 +38,7 @@ enum class Marker : uint8_t {
   SECTION_DELTA = 0x26,
   SECTION_EPOCH_HISTORY = 0x27,
   SECTION_OFFSETS = 0x42,
+  SECTION_SCHEMAS = 0x43,
 
   DELTA_VERTEX_CREATE = 0x50,
   DELTA_VERTEX_DELETE = 0x51,
@@ -56,6 +57,8 @@ enum class Marker : uint8_t {
   DELTA_EXISTENCE_CONSTRAINT_DROP = 0x5e,
   DELTA_UNIQUE_CONSTRAINT_CREATE = 0x5f,
   DELTA_UNIQUE_CONSTRAINT_DROP = 0x60,
+  DELTA_SCHEMA_CREATE = 0x61,
+  DELTA_SCHEMA_DROP = 0x62,
 
   VALUE_FALSE = 0x00,
   VALUE_TRUE = 0xff,
@@ -63,7 +66,7 @@ enum class Marker : uint8_t {
 
 /// List of all available markers.
 /// IMPORTANT: Don't forget to update this list when you add a new Marker.
-static const Marker kMarkersAll[] = {
+constexpr Marker kMarkersAll[] = {
     Marker::TYPE_NULL,
     Marker::TYPE_BOOL,
     Marker::TYPE_INT,
@@ -99,6 +102,8 @@ static const Marker kMarkersAll[] = {
     Marker::DELTA_EXISTENCE_CONSTRAINT_DROP,
     Marker::DELTA_UNIQUE_CONSTRAINT_CREATE,
     Marker::DELTA_UNIQUE_CONSTRAINT_DROP,
+    Marker::DELTA_SCHEMA_CREATE,
+    Marker::DELTA_SCHEMA_DROP,
     Marker::VALUE_FALSE,
     Marker::VALUE_TRUE,
 };

src/storage/v2/durability/serialization.cpp

@@ -333,6 +333,7 @@ std::optional<PropertyValue> Decoder::ReadPropertyValue() {
     case Marker::SECTION_DELTA:
     case Marker::SECTION_EPOCH_HISTORY:
     case Marker::SECTION_OFFSETS:
+    case Marker::SECTION_SCHEMAS:
     case Marker::DELTA_VERTEX_CREATE:
     case Marker::DELTA_VERTEX_DELETE:
     case Marker::DELTA_VERTEX_ADD_LABEL:
@@ -350,6 +351,8 @@ std::optional<PropertyValue> Decoder::ReadPropertyValue() {
     case Marker::DELTA_EXISTENCE_CONSTRAINT_DROP:
     case Marker::DELTA_UNIQUE_CONSTRAINT_CREATE:
     case Marker::DELTA_UNIQUE_CONSTRAINT_DROP:
+    case Marker::DELTA_SCHEMA_CREATE:
+    case Marker::DELTA_SCHEMA_DROP:
     case Marker::VALUE_FALSE:
     case Marker::VALUE_TRUE:
       return std::nullopt;
@@ -432,6 +435,7 @@ bool Decoder::SkipPropertyValue() {
     case Marker::SECTION_DELTA:
     case Marker::SECTION_EPOCH_HISTORY:
     case Marker::SECTION_OFFSETS:
+    case Marker::SECTION_SCHEMAS:
     case Marker::DELTA_VERTEX_CREATE:
     case Marker::DELTA_VERTEX_DELETE:
     case Marker::DELTA_VERTEX_ADD_LABEL:
@@ -449,6 +453,8 @@ bool Decoder::SkipPropertyValue() {
     case Marker::DELTA_EXISTENCE_CONSTRAINT_DROP:
     case Marker::DELTA_UNIQUE_CONSTRAINT_CREATE:
     case Marker::DELTA_UNIQUE_CONSTRAINT_DROP:
+    case Marker::DELTA_SCHEMA_CREATE:
+    case Marker::DELTA_SCHEMA_DROP:
     case Marker::VALUE_FALSE:
     case Marker::VALUE_TRUE:
       return false;

src/storage/v2/durability/snapshot.hpp

@@ -30,6 +30,7 @@ namespace memgraph::storage::durability {
 
 /// Structure used to hold information about a snapshot.
 struct SnapshotInfo {
+  uint64_t offset_schemas;
   uint64_t offset_edges;
   uint64_t offset_vertices;
   uint64_t offset_indices;

src/storage/v2/durability/wal.cpp

@@ -69,6 +69,10 @@ namespace memgraph::storage::durability {
 //         * unique constraint create, unique constraint drop
 //              * label name
 //              * property names
+//         * schema create, schema drop
+//              * label name
+//              * property names
+//              * property type
 //
 // IMPORTANT: When changing WAL encoding/decoding bump the snapshot/WAL version
 // in `version.hpp`.
@@ -93,6 +97,10 @@ Marker OperationToMarker(StorageGlobalOperation operation) {
       return Marker::DELTA_UNIQUE_CONSTRAINT_CREATE;
     case StorageGlobalOperation::UNIQUE_CONSTRAINT_DROP:
       return Marker::DELTA_UNIQUE_CONSTRAINT_DROP;
+    case StorageGlobalOperation::SCHEMA_CREATE:
+      return Marker::DELTA_SCHEMA_CREATE;
+    case StorageGlobalOperation::SCHEMA_DROP:
+      return Marker::DELTA_SCHEMA_DROP;
   }
 }
 
@@ -122,7 +130,7 @@ Marker VertexActionToMarker(Delta::Action action) {
   }
 }
 
-// This function convertes a Marker to a WalDeltaData::Type. It checks for the
+// This function converts a Marker to a WalDeltaData::Type. It checks for the
 // validity of the marker and throws if an invalid marker is specified.
 // @throw RecoveryFailure
 WalDeltaData::Type MarkerToWalDeltaDataType(Marker marker) {
@@ -161,6 +169,10 @@ WalDeltaData::Type MarkerToWalDeltaDataType(Marker marker) {
       return WalDeltaData::Type::UNIQUE_CONSTRAINT_CREATE;
     case Marker::DELTA_UNIQUE_CONSTRAINT_DROP:
       return WalDeltaData::Type::UNIQUE_CONSTRAINT_DROP;
+    case Marker::DELTA_SCHEMA_CREATE:
+      return WalDeltaData::Type::UNIQUE_CONSTRAINT_CREATE;
+    case Marker::DELTA_SCHEMA_DROP:
+      return WalDeltaData::Type::UNIQUE_CONSTRAINT_DROP;
 
     case Marker::TYPE_NULL:
     case Marker::TYPE_BOOL:
@@ -309,6 +321,11 @@ WalDeltaData ReadSkipWalDeltaData(BaseDecoder *decoder) {
           if (!decoder->SkipString()) throw RecoveryFailure("Invalid WAL data!");
         }
       }
+      break;
+    }
+    case WalDeltaData::Type::SCHEMA_CREATE:
+    case WalDeltaData::Type::SCHEMA_DROP: {
+      break;
     }
   }
 
@@ -456,6 +473,10 @@ bool operator==(const WalDeltaData &a, const WalDeltaData &b) {
     case WalDeltaData::Type::UNIQUE_CONSTRAINT_DROP:
       return a.operation_label_properties.label == b.operation_label_properties.label &&
              a.operation_label_properties.properties == b.operation_label_properties.properties;
+    case WalDeltaData::Type::SCHEMA_CREATE:
+    case WalDeltaData::Type::SCHEMA_DROP: {
+      return a.operation_label_create_schema.label == b.operation_label_create_schema.label;
+    }
   }
 }
 bool operator!=(const WalDeltaData &a, const WalDeltaData &b) { return !(a == b); }
@@ -615,6 +636,49 @@ void EncodeOperation(BaseEncoder *encoder, NameIdMapper *name_id_mapper, Storage
       }
       break;
     }
+    case StorageGlobalOperation::SCHEMA_CREATE:
+    case StorageGlobalOperation::SCHEMA_DROP: {
+      MG_ASSERT(!properties.empty(), "Invalid function call!");
+      encoder->WriteMarker(OperationToMarker(operation));
+      encoder->WriteString(name_id_mapper->IdToName(label.AsUint()));
+      encoder->WriteUint(properties.size());
+      for (const auto &property : properties) {
+        encoder->WriteString(name_id_mapper->IdToName(property.AsUint()));
+      }
+      break;
+    }
+  }
+}
+
+void EncodeOperation(BaseEncoder *encoder, NameIdMapper *name_id_mapper, StorageGlobalOperation operation,
+                     const Schemas::Schema &schema, uint64_t timestamp) {
+  encoder->WriteMarker(Marker::SECTION_DELTA);
+  encoder->WriteUint(timestamp);
+  switch (operation) {
+    case StorageGlobalOperation::LABEL_INDEX_CREATE:
+    case StorageGlobalOperation::LABEL_INDEX_DROP:
+    case StorageGlobalOperation::LABEL_PROPERTY_INDEX_CREATE:
+    case StorageGlobalOperation::LABEL_PROPERTY_INDEX_DROP:
+    case StorageGlobalOperation::EXISTENCE_CONSTRAINT_CREATE:
+    case StorageGlobalOperation::EXISTENCE_CONSTRAINT_DROP:
+    case StorageGlobalOperation::UNIQUE_CONSTRAINT_CREATE:
+    case StorageGlobalOperation::UNIQUE_CONSTRAINT_DROP:
+    case StorageGlobalOperation::SCHEMA_DROP: {
+      throw RecoveryFailure("Unsupported action!");
+    }
+    case StorageGlobalOperation::SCHEMA_CREATE: {
+      encoder->WriteMarker(OperationToMarker(operation));
+      encoder->WriteString(name_id_mapper->IdToName(schema.first.AsUint()));
+      encoder->WriteUint(schema.second.size());
+      for (const auto &schema_type : schema.second) {
+        encoder->WriteString(name_id_mapper->IdToName(schema_type.property_id.AsUint()));
+      }
+      encoder->WriteUint(schema.second.size());
+      for (const auto &schema_type : schema.second) {
+        encoder->WriteUint(static_cast<uint64_t>(schema_type.type));
+      }
+      break;
+    }
   }
 }
 
@@ -847,6 +911,10 @@ RecoveryInfo LoadWal(const std::filesystem::path &path, RecoveredIndicesAndConst
                                          "The unique constraint doesn't exist!");
           break;
         }
+        case WalDeltaData::Type::SCHEMA_CREATE:
+        case WalDeltaData::Type::SCHEMA_DROP: {
+          break;
+        }
       }
       ret.next_timestamp = std::max(ret.next_timestamp, timestamp + 1);
       ++deltas_applied;
@@ -967,6 +1035,11 @@ void WalFile::AppendOperation(StorageGlobalOperation operation, LabelId label, c
   UpdateStats(timestamp);
 }
 
+void WalFile::AppendOperation(StorageGlobalOperation operation, const Schemas::Schema &schema, uint64_t timestamp) {
+  EncodeOperation(&wal_, name_id_mapper_, operation, schema, timestamp);
+  UpdateStats(timestamp);
+}
+
 void WalFile::Sync() { wal_.Sync(); }
 
 uint64_t WalFile::GetSize() { return wal_.GetSize(); }

src/storage/v2/durability/wal.hpp

@@ -24,6 +24,7 @@
 #include "storage/v2/id_types.hpp"
 #include "storage/v2/name_id_mapper.hpp"
 #include "storage/v2/property_value.hpp"
+#include "storage/v2/schemas.hpp"
 #include "storage/v2/vertex.hpp"
 #include "utils/file_locker.hpp"
 #include "utils/skip_list.hpp"
@@ -63,6 +64,8 @@ struct WalDeltaData {
     EXISTENCE_CONSTRAINT_DROP,
     UNIQUE_CONSTRAINT_CREATE,
     UNIQUE_CONSTRAINT_DROP,
+    SCHEMA_CREATE,
+    SCHEMA_DROP,
   };
 
   Type type{Type::TRANSACTION_END};
@@ -102,6 +105,11 @@ struct WalDeltaData {
     std::string label;
     std::set<std::string> properties;
   } operation_label_properties;
+
+  struct {
+    std::string label;
+    std::vector<SchemaPropertyType> schema_properties_types;
+  } operation_label_create_schema;
 };
 
 bool operator==(const WalDeltaData &a, const WalDeltaData &b);
@@ -117,6 +125,8 @@ enum class StorageGlobalOperation {
   EXISTENCE_CONSTRAINT_DROP,
   UNIQUE_CONSTRAINT_CREATE,
   UNIQUE_CONSTRAINT_DROP,
+  SCHEMA_CREATE,
+  SCHEMA_DROP,
 };
 
 constexpr bool IsWalDeltaDataTypeTransactionEnd(const WalDeltaData::Type type) {
@@ -148,6 +158,8 @@ constexpr bool IsWalDeltaDataTypeTransactionEnd(const WalDeltaData::Type type) {
     case WalDeltaData::Type::EXISTENCE_CONSTRAINT_DROP:
     case WalDeltaData::Type::UNIQUE_CONSTRAINT_CREATE:
     case WalDeltaData::Type::UNIQUE_CONSTRAINT_DROP:
+    case WalDeltaData::Type::SCHEMA_CREATE:
+    case WalDeltaData::Type::SCHEMA_DROP:
       return true;
   }
 }
@@ -188,6 +200,10 @@ void EncodeTransactionEnd(BaseEncoder *encoder, uint64_t timestamp);
 void EncodeOperation(BaseEncoder *encoder, NameIdMapper *name_id_mapper, StorageGlobalOperation operation,
                      LabelId label, const std::set<PropertyId> &properties, uint64_t timestamp);
 
+/// Function used to encode non-transactional operation related.
+void EncodeOperation(BaseEncoder *encoder, NameIdMapper *name_id_mapper, StorageGlobalOperation operation,
+                     const Schemas::Schema &schema, uint64_t timestamp);
+
 /// Function used to load the WAL data into the storage.
 /// @throw RecoveryFailure
 RecoveryInfo LoadWal(const std::filesystem::path &path, RecoveredIndicesAndConstraints *indices_constraints,
@@ -218,6 +234,8 @@ class WalFile {
   void AppendOperation(StorageGlobalOperation operation, LabelId label, const std::set<PropertyId> &properties,
                        uint64_t timestamp);
 
+  void AppendOperation(StorageGlobalOperation operation, const Schemas::Schema &schema, uint64_t timestamp);
+
   void Sync();
 
   uint64_t GetSize();

src/storage/v2/replication/config.hpp

@@ -16,10 +16,6 @@
 namespace memgraph::storage::replication {
 struct ReplicationClientConfig {
   std::optional<double> timeout;
-  // The default delay between main checking/pinging replicas is 1s because
-  // that seems like a reasonable timeframe in which main should notice a
-  // replica is down.
-  std::chrono::seconds replica_check_frequency{1};
 
   struct SSL {
     std::string key_file = "";

src/storage/v2/replication/replication_client.cpp

@@ -41,49 +41,12 @@ Storage::ReplicationClient::ReplicationClient(std::string name, Storage *storage
   }
 
   rpc_client_.emplace(endpoint, &*rpc_context_);
-  TryInitializeClientSync();
+  TryInitializeClient();
 
   if (config.timeout && replica_state_ != replication::ReplicaState::INVALID) {
     timeout_.emplace(*config.timeout);
     timeout_dispatcher_.emplace();
   }
-
-  // Help the user to get the most accurate replica state possible.
-  if (config.replica_check_frequency > std::chrono::seconds(0)) {
-    replica_checker_.Run("Replica Checker", config.replica_check_frequency, [&] { FrequentCheck(); });
-  }
-}
-
-void Storage::ReplicationClient::TryInitializeClientAsync() {
-  thread_pool_.AddTask([this] {
-    rpc_client_->Abort();
-    this->TryInitializeClientSync();
-  });
-}
-
-void Storage::ReplicationClient::FrequentCheck() {
-  const auto is_success = std::invoke([this]() {
-    try {
-      auto stream{rpc_client_->Stream<replication::FrequentHeartbeatRpc>()};
-      const auto response = stream.AwaitResponse();
-      return response.success;
-    } catch (const rpc::RpcFailedException &) {
-      return false;
-    }
-  });
-  // States: READY, REPLICATING, RECOVERY, INVALID
-  // If success && ready, replicating, recovery -> stay the same because something good is going on.
-  // If success && INVALID -> [it's possible that replica came back to life] -> TryInitializeClient.
-  // If fail -> [replica is not reachable at all] -> INVALID state.
-  // NOTE: TryInitializeClient might return nothing if there is a branching point.
-  // NOTE: The early return pattern simplified the code, but the behavior should be as explained.
-  if (!is_success) {
-    replica_state_.store(replication::ReplicaState::INVALID);
-    return;
-  }
-  if (replica_state_.load() == replication::ReplicaState::INVALID) {
-    TryInitializeClientAsync();
-  }
 }
 
 /// @throws rpc::RpcFailedException
@@ -137,7 +100,7 @@ void Storage::ReplicationClient::InitializeClient() {
   }
 }
 
-void Storage::ReplicationClient::TryInitializeClientSync() {
+void Storage::ReplicationClient::TryInitializeClient() {
   try {
     InitializeClient();
   } catch (const rpc::RpcFailedException &) {
@@ -150,7 +113,10 @@ void Storage::ReplicationClient::TryInitializeClientSync() {
 
 void Storage::ReplicationClient::HandleRpcFailure() {
   spdlog::error(utils::MessageWithLink("Couldn't replicate data to {}.", name_, "https://memgr.ph/replication"));
-  TryInitializeClientAsync();
+  thread_pool_.AddTask([this] {
+    rpc_client_->Abort();
+    this->TryInitializeClient();
+  });
 }
 
 replication::SnapshotRes Storage::ReplicationClient::TransferSnapshot(const std::filesystem::path &path) {
@@ -227,18 +193,17 @@ void Storage::ReplicationClient::IfStreamingTransaction(const std::function<void
   }
 }
 
-std::optional<bool> Storage::ReplicationClient::FinalizeTransactionReplication() {
+void Storage::ReplicationClient::FinalizeTransactionReplication() {
   // We can only check the state because it guarantees to be only
   // valid during a single transaction replication (if the assumption
   // that this and other transaction replication functions can only be
   // called from a one thread stands)
   if (replica_state_ != replication::ReplicaState::REPLICATING) {
-    return std::nullopt;
+    return;
   }
 
   if (mode_ == replication::ReplicationMode::ASYNC) {
-    thread_pool_.AddTask([this] { [[maybe_unused]] auto finalized = this->FinalizeTransactionReplicationInternal(); });
-    return true;
+    thread_pool_.AddTask([this] { this->FinalizeTransactionReplicationInternal(); });
   } else if (timeout_) {
     MG_ASSERT(mode_ == replication::ReplicationMode::SYNC, "Only SYNC replica can have a timeout.");
     MG_ASSERT(timeout_dispatcher_, "Timeout thread is missing");
@@ -246,7 +211,7 @@ std::optional<bool> Storage::ReplicationClient::FinalizeTransactionReplication()
 
     timeout_dispatcher_->active = true;
     thread_pool_.AddTask([&, this] {
-      [[maybe_unused]] auto finalized = this->FinalizeTransactionReplicationInternal();
+      this->FinalizeTransactionReplicationInternal();
       std::unique_lock main_guard(timeout_dispatcher_->main_lock);
       // TimerThread can finish waiting for timeout
       timeout_dispatcher_->active = false;
@@ -274,13 +239,12 @@ std::optional<bool> Storage::ReplicationClient::FinalizeTransactionReplication()
       // and acces the `active` variable`
       thread_pool_.AddTask([this] { timeout_dispatcher_.reset(); });
     }
-    return true;
   } else {
-    return FinalizeTransactionReplicationInternal();
+    FinalizeTransactionReplicationInternal();
   }
 }
 
-bool Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
+void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
   MG_ASSERT(replica_stream_, "Missing stream for transaction deltas");
   try {
     auto response = replica_stream_->Finalize();
@@ -291,7 +255,6 @@ bool Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
       thread_pool_.AddTask([&, this] { this->RecoverReplica(response.current_commit_timestamp); });
     } else {
       replica_state_.store(replication::ReplicaState::READY);
-      return true;
     }
   } catch (const rpc::RpcFailedException &) {
     replica_stream_.reset();
@@ -301,7 +264,6 @@ bool Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
     }
     HandleRpcFailure();
   }
-  return false;
 }
 
 void Storage::ReplicationClient::RecoverReplica(uint64_t replica_commit) {
@@ -600,6 +562,12 @@ void Storage::ReplicationClient::ReplicaStream::AppendOperation(durability::Stor
   EncodeOperation(&encoder, &self_->storage_->name_id_mapper_, operation, label, properties, timestamp);
 }
 
+void Storage::ReplicationClient::ReplicaStream::AppendOperation(durability::StorageGlobalOperation operation,
+                                                                const Schemas::Schema &schema, uint64_t timestamp) {
+  replication::Encoder encoder(stream_.GetBuilder());
+  EncodeOperation(&encoder, &self_->storage_->name_id_mapper_, operation, schema, timestamp);
+}
+
 replication::AppendDeltasRes Storage::ReplicationClient::ReplicaStream::Finalize() { return stream_.AwaitResponse(); }
 
 ////// CurrentWalHandler //////

src/storage/v2/replication/replication_client.hpp

@@ -28,6 +28,7 @@
 #include "storage/v2/replication/enums.hpp"
 #include "storage/v2/replication/rpc.hpp"
 #include "storage/v2/replication/serialization.hpp"
+#include "storage/v2/schemas.hpp"
 #include "storage/v2/storage.hpp"
 #include "utils/file.hpp"
 #include "utils/file_locker.hpp"
@@ -62,6 +63,10 @@ class Storage::ReplicationClient {
     void AppendOperation(durability::StorageGlobalOperation operation, LabelId label,
                          const std::set<PropertyId> &properties, uint64_t timestamp);
 
+    /// @throw rpc::RpcFailedException
+    void AppendOperation(durability::StorageGlobalOperation operation, const Schemas::Schema &schema,
+                         uint64_t timestamp);
+
    private:
     /// @throw rpc::RpcFailedException
     replication::AppendDeltasRes Finalize();
@@ -103,10 +108,7 @@ class Storage::ReplicationClient {
   // StartTransactionReplication, stream is created.
   void IfStreamingTransaction(const std::function<void(ReplicaStream &handler)> &callback);
 
-  // Return none  -> OK
-  // Return true  -> OK
-  // Return false -> FAIL
-  std::optional<bool> FinalizeTransactionReplication();
+  void FinalizeTransactionReplication();
 
   // Transfer the snapshot file.
   // @param path Path of the snapshot file.
@@ -128,7 +130,7 @@ class Storage::ReplicationClient {
   const auto &Endpoint() const { return rpc_client_->Endpoint(); }
 
  private:
-  [[nodiscard]] bool FinalizeTransactionReplicationInternal();
+  void FinalizeTransactionReplicationInternal();
 
   void RecoverReplica(uint64_t replica_commit);
 
@@ -145,14 +147,16 @@ class Storage::ReplicationClient {
 
   std::vector<RecoveryStep> GetRecoverySteps(uint64_t replica_commit, utils::FileRetainer::FileLocker *file_locker);
 
-  void FrequentCheck();
   void InitializeClient();
-  void TryInitializeClientSync();
-  void TryInitializeClientAsync();
+
+  void TryInitializeClient();
+
   void HandleRpcFailure();
 
   std::string name_;
+
   Storage *storage_;
+
   std::optional<communication::ClientContext> rpc_context_;
   std::optional<rpc::Client> rpc_client_;
 
@@ -199,8 +203,6 @@ class Storage::ReplicationClient {
   //    to ignore concurrency problems inside the client.
   utils::ThreadPool thread_pool_{1};
   std::atomic<replication::ReplicaState> replica_state_{replication::ReplicaState::INVALID};
-
-  utils::Scheduler replica_checker_;
 };
 
 }  // namespace memgraph::storage

src/storage/v2/replication/replication_server.cpp

@@ -60,10 +60,6 @@ Storage::ReplicationServer::ReplicationServer(Storage *storage, io::network::End
     spdlog::debug("Received HeartbeatRpc");
     this->HeartbeatHandler(req_reader, res_builder);
   });
-  rpc_server_->Register<replication::FrequentHeartbeatRpc>([](auto *req_reader, auto *res_builder) {
-    spdlog::debug("Received FrequentHeartbeatRpc");
-    FrequentHeartbeatHandler(req_reader, res_builder);
-  });
   rpc_server_->Register<replication::AppendDeltasRpc>([this](auto *req_reader, auto *res_builder) {
     spdlog::debug("Received AppendDeltasRpc");
     this->AppendDeltasHandler(req_reader, res_builder);
@@ -90,13 +86,6 @@ void Storage::ReplicationServer::HeartbeatHandler(slk::Reader *req_reader, slk::
   slk::Save(res, res_builder);
 }
 
-void Storage::ReplicationServer::FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
-  replication::FrequentHeartbeatReq req;
-  slk::Load(&req, req_reader);
-  replication::FrequentHeartbeatRes res{true};
-  slk::Save(res, res_builder);
-}
-
 void Storage::ReplicationServer::AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
   replication::AppendDeltasReq req;
   slk::Load(&req, req_reader);
@@ -323,13 +312,13 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
     switch (delta.type) {
       case durability::WalDeltaData::Type::VERTEX_CREATE: {
         spdlog::trace("       Create vertex {}", delta.vertex_create_delete.gid.AsUint());
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         transaction->CreateVertex(delta.vertex_create_delete.gid);
         break;
       }
       case durability::WalDeltaData::Type::VERTEX_DELETE: {
         spdlog::trace("       Delete vertex {}", delta.vertex_create_delete.gid.AsUint());
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         auto vertex = transaction->FindVertex(delta.vertex_create_delete.gid, storage::View::NEW);
         if (!vertex) throw utils::BasicException("Invalid transaction!");
         auto ret = transaction->DeleteVertex(&*vertex);
@@ -339,7 +328,7 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
       case durability::WalDeltaData::Type::VERTEX_ADD_LABEL: {
         spdlog::trace("       Vertex {} add label {}", delta.vertex_add_remove_label.gid.AsUint(),
                       delta.vertex_add_remove_label.label);
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         auto vertex = transaction->FindVertex(delta.vertex_add_remove_label.gid, storage::View::NEW);
         if (!vertex) throw utils::BasicException("Invalid transaction!");
         auto ret = vertex->AddLabel(transaction->NameToLabel(delta.vertex_add_remove_label.label));
@@ -349,7 +338,7 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
       case durability::WalDeltaData::Type::VERTEX_REMOVE_LABEL: {
         spdlog::trace("       Vertex {} remove label {}", delta.vertex_add_remove_label.gid.AsUint(),
                       delta.vertex_add_remove_label.label);
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         auto vertex = transaction->FindVertex(delta.vertex_add_remove_label.gid, storage::View::NEW);
         if (!vertex) throw utils::BasicException("Invalid transaction!");
         auto ret = vertex->RemoveLabel(transaction->NameToLabel(delta.vertex_add_remove_label.label));
@@ -359,7 +348,7 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
       case durability::WalDeltaData::Type::VERTEX_SET_PROPERTY: {
         spdlog::trace("       Vertex {} set property {} to {}", delta.vertex_edge_set_property.gid.AsUint(),
                       delta.vertex_edge_set_property.property, delta.vertex_edge_set_property.value);
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         auto vertex = transaction->FindVertex(delta.vertex_edge_set_property.gid, storage::View::NEW);
         if (!vertex) throw utils::BasicException("Invalid transaction!");
         auto ret = vertex->SetProperty(transaction->NameToProperty(delta.vertex_edge_set_property.property),
@@ -371,7 +360,7 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
         spdlog::trace("       Create edge {} of type {} from vertex {} to vertex {}",
                       delta.edge_create_delete.gid.AsUint(), delta.edge_create_delete.edge_type,
                       delta.edge_create_delete.from_vertex.AsUint(), delta.edge_create_delete.to_vertex.AsUint());
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         auto from_vertex = transaction->FindVertex(delta.edge_create_delete.from_vertex, storage::View::NEW);
         if (!from_vertex) throw utils::BasicException("Invalid transaction!");
         auto to_vertex = transaction->FindVertex(delta.edge_create_delete.to_vertex, storage::View::NEW);
@@ -386,7 +375,7 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
         spdlog::trace("       Delete edge {} of type {} from vertex {} to vertex {}",
                       delta.edge_create_delete.gid.AsUint(), delta.edge_create_delete.edge_type,
                       delta.edge_create_delete.from_vertex.AsUint(), delta.edge_create_delete.to_vertex.AsUint());
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
         auto from_vertex = transaction->FindVertex(delta.edge_create_delete.from_vertex, storage::View::NEW);
         if (!from_vertex) throw utils::BasicException("Invalid transaction!");
         auto to_vertex = transaction->FindVertex(delta.edge_create_delete.to_vertex, storage::View::NEW);
@@ -409,7 +398,7 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
               "Can't set properties on edges because properties on edges "
               "are disabled!");
 
-        auto transaction = get_transaction(timestamp);
+        auto *transaction = get_transaction(timestamp);
 
         // The following block of code effectively implements `FindEdge` and
         // yields an accessor that is only valid for managing the edge's
@@ -561,6 +550,31 @@ uint64_t Storage::ReplicationServer::ReadAndApplyDelta(durability::BaseDecoder *
         if (ret != UniqueConstraints::DeletionStatus::SUCCESS) throw utils::BasicException("Invalid transaction!");
         break;
       }
+      case durability::WalDeltaData::Type::SCHEMA_CREATE: {
+        // std::stringstream ss;
+        // utils::PrintIterable(ss, delta.operation_label_create_schema);
+        // spdlog::trace("       Create schema on label :{}", delta.operation_label_create_schema.label, ss.str());
+        // if (commit_timestamp_and_accessor) {
+        //   throw utils::BasicException("Invalid transaction!");
+        // }
+        // if (!storage_->CreateSchema(storage_->NameToLabel(delta.operation_label_create_schema.label),
+        //                             delta.operation_label_create_schema.schema_properties_types, timestamp)) {
+        //   throw utils::BasicException("Invalid transaction!");
+        // }
+        break;
+      }
+      case durability::WalDeltaData::Type::SCHEMA_DROP: {
+        // std::stringstream ss;
+        // utils::PrintIterable(ss, delta.operation_label);
+        // spdlog::trace("       Drop schema on label :{}", delta.operation_label.label, ss.str());
+        // if (commit_timestamp_and_accessor) {
+        //   throw utils::BasicException("Invalid transaction!");
+        // }
+        // if (!storage_->DropSchema(storage_->NameToLabel(delta.operation_label.label), timestamp)) {
+        //   throw utils::BasicException("Invalid transaction!");
+        // }
+        break;
+      }
     }
   }
 

src/storage/v2/replication/replication_server.hpp

@@ -29,7 +29,6 @@ class Storage::ReplicationServer {
  private:
   // RPC handlers
   void HeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
-  static void FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
   void AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder);
   void SnapshotHandler(slk::Reader *req_reader, slk::Builder *res_builder);
   void WalFilesHandler(slk::Reader *req_reader, slk::Builder *res_builder);

src/storage/v2/replication/rpc.lcp

@@ -43,12 +43,6 @@ cpp<#
      (current-commit-timestamp :uint64_t)
      (epoch-id "std::string"))))
 
-;; FrequentHearthbeat is required because calling Heartbeat takes the storage lock.
-;; Configured by `replication_replica_check_delay`.
-(lcp:define-rpc frequent-heartbeat
-  (:request ())
-  (:response ((success :bool))))
-
 (lcp:define-rpc snapshot
   (:request ())
   (:response

src/storage/v2/replication/serialization.hpp

@@ -31,6 +31,8 @@ class Encoder final : public durability::BaseEncoder {
 
   void WriteUint(uint64_t value) override;
 
+  // void WriteUint(uint8_t value) override;
+
   void WriteDouble(double value) override;
 
   void WriteString(const std::string_view &value) override;

src/storage/v2/schemas.cpp

@@ -0,0 +1,83 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "storage/v2/property_value.hpp"
+#include "storage/v2/schemas.hpp"
+
+namespace memgraph::storage {
+
+SchemaViolation::SchemaViolation(ValidationStatus status, LabelId label) : status{status}, label{label} {}
+SchemaViolation::SchemaViolation(ValidationStatus status, LabelId label, SchemaPropertyType violated_type)
+    : status{status}, label{label}, violated_type{violated_type} {}
+
+SchemaViolation::SchemaViolation(ValidationStatus status, LabelId label, SchemaPropertyType violated_type,
+                                 PropertyValue violated_property_value)
+    : status{status}, label{label}, violated_type{violated_type}, violated_property_value{violated_property_value} {}
+
+Schemas::SchemasList Schemas::ListSchemas() const {
+  Schemas::SchemasList ret;
+  ret.reserve(schemas_.size());
+  for (const auto &[label_props, schema_property] : schemas_) {
+    ret.emplace_back(label_props, schema_property);
+  }
+  return ret;
+}
+
+std::optional<Schemas::Schema> Schemas::GetSchema(const LabelId primary_label) const {
+  if (auto schema_map = schemas_.find(primary_label); schema_map != schemas_.end()) {
+    return Schema{schema_map->first, schema_map->second};
+  }
+  return std::nullopt;
+}
+
+bool Schemas::CreateSchema(const LabelId primary_label, const std::vector<SchemaPropertyType> &schemas_types) {
+  if (schemas_.contains(primary_label)) {
+    return false;
+  }
+  schemas_.insert({primary_label, schemas_types});
+  return true;
+}
+
+bool Schemas::DropSchema(const LabelId primary_label) { return schemas_.erase(primary_label); }
+
+std::optional<SchemaViolation> Schemas::ValidateVertex(const LabelId primary_label, const Vertex &vertex) {
+  // TODO Check for multiple defined primary labels
+  const auto schema = schemas_.find(primary_label);
+  if (schema == schemas_.end()) {
+    return SchemaViolation(SchemaViolation::ValidationStatus::NO_SCHEMA_DEFINED_FOR_LABEL, primary_label);
+  }
+  if (!utils::Contains(vertex.labels, primary_label)) {
+    return SchemaViolation(SchemaViolation::ValidationStatus::VERTEX_HAS_NO_PRIMARY_LABEL, primary_label);
+  }
+
+  for (const auto &schema_type : schema->second) {
+    if (!vertex.properties.HasProperty(schema_type.property_id)) {
+      return SchemaViolation(SchemaViolation::ValidationStatus::VERTEX_HAS_NO_PROPERTY, primary_label, schema_type);
+    }
+    // Property type check
+    //  TODO Can this be replaced with just property id check?
+    if (auto vertex_property = vertex.properties.GetProperty(schema_type.property_id);
+        PropertyTypeToSchemaType(vertex_property) != schema_type.type) {
+      return SchemaViolation(SchemaViolation::ValidationStatus::VERTEX_PROPERTY_WRONG_TYPE, primary_label, schema_type,
+                             vertex_property);
+    }
+  }
+  // TODO after the introduction of vertex hashing introduce check for vertex
+  // primary key uniqueness
+
+  return std::nullopt;
+}
+
+}  // namespace memgraph::storage

src/storage/v2/schemas.hpp

@@ -0,0 +1,156 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "common/types.hpp"
+#include "storage/v2/id_types.hpp"
+#include "storage/v2/indices.hpp"
+#include "storage/v2/property_value.hpp"
+#include "storage/v2/temporal.hpp"
+#include "storage/v2/transaction.hpp"
+#include "storage/v2/vertex.hpp"
+#include "utils/result.hpp"
+
+namespace memgraph::storage {
+
+class SchemaViolationException : public utils::BasicException {
+  using utils::BasicException::BasicException;
+};
+
+struct SchemaPropertyType {
+  common::SchemaType type;
+  PropertyId property_id;
+};
+
+struct SchemaViolation {
+  enum class ValidationStatus : uint8_t {
+    VERTEX_HAS_NO_PRIMARY_LABEL,
+    VERTEX_HAS_NO_PROPERTY,
+    NO_SCHEMA_DEFINED_FOR_LABEL,
+    VERTEX_PROPERTY_WRONG_TYPE
+  };
+
+  SchemaViolation(ValidationStatus status, LabelId label);
+
+  SchemaViolation(ValidationStatus status, LabelId label, SchemaPropertyType violated_type);
+
+  SchemaViolation(ValidationStatus status, LabelId label, SchemaPropertyType violated_type,
+                  PropertyValue violated_property_value);
+
+  ValidationStatus status;
+  LabelId label;
+  std::optional<SchemaPropertyType> violated_type;
+  std::optional<PropertyValue> violated_property_value;
+};
+
+/// Structure that represents a collection of schemas
+/// Schema can be mapped under only one label => primary label
+class Schemas {
+ public:
+  using Schema = std::pair<LabelId, std::vector<SchemaPropertyType>>;
+  using SchemasMap = std::unordered_map<LabelId, std::vector<SchemaPropertyType>>;
+  using SchemasList = std::vector<Schema>;
+
+  Schemas() = default;
+  Schemas(const Schemas &) = delete;
+  Schemas(Schemas &&) = delete;
+  Schemas &operator=(const Schemas &) = delete;
+  Schemas &operator=(Schemas &&) = delete;
+  ~Schemas() = default;
+
+  [[nodiscard]] SchemasList ListSchemas() const;
+
+  [[nodiscard]] std::optional<Schemas::Schema> GetSchema(LabelId primary_label) const;
+
+  // Returns true if it was successfully created or false if the schema
+  // already exists
+  [[nodiscard]] bool CreateSchema(LabelId label, const std::vector<SchemaPropertyType> &schemas_types);
+
+  // Returns true if it was successfully dropped or false if the schema
+  // does not exist
+  [[nodiscard]] bool DropSchema(LabelId label);
+
+  [[nodiscard]] std::optional<SchemaViolation> ValidateVertex(LabelId primary_label, const Vertex &vertex);
+
+ private:
+  SchemasMap schemas_;
+};
+
+inline std::optional<common::SchemaType> PropertyTypeToSchemaType(const PropertyValue &property_value) {
+  switch (property_value.type()) {
+    case PropertyValue::Type::Bool: {
+      return common::SchemaType::BOOL;
+    }
+    case PropertyValue::Type::Int: {
+      return common::SchemaType::INT;
+    }
+    case PropertyValue::Type::String: {
+      return common::SchemaType::STRING;
+    }
+    case PropertyValue::Type::TemporalData: {
+      switch (property_value.ValueTemporalData().type) {
+        case TemporalType::Date: {
+          return common::SchemaType::DATE;
+        }
+        case TemporalType::LocalDateTime: {
+          return common::SchemaType::LOCALDATETIME;
+        }
+        case TemporalType::LocalTime: {
+          return common::SchemaType::LOCALTIME;
+        }
+        case TemporalType::Duration: {
+          return common::SchemaType::DURATION;
+        }
+      }
+    }
+    case PropertyValue::Type::Double:
+    case PropertyValue::Type::Null:
+    case PropertyValue::Type::Map:
+    case PropertyValue::Type::List: {
+      return std::nullopt;
+    }
+  }
+}
+
+inline std::string SchemaTypeToString(const common::SchemaType type) {
+  switch (type) {
+    case common::SchemaType::BOOL: {
+      return "Bool";
+    }
+    case common::SchemaType::INT: {
+      return "Integer";
+    }
+    case common::SchemaType::STRING: {
+      return "String";
+    }
+    case common::SchemaType::DATE: {
+      return "Date";
+    }
+    case common::SchemaType::LOCALTIME: {
+      return "LocalTime";
+    }
+    case common::SchemaType::LOCALDATETIME: {
+      return "LocalDateTime";
+    }
+    case common::SchemaType::DURATION: {
+      return "Duration";
+    }
+  }
+}
+
+}  // namespace memgraph::storage

src/storage/v2/storage.cpp

@@ -28,6 +28,7 @@
 #include "storage/v2/indices.hpp"
 #include "storage/v2/mvcc.hpp"
 #include "storage/v2/replication/config.hpp"
+#include "storage/v2/schemas.hpp"
 #include "storage/v2/transaction.hpp"
 #include "storage/v2/vertex_accessor.hpp"
 #include "utils/file.hpp"
@@ -37,10 +38,10 @@
 #include "utils/rw_lock.hpp"
 #include "utils/spin_lock.hpp"
 #include "utils/stat.hpp"
+#include "utils/synchronized.hpp"
 #include "utils/uuid.hpp"
 
 /// REPLICATION ///
-#include "storage/v2/commit_error.hpp"
 #include "storage/v2/replication/replication_client.hpp"
 #include "storage/v2/replication/replication_server.hpp"
 #include "storage/v2/replication/rpc.hpp"
@@ -457,12 +458,13 @@ VertexAccessor Storage::Accessor::CreateVertex() {
   OOMExceptionEnabler oom_exception;
   auto gid = storage_->vertex_id_.fetch_add(1, std::memory_order_acq_rel);
   auto acc = storage_->vertices_.access();
-  auto delta = CreateDeleteObjectDelta(&transaction_);
+  auto *delta = CreateDeleteObjectDelta(&transaction_);
   auto [it, inserted] = acc.insert(Vertex{storage::Gid::FromUint(gid), delta});
   MG_ASSERT(inserted, "The vertex must be inserted here!");
   MG_ASSERT(it != acc.end(), "Invalid Vertex accessor!");
+
   delta->prev.Set(&*it);
-  return VertexAccessor(&*it, &transaction_, &storage_->indices_, &storage_->constraints_, config_);
+  return {&*it, &transaction_, &storage_->indices_, &storage_->constraints_, config_};
 }
 
 VertexAccessor Storage::Accessor::CreateVertex(storage::Gid gid) {
@@ -814,7 +816,7 @@ EdgeTypeId Storage::Accessor::NameToEdgeType(const std::string_view &name) { ret
 
 void Storage::Accessor::AdvanceCommand() { ++transaction_.command_id; }
 
-utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
+utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
     const std::optional<uint64_t> desired_commit_timestamp) {
   MG_ASSERT(is_transaction_active_, "The transaction is already terminated!");
   MG_ASSERT(!transaction_.must_abort, "The transaction can't be committed!");
@@ -837,8 +839,7 @@ utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
       auto validation_result = ValidateExistenceConstraints(*prev.vertex, storage_->constraints_);
       if (validation_result) {
         Abort();
-        return CommitError{.type = CommitError::Type::CONSTRAINT_VIOLATION,
-                           .maybe_constraint_violation = *validation_result};
+        return *validation_result;
       }
     }
 
@@ -850,29 +851,6 @@ utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
     // Save these so we can mark them used in the commit log.
     uint64_t start_timestamp = transaction_.start_timestamp;
 
-    // Aboring here and after the locked block obviously has an issue if
-    // replica goes down and back up during the execution of the locked block of
-    // code.
-    // Not enough, before the actual commit check all SYNC replicas for availability.
-    bool unable_to_sync_replicate = false;
-    const auto check_replicas = [&]() {
-      storage_->replication_clients_.WithLock([&](auto &clients) {
-        for (auto &client : clients) {
-          // Exclusively SYNC replicas have to be available to commit the transaction.
-          if (client->Mode() == replication::ReplicationMode::SYNC && !client->Timeout().has_value() &&
-              client->State() == replication::ReplicaState::INVALID) {
-            unable_to_sync_replicate = true;
-          }
-        }
-      });
-    };
-
-    check_replicas();
-    if (unable_to_sync_replicate) {
-      Abort();
-      return CommitError{.type = CommitError::Type::UNABLE_TO_SYNC_REPLICATE};
-    }
-
     {
       std::unique_lock<utils::SpinLock> engine_guard(storage_->engine_lock_);
       commit_timestamp_.emplace(storage_->CommitTimestamp(desired_commit_timestamp));
@@ -918,8 +896,6 @@ utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
         // Replica can log only the write transaction received from Main
         // so the Wal files are consistent
         if (storage_->replication_role_ == ReplicationRole::MAIN || desired_commit_timestamp.has_value()) {
-          // TODO(gitbuda): Possible to abort data operation because in this context there is an abort operation.
-          // TODO(gitbuda): If AppendToWal returns false, we can exit this block and Abort.
           storage_->AppendToWal(transaction_, *commit_timestamp_);
         }
 
@@ -942,27 +918,13 @@ utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
           engine_guard.unlock();
         });
 
-        // NOTE: This will finish/commit the transaction.
         storage_->commit_log_->MarkFinished(start_timestamp);
-        // TODO(gitbuda): Maybe the solution here is to 1. mark 2. check all
-        // the replicas 3. unmark if required... it's not possible to unmark
-        // easily because mark contains mark + update_latest_active which is
-        // not reversable operation
       }
     }
 
-    // TODO(gitbuda): This doesn't have any effect because (it will only if the
-    // constraints are also violated).
-    check_replicas();
-    if (unable_to_sync_replicate) {
-      Abort();
-      return CommitError{.type = CommitError::Type::UNABLE_TO_SYNC_REPLICATE};
-    }
-
     if (unique_constraint_violation) {
       Abort();
-      return CommitError{.type = CommitError::Type::CONSTRAINT_VIOLATION,
-                         .maybe_constraint_violation = *unique_constraint_violation};
+      return *unique_constraint_violation;
     }
   }
   is_transaction_active_ = false;
@@ -1165,18 +1127,6 @@ EdgeTypeId Storage::NameToEdgeType(const std::string_view &name) {
   return EdgeTypeId::FromUint(name_id_mapper_.NameToId(name));
 }
 
-// TODO(gitbuda): Hard to abort global operations in SYNC replication mode
-// because there is no an abort op for that yet, one idea is to just apply
-// reverse operation, e.g., CreateIndex <-> DropIndex.
-//
-// Another idea is to double check that all replicas have relevant data prior
-// to calling MarkFinished. That approach would work in both data replication
-// and global operation cases.
-//
-// EDGE CASE 1: What if the first SYNC replica is alive, receives the delta
-// object, while the second SYNC replica is dead? (replication clients are
-// stored in a vector and accessed one by one)
-
 bool Storage::CreateIndex(LabelId label, const std::optional<uint64_t> desired_commit_timestamp) {
   std::unique_lock<utils::RWLock> storage_guard(main_lock_);
   if (!indices_.label_index.CreateIndex(label, vertices_.access())) return false;
@@ -1280,6 +1230,49 @@ ConstraintsInfo Storage::ListAllConstraints() const {
   return {ListExistenceConstraints(constraints_), constraints_.unique_constraints.ListConstraints()};
 }
 
+SchemasInfo Storage::ListAllSchemas() const {
+  std::shared_lock<utils::RWLock> storage_guard_(main_lock_);
+  return {schemas_.ListSchemas()};
+}
+
+SchemasInfo Storage::GetSchema(const LabelId primary_label) const {
+  std::shared_lock<utils::RWLock> storage_guard_(main_lock_);
+  if (const auto schema = schemas_.GetSchema(primary_label); schema) {
+    return {{*schema}};
+  }
+  return {};
+}
+
+bool Storage::CreateSchema(const LabelId primary_label, const std::vector<SchemaPropertyType> &schemas_types,
+                           const std::optional<uint64_t> desired_commit_timestamp) {
+  std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+  auto ret = schemas_.CreateSchema(primary_label, schemas_types);
+  if (!ret) {
+    return ret;
+  }
+  const auto commit_timestamp = CommitTimestamp(desired_commit_timestamp);
+  AppendToWal(durability::StorageGlobalOperation::EXISTENCE_CONSTRAINT_CREATE, *schemas_.GetSchema(primary_label),
+              commit_timestamp);
+  commit_log_->MarkFinished(commit_timestamp);
+  last_commit_timestamp_ = commit_timestamp;
+
+  return schemas_.CreateSchema(primary_label, schemas_types);
+}
+
+bool Storage::DropSchema(const LabelId primary_label, std::optional<uint64_t> desired_commit_timestamp) {
+  std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+  auto res = schemas_.DropSchema(primary_label);
+  if (!res) {
+    return res;
+  }
+  const auto commit_timestamp = CommitTimestamp(desired_commit_timestamp);
+  AppendToWal(durability::StorageGlobalOperation::SCHEMA_DROP, primary_label, {}, commit_timestamp);
+  commit_log_->MarkFinished(commit_timestamp);
+  last_commit_timestamp_ = commit_timestamp;
+
+  return res;
+}
+
 StorageInfo Storage::GetInfo() const {
   auto vertex_count = vertices_.size();
   auto edge_count = edge_count_.load(std::memory_order_acquire);
@@ -1625,9 +1618,6 @@ void Storage::FinalizeWalFile() {
   }
 }
 
-// TODO(gitbuda): Hard to abort data operation in SYNC replication mode because:
-//   * Just calling Abort inside AppendToWal for some reason causes infinite loop.
-
 void Storage::AppendToWal(const Transaction &transaction, uint64_t final_commit_timestamp) {
   if (!InitializeWalFile()) return;
   // Traverse deltas and append them to the WAL file.
@@ -1799,15 +1789,10 @@ void Storage::AppendToWal(const Transaction &transaction, uint64_t final_commit_
   FinalizeWalFile();
 
   replication_clients_.WithLock([&](auto &clients) {
-    bool all_sync_replicas_ok = true;
     for (auto &client : clients) {
-      // TODO(gitbuda): SEMI-SYNC should be exculded from here.
-      if (client->Mode() == replication::ReplicationMode::SYNC)
       client->IfStreamingTransaction([&](auto &stream) { stream.AppendTransactionEnd(final_commit_timestamp); });
-      // TODO(gitbuda): FinalizeTransactionReplication should also indicate that eveything went well for SYNC replicas.
       client->FinalizeTransactionReplication();
     }
-    return all_sync_replicas_ok;
   });
 }
 
@@ -1830,6 +1815,25 @@ void Storage::AppendToWal(durability::StorageGlobalOperation operation, LabelId
   FinalizeWalFile();
 }
 
+void Storage::AppendToWal(durability::StorageGlobalOperation operation, const Schemas::Schema &schema,
+                          uint64_t final_commit_timestamp) {
+  if (!InitializeWalFile()) return;
+  wal_file_->AppendOperation(operation, schema, final_commit_timestamp);
+  {
+    if (replication_role_.load() == ReplicationRole::MAIN) {
+      replication_clients_.WithLock([&](auto &clients) {
+        for (auto &client : clients) {
+          client->StartTransactionReplication(wal_file_->SequenceNumber());
+          client->IfStreamingTransaction(
+              [&](auto &stream) { stream.AppendOperation(operation, schema, final_commit_timestamp); });
+          client->FinalizeTransactionReplication();
+        }
+      });
+    }
+  }
+  FinalizeWalFile();
+}
+
 utils::BasicResult<Storage::CreateSnapshotError> Storage::CreateSnapshot() {
   if (replication_role_.load() != ReplicationRole::MAIN) {
     return CreateSnapshotError::DisabledForReplica;

src/storage/v2/storage.hpp

@@ -16,6 +16,7 @@
 #include <optional>
 #include <shared_mutex>
 #include <variant>
+#include <vector>
 
 #include "io/network/endpoint.hpp"
 #include "storage/v2/commit_log.hpp"
@@ -25,14 +26,18 @@
 #include "storage/v2/durability/wal.hpp"
 #include "storage/v2/edge.hpp"
 #include "storage/v2/edge_accessor.hpp"
+#include "storage/v2/id_types.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/property_value.hpp"
 #include "storage/v2/result.hpp"
+#include "storage/v2/schemas.hpp"
 #include "storage/v2/transaction.hpp"
 #include "storage/v2/vertex.hpp"
 #include "storage/v2/vertex_accessor.hpp"
+#include "utils/exceptions.hpp"
 #include "utils/file_locker.hpp"
 #include "utils/on_scope_exit.hpp"
 #include "utils/rw_lock.hpp"
@@ -43,7 +48,6 @@
 
 /// REPLICATION ///
 #include "rpc/server.hpp"
-#include "storage/v2/commit_error.hpp"
 #include "storage/v2/replication/config.hpp"
 #include "storage/v2/replication/enums.hpp"
 #include "storage/v2/replication/rpc.hpp"
@@ -174,6 +178,11 @@ struct ConstraintsInfo {
   std::vector<std::pair<LabelId, std::set<PropertyId>>> unique;
 };
 
+/// Structure used to return information about existing schemas in the storage
+struct SchemasInfo {
+  Schemas::SchemasList schemas;
+};
+
 /// Structure used to return information about the storage.
 struct StorageInfo {
   uint64_t vertex_count;
@@ -309,12 +318,11 @@ class Storage final {
 
     void AdvanceCommand();
 
-    /// Commit returns `CommitError` if the changes made by this transaction
-    /// violate an existence, unique constraint or data could NOT be replicated
-    /// to SYNC replica. In that case the transaction is automatically aborted.
-    /// Otherwise, void is returned.
+    /// Commit returns `ConstraintViolation` if the changes made by this
+    /// transaction violate an existence or unique constraint. In that case the
+    /// transaction is automatically aborted. Otherwise, void is returned.
     /// @throw std::bad_alloc
-    utils::BasicResult<CommitError, void> Commit(std::optional<uint64_t> desired_commit_timestamp = {});
+    utils::BasicResult<ConstraintViolation, void> Commit(std::optional<uint64_t> desired_commit_timestamp = {});
 
     /// @throw std::bad_alloc
     void Abort();
@@ -366,7 +374,7 @@ class Storage final {
   IndicesInfo ListAllIndices() const;
 
   /// Creates an existence constraint. Returns true if the constraint was
-  /// successfuly added, false if it already exists and a `ConstraintViolation`
+  /// successfully added, false if it already exists and a `ConstraintViolation`
   /// if there is an existing vertex violating the constraint.
   ///
   /// @throw std::bad_alloc
@@ -404,6 +412,15 @@ class Storage final {
 
   ConstraintsInfo ListAllConstraints() const;
 
+  SchemasInfo ListAllSchemas() const;
+
+  SchemasInfo GetSchema(LabelId primary_label) const;
+
+  bool CreateSchema(LabelId primary_label, const std::vector<SchemaPropertyType> &schemas_types,
+                    std::optional<uint64_t> desired_commit_timestamp = {});
+
+  bool DropSchema(LabelId primary_label, std::optional<uint64_t> desired_commit_timestamp = {});
+
   StorageInfo GetInfo() const;
 
   bool LockPath();
@@ -468,6 +485,8 @@ class Storage final {
   void AppendToWal(const Transaction &transaction, uint64_t final_commit_timestamp);
   void AppendToWal(durability::StorageGlobalOperation operation, LabelId label, const std::set<PropertyId> &properties,
                    uint64_t final_commit_timestamp);
+  void AppendToWal(durability::StorageGlobalOperation operation, const Schemas::Schema &schema,
+                   uint64_t final_commit_timestamp);
 
   uint64_t CommitTimestamp(std::optional<uint64_t> desired_commit_timestamp = {});
 
@@ -493,6 +512,7 @@ class Storage final {
 
   Constraints constraints_;
   Indices indices_;
+  Schemas schemas_;
 
   // Transaction engine
   utils::SpinLock engine_lock_;

tests/e2e/replication/CMakeLists.txt

@@ -5,7 +5,3 @@ target_link_libraries(memgraph__e2e__replication__constraints gflags mgclient mg
 
 add_executable(memgraph__e2e__replication__read_write_benchmark read_write_benchmark.cpp)
 target_link_libraries(memgraph__e2e__replication__read_write_benchmark gflags json mgclient mg-utils mg-io Threads::Threads)
-
-copy_e2e_python_files(replication_show common.py)
-copy_e2e_python_files(replication_show conftest.py)
-copy_e2e_python_files(replication_show show.py)

tests/e2e/replication/common.py

@@ -1,26 +0,0 @@
-# Copyright 2022 Memgraph Ltd.
-#
-# Use of this software is governed by the Business Source License
-# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
-# License, and you may not use this file except in compliance with the Business Source License.
-#
-# As of the Change Date specified in that file, in accordance with
-# the Business Source License, use of this software will be governed
-# by the Apache License, Version 2.0, included in the file
-# licenses/APL.txt.
-
-import mgclient
-import typing
-
-
-def execute_and_fetch_all(
-    cursor: mgclient.Cursor, query: str, params: dict = {}
-) -> typing.List[tuple]:
-    cursor.execute(query, params)
-    return cursor.fetchall()
-
-
-def connect(**kwargs) -> mgclient.Connection:
-    connection = mgclient.connect(**kwargs)
-    connection.autocommit = True
-    return connection

tests/e2e/replication/conftest.py

@@ -1,44 +0,0 @@
-# Copyright 2022 Memgraph Ltd.
-#
-# Use of this software is governed by the Business Source License
-# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
-# License, and you may not use this file except in compliance with the Business Source License.
-#
-# As of the Change Date specified in that file, in accordance with
-# the Business Source License, use of this software will be governed
-# by the Apache License, Version 2.0, included in the file
-# licenses/APL.txt.
-
-import pytest
-
-from common import execute_and_fetch_all, connect
-
-
-# The fixture here is more complex because the connection has to be
-# parameterized based on the test parameters (info has to be available on both
-# sides).
-#
-# https://docs.pytest.org/en/latest/example/parametrize.html#indirect-parametrization
-# is not an elegant/feasible solution here.
-#
-# The solution was independently developed and then I stumbled upon the same
-# approach here https://stackoverflow.com/a/68286553/4888809 which I think is
-# optimal.
-@pytest.fixture(scope="function")
-def connection():
-    connection_holder = None
-    role_holder = None
-
-    def inner_connection(port, role):
-        nonlocal connection_holder, role_holder
-        connection_holder = connect(host="localhost", port=port)
-        role_holder = role
-        return connection_holder
-
-    yield inner_connection
-
-    # Only main instance can be cleaned up because replicas do NOT accept
-    # writes.
-    if role_holder == "main":
-        cursor = connection_holder.cursor()
-        execute_and_fetch_all(cursor, "MATCH (n) DETACH DELETE n;")

tests/e2e/replication/show.py

@@ -1,46 +0,0 @@
-# Copyright 2022 Memgraph Ltd.
-#
-# Use of this software is governed by the Business Source License
-# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
-# License, and you may not use this file except in compliance with the Business Source License.
-#
-# As of the Change Date specified in that file, in accordance with
-# the Business Source License, use of this software will be governed
-# by the Apache License, Version 2.0, included in the file
-# licenses/APL.txt.
-
-import sys
-
-import pytest
-from common import execute_and_fetch_all
-
-
-@pytest.mark.parametrize(
-    "port, role",
-    [(7687, "main"), (7688, "replica"), (7689, "replica"), (7690, "replica")],
-)
-def test_show_replication_role(port, role, connection):
-    cursor = connection(port, role).cursor()
-    data = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;")
-    assert cursor.description[0].name == "replication role"
-    assert data[0][0] == role
-
-
-def test_show_replicas(connection):
-    cursor = connection(7687, "main").cursor()
-    actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
-
-    expected_column_names = {"name", "socket_address", "sync_mode", "timeout"}
-    actual_column_names = {x.name for x in cursor.description}
-    assert expected_column_names == actual_column_names
-
-    expected_data = {
-        ("replica_1", "127.0.0.1:10001", "sync", 0),
-        ("replica_2", "127.0.0.1:10002", "sync", 1.0),
-        ("replica_3", "127.0.0.1:10003", "async", None),
-    }
-    assert expected_data == actual_data
-
-
-if __name__ == "__main__":
-    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/replication/workloads.yaml

@@ -46,31 +46,4 @@ workloads:
     args: []
     <<: *template_cluster
 
-  - name: "Show"
-    binary: "tests/e2e/pytest_runner.sh"
-    args: ["replication/show.py"]
-    cluster:
-      replica_1:
-        args: ["--bolt-port", "7688", "--log-level=TRACE"]
-        log_file: "replication-e2e-replica1.log"
-        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10001;"]
-        validation_queries: []
-      replica_2:
-        args: ["--bolt-port", "7689", "--log-level=TRACE"]
-        log_file: "replication-e2e-replica2.log"
-        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10002;"]
-        validation_queries: []
-      replica_3:
-        args: ["--bolt-port", "7690", "--log-level=TRACE"]
-        log_file: "replication-e2e-replica3.log"
-        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10003;"]
-        validation_queries: []
-      main:
-        args: ["--bolt-port", "7687", "--log-level=TRACE"]
-        log_file: "replication-e2e-main.log"
-        setup_queries: [
-          "REGISTER REPLICA replica_1 SYNC WITH TIMEOUT 0 TO '127.0.0.1:10001'",
-          "REGISTER REPLICA replica_2 SYNC WITH TIMEOUT 1 TO '127.0.0.1:10002'",
-          "REGISTER REPLICA replica_3 ASYNC TO '127.0.0.1:10003'"
-        ]
-        validation_queries: []
+

tests/unit/cypher_main_visitor.cpp

@@ -2213,6 +2213,8 @@ TEST_P(CypherMainVisitorTest, GrantPrivilege) {
                    {AuthQuery::Privilege::MODULE_READ});
   check_auth_query(&ast_generator, "GRANT MODULE_WRITE TO user", AuthQuery::Action::GRANT_PRIVILEGE, "", "", "user", {},
                    {AuthQuery::Privilege::MODULE_WRITE});
+  check_auth_query(&ast_generator, "GRANT SCHEMA TO user", AuthQuery::Action::GRANT_PRIVILEGE, "", "", "user", {},
+                   {AuthQuery::Privilege::SCHEMA});
 }
 
 TEST_P(CypherMainVisitorTest, DenyPrivilege) {
@@ -2253,6 +2255,8 @@ TEST_P(CypherMainVisitorTest, DenyPrivilege) {
                    {AuthQuery::Privilege::MODULE_READ});
   check_auth_query(&ast_generator, "DENY MODULE_WRITE TO user", AuthQuery::Action::DENY_PRIVILEGE, "", "", "user", {},
                    {AuthQuery::Privilege::MODULE_WRITE});
+  check_auth_query(&ast_generator, "DENY SCHEMA TO user", AuthQuery::Action::DENY_PRIVILEGE, "", "", "user", {},
+                   {AuthQuery::Privilege::SCHEMA});
 }
 
 TEST_P(CypherMainVisitorTest, RevokePrivilege) {
@@ -2295,6 +2299,8 @@ TEST_P(CypherMainVisitorTest, RevokePrivilege) {
                    {}, {AuthQuery::Privilege::MODULE_READ});
   check_auth_query(&ast_generator, "REVOKE MODULE_WRITE FROM user", AuthQuery::Action::REVOKE_PRIVILEGE, "", "", "user",
                    {}, {AuthQuery::Privilege::MODULE_WRITE});
+  check_auth_query(&ast_generator, "REVOKE SCHEMA FROM user", AuthQuery::Action::REVOKE_PRIVILEGE, "", "", "user", {},
+                   {AuthQuery::Privilege::SCHEMA});
 }
 
 TEST_P(CypherMainVisitorTest, ShowPrivileges) {
@@ -4211,3 +4217,76 @@ TEST_P(CypherMainVisitorTest, Foreach) {
     ASSERT_TRUE(dynamic_cast<RemoveProperty *>(*++clauses.begin()));
   }
 }
+
+TEST_P(CypherMainVisitorTest, TestShowSchemas) {
+  auto &ast_generator = *GetParam();
+  auto *query = dynamic_cast<SchemaQuery *>(ast_generator.ParseQuery("SHOW SCHEMAS"));
+  ASSERT_TRUE(query);
+  EXPECT_EQ(query->action_, SchemaQuery::Action::SHOW_SCHEMAS);
+}
+
+TEST_P(CypherMainVisitorTest, TestShowSchema) {
+  auto &ast_generator = *GetParam();
+  EXPECT_THROW(ast_generator.ParseQuery("SHOW SCHEMA ON label"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("SHOW SCHEMA :label"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("SHOW SCHEMA label"), SyntaxException);
+
+  auto *query = dynamic_cast<SchemaQuery *>(ast_generator.ParseQuery("SHOW SCHEMA ON :label"));
+  ASSERT_TRUE(query);
+  EXPECT_EQ(query->action_, SchemaQuery::Action::SHOW_SCHEMA);
+  EXPECT_EQ(query->label_.name, "label");
+}
+
+TEST_P(CypherMainVisitorTest, TestCreateSchema) {
+  auto &ast_generator = *GetParam();
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label()"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label(123 INTEGER)"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label(name TYPE)"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label(name, age)"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label(name, DURATION)"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON label(name INTEGER)"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label(name INTEGER, name INTEGER)"), SemanticException);
+  EXPECT_THROW(ast_generator.ParseQuery("CREATE SCHEMA ON :label(name INTEGER, name STRING)"), SemanticException);
+
+  {
+    auto *query = dynamic_cast<SchemaQuery *>(ast_generator.ParseQuery("CREATE SCHEMA ON :label1(name STRING)"));
+    ASSERT_TRUE(query);
+    EXPECT_EQ(query->action_, SchemaQuery::Action::CREATE_SCHEMA);
+    EXPECT_EQ(query->label_.name, "label1");
+  }
+  {
+    auto *query = dynamic_cast<SchemaQuery *>(ast_generator.ParseQuery("CREATE SCHEMA ON :label2(name string)"));
+    ASSERT_TRUE(query);
+    EXPECT_EQ(query->action_, SchemaQuery::Action::CREATE_SCHEMA);
+    EXPECT_EQ(query->label_.name, "label2");
+  }
+  {
+    auto *query = dynamic_cast<SchemaQuery *>(
+        ast_generator.ParseQuery("CREATE SCHEMA ON :label3(first_name STRING, last_name STRING)"));
+    ASSERT_TRUE(query);
+    EXPECT_EQ(query->action_, SchemaQuery::Action::CREATE_SCHEMA);
+    EXPECT_EQ(query->label_.name, "label3");
+  }
+  {
+    auto *query = dynamic_cast<SchemaQuery *>(
+        ast_generator.ParseQuery("CREATE SCHEMA ON :label4(name STRING, age INTEGER, dur DURATION, birthday "
+                                 "LOCALDATETIME, some_time LOCALTIME, speaks_truth BOOL)"));
+    ASSERT_TRUE(query);
+    EXPECT_EQ(query->action_, SchemaQuery::Action::CREATE_SCHEMA);
+    EXPECT_EQ(query->label_.name, "label4");
+  }
+}
+
+TEST_P(CypherMainVisitorTest, TestDropSchema) {
+  auto &ast_generator = *GetParam();
+  EXPECT_THROW(ast_generator.ParseQuery("DROP SCHEMA"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("DROP SCHEMA ON label"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("DROP SCHEMA :label"), SyntaxException);
+  EXPECT_THROW(ast_generator.ParseQuery("DROP SCHEMA ON :label()"), SyntaxException);
+
+  auto *query = dynamic_cast<SchemaQuery *>(ast_generator.ParseQuery("DROP SCHEMA ON :label"));
+  ASSERT_TRUE(query);
+  EXPECT_EQ(query->action_, SchemaQuery::Action::DROP_SCHEMA);
+  EXPECT_EQ(query->label_.name, "label");
+}

tests/unit/interpreter.cpp

@@ -10,8 +10,10 @@
 // licenses/APL.txt.
 
 #include <algorithm>
+#include <cstddef>
 #include <cstdlib>
 #include <filesystem>
+#include <unordered_set>
 
 #include "communication/bolt/v1/value.hpp"
 #include "communication/result_stream_faker.hpp"
@@ -38,6 +40,12 @@ auto ToEdgeList(const memgraph::communication::bolt::Value &v) {
     list.push_back(x.ValueEdge());
   }
   return list;
+}
+
+auto StringToUnorderedSet(const std::string &element, const size_t number_of_split_elements) {
+  const auto element_split = memgraph::utils::Split(element, ", ");
+  MG_ASSERT(element_split.size() == number_of_split_elements);
+  return std::unordered_set<std::string>(element_split.begin(), element_split.end());
 };
 
 struct InterpreterFaker {
@@ -1465,3 +1473,148 @@ TEST_F(InterpreterTest, LoadCsvClauseNotification) {
             "conversion functions such as ToInteger, ToFloat, ToBoolean etc.");
   ASSERT_EQ(notification["description"].ValueString(), "");
 }
+
+TEST_F(InterpreterTest, CreateSchemaMulticommandTransaction) {
+  Interpret("BEGIN");
+  ASSERT_THROW(Interpret("CREATE SCHEMA ON :label(name STRING, age INTEGER)"),
+               memgraph::query::ConstraintInMulticommandTxException);
+  Interpret("ROLLBACK");
+}
+
+TEST_F(InterpreterTest, ShowSchemasMulticommandTransaction) {
+  Interpret("BEGIN");
+  ASSERT_THROW(Interpret("SHOW SCHEMAS"), memgraph::query::ConstraintInMulticommandTxException);
+  Interpret("ROLLBACK");
+}
+
+TEST_F(InterpreterTest, ShowSchemaMulticommandTransaction) {
+  Interpret("BEGIN");
+  ASSERT_THROW(Interpret("SHOW SCHEMA ON :label"), memgraph::query::ConstraintInMulticommandTxException);
+  Interpret("ROLLBACK");
+}
+
+TEST_F(InterpreterTest, DropSchemaMulticommandTransaction) {
+  Interpret("BEGIN");
+  ASSERT_THROW(Interpret("DROP SCHEMA ON :label"), memgraph::query::ConstraintInMulticommandTxException);
+  Interpret("ROLLBACK");
+}
+
+TEST_F(InterpreterTest, SchemaTestCreateAndShow) {
+  // Empty schema type map should result with syntax exception.
+  ASSERT_THROW(Interpret("CREATE SCHEMA ON :label();"), memgraph::query::SyntaxException);
+
+  // Duplicate properties are should also cause an exception
+  ASSERT_THROW(Interpret("CREATE SCHEMA ON :label(name STRING, name STRING);"), memgraph::query::SemanticException);
+  ASSERT_THROW(Interpret("CREATE SCHEMA ON :label(name STRING, name INTEGER);"), memgraph::query::SemanticException);
+
+  {
+    // Cannot create same schema twice
+    Interpret("CREATE SCHEMA ON :label(name STRING, age INTEGER)");
+    ASSERT_THROW(Interpret("CREATE SCHEMA ON :label(name STRING);"), memgraph::query::QueryException);
+  }
+  // Show schema
+  {
+    auto stream = Interpret("SHOW SCHEMA ON :label");
+    ASSERT_EQ(stream.GetHeader().size(), 2U);
+    const auto &header = stream.GetHeader();
+    ASSERT_EQ(header[0], "property_name");
+    ASSERT_EQ(header[1], "property_type");
+    ASSERT_EQ(stream.GetResults().size(), 2U);
+    std::unordered_map<std::string, std::string> result_table{{"age", "Integer"}, {"name", "String"}};
+
+    const auto &result = stream.GetResults().front();
+    ASSERT_EQ(result.size(), 2U);
+    const auto key1 = result[0].ValueString();
+    ASSERT_TRUE(result_table.contains(key1));
+    ASSERT_EQ(result[1].ValueString(), result_table[key1]);
+
+    const auto &result2 = stream.GetResults().front();
+    ASSERT_EQ(result2.size(), 2U);
+    const auto key2 = result2[0].ValueString();
+    ASSERT_TRUE(result_table.contains(key2));
+    ASSERT_EQ(result[1].ValueString(), result_table[key2]);
+  }
+  // Create Another Schema
+  Interpret("CREATE SCHEMA ON :label2(place STRING, dur DURATION)");
+
+  // Show schemas
+  {
+    auto stream = Interpret("SHOW SCHEMAS");
+    ASSERT_EQ(stream.GetHeader().size(), 3U);
+    const auto &header = stream.GetHeader();
+    ASSERT_EQ(header[0], "label");
+    ASSERT_EQ(header[1], "primary_key");
+    ASSERT_EQ(header[2], "primary_key_type");
+    ASSERT_EQ(stream.GetResults().size(), 2U);
+    std::unordered_map<std::string, std::pair<std::unordered_set<std::string>, std::string>> result_table{
+        {"label", {{"name::String", "age::Integer"}, "Composite"}},
+        {"label2", {{"place::String", "dur::Duration"}, "Composite"}}};
+
+    const auto &result = stream.GetResults().front();
+    ASSERT_EQ(result.size(), 3U);
+    const auto key1 = result[0].ValueString();
+    ASSERT_TRUE(result_table.contains(key1));
+    const auto primary_key_split = StringToUnorderedSet(result[1].ValueString(), 2);
+    ASSERT_TRUE(primary_key_split == result_table[key1].first);
+    ASSERT_EQ(result[2].ValueString(), result_table[key1].second);
+
+    const auto &result2 = stream.GetResults().front();
+    ASSERT_EQ(result2.size(), 3U);
+    const auto key2 = result2[0].ValueString();
+    ASSERT_TRUE(result_table.contains(key2));
+    const auto primary_key_split2 = StringToUnorderedSet(result2[1].ValueString(), 2);
+    ASSERT_TRUE(primary_key_split2 == result_table[key2].first);
+    ASSERT_EQ(result2[2].ValueString(), result_table[key2].second);
+  }
+}
+
+TEST_F(InterpreterTest, SchemaTestCreateDropAndShow) {
+  Interpret("CREATE SCHEMA ON :label(name STRING, age INTEGER)");
+  // Wrong syntax for dropping schema.
+  ASSERT_THROW(Interpret("DROP SCHEMA ON :label();"), memgraph::query::SyntaxException);
+  // Cannot drop non existant schema.
+  ASSERT_THROW(Interpret("DROP SCHEMA ON :label1;"), memgraph::query::QueryException);
+
+  // Create Schema and Drop
+  auto get_number_of_schemas = [this]() {
+    auto stream = Interpret("SHOW SCHEMAS");
+    return stream.GetResults().size();
+  };
+
+  ASSERT_EQ(get_number_of_schemas(), 1);
+  Interpret("CREATE SCHEMA ON :label1(name STRING, age INTEGER)");
+  ASSERT_EQ(get_number_of_schemas(), 2);
+  Interpret("CREATE SCHEMA ON :label2(name STRING, sex BOOL)");
+  ASSERT_EQ(get_number_of_schemas(), 3);
+  Interpret("DROP SCHEMA ON :label1");
+  ASSERT_EQ(get_number_of_schemas(), 2);
+  Interpret("CREATE SCHEMA ON :label3(name STRING, birthday LOCALDATETIME)");
+  ASSERT_EQ(get_number_of_schemas(), 3);
+  Interpret("DROP SCHEMA ON :label2");
+  ASSERT_EQ(get_number_of_schemas(), 2);
+  Interpret("CREATE SCHEMA ON :label4(name STRING, age DURATION)");
+  ASSERT_EQ(get_number_of_schemas(), 3);
+  Interpret("DROP SCHEMA ON :label3");
+  ASSERT_EQ(get_number_of_schemas(), 2);
+  Interpret("DROP SCHEMA ON :label");
+  ASSERT_EQ(get_number_of_schemas(), 1);
+
+  // Show schemas
+  auto stream = Interpret("SHOW SCHEMAS");
+  ASSERT_EQ(stream.GetHeader().size(), 3U);
+  const auto &header = stream.GetHeader();
+  ASSERT_EQ(header[0], "label");
+  ASSERT_EQ(header[1], "primary_key");
+  ASSERT_EQ(header[2], "primary_key_type");
+  ASSERT_EQ(stream.GetResults().size(), 1U);
+  std::unordered_map<std::string, std::pair<std::unordered_set<std::string>, std::string>> result_table{
+      {"label4", {{"name::String", "age::Duration"}, "Composite"}}};
+
+  const auto &result = stream.GetResults().front();
+  ASSERT_EQ(result.size(), 3U);
+  const auto key1 = result[0].ValueString();
+  ASSERT_TRUE(result_table.contains(key1));
+  const auto primary_key_split = StringToUnorderedSet(result[1].ValueString(), 2);
+  ASSERT_TRUE(primary_key_split == result_table[key1].first);
+  ASSERT_EQ(result[2].ValueString(), result_table[key1].second);
+}

tests/unit/query_required_privileges.cpp

@@ -192,6 +192,11 @@ TEST_F(TestPrivilegeExtractor, ShowVersion) {
   EXPECT_THAT(GetRequiredPrivileges(query), UnorderedElementsAre(AuthQuery::Privilege::STATS));
 }
 
+TEST_F(TestPrivilegeExtractor, SchemaQuery) {
+  auto *query = storage.Create<SchemaQuery>();
+  EXPECT_THAT(GetRequiredPrivileges(query), UnorderedElementsAre(AuthQuery::Privilege::SCHEMA));
+}
+
 TEST_F(TestPrivilegeExtractor, CallProcedureQuery) {
   {
     auto *query = QUERY(SINGLE_QUERY(CALL_PROCEDURE("mg.get_module_files")));

tests/unit/storage_v2_constraints.cpp

@@ -137,10 +137,8 @@ TEST_F(ConstraintsTest, ExistenceConstraintsViolationOnCommit) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(
-        res.GetError(),
-        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                     ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}));
   }
 
   {
@@ -159,10 +157,8 @@ TEST_F(ConstraintsTest, ExistenceConstraintsViolationOnCommit) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(
-        res.GetError(),
-        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                     ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}));
   }
 
   {
@@ -462,10 +458,8 @@ TEST_F(ConstraintsTest, UniqueConstraintsViolationOnCommit1) {
     ASSERT_NO_ERROR(vertex2.SetProperty(prop1, PropertyValue(1)));
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(
-        res.GetError(),
-        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
   }
 }
 
@@ -506,10 +500,8 @@ TEST_F(ConstraintsTest, UniqueConstraintsViolationOnCommit2) {
     ASSERT_NO_ERROR(acc2.Commit());
     auto res = acc3.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(
-        res.GetError(),
-        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
   }
 }
 
@@ -553,16 +545,12 @@ TEST_F(ConstraintsTest, UniqueConstraintsViolationOnCommit3) {
 
     auto res = acc2.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(
-        res.GetError(),
-        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
     res = acc3.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(
-        res.GetError(),
-        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
   }
 }
 
@@ -632,9 +620,7 @@ TEST_F(ConstraintsTest, UniqueConstraintsLabelAlteration) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}}));
+    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}));
   }
 
   {
@@ -668,9 +654,7 @@ TEST_F(ConstraintsTest, UniqueConstraintsLabelAlteration) {
 
     auto res = acc1.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}}));
+    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}));
   }
 }
 
@@ -765,9 +749,8 @@ TEST_F(ConstraintsTest, UniqueConstraintsMultipleProperties) {
     ASSERT_NO_ERROR(vertex2->SetProperty(prop2, PropertyValue(2)));
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(), (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1,
-                                                               std::set<PropertyId>{prop1, prop2}}}));
+    EXPECT_EQ(res.GetError(),
+              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1, prop2}}));
   }
 
   // Then change the second property of both vertex to null. Property values of
@@ -878,9 +861,7 @@ TEST_F(ConstraintsTest, UniqueConstraintsInsertRemoveAbortInsert) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1, prop2}}}));
+    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1, prop2}}));
   }
 }
 
@@ -919,9 +900,7 @@ TEST_F(ConstraintsTest, UniqueConstraintsDeleteVertexSetProperty) {
 
     auto res = acc1.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
-                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}}));
+    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}));
 
     ASSERT_NO_ERROR(acc2.Commit());
   }