Fix shared module loading

* Moving function add_query_module from CMakeLists from tests/e2e/magic_functions to tests/e2e

* Adding failing test copying behavior when loading c module (.so) into memgraph.

* Fixing issue where NO_ERROR status returned  MgpTransAddFixedResult was converted to false

* Removing unnecessary transformation

* removing incorrect parameterization of test

* re-adding parametrized transformation

.github/workflows/release_docker.yaml

@@ -0,0 +1,49 @@
+name: Publish Docker images
+
+on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: "Memgraph binary version to publish on Dockerhub."
+        required: true
+
+jobs:
+  docker_publish:
+    runs-on: ubuntu-latest
+    env:
+      DOCKER_ORGANIZATION_NAME: memgraph
+      DOCKER_REPOSITORY_NAME: memgraph
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v1
+
+      - name: Set up Docker Buildx
+        id: buildx
+        uses: docker/setup-buildx-action@v1
+
+      - name: Log in to Docker Hub
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+
+      - name: Download memgraph binary
+        run: |
+          cd release/docker
+          curl -L https://download.memgraph.com/memgraph/v${{ github.event.inputs.version }}/debian-11/memgraph_${{ github.event.inputs.version }}-1_amd64.deb > memgraph-amd64.deb
+          curl -L https://download.memgraph.com/memgraph/v${{ github.event.inputs.version }}/debian-11-aarch64/memgraph_${{ github.event.inputs.version }}-1_arm64.deb > memgraph-arm64.deb
+
+      - name: Build & push docker images
+        run: |
+          cd release/docker
+          docker buildx build \
+          --build-arg BINARY_NAME="memgraph-" \
+          --build-arg EXTENSION="deb" \
+          --platform linux/amd64,linux/arm64 \
+          --tag $DOCKER_ORGANIZATION_NAME/$DOCKER_REPOSITORY_NAME:${{ github.event.inputs.version }} \
+          --tag $DOCKER_ORGANIZATION_NAME/$DOCKER_REPOSITORY_NAME:latest \
+          --file memgraph_deb.dockerfile \
+          --push .

CMakeLists.txt

@@ -54,7 +54,7 @@ option(MG_ENTERPRISE "Build Memgraph Enterprise Edition" ON)
 # Set the current version here to override the automatic version detection. The
 # version must be specified as `X.Y.Z`. Primarily used when building new patch
 # versions.
-set(MEMGRAPH_OVERRIDE_VERSION "2.2.1")
+set(MEMGRAPH_OVERRIDE_VERSION "")
 
 # Custom suffix that this version should have. The suffix can be any arbitrary
 # string. Primarily used when building a version for a specific customer.
@@ -184,7 +184,8 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall \
     -Werror=switch -Werror=switch-bool -Werror=return-type \
     -Werror=return-stack-address \
-    -Wno-c99-designator")
+    -Wno-c99-designator \
+    -DBOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT")
 
 # Don't omit frame pointer in RelWithDebInfo, for additional callchain debug.
 set(CMAKE_CXX_FLAGS_RELWITHDEBINFO

README.md

@@ -54,6 +54,18 @@ Memgraph is implemented in C/C++ and leverages an in-memory first architecture
 to ensure that you’re getting the best possible performance consistently and
 without surprises. It’s also ACID-compliant and highly available.
 
+## :video_game: Memgraph Playground
+
+You don't need to install anything to try out Memgraph. Check out 
+our **[Memgraph Playground](https://playground.memgraph.com/)** sandboxes in 
+your browser.
+
+<p align="left">
+  <a href="https://playground.memgraph.com/">
+    <img width="450px" alt="Memgraph Playground" src="https://download.memgraph.com/asset/github/memgraph/memgraph-playground.png">
+  </a>
+</p>
+
 ## :floppy_disk: Download & Install
 
 ### Windows

environment/os/debian-11-arm.sh

@@ -0,0 +1,104 @@
+#!/bin/bash
+
+set -Eeuo pipefail
+
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+source "$DIR/../util.sh"
+
+TOOLCHAIN_BUILD_DEPS=(
+    coreutils gcc g++ build-essential make # generic build tools
+    wget # used for archive download
+    gnupg # used for archive signature verification
+    tar gzip bzip2 xz-utils unzip # used for archive unpacking
+    zlib1g-dev # zlib library used for all builds
+    libexpat1-dev liblzma-dev python3-dev texinfo # for gdb
+    libcurl4-openssl-dev # for cmake
+    libreadline-dev # for cmake and llvm
+    libffi-dev libxml2-dev # for llvm
+    libedit-dev libpcre3-dev automake bison # for swig
+    curl # snappy
+    file # for libunwind
+    libssl-dev # for libevent
+    libgmp-dev
+    gperf # for proxygen
+    git # for fbthrift
+)
+
+TOOLCHAIN_RUN_DEPS=(
+    make # generic build tools
+    tar gzip bzip2 xz-utils # used for archive unpacking
+    zlib1g # zlib library used for all builds
+    libexpat1 liblzma5 python3 # for gdb
+    libcurl4 # for cmake
+    file # for CPack
+    libreadline8 # for cmake and llvm
+    libffi7 libxml2 # for llvm
+    libssl-dev # for libevent
+)
+
+MEMGRAPH_BUILD_DEPS=(
+    git # source code control
+    make pkg-config # build system
+    curl wget # for downloading libs
+    uuid-dev default-jre-headless # required by antlr
+    libreadline-dev # for memgraph console
+    libpython3-dev python3-dev # for query modules
+    libssl-dev
+    libseccomp-dev
+    netcat # tests are using nc to wait for memgraph
+    python3 virtualenv python3-virtualenv python3-pip # for qa, macro_benchmark and stress tests
+    python3-yaml # for the configuration generator
+    libcurl4-openssl-dev # mg-requests
+    sbcl # for custom Lisp C++ preprocessing
+    doxygen graphviz # source documentation generators
+    mono-runtime mono-mcs zip unzip default-jdk-headless # for driver tests
+    golang nodejs npm
+    autoconf # for jemalloc code generation
+    libtool  # for protobuf code generation
+)
+
+list() {
+    echo "$1"
+}
+
+check() {
+    check_all_dpkg "$1"
+}
+
+install() {
+    cat >/etc/apt/sources.list <<EOF
+deb http://deb.debian.org/debian bullseye main
+deb-src http://deb.debian.org/debian bullseye main
+
+deb http://deb.debian.org/debian-security/ bullseye-security main
+deb-src http://deb.debian.org/debian-security/ bullseye-security main
+
+deb http://deb.debian.org/debian bullseye-updates main
+deb-src http://deb.debian.org/debian bullseye-updates main
+EOF
+    cd "$DIR"
+    apt update
+    # If GitHub Actions runner is installed, append LANG to the environment.
+    # Python related tests doesn't work the LANG export.
+    if [ -d "/home/gh/actions-runner" ]; then
+        echo "LANG=en_US.utf8" >> /home/gh/actions-runner/.env
+    else
+        echo "NOTE: export LANG=en_US.utf8"
+    fi
+    apt install -y wget
+    for pkg in $1; do
+        if [ "$pkg" == dotnet-sdk-3.1 ]; then
+            if ! dpkg -s "$pkg" 2>/dev/null >/dev/null; then
+                wget -nv https://packages.microsoft.com/config/debian/10/packages-microsoft-prod.deb -O packages-microsoft-prod.deb
+                dpkg -i packages-microsoft-prod.deb
+                apt-get update
+                apt-get install -y apt-transport-https dotnet-sdk-3.1
+            fi
+            continue
+        fi
+        apt install -y "$pkg"
+    done
+}
+
+deps=$2"[*]"
+"$1" "${!deps}"

environment/util.sh

@@ -5,6 +5,10 @@ operating_system() {
     sort | cut -d '=' -f 2- | sed 's/"//g' | paste -s -d '-'
 }
 
+architecture() {
+    uname -m
+}
+
 check_all_yum() {
     local missing=""
     for pkg in $1; do

include/mg_procedure.h

@@ -1,4 +1,4 @@
-// Copyright 2021 Memgraph Ltd.
+// 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
@@ -549,6 +549,8 @@ enum mgp_error mgp_path_equal(struct mgp_path *p1, struct mgp_path *p2, int *res
 struct mgp_result;
 /// Represents a record of resulting field values.
 struct mgp_result_record;
+/// Represents a return type for magic functions
+struct mgp_func_result;
 
 /// Set the error as the result of the procedure.
 /// Return MGP_ERROR_UNABLE_TO_ALLOCATE ff there's no memory for copying the error message.
@@ -1290,6 +1292,9 @@ struct mgp_module;
 /// Describes a procedure of a query module.
 struct mgp_proc;
 
+/// Describes a Memgraph magic function.
+struct mgp_func;
+
 /// Entry-point for a query module read procedure, invoked through openCypher.
 ///
 /// Passed in arguments will not live longer than the callback's execution.
@@ -1502,6 +1507,84 @@ typedef void (*mgp_trans_cb)(struct mgp_messages *, struct mgp_graph *, struct m
 enum mgp_error mgp_module_add_transformation(struct mgp_module *module, const char *name, mgp_trans_cb cb);
 /// @}
 
+/// @name Memgraph Magic Functions API
+///
+/// API for creating the Memgraph magic functions. It is used to create external-source stateless methods which can
+/// be called by using openCypher query language. These methods should not modify the original graph and should use only
+/// the values provided as arguments to the method.
+///
+///@{
+
+/// Add a required argument to a function.
+///
+/// The order of the added arguments corresponds to the signature of the openCypher function.
+/// Note, that required arguments are followed by optional arguments.
+///
+/// The `name` must be a valid identifier, following the same rules as the
+/// function `name` in mgp_module_add_function.
+///
+/// Passed in `type` describes what kind of values can be used as the argument.
+///
+/// Return MGP_ERROR_UNABLE_TO_ALLOCATE if unable to allocate memory for an argument.
+/// Return MGP_ERROR_INVALID_ARGUMENT if `name` is not a valid argument name.
+/// Return MGP_ERROR_LOGIC_ERROR if the function already has any optional argument.
+enum mgp_error mgp_func_add_arg(struct mgp_func *func, const char *name, struct mgp_type *type);
+
+/// Add an optional argument with a default value to a function.
+///
+/// The order of the added arguments corresponds to the signature of the openCypher function.
+/// Note, that required arguments are followed by optional arguments.
+///
+/// The `name` must be a valid identifier, following the same rules as the
+/// function `name` in mgp_module_add_function.
+///
+/// Passed in `type` describes what kind of values can be used as the argument.
+///
+/// `default_value` is copied and set as the default value for the argument.
+/// Don't forget to call mgp_value_destroy when you are done using
+/// `default_value`. When the function is called, if this argument is not
+/// provided, `default_value` will be used instead. `default_value` must not be
+/// a graph element (node, relationship, path) and it must satisfy the given
+/// `type`.
+///
+/// Return MGP_ERROR_UNABLE_TO_ALLOCATE if unable to allocate memory for an argument.
+/// Return MGP_ERROR_INVALID_ARGUMENT if `name` is not a valid argument name.
+/// Return MGP_ERROR_VALUE_CONVERSION if `default_value` is a graph element (vertex, edge or path).
+/// Return MGP_ERROR_LOGIC_ERROR if `default_value` does not satisfy `type`.
+enum mgp_error mgp_func_add_opt_arg(struct mgp_func *func, const char *name, struct mgp_type *type,
+                                    struct mgp_value *default_value);
+
+/// Entry-point for a custom Memgraph awesome function.
+///
+/// Passed in arguments will not live longer than the callback's execution.
+/// Therefore, you must not store them globally or use the passed in mgp_memory
+/// to allocate global resources.
+typedef void (*mgp_func_cb)(struct mgp_list *, struct mgp_func_context *, struct mgp_func_result *,
+                            struct mgp_memory *);
+
+/// Register a Memgraph magic function
+///
+/// The `name` must be a sequence of digits, underscores, lowercase and
+/// uppercase Latin letters. The name must begin with a non-digit character.
+/// Note that Unicode characters are not allowed.
+///
+/// Return MGP_ERROR_UNABLE_TO_ALLOCATE if unable to allocate memory for mgp_func.
+/// Return MGP_ERROR_INVALID_ARGUMENT if `name` is not a valid function name.
+/// RETURN MGP_ERROR_LOGIC_ERROR if a function with the same name was already registered.
+enum mgp_error mgp_module_add_function(struct mgp_module *module, const char *name, mgp_func_cb cb,
+                                       struct mgp_func **result);
+
+/// Set an error message as an output to the Magic function
+/// Return MGP_ERROR_UNABLE_TO_ALLOCATE if there's no memory for copying the error message.
+enum mgp_error mgp_func_result_set_error_msg(struct mgp_func_result *result, const char *error_msg,
+                                             struct mgp_memory *memory);
+
+/// Set an output value for the Magic function
+/// Return MGP_ERROR_UNABLE_TO_ALLOCATE if unable to allocate memory to copy the mgp_value to mgp_func_result.
+enum mgp_error mgp_func_result_set_value(struct mgp_func_result *result, struct mgp_value *value,
+                                         struct mgp_memory *memory);
+/// @}
+
 #ifdef __cplusplus
 }  // extern "C"
 #endif

include/mgp.py

@@ -40,6 +40,7 @@ class InvalidContextError(Exception):
     """
     Signals using a graph element instance outside of the registered procedure.
     """
+
     pass
 
 
@@ -47,6 +48,7 @@ class UnknownError(_mgp.UnknownError):
     """
     Signals unspecified failure.
     """
+
     pass
 
 
@@ -54,6 +56,7 @@ class UnableToAllocateError(_mgp.UnableToAllocateError):
     """
     Signals failed memory allocation.
     """
+
     pass
 
 
@@ -61,6 +64,7 @@ class InsufficientBufferError(_mgp.InsufficientBufferError):
     """
     Signals that some buffer is not big enough.
     """
+
     pass
 
 
@@ -69,6 +73,7 @@ class OutOfRangeError(_mgp.OutOfRangeError):
     Signals that an index-like parameter has a value that is outside its
     possible values.
     """
+
     pass
 
 
@@ -77,6 +82,7 @@ class LogicErrorError(_mgp.LogicErrorError):
     Signals faulty logic within the program such as violating logical
     preconditions or class invariants and may be preventable.
     """
+
     pass
 
 
@@ -84,6 +90,7 @@ class DeletedObjectError(_mgp.DeletedObjectError):
     """
     Signals accessing an already deleted object.
     """
+
     pass
 
 
@@ -91,6 +98,7 @@ class InvalidArgumentError(_mgp.InvalidArgumentError):
     """
     Signals that some of the arguments have invalid values.
     """
+
     pass
 
 
@@ -98,6 +106,7 @@ class KeyAlreadyExistsError(_mgp.KeyAlreadyExistsError):
     """
     Signals that a key already exists in a container-like object.
     """
+
     pass
 
 
@@ -105,6 +114,7 @@ class ImmutableObjectError(_mgp.ImmutableObjectError):
     """
     Signals modification of an immutable object.
     """
+
     pass
 
 
@@ -112,6 +122,7 @@ class ValueConversionError(_mgp.ValueConversionError):
     """
     Signals that the conversion failed between python and cypher values.
     """
+
     pass
 
 
@@ -120,12 +131,14 @@ class SerializationError(_mgp.SerializationError):
     Signals serialization error caused by concurrent modifications from
     different transactions.
     """
+
     pass
 
 
 class Label:
     """Label of a Vertex."""
-    __slots__ = ('_name',)
+
+    __slots__ = ("_name",)
 
     def __init__(self, name: str):
         self._name = name
@@ -145,19 +158,22 @@ class Label:
 # Named property value of a Vertex or an Edge.
 # It would be better to use typing.NamedTuple with typed fields, but that is
 # not available in Python 3.5.
-Property = namedtuple('Property', ('name', 'value'))
+Property = namedtuple("Property", ("name", "value"))
 
 
 class Properties:
     """
     A collection of properties either on a Vertex or an Edge.
     """
-    __slots__ = ('_vertex_or_edge', '_len',)
+
+    __slots__ = (
+        "_vertex_or_edge",
+        "_len",
+    )
 
     def __init__(self, vertex_or_edge):
         if not isinstance(vertex_or_edge, (_mgp.Vertex, _mgp.Edge)):
-            raise TypeError("Expected '_mgp.Vertex' or '_mgp.Edge', \
-                            got {}".format(type(vertex_or_edge)))
+            raise TypeError("Expected '_mgp.Vertex' or '_mgp.Edge', got {}".format(type(vertex_or_edge)))
         self._len = None
         self._vertex_or_edge = vertex_or_edge
 
@@ -330,7 +346,8 @@ class Properties:
 
 class EdgeType:
     """Type of an Edge."""
-    __slots__ = ('_name',)
+
+    __slots__ = ("_name",)
 
     def __init__(self, name):
         self._name = name
@@ -348,7 +365,7 @@ class EdgeType:
 
 
 if sys.version_info >= (3, 5, 2):
-    EdgeId = typing.NewType('EdgeId', int)
+    EdgeId = typing.NewType("EdgeId", int)
 else:
     EdgeId = int
 
@@ -360,12 +377,12 @@ class Edge:
     a query. You should not globally store an instance of an Edge. Using an
     invalid Edge instance will raise InvalidContextError.
     """
-    __slots__ = ('_edge',)
+
+    __slots__ = ("_edge",)
 
     def __init__(self, edge):
         if not isinstance(edge, _mgp.Edge):
-            raise TypeError(
-                "Expected '_mgp.Edge', got '{}'".format(type(edge)))
+            raise TypeError("Expected '_mgp.Edge', got '{}'".format(type(edge)))
         self._edge = edge
 
     def __deepcopy__(self, memo):
@@ -408,7 +425,7 @@ class Edge:
         return EdgeType(self._edge.get_type_name())
 
     @property
-    def from_vertex(self) -> 'Vertex':
+    def from_vertex(self) -> "Vertex":
         """
         Get the source vertex.
 
@@ -419,7 +436,7 @@ class Edge:
         return Vertex(self._edge.from_vertex())
 
     @property
-    def to_vertex(self) -> 'Vertex':
+    def to_vertex(self) -> "Vertex":
         """
         Get the destination vertex.
 
@@ -453,7 +470,7 @@ class Edge:
 
 
 if sys.version_info >= (3, 5, 2):
-    VertexId = typing.NewType('VertexId', int)
+    VertexId = typing.NewType("VertexId", int)
 else:
     VertexId = int
 
@@ -465,12 +482,12 @@ class Vertex:
     in a query. You should not globally store an instance of a Vertex. Using an
     invalid Vertex instance will raise InvalidContextError.
     """
-    __slots__ = ('_vertex',)
+
+    __slots__ = ("_vertex",)
 
     def __init__(self, vertex):
         if not isinstance(vertex, _mgp.Vertex):
-            raise TypeError(
-                "Expected '_mgp.Vertex', got '{}'".format(type(vertex)))
+            raise TypeError("Expected '_mgp.Vertex', got '{}'".format(type(vertex)))
         self._vertex = vertex
 
     def __deepcopy__(self, memo):
@@ -513,8 +530,7 @@ class Vertex:
         """
         if not self.is_valid():
             raise InvalidContextError()
-        return tuple(Label(self._vertex.label_at(i))
-                     for i in range(self._vertex.labels_count()))
+        return tuple(Label(self._vertex.label_at(i)) for i in range(self._vertex.labels_count()))
 
     def add_label(self, label: str) -> None:
         """
@@ -615,7 +631,8 @@ class Vertex:
 
 class Path:
     """Path containing Vertex and Edge instances."""
-    __slots__ = ('_path', '_vertices', '_edges')
+
+    __slots__ = ("_path", "_vertices", "_edges")
 
     def __init__(self, starting_vertex_or_path: typing.Union[_mgp.Path, Vertex]):
         """Initialize with a starting Vertex.
@@ -636,8 +653,7 @@ class Path:
                 raise InvalidContextError()
             self._path = _mgp.Path.make_with_start(vertex)
         else:
-            raise TypeError("Expected '_mgp.Vertex' or '_mgp.Path', got '{}'"
-                            .format(type(starting_vertex_or_path)))
+            raise TypeError("Expected '_mgp.Vertex' or '_mgp.Path', got '{}'".format(type(starting_vertex_or_path)))
 
     def __copy__(self):
         if not self.is_valid():
@@ -678,8 +694,7 @@ class Path:
         extension.
         """
         if not isinstance(edge, Edge):
-            raise TypeError(
-                "Expected '_mgp.Edge', got '{}'".format(type(edge)))
+            raise TypeError("Expected '_mgp.Edge', got '{}'".format(type(edge)))
         if not self.is_valid() or not edge.is_valid():
             raise InvalidContextError()
         self._path.expand(edge._edge)
@@ -698,8 +713,7 @@ class Path:
             raise InvalidContextError()
         if self._vertices is None:
             num_vertices = self._path.size() + 1
-            self._vertices = tuple(Vertex(self._path.vertex_at(i))
-                                   for i in range(num_vertices))
+            self._vertices = tuple(Vertex(self._path.vertex_at(i)) for i in range(num_vertices))
         return self._vertices
 
     @property
@@ -713,14 +727,14 @@ class Path:
             raise InvalidContextError()
         if self._edges is None:
             num_edges = self._path.size()
-            self._edges = tuple(Edge(self._path.edge_at(i))
-                                for i in range(num_edges))
+            self._edges = tuple(Edge(self._path.edge_at(i)) for i in range(num_edges))
         return self._edges
 
 
 class Record:
     """Represents a record of resulting field values."""
-    __slots__ = ('fields',)
+
+    __slots__ = ("fields",)
 
     def __init__(self, **kwargs):
         """Initialize with name=value fields in kwargs."""
@@ -729,12 +743,12 @@ class Record:
 
 class Vertices:
     """Iterable over vertices in a graph."""
-    __slots__ = ('_graph', '_len')
+
+    __slots__ = ("_graph", "_len")
 
     def __init__(self, graph):
         if not isinstance(graph, _mgp.Graph):
-            raise TypeError(
-                "Expected '_mgp.Graph', got '{}'".format(type(graph)))
+            raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
         self._graph = graph
         self._len = None
 
@@ -791,12 +805,12 @@ class Vertices:
 
 class Graph:
     """State of the graph database in current ProcCtx."""
-    __slots__ = ('_graph',)
+
+    __slots__ = ("_graph",)
 
     def __init__(self, graph):
         if not isinstance(graph, _mgp.Graph):
-            raise TypeError(
-                "Expected '_mgp.Graph', got '{}'".format(type(graph)))
+            raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
         self._graph = graph
 
     def __deepcopy__(self, memo):
@@ -885,8 +899,7 @@ class Graph:
             raise InvalidContextError()
         self._graph.detach_delete_vertex(vertex._vertex)
 
-    def create_edge(self, from_vertex: Vertex, to_vertex: Vertex,
-                    edge_type: EdgeType) -> None:
+    def create_edge(self, from_vertex: Vertex, to_vertex: Vertex, edge_type: EdgeType) -> None:
         """
         Create an edge.
 
@@ -899,8 +912,7 @@ class Graph:
         """
         if not self.is_valid():
             raise InvalidContextError()
-        return Edge(self._graph.create_edge(from_vertex._vertex,
-                                            to_vertex._vertex, edge_type.name))
+        return Edge(self._graph.create_edge(from_vertex._vertex, to_vertex._vertex, edge_type.name))
 
     def delete_edge(self, edge: Edge) -> None:
         """
@@ -918,6 +930,7 @@ class Graph:
 
 class AbortError(Exception):
     """Signals that the procedure was asked to abort its execution."""
+
     pass
 
 
@@ -927,12 +940,12 @@ class ProcCtx:
     Access to a ProcCtx is only valid during a single execution of a procedure
     in a query. You should not globally store a ProcCtx instance.
     """
-    __slots__ = ('_graph',)
+
+    __slots__ = ("_graph",)
 
     def __init__(self, graph):
         if not isinstance(graph, _mgp.Graph):
-            raise TypeError(
-                "Expected '_mgp.Graph', got '{}'".format(type(graph)))
+            raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
         self._graph = Graph(graph)
 
     def is_valid(self) -> bool:
@@ -969,8 +982,7 @@ LocalDateTime = datetime.datetime
 
 Duration = datetime.timedelta
 
-Any = typing.Union[bool, str, Number, Map, Path,
-                   list, Date, LocalTime, LocalDateTime, Duration]
+Any = typing.Union[bool, str, Number, Map, Path, list, Date, LocalTime, LocalDateTime, Duration]
 
 List = typing.List
 
@@ -1003,7 +1015,7 @@ def _typing_to_cypher_type(type_):
         Date: _mgp.type_date(),
         LocalTime: _mgp.type_local_time(),
         LocalDateTime: _mgp.type_local_date_time(),
-        Duration: _mgp.type_duration()
+        Duration: _mgp.type_duration(),
     }
     try:
         return simple_types[type_]
@@ -1021,14 +1033,14 @@ def _typing_to_cypher_type(type_):
             if type(None) in type_args:
                 types = tuple(t for t in type_args if t is not type(None))  # noqa E721
                 if len(types) == 1:
-                    type_arg, = types
+                    (type_arg,) = types
                 else:
                     # We cannot do typing.Union[*types], so do the equivalent
                     # with __getitem__ which does not even need arg unpacking.
                     type_arg = typing.Union.__getitem__(types)
                 return _mgp.type_nullable(_typing_to_cypher_type(type_arg))
         elif complex_type == list:
-            type_arg, = type_args
+            (type_arg,) = type_args
             return _mgp.type_list(_typing_to_cypher_type(type_arg))
         raise UnsupportedTypingError(type_)
     else:
@@ -1038,13 +1050,17 @@ def _typing_to_cypher_type(type_):
         # printed the same way. `typing.List[type]` is printed as such, while
         # `typing.Optional[type]` is printed as 'typing.Union[type, NoneType]'
         def parse_type_args(type_as_str):
-            return tuple(map(str.strip,
-                             type_as_str[type_as_str.index('[') + 1: -1].split(',')))
+            return tuple(
+                map(
+                    str.strip,
+                    type_as_str[type_as_str.index("[") + 1 : -1].split(","),
+                )
+            )
 
         def fully_qualified_name(cls):
-            if cls.__module__ is None or cls.__module__ == 'builtins':
+            if cls.__module__ is None or cls.__module__ == "builtins":
                 return cls.__name__
-            return cls.__module__ + '.' + cls.__name__
+            return cls.__module__ + "." + cls.__name__
 
         def get_simple_type(type_as_str):
             for simple_type, cypher_type in simple_types.items():
@@ -1060,28 +1076,26 @@ def _typing_to_cypher_type(type_):
                     pass
 
         def parse_typing(type_as_str):
-            if type_as_str.startswith('typing.Union'):
+            if type_as_str.startswith("typing.Union"):
                 type_args_as_str = parse_type_args(type_as_str)
                 none_type_as_str = type(None).__name__
                 if none_type_as_str in type_args_as_str:
-                    types = tuple(
-                        t for t in type_args_as_str if t != none_type_as_str)
+                    types = tuple(t for t in type_args_as_str if t != none_type_as_str)
                     if len(types) == 1:
-                        type_arg_as_str, = types
+                        (type_arg_as_str,) = types
                     else:
-                        type_arg_as_str = 'typing.Union[' + \
-                            ', '.join(types) + ']'
+                        type_arg_as_str = "typing.Union[" + ", ".join(types) + "]"
                     simple_type = get_simple_type(type_arg_as_str)
                     if simple_type is not None:
                         return _mgp.type_nullable(simple_type)
                     return _mgp.type_nullable(parse_typing(type_arg_as_str))
-            elif type_as_str.startswith('typing.List'):
+            elif type_as_str.startswith("typing.List"):
                 type_arg_as_str = parse_type_args(type_as_str)
 
                 if len(type_arg_as_str) > 1:
                     # Nested object could be a type consisting of a list of types (e.g. mgp.Map)
                     # so we need to join the parts.
-                    type_arg_as_str = ', '.join(type_arg_as_str)
+                    type_arg_as_str = ", ".join(type_arg_as_str)
                 else:
                     type_arg_as_str = type_arg_as_str[0]
 
@@ -1096,9 +1110,11 @@ def _typing_to_cypher_type(type_):
 
 # Procedure registration
 
+
 class Deprecated:
     """Annotate a resulting Record's field as deprecated."""
-    __slots__ = ('field_type',)
+
+    __slots__ = ("field_type",)
 
     def __init__(self, type_):
         self.field_type = type_
@@ -1106,8 +1122,7 @@ class Deprecated:
 
 def raise_if_does_not_meet_requirements(func: typing.Callable[..., Record]):
     if not callable(func):
-        raise TypeError("Expected a callable object, got an instance of '{}'"
-                        .format(type(func)))
+        raise TypeError("Expected a callable object, got an instance of '{}'".format(type(func)))
     if inspect.iscoroutinefunction(func):
         raise TypeError("Callable must not be 'async def' function")
     if sys.version_info >= (3, 6):
@@ -1117,24 +1132,25 @@ def raise_if_does_not_meet_requirements(func: typing.Callable[..., Record]):
         raise NotImplementedError("Generator functions are not supported")
 
 
-def _register_proc(func: typing.Callable[..., Record],
-                   is_write: bool):
+def _register_proc(func: typing.Callable[..., Record], is_write: bool):
     raise_if_does_not_meet_requirements(func)
-    register_func = (
-        _mgp.Module.add_write_procedure if is_write
-        else _mgp.Module.add_read_procedure)
+    register_func = _mgp.Module.add_write_procedure if is_write else _mgp.Module.add_read_procedure
     sig = inspect.signature(func)
     params = tuple(sig.parameters.values())
     if params and params[0].annotation is ProcCtx:
+
         @wraps(func)
         def wrapper(graph, args):
             return func(ProcCtx(graph), *args)
+
         params = params[1:]
         mgp_proc = register_func(_mgp._MODULE, wrapper)
     else:
+
         @wraps(func)
         def wrapper(graph, args):
             return func(*args)
+
         mgp_proc = register_func(_mgp._MODULE, wrapper)
     for param in params:
         name = param.name
@@ -1149,8 +1165,7 @@ def _register_proc(func: typing.Callable[..., Record],
     if sig.return_annotation is not sig.empty:
         record = sig.return_annotation
         if not isinstance(record, Record):
-            raise TypeError("Expected '{}' to return 'mgp.Record', got '{}'"
-                            .format(func.__name__, type(record)))
+            raise TypeError("Expected '{}' to return 'mgp.Record', got '{}'".format(func.__name__, type(record)))
         for name, type_ in record.fields.items():
             if isinstance(type_, Deprecated):
                 cypher_type = _typing_to_cypher_type(type_.field_type)
@@ -1257,20 +1272,22 @@ class InvalidMessageError(Exception):
     """
     Signals using a message instance outside of the registered transformation.
     """
+
     pass
 
 
 SOURCE_TYPE_KAFKA = _mgp.SOURCE_TYPE_KAFKA
 SOURCE_TYPE_PULSAR = _mgp.SOURCE_TYPE_PULSAR
 
+
 class Message:
     """Represents a message from a stream."""
-    __slots__ = ('_message',)
+
+    __slots__ = ("_message",)
 
     def __init__(self, message):
         if not isinstance(message, _mgp.Message):
-            raise TypeError(
-                "Expected '_mgp.Message', got '{}'".format(type(message)))
+            raise TypeError("Expected '_mgp.Message', got '{}'".format(type(message)))
         self._message = message
 
     def __deepcopy__(self, memo):
@@ -1353,17 +1370,18 @@ class Message:
 
 class InvalidMessagesError(Exception):
     """Signals using a messages instance outside of the registered transformation."""
+
     pass
 
 
 class Messages:
     """Represents a list of messages from a stream."""
-    __slots__ = ('_messages',)
+
+    __slots__ = ("_messages",)
 
     def __init__(self, messages):
         if not isinstance(messages, _mgp.Messages):
-            raise TypeError(
-                "Expected '_mgp.Messages', got '{}'".format(type(messages)))
+            raise TypeError("Expected '_mgp.Messages', got '{}'".format(type(messages)))
         self._messages = messages
 
     def __deepcopy__(self, memo):
@@ -1395,12 +1413,12 @@ class TransCtx:
     Access to a TransCtx is only valid during a single execution of a transformation.
     You should not globally store a TransCtx instance.
     """
-    __slots__ = ('_graph')
+
+    __slots__ = "_graph"
 
     def __init__(self, graph):
         if not isinstance(graph, _mgp.Graph):
-            raise TypeError(
-                "Expected '_mgp.Graph', got '{}'".format(type(graph)))
+            raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
         self._graph = Graph(graph)
 
     def is_valid(self) -> bool:
@@ -1420,21 +1438,76 @@ def transformation(func: typing.Callable[..., Record]):
     params = tuple(sig.parameters.values())
     if not params or not params[0].annotation is Messages:
         if not len(params) == 2 or not params[1].annotation is Messages:
-            raise NotImplementedError(
-                "Valid signatures for transformations are (TransCtx, Messages) or (Messages)")
+            raise NotImplementedError("Valid signatures for transformations are (TransCtx, Messages) or (Messages)")
     if params[0].annotation is TransCtx:
+
         @wraps(func)
         def wrapper(graph, messages):
             return func(TransCtx(graph), messages)
+
         _mgp._MODULE.add_transformation(wrapper)
     else:
+
         @wraps(func)
         def wrapper(graph, messages):
             return func(messages)
+
         _mgp._MODULE.add_transformation(wrapper)
     return func
 
 
+class FuncCtx:
+    """Context of a function being executed.
+
+    Access to a FuncCtx is only valid during a single execution of a transformation.
+    You should not globally store a FuncCtx instance.
+    """
+
+    __slots__ = "_graph"
+
+    def __init__(self, graph):
+        if not isinstance(graph, _mgp.Graph):
+            raise TypeError("Expected '_mgp.Graph', got '{}'".format(type(graph)))
+        self._graph = Graph(graph)
+
+    def is_valid(self) -> bool:
+        return self._graph.is_valid()
+
+
+def function(func: typing.Callable):
+    raise_if_does_not_meet_requirements(func)
+    register_func = _mgp.Module.add_function
+    sig = inspect.signature(func)
+    params = tuple(sig.parameters.values())
+    if params and params[0].annotation is FuncCtx:
+
+        @wraps(func)
+        def wrapper(graph, args):
+            return func(FuncCtx(graph), *args)
+
+        params = params[1:]
+        mgp_func = register_func(_mgp._MODULE, wrapper)
+    else:
+
+        @wraps(func)
+        def wrapper(graph, args):
+            return func(*args)
+
+        mgp_func = register_func(_mgp._MODULE, wrapper)
+
+    for param in params:
+        name = param.name
+        type_ = param.annotation
+        if type_ is param.empty:
+            type_ = object
+        cypher_type = _typing_to_cypher_type(type_)
+        if param.default is param.empty:
+            mgp_func.add_arg(name, cypher_type)
+        else:
+            mgp_func.add_opt_arg(name, cypher_type, param.default)
+    return func
+
+
 def _wrap_exceptions():
     def wrap_function(func):
         @wraps(func)
@@ -1463,6 +1536,7 @@ def _wrap_exceptions():
                 raise ValueConversionError(e)
             except _mgp.SerializationError as e:
                 raise SerializationError(e)
+
         return wrapped_func
 
     def wrap_prop_func(func):
@@ -1473,11 +1547,16 @@ def _wrap_exceptions():
             if inspect.isfunction(obj):
                 setattr(cls, name, wrap_function(obj))
             elif isinstance(obj, property):
-                setattr(cls, name, property(
-                    wrap_prop_func(obj.fget),
-                    wrap_prop_func(obj.fset),
-                    wrap_prop_func(obj.fdel),
-                    obj.__doc__))
+                setattr(
+                    cls,
+                    name,
+                    property(
+                        wrap_prop_func(obj.fget),
+                        wrap_prop_func(obj.fset),
+                        wrap_prop_func(obj.fdel),
+                        obj.__doc__,
+                    ),
+                )
 
     def defined_in_this_module(obj: object):
         return getattr(obj, "__module__", "") == __name__

init

@@ -65,8 +65,14 @@ else
 fi
 
 DISTRO=$(operating_system)
-echo "ALL BUILD PACKAGES: $($DIR/environment/os/$DISTRO.sh list MEMGRAPH_BUILD_DEPS)"
-$DIR/environment/os/$DISTRO.sh check MEMGRAPH_BUILD_DEPS
+ARCHITECTURE=$(architecture)
+if [ "${ARCHITECTURE}" = "arm64" ]; then
+  OS_SCRIPT=$DIR/environment/os/$DISTRO-arm.sh
+else
+  OS_SCRIPT=$DIR/environment/os/$DISTRO.sh
+fi
+echo "ALL BUILD PACKAGES: $($OS_SCRIPT list MEMGRAPH_BUILD_DEPS)"
+$OS_SCRIPT check MEMGRAPH_BUILD_DEPS
 echo "All packages are in-place..."
 
 # create a default build directory

licenses/BSL.txt

@@ -36,7 +36,7 @@ ADDITIONAL USE GRANT:   You may use the Licensed Work in accordance with the
                         3. using the Licensed Work to create a work or solution
                         which competes (or might reasonably be expected to
                         compete) with the Licensed Work.
-CHANGE DATE:            2026-17-03
+CHANGE DATE:            2026-27-04
 CHANGE LICENSE:         Apache License, Version 2.0
 
 For information about alternative licensing arrangements, please visit: https://memgraph.com/legal.

release/CMakeLists.txt

@@ -10,6 +10,14 @@ set(CPACK_PACKAGE_VENDOR "Memgraph Ltd.")
 set(CPACK_PACKAGE_DESCRIPTION_SUMMARY
     "High performance, in-memory, transactional graph database")
 
+# Setting arhitecture extension for deb packages
+set(MG_ARCH_EXTENSION_DEB "all")
+if (${MG_ARCH} STREQUAL "x86_64")
+    set(MG_ARCH_EXTENSION_DEB "amd64")
+elseif (${MG_ARCH} STREQUAL "ARM64")
+    set(MG_ARCH_EXTENSION_DEB  "arm64")
+endif()
+
 # DEB specific
 # Instead of using "name <email>" format, we use "email (name)" to prevent
 # errors due to full stop, '.' at the end of "Ltd". (See: RFC 822)
@@ -17,7 +25,7 @@ set(CPACK_DEBIAN_PACKAGE_MAINTAINER "tech@memgraph.com (Memgraph Ltd.)")
 set(CPACK_DEBIAN_PACKAGE_SECTION non-free/database)
 set(CPACK_DEBIAN_PACKAGE_HOMEPAGE https://memgraph.com)
 set(CPACK_DEBIAN_PACKAGE_VERSION "${MEMGRAPH_VERSION_DEB}")
-set(CPACK_DEBIAN_FILE_NAME "memgraph_${MEMGRAPH_VERSION_DEB}_amd64.deb")
+set(CPACK_DEBIAN_FILE_NAME "memgraph_${MEMGRAPH_VERSION_DEB}_${MG_ARCH_EXTENSION_DEB}.deb")
 set(CPACK_DEBIAN_PACKAGE_CONTROL_EXTRA
     "${CMAKE_CURRENT_SOURCE_DIR}/debian/conffiles;"
     "${CMAKE_CURRENT_SOURCE_DIR}/debian/copyright;"
@@ -35,19 +43,18 @@ set(CPACK_DEBIAN_PACKAGE_DESCRIPTION "${CPACK_PACKAGE_DESCRIPTION_SUMMARY}
 # We also depend on `python3` because we embed it in Memgraph.
 set(CPACK_DEBIAN_PACKAGE_DEPENDS "openssl (>= 1.1.0), python3 (>= 3.5.0)")
 
-# RPM specific
-
-set(MG_ARCH_EXTENSION "noarch")
-
+# Setting arhitecture extension for rpm packages
+set(MG_ARCH_EXTENSION_RPM "noarch")
 if (${MG_ARCH} STREQUAL "x86_64")
-    set(MG_ARCH_EXTENSION "x86_64")
+    set(MG_ARCH_EXTENSION_RPM "x86_64")
 elseif (${MG_ARCH} STREQUAL "ARM64")
-    set(MG_ARCH_EXTENSION  "aarch64")
+    set(MG_ARCH_EXTENSION_RPM  "aarch64")
 endif()
 
+# RPM specific
 set(CPACK_RPM_PACKAGE_URL https://memgraph.com)
 set(CPACK_RPM_PACKAGE_VERSION "${MEMGRAPH_VERSION_RPM}")
-set(CPACK_RPM_FILE_NAME "memgraph-${MEMGRAPH_VERSION_RPM}-1.${MG_ARCH_EXTENSION}.rpm")
+set(CPACK_RPM_FILE_NAME "memgraph-${MEMGRAPH_VERSION_RPM}-1.${MG_ARCH_EXTENSION_RPM}.rpm")
 set(CPACK_RPM_EXCLUDE_FROM_AUTO_FILELIST_ADDITION
   /var /var/lib /var/log /etc/logrotate.d
   /lib /lib/systemd /lib/systemd/system /lib/systemd/system/memgraph.service)

release/docker/memgraph_deb.dockerfile

@@ -1,19 +1,21 @@
 FROM debian:bullseye
 # NOTE: If you change the base distro update release/package as well.
 
-ARG release
+ARG BINARY_NAME
+ARG EXTENSION
+ARG TARGETARCH
 
 RUN apt-get update && apt-get install -y \
-    openssl libcurl4 libssl1.1 libseccomp2 python3 libpython3.9 python3-pip \
-    --no-install-recommends \
+  openssl libcurl4 libssl1.1 libseccomp2 python3 libpython3.9 python3-pip \
+  --no-install-recommends \
   && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
 
 RUN pip3 install networkx==2.4 numpy==1.21.4 scipy==1.7.3
 
-COPY ${release} /
+COPY "${BINARY_NAME}${TARGETARCH}.${EXTENSION}" /
 
 # Install memgraph package
-RUN dpkg -i ${release}
+RUN dpkg -i "${BINARY_NAME}${TARGETARCH}.deb"
 
 # Memgraph listens for Bolt Protocol on this port by default.
 EXPOSE 7687

release/docker/package_docker

@@ -55,7 +55,10 @@ image_name="memgraph:${version}"
 image_package_name="memgraph-${version}-docker.tar.gz"
 
 # Build docker image.
-docker build -t ${image_name} ${tag_latest} -f ${dockerfile_path} --build-arg release=${package_name}.${extension} .
+docker build -t ${image_name} ${tag_latest} -f ${dockerfile_path} \
+--build-arg BINARY_NAME=${package_name} \
+--build-arg EXTENSION=${extension} \
+--build-arg TARGETARCH="" .
 docker save ${image_name} ${latest_image} | gzip > ${image_package_name}
 rm "${package_name}.${extension}"
 echo "Built Docker image at '${working_dir}/${image_package_name}'"

src/auth/models.cpp

@@ -22,7 +22,7 @@
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_bool(auth_password_permit_null, true, "Set to false to disable null passwords.");
 
-constexpr std::string_view default_password_regex = ".+";
+inline constexpr std::string_view default_password_regex = ".+";
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_string(auth_password_strength_regex, default_password_regex.data(),
               "The regular expression that should be used to match the entire "

src/communication/bolt/v1/codes.hpp

@@ -15,8 +15,8 @@
 
 namespace memgraph::communication::bolt {
 
-static constexpr uint8_t kPreamble[4] = {0x60, 0x60, 0xB0, 0x17};
-static constexpr uint8_t kProtocol[4] = {0x00, 0x00, 0x00, 0x01};
+inline constexpr uint8_t kPreamble[4] = {0x60, 0x60, 0xB0, 0x17};
+inline constexpr uint8_t kProtocol[4] = {0x00, 0x00, 0x00, 0x01};
 
 enum class Signature : uint8_t {
   Noop = 0x00,
@@ -95,9 +95,9 @@ enum class Marker : uint8_t {
   Struct16 = 0xDD,
 };
 
-static constexpr uint8_t MarkerString = 0, MarkerList = 1, MarkerMap = 2;
-static constexpr Marker MarkerTiny[3] = {Marker::TinyString, Marker::TinyList, Marker::TinyMap};
-static constexpr Marker Marker8[3] = {Marker::String8, Marker::List8, Marker::Map8};
-static constexpr Marker Marker16[3] = {Marker::String16, Marker::List16, Marker::Map16};
-static constexpr Marker Marker32[3] = {Marker::String32, Marker::List32, Marker::Map32};
+inline constexpr uint8_t MarkerString = 0, MarkerList = 1, MarkerMap = 2;
+inline constexpr Marker MarkerTiny[3] = {Marker::TinyString, Marker::TinyList, Marker::TinyMap};
+inline constexpr Marker Marker8[3] = {Marker::String8, Marker::List8, Marker::Map8};
+inline constexpr Marker Marker16[3] = {Marker::String16, Marker::List16, Marker::Map16};
+inline constexpr Marker Marker32[3] = {Marker::String32, Marker::List32, Marker::Map32};
 }  // namespace memgraph::communication::bolt

src/communication/bolt/v1/constants.hpp

@@ -19,17 +19,17 @@ namespace memgraph::communication::bolt {
 /**
  * Sizes related to the chunk defined in Bolt protocol.
  */
-static constexpr size_t kChunkHeaderSize = 2;
-static constexpr size_t kChunkMaxDataSize = 65535;
-static constexpr size_t kChunkWholeSize = kChunkHeaderSize + kChunkMaxDataSize;
+inline constexpr size_t kChunkHeaderSize = 2;
+inline constexpr size_t kChunkMaxDataSize = 65535;
+inline constexpr size_t kChunkWholeSize = kChunkHeaderSize + kChunkMaxDataSize;
 
 /**
  * Handshake size defined in the Bolt protocol.
  */
-static constexpr size_t kHandshakeSize = 20;
+inline constexpr size_t kHandshakeSize = 20;
 
-static constexpr uint16_t kSupportedVersions[] = {0x0100, 0x0400, 0x0401, 0x0403};
+inline constexpr uint16_t kSupportedVersions[] = {0x0100, 0x0400, 0x0401, 0x0403};
 
-static constexpr int kPullAll = -1;
-static constexpr int kPullLast = -1;
+inline constexpr int kPullAll = -1;
+inline constexpr int kPullLast = -1;
 }  // namespace memgraph::communication::bolt

src/communication/context.cpp

@@ -78,14 +78,18 @@ bool ClientContext::use_ssl() { return use_ssl_; }
 
 ServerContext::ServerContext(const std::string &key_file, const std::string &cert_file, const std::string &ca_file,
                              bool verify_peer) {
-  ctx_.emplace(boost::asio::ssl::context::tls_server);
+  namespace ssl = boost::asio::ssl;
+  ctx_.emplace(ssl::context::tls_server);
+  // NOLINTNEXTLINE(hicpp-signed-bitwise)
+  ctx_->set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::no_sslv3 |
+                    ssl::context::single_dh_use);
   ctx_->set_default_verify_paths();
   // TODO: add support for encrypted private keys
   // TODO: add certificate revocation list (CRL)
   boost::system::error_code ec;
   ctx_->use_certificate_chain_file(cert_file, ec);
   MG_ASSERT(!ec, "Couldn't load server certificate from file: {}", cert_file);
-  ctx_->use_private_key_file(key_file, boost::asio::ssl::context::pem, ec);
+  ctx_->use_private_key_file(key_file, ssl::context::pem, ec);
   MG_ASSERT(!ec, "Couldn't load server private key from file: {}", key_file);
 
   ctx_->set_options(SSL_OP_NO_SSLv3, ec);
@@ -100,7 +104,7 @@ ServerContext::ServerContext(const std::string &key_file, const std::string &cer
     if (verify_peer) {
       // Enable verification of the client certificate.
       // NOLINTNEXTLINE(hicpp-signed-bitwise)
-      ctx_->set_verify_mode(boost::asio::ssl::verify_peer | boost::asio::ssl::verify_fail_if_no_peer_cert, ec);
+      ctx_->set_verify_mode(ssl::verify_peer | ssl::verify_fail_if_no_peer_cert, ec);
       MG_ASSERT(!ec, "Setting SSL verification mode failed!");
     }
   }

src/communication/v2/listener.hpp

@@ -0,0 +1,135 @@
+// 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 <atomic>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string_view>
+#include <thread>
+#include <vector>
+
+#include <spdlog/spdlog.h>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/beast/core.hpp>
+#include <boost/system/detail/error_code.hpp>
+
+#include "communication/context.hpp"
+#include "communication/v2/pool.hpp"
+#include "communication/v2/session.hpp"
+#include "utils/spin_lock.hpp"
+#include "utils/synchronized.hpp"
+
+namespace memgraph::communication::v2 {
+
+template <class TSession, class TSessionData>
+class Listener final : public std::enable_shared_from_this<Listener<TSession, TSessionData>> {
+  using tcp = boost::asio::ip::tcp;
+  using SessionHandler = Session<TSession, TSessionData>;
+  using std::enable_shared_from_this<Listener<TSession, TSessionData>>::shared_from_this;
+
+ public:
+  Listener(const Listener &) = delete;
+  Listener(Listener &&) = delete;
+  Listener &operator=(const Listener &) = delete;
+  Listener &operator=(Listener &&) = delete;
+  ~Listener() {}
+
+  template <typename... Args>
+  static std::shared_ptr<Listener> Create(Args &&...args) {
+    return std::shared_ptr<Listener>{new Listener(std::forward<Args>(args)...)};
+  }
+
+  void Start() { DoAccept(); }
+
+  bool IsRunning() const noexcept { return alive_.load(std::memory_order_relaxed); }
+
+ private:
+  Listener(boost::asio::io_context &io_context, TSessionData *data, ServerContext *server_context,
+           tcp::endpoint &endpoint, const std::string_view service_name, const uint64_t inactivity_timeout_sec)
+      : io_context_(io_context),
+        data_(data),
+        server_context_(server_context),
+        acceptor_(io_context_),
+        endpoint_{endpoint},
+        service_name_{service_name},
+        inactivity_timeout_{inactivity_timeout_sec} {
+    boost::system::error_code ec;
+    // Open the acceptor
+    acceptor_.open(endpoint.protocol(), ec);
+    if (ec) {
+      OnError(ec, "open");
+      return;
+    }
+
+    // Allow address reuse
+    acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
+    if (ec) {
+      OnError(ec, "set_option");
+      return;
+    }
+
+    // Bind to the server address
+    acceptor_.bind(endpoint, ec);
+    if (ec) {
+      spdlog::error(
+          utils::MessageWithLink("Cannot bind to socket on endpoint {}.", endpoint, "https://memgr.ph/socket"));
+      OnError(ec, "bind");
+      return;
+    }
+
+    acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
+    if (ec) {
+      OnError(ec, "listen");
+      return;
+    }
+  }
+
+  void DoAccept() {
+    acceptor_.async_accept(io_context_,
+                           [shared_this = shared_from_this()](auto ec, boost::asio::ip::tcp::socket &&socket) {
+                             shared_this->OnAccept(ec, std::move(socket));
+                           });
+  }
+
+  void OnAccept(boost::system::error_code ec, tcp::socket socket) {
+    if (ec) {
+      return OnError(ec, "accept");
+    }
+
+    auto session = SessionHandler::Create(std::move(socket), data_, *server_context_, endpoint_, inactivity_timeout_,
+                                          service_name_);
+    session->Start();
+    DoAccept();
+  }
+
+  void OnError(const boost::system::error_code &ec, const std::string_view what) {
+    spdlog::error("Listener failed on {}: {}", what, ec.message());
+    alive_.store(false, std::memory_order_relaxed);
+  }
+
+  boost::asio::io_context &io_context_;
+  TSessionData *data_;
+  ServerContext *server_context_;
+  tcp::acceptor acceptor_;
+
+  tcp::endpoint endpoint_;
+  std::string_view service_name_;
+  std::chrono::seconds inactivity_timeout_;
+
+  std::atomic<bool> alive_;
+};
+}  // namespace memgraph::communication::v2

src/communication/v2/pool.hpp

@@ -0,0 +1,68 @@
+// 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 <cstddef>
+#include <thread>
+#include <vector>
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+
+#include "utils/logging.hpp"
+
+namespace memgraph::communication::v2 {
+
+class IOContextThreadPool final {
+ private:
+  using IOContext = boost::asio::io_context;
+  using IOContextGuard = boost::asio::executor_work_guard<boost::asio::io_context::executor_type>;
+
+ public:
+  explicit IOContextThreadPool(size_t pool_size) : guard_{io_context_.get_executor()}, pool_size_{pool_size} {
+    MG_ASSERT(pool_size != 0, "Pool size must be greater then 0!");
+  }
+
+  IOContextThreadPool(const IOContextThreadPool &) = delete;
+  IOContextThreadPool &operator=(const IOContextThreadPool &) = delete;
+  IOContextThreadPool(IOContextThreadPool &&) = delete;
+  IOContextThreadPool &operator=(IOContextThreadPool &&) = delete;
+  ~IOContextThreadPool() = default;
+
+  void Run() {
+    background_threads_.reserve(pool_size_);
+    for (size_t i = 0; i < pool_size_; ++i) {
+      background_threads_.emplace_back([this]() { io_context_.run(); });
+    }
+    running_ = true;
+  }
+
+  void Shutdown() {
+    io_context_.stop();
+    running_ = false;
+  }
+
+  void AwaitShutdown() { background_threads_.clear(); }
+
+  bool IsRunning() const noexcept { return running_; }
+
+  IOContext &GetIOContext() noexcept { return io_context_; }
+
+ private:
+  /// The pool of io_context.
+  IOContext io_context_;
+  IOContextGuard guard_;
+  size_t pool_size_;
+  std::vector<std::jthread> background_threads_;
+  bool running_{false};
+};
+}  // namespace memgraph::communication::v2

src/communication/v2/server.hpp

@@ -0,0 +1,128 @@
+// 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 <algorithm>
+#include <cstdint>
+#include <iostream>
+#include <memory>
+#include <optional>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include <fmt/format.h>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/address.hpp>
+#include <boost/asio/ip/tcp.hpp>
+
+#include "communication/context.hpp"
+#include "communication/init.hpp"
+#include "communication/v2/listener.hpp"
+#include "communication/v2/pool.hpp"
+#include "utils/logging.hpp"
+#include "utils/message.hpp"
+#include "utils/thread.hpp"
+
+namespace memgraph::communication::v2 {
+
+using Socket = boost::asio::ip::tcp::socket;
+using ServerEndpoint = boost::asio::ip::tcp::endpoint;
+/**
+ * Communication server.
+ *
+ * Listens for incoming connections on the server port and assigns them to the
+ * connection listener. The listener and session are implemented using asio
+ * async model. Currently the implemented model is thread per core model
+ * opposed to io_context per core. The reasoning for opting for the former model
+ * is the robustness to the multiple resource demanding queries that can be split
+ * across multiple threads, and then a single thread would not block io_context,
+ * unlike in the latter model where it is possible that thread that accepts
+ * request is being blocked by demanding query.
+ * All logic is contained within handlers that are being dispatched
+ * on a single strand per session. The only exception is write which is
+ * synchronous since the nature of the clients conenction is synchronous as
+ * well.
+ *
+ * Current Server architecture:
+ * incoming connection -> server -> listener -> session
+
+ *
+ * @tparam TSession the server can handle different Sessions, each session
+ *         represents a different protocol so the same network infrastructure
+ *         can be used for handling different protocols
+ * @tparam TSessionData the class with objects that will be forwarded to the
+ *         session
+ */
+template <typename TSession, typename TSessionData>
+class Server final {
+  using ServerHandler = Server<TSession, TSessionData>;
+
+ public:
+  /**
+   * Constructs and binds server to endpoint, operates on session data and
+   * invokes workers_count workers
+   */
+  Server(ServerEndpoint &endpoint, TSessionData *session_data, ServerContext *server_context,
+         const int inactivity_timeout_sec, const std::string_view service_name,
+         size_t workers_count = std::thread::hardware_concurrency())
+      : endpoint_{endpoint},
+        service_name_{service_name},
+        context_thread_pool_{workers_count},
+        listener_{Listener<TSession, TSessionData>::Create(context_thread_pool_.GetIOContext(), session_data,
+                                                           server_context, endpoint_, service_name_,
+                                                           inactivity_timeout_sec)} {}
+
+  ~Server() { MG_ASSERT(!IsRunning(), "Server wasn't shutdown properly"); }
+
+  Server(const Server &) = delete;
+  Server(Server &&) = delete;
+  Server &operator=(const Server &) = delete;
+  Server &operator=(Server &&) = delete;
+
+  const auto &Endpoint() const {
+    MG_ASSERT(IsRunning(), "You can't get the server endpoint when it's not running!");
+    return endpoint_;
+  }
+
+  bool Start() {
+    if (IsRunning()) {
+      spdlog::error("The server is already running");
+      return false;
+    }
+    listener_->Start();
+
+    spdlog::info("{} server is fully armed and operational", service_name_);
+    spdlog::info("{} listening on {}", service_name_, endpoint_.address());
+    context_thread_pool_.Run();
+
+    return true;
+  }
+
+  void Shutdown() {
+    context_thread_pool_.Shutdown();
+    spdlog::info("{} shutting down...", service_name_);
+  }
+
+  void AwaitShutdown() { context_thread_pool_.AwaitShutdown(); }
+
+  bool IsRunning() const noexcept { return context_thread_pool_.IsRunning() && listener_->IsRunning(); }
+
+ private:
+  ServerEndpoint endpoint_;
+  std::string service_name_;
+
+  IOContextThreadPool context_thread_pool_;
+  std::shared_ptr<Listener<TSession, TSessionData>> listener_;
+};
+
+}  // namespace memgraph::communication::v2

src/communication/v2/session.hpp

@@ -0,0 +1,513 @@
+// 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 <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <deque>
+#include <functional>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <variant>
+
+#include <spdlog/spdlog.h>
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/read.hpp>
+#include <boost/asio/socket_base.hpp>
+#include <boost/asio/ssl/stream.hpp>
+#include <boost/asio/ssl/stream_base.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/system_context.hpp>
+#include <boost/asio/write.hpp>
+#include <boost/beast/core/tcp_stream.hpp>
+#include <boost/beast/http.hpp>
+#include <boost/beast/websocket.hpp>
+#include <boost/beast/websocket/rfc6455.hpp>
+#include <boost/system/detail/error_code.hpp>
+
+#include "communication/context.hpp"
+#include "communication/exceptions.hpp"
+#include "utils/logging.hpp"
+#include "utils/variant_helpers.hpp"
+
+namespace memgraph::communication::v2 {
+
+/**
+ * This is used to provide input to user Sessions. All Sessions used with the
+ * network stack should use this class as their input stream.
+ */
+using InputStream = communication::Buffer::ReadEnd;
+using tcp = boost::asio::ip::tcp;
+
+/**
+ * This is used to provide output from user Sessions. All Sessions used with the
+ * network stack should use this class for their output stream.
+ */
+class OutputStream final {
+ public:
+  explicit OutputStream(std::function<bool(const uint8_t *, size_t, bool)> write_function)
+      : write_function_(write_function) {}
+
+  OutputStream(const OutputStream &) = delete;
+  OutputStream(OutputStream &&) = delete;
+  OutputStream &operator=(const OutputStream &) = delete;
+  OutputStream &operator=(OutputStream &&) = delete;
+  ~OutputStream() = default;
+
+  bool Write(const uint8_t *data, size_t len, bool have_more = false) { return write_function_(data, len, have_more); }
+
+  bool Write(const std::string &str, bool have_more = false) {
+    return Write(reinterpret_cast<const uint8_t *>(str.data()), str.size(), have_more);
+  }
+
+ private:
+  std::function<bool(const uint8_t *, size_t, bool)> write_function_;
+};
+
+/**
+ * This class is used internally in the communication stack to handle all user
+ * Websocket Sessions. It handles socket ownership, inactivity timeout and protocol
+ * wrapping.
+ */
+template <typename TSession, typename TSessionData>
+class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TSession, TSessionData>> {
+  using WebSocket = boost::beast::websocket::stream<boost::beast::tcp_stream>;
+  using std::enable_shared_from_this<WebsocketSession<TSession, TSessionData>>::shared_from_this;
+
+ public:
+  template <typename... Args>
+  static std::shared_ptr<WebsocketSession> Create(Args &&...args) {
+    return std::shared_ptr<WebsocketSession>(new WebsocketSession(std::forward<Args>(args)...));
+  }
+
+  // Start the asynchronous accept operation
+  template <class Body, class Allocator>
+  void DoAccept(boost::beast::http::request<Body, boost::beast::http::basic_fields<Allocator>> req) {
+    execution_active_ = true;
+    // Set suggested timeout settings for the websocket
+    ws_.set_option(boost::beast::websocket::stream_base::timeout::suggested(boost::beast::role_type::server));
+    boost::asio::socket_base::keep_alive option(true);
+
+    // Set a decorator to change the Server of the handshake
+    ws_.set_option(boost::beast::websocket::stream_base::decorator([](boost::beast::websocket::response_type &res) {
+      res.set(boost::beast::http::field::server, std::string("Memgraph Bolt WS"));
+      res.set(boost::beast::http::field::sec_websocket_protocol, "binary");
+    }));
+    ws_.binary(true);
+
+    // Accept the websocket handshake
+    ws_.async_accept(
+        req, boost::asio::bind_executor(strand_, std::bind_front(&WebsocketSession::OnAccept, shared_from_this())));
+  }
+
+  bool Write(const uint8_t *data, size_t len) {
+    if (!IsConnected()) {
+      return false;
+    }
+
+    boost::system::error_code ec;
+    ws_.write(boost::asio::buffer(data, len), ec);
+    if (ec) {
+      OnError(ec, "write");
+      return false;
+    }
+    return true;
+  }
+
+ private:
+  // Take ownership of the socket
+  explicit WebsocketSession(tcp::socket &&socket, TSessionData *data, tcp::endpoint endpoint,
+                            std::string_view service_name)
+      : ws_(std::move(socket)),
+        strand_{boost::asio::make_strand(ws_.get_executor())},
+        output_stream_([this](const uint8_t *data, size_t len, bool /*have_more*/) { return Write(data, len); }),
+        session_(data, endpoint, input_buffer_.read_end(), &output_stream_),
+        endpoint_{endpoint},
+        remote_endpoint_{ws_.next_layer().socket().remote_endpoint()},
+        service_name_{service_name} {}
+
+  void OnAccept(boost::beast::error_code ec) {
+    if (ec) {
+      return OnError(ec, "accept");
+    }
+
+    // Read a message
+    DoRead();
+  }
+
+  void DoRead() {
+    if (!IsConnected()) {
+      return;
+    }
+    // Read a message into our buffer
+    auto buffer = input_buffer_.write_end()->Allocate();
+    ws_.async_read_some(
+        boost::asio::buffer(buffer.data, buffer.len),
+        boost::asio::bind_executor(strand_, std::bind_front(&WebsocketSession::OnRead, shared_from_this())));
+  }
+
+  void OnRead(const boost::system::error_code &ec, [[maybe_unused]] const size_t bytes_transferred) {
+    // This indicates that the WebsocketSession was closed
+    if (ec == boost::beast::websocket::error::closed) {
+      return;
+    }
+    if (ec) {
+      OnError(ec, "read");
+    }
+    input_buffer_.write_end()->Written(bytes_transferred);
+
+    try {
+      session_.Execute();
+      DoRead();
+    } catch (const SessionClosedException &e) {
+      spdlog::info("{} client {}:{} closed the connection.", service_name_, remote_endpoint_.address(),
+                   remote_endpoint_.port());
+      DoClose();
+    } catch (const std::exception &e) {
+      spdlog::error(
+          "Exception was thrown while processing event in {} session "
+          "associated with {}:{}",
+          service_name_, remote_endpoint_.address(), remote_endpoint_.port());
+      spdlog::debug("Exception message: {}", e.what());
+      DoClose();
+    }
+  }
+
+  void OnError(const boost::system::error_code &ec, const std::string_view action) {
+    spdlog::error("Websocket Bolt session error: {} on {}", ec.message(), action);
+
+    DoClose();
+  }
+
+  void DoClose() {
+    ws_.async_close(
+        boost::beast::websocket::close_code::normal,
+        boost::asio::bind_executor(
+            strand_, [shared_this = shared_from_this()](boost::beast::error_code ec) { shared_this->OnClose(ec); }));
+  }
+
+  void OnClose(const boost::system::error_code &ec) {
+    if (!IsConnected()) {
+      return;
+    }
+    if (ec) {
+      return OnError(ec, "close");
+    }
+  }
+
+  bool IsConnected() const { return ws_.is_open() && execution_active_; }
+
+  WebSocket ws_;
+  boost::asio::strand<WebSocket::executor_type> strand_;
+
+  communication::Buffer input_buffer_;
+  OutputStream output_stream_;
+  TSession session_;
+  tcp::endpoint endpoint_;
+  tcp::endpoint remote_endpoint_;
+  std::string_view service_name_;
+  bool execution_active_{false};
+};
+
+/**
+ * This class is used internally in the communication stack to handle all user
+ * Sessions. It handles socket ownership, inactivity timeout and protocol
+ * wrapping.
+ */
+template <typename TSession, typename TSessionData>
+class Session final : public std::enable_shared_from_this<Session<TSession, TSessionData>> {
+  using TCPSocket = tcp::socket;
+  using SSLSocket = boost::asio::ssl::stream<TCPSocket>;
+  using std::enable_shared_from_this<Session<TSession, TSessionData>>::shared_from_this;
+
+ public:
+  template <typename... Args>
+  static std::shared_ptr<Session> Create(Args &&...args) {
+    return std::shared_ptr<Session>(new Session(std::forward<Args>(args)...));
+  }
+
+  Session(const Session &) = delete;
+  Session(Session &&) = delete;
+  Session &operator=(const Session &) = delete;
+  Session &operator=(Session &&) = delete;
+  ~Session() {
+    if (IsConnected()) {
+      spdlog::error("Session: Destructor called while execution is active");
+    }
+  }
+
+  bool Start() {
+    if (execution_active_) {
+      return false;
+    }
+    execution_active_ = true;
+    timeout_timer_.async_wait(boost::asio::bind_executor(strand_, std::bind(&Session::OnTimeout, shared_from_this())));
+
+    if (std::holds_alternative<SSLSocket>(socket_)) {
+      boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoHandshake(); });
+    } else {
+      boost::asio::dispatch(strand_, [shared_this = shared_from_this()] { shared_this->DoRead(); });
+    }
+    return true;
+  }
+
+  bool Write(const uint8_t *data, size_t len, bool have_more = false) {
+    if (!IsConnected()) {
+      return false;
+    }
+    return std::visit(
+        utils::Overloaded{[shared_this = shared_from_this(), data, len, have_more](TCPSocket &socket) mutable {
+                            boost::system::error_code ec;
+                            while (len > 0) {
+                              const auto sent = socket.send(boost::asio::buffer(data, len),
+                                                            MSG_NOSIGNAL | (have_more ? MSG_MORE : 0), ec);
+                              if (ec) {
+                                shared_this->OnError(ec);
+                                return false;
+                              }
+                              data += sent;
+                              len -= sent;
+                            }
+                            return true;
+                          },
+                          [shared_this = shared_from_this(), data, len](SSLSocket &socket) mutable {
+                            boost::system::error_code ec;
+                            while (len > 0) {
+                              const auto sent = socket.write_some(boost::asio::buffer(data, len), ec);
+                              if (ec) {
+                                shared_this->OnError(ec);
+                                return false;
+                              }
+                              data += sent;
+                              len -= sent;
+                            }
+                            return true;
+                          }},
+        socket_);
+  }
+
+  bool IsConnected() const {
+    return std::visit([this](const auto &socket) { return execution_active_ && socket.lowest_layer().is_open(); },
+                      socket_);
+  }
+
+ private:
+  explicit Session(tcp::socket &&socket, TSessionData *data, ServerContext &server_context, tcp::endpoint endpoint,
+                   const std::chrono::seconds inactivity_timeout_sec, std::string_view service_name)
+      : socket_(CreateSocket(std::move(socket), server_context)),
+        strand_{boost::asio::make_strand(GetExecutor())},
+        output_stream_([this](const uint8_t *data, size_t len, bool have_more) { return Write(data, len, have_more); }),
+        session_(data, endpoint, input_buffer_.read_end(), &output_stream_),
+        data_{data},
+        endpoint_{endpoint},
+        remote_endpoint_{GetRemoteEndpoint()},
+        service_name_{service_name},
+        timeout_seconds_(inactivity_timeout_sec),
+        timeout_timer_(GetExecutor()) {
+    ExecuteForSocket([](auto &&socket) {
+      socket.lowest_layer().set_option(tcp::no_delay(true));                         // enable PSH
+      socket.lowest_layer().set_option(boost::asio::socket_base::keep_alive(true));  // enable SO_KEEPALIVE
+      socket.lowest_layer().non_blocking(false);
+    });
+    timeout_timer_.expires_at(boost::asio::steady_timer::time_point::max());
+    spdlog::info("Accepted a connection from {}:", service_name_, remote_endpoint_.address(), remote_endpoint_.port());
+  }
+
+  void DoRead() {
+    if (!IsConnected()) {
+      return;
+    }
+    timeout_timer_.expires_after(timeout_seconds_);
+    ExecuteForSocket([this](auto &&socket) {
+      auto buffer = input_buffer_.write_end()->Allocate();
+      socket.async_read_some(
+          boost::asio::buffer(buffer.data, buffer.len),
+          boost::asio::bind_executor(strand_, std::bind_front(&Session::OnRead, shared_from_this())));
+    });
+  }
+
+  bool IsWebsocketUpgrade(boost::beast::http::request_parser<boost::beast::http::string_body> &parser) {
+    boost::system::error_code error_code_parsing;
+    parser.put(boost::asio::buffer(input_buffer_.read_end()->data(), input_buffer_.read_end()->size()),
+               error_code_parsing);
+    if (error_code_parsing) {
+      return false;
+    }
+
+    return boost::beast::websocket::is_upgrade(parser.get());
+  }
+
+  void OnRead(const boost::system::error_code &ec, const size_t bytes_transferred) {
+    if (ec) {
+      return OnError(ec);
+    }
+    input_buffer_.write_end()->Written(bytes_transferred);
+
+    // Can be a websocket connection only on the first read, since it is not
+    // expected from clients to upgrade from tcp to websocket
+    if (!has_received_msg_) {
+      has_received_msg_ = true;
+      boost::beast::http::request_parser<boost::beast::http::string_body> parser;
+
+      if (IsWebsocketUpgrade(parser)) {
+        spdlog::info("Switching {} to websocket connection", remote_endpoint_);
+        if (std::holds_alternative<TCPSocket>(socket_)) {
+          auto sock = std::get<TCPSocket>(std::move(socket_));
+          WebsocketSession<TSession, TSessionData>::Create(std::move(sock), data_, endpoint_, service_name_)
+              ->DoAccept(parser.release());
+          execution_active_ = false;
+          return;
+        }
+        spdlog::error("Error while upgrading connection to websocket");
+        DoShutdown();
+      }
+    }
+
+    try {
+      session_.Execute();
+      DoRead();
+    } catch (const SessionClosedException &e) {
+      spdlog::info("{} client {}:{} closed the connection.", service_name_, remote_endpoint_.address(),
+                   remote_endpoint_.port());
+      DoShutdown();
+    } catch (const std::exception &e) {
+      spdlog::error(
+          "Exception was thrown while processing event in {} session "
+          "associated with {}:{}",
+          service_name_, remote_endpoint_.address(), remote_endpoint_.port());
+      spdlog::debug("Exception message: {}", e.what());
+      DoShutdown();
+    }
+  }
+
+  void OnError(const boost::system::error_code &ec) {
+    if (ec == boost::asio::error::operation_aborted) {
+      return;
+    }
+    execution_active_ = false;
+
+    if (ec == boost::asio::error::eof) {
+      spdlog::info("Session closed by peer");
+    } else {
+      spdlog::error("Session error: {}", ec.message());
+    }
+
+    DoShutdown();
+  }
+
+  void DoShutdown() {
+    if (!IsConnected()) {
+      return;
+    }
+    execution_active_ = false;
+    timeout_timer_.cancel();
+    ExecuteForSocket([](auto &socket) {
+      boost::system::error_code ec;
+      auto &lowest_layer = socket.lowest_layer();
+      lowest_layer.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
+      if (ec) {
+        spdlog::error("Session shutdown failed: {}", ec.what());
+      }
+      lowest_layer.close();
+    });
+  }
+
+  void DoHandshake() {
+    if (!IsConnected()) {
+      return;
+    }
+    if (auto *socket = std::get_if<SSLSocket>(&socket_); socket) {
+      socket->async_handshake(
+          boost::asio::ssl::stream_base::server,
+          boost::asio::bind_executor(strand_, std::bind_front(&Session::OnHandshake, shared_from_this())));
+    }
+  }
+
+  void OnHandshake(const boost::system::error_code &ec) {
+    if (ec) {
+      return OnError(ec);
+    }
+    DoRead();
+  }
+
+  void OnClose(const boost::system::error_code &ec) {
+    if (ec) {
+      return OnError(ec);
+    }
+  }
+
+  void OnTimeout() {
+    if (!IsConnected()) {
+      return;
+    }
+    // Check whether the deadline has passed. We compare the deadline against
+    // the current time since a new asynchronous operation may have moved the
+    // deadline before this actor had a chance to run.
+    if (timeout_timer_.expiry() <= boost::asio::steady_timer::clock_type::now()) {
+      // The deadline has passed. Stop the session. The other actors will
+      // terminate as soon as possible.
+      spdlog::info("Shutting down session after {} of inactivity", timeout_seconds_);
+      DoShutdown();
+    } else {
+      // Put the actor back to sleep.
+      timeout_timer_.async_wait(
+          boost::asio::bind_executor(strand_, std::bind(&Session::OnTimeout, shared_from_this())));
+    }
+  }
+
+  std::variant<TCPSocket, SSLSocket> CreateSocket(tcp::socket &&socket, ServerContext &context) {
+    if (context.use_ssl()) {
+      ssl_context_.emplace(context.context_clone());
+      return SSLSocket{std::move(socket), *ssl_context_};
+    }
+
+    return TCPSocket{std::move(socket)};
+  }
+
+  auto GetExecutor() {
+    return std::visit(utils::Overloaded{[](auto &&socket) { return socket.get_executor(); }}, socket_);
+  }
+
+  auto GetRemoteEndpoint() const {
+    return std::visit(utils::Overloaded{[](const auto &socket) { return socket.lowest_layer().remote_endpoint(); }},
+                      socket_);
+  }
+
+  template <typename F>
+  decltype(auto) ExecuteForSocket(F &&fun) {
+    return std::visit(utils::Overloaded{std::forward<F>(fun)}, socket_);
+  }
+
+  std::variant<TCPSocket, SSLSocket> socket_;
+  std::optional<std::reference_wrapper<boost::asio::ssl::context>> ssl_context_;
+  boost::asio::strand<tcp::socket::executor_type> strand_;
+
+  communication::Buffer input_buffer_;
+  OutputStream output_stream_;
+  TSession session_;
+  TSessionData *data_;
+  tcp::endpoint endpoint_;
+  tcp::endpoint remote_endpoint_;
+  std::string_view service_name_;
+  std::chrono::seconds timeout_seconds_;
+  boost::asio::steady_timer timeout_timer_;
+  bool execution_active_{false};
+  bool has_received_msg_{false};
+};
+}  // namespace memgraph::communication::v2

src/communication/websocket/listener.hpp

@@ -11,8 +11,6 @@
 
 #pragma once
 
-#define BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT
-
 #include <list>
 #include <memory>
 

src/communication/websocket/server.cpp

@@ -44,10 +44,10 @@ class QuoteEscapeFormatter : public spdlog::custom_flag_formatter {
   void format(const spdlog::details::log_msg &msg, const std::tm & /*time*/, spdlog::memory_buf_t &dest) override {
     for (const auto c : msg.payload) {
       if (c == '"') {
-        constexpr std::string_view escaped_quote = "\\\"";
+        static constexpr std::string_view escaped_quote = "\\\"";
         dest.append(escaped_quote.data(), escaped_quote.data() + escaped_quote.size());
       } else if (c == '\n') {
-        constexpr std::string_view escaped_newline = "\\n";
+        static constexpr std::string_view escaped_newline = "\\n";
         dest.append(escaped_newline.data(), escaped_newline.data() + escaped_newline.size());
       } else {
         dest.push_back(c);

src/communication/websocket/server.hpp

@@ -11,8 +11,6 @@
 
 #pragma once
 
-#define BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT
-
 #include <thread>
 
 #include <spdlog/sinks/base_sink.h>

src/communication/websocket/session.hpp

@@ -11,8 +11,6 @@
 
 #pragma once
 
-#define BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT
-
 #include <deque>
 #include <memory>
 #include <optional>

src/integrations/constants.hpp

@@ -16,10 +16,10 @@
 
 namespace memgraph::integrations {
 
-constexpr int64_t kDefaultCheckBatchLimit{1};
-constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000};
-constexpr std::chrono::milliseconds kMinimumInterval{1};
-constexpr int64_t kMinimumSize{1};
+inline constexpr int64_t kDefaultCheckBatchLimit{1};
+inline constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000};
+inline constexpr std::chrono::milliseconds kMinimumInterval{1};
+inline constexpr int64_t kMinimumSize{1};
 const std::string kReducted{"<REDUCTED>"};
 
 }  // namespace memgraph::integrations

src/integrations/kafka/consumer.cpp

@@ -185,8 +185,10 @@ Consumer::Consumer(ConsumerInfo info, ConsumerFunction consumer_function)
                  std::inserter(topic_names_from_metadata, topic_names_from_metadata.begin()),
                  [](const auto topic_metadata) { return topic_metadata->topic(); });
 
-  constexpr size_t max_topic_name_length = 249;
-  constexpr auto is_valid_topic_name = [](const auto c) { return std::isalnum(c) || c == '.' || c == '_' || c == '-'; };
+  static constexpr size_t max_topic_name_length = 249;
+  static constexpr auto is_valid_topic_name = [](const auto c) {
+    return std::isalnum(c) || c == '.' || c == '_' || c == '-';
+  };
 
   for (const auto &topic_name : info_.topics) {
     if (topic_name.size() > max_topic_name_length ||
@@ -351,7 +353,7 @@ void Consumer::StartConsuming() {
   }
 
   thread_ = std::thread([this] {
-    constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
+    static constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
     const auto full_thread_name = "Cons#" + info_.consumer_name;
 
     utils::ThreadSetName(full_thread_name.substr(0, kMaxThreadNameSize));

src/integrations/pulsar/consumer.cpp

@@ -231,7 +231,7 @@ void Consumer::StartConsuming() {
   is_running_.store(true);
 
   thread_ = std::thread([this] {
-    constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
+    static constexpr auto kMaxThreadNameSize = utils::GetMaxThreadNameSize();
     const auto full_thread_name = "Cons#" + info_.consumer_name;
 
     utils::ThreadSetName(full_thread_name.substr(0, kMaxThreadNameSize));

src/io/network/addrinfo.cpp

@@ -9,34 +9,52 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
-#include <netdb.h>
-#include <cstring>
-
 #include "io/network/addrinfo.hpp"
 
+#include <concepts>
+#include <iterator>
+
 #include "io/network/network_error.hpp"
 
 namespace memgraph::io::network {
 
-AddrInfo::AddrInfo(struct addrinfo *info) : info(info) {}
+static_assert(std::forward_iterator<AddrInfo::Iterator> && std::equality_comparable<AddrInfo::Iterator>);
 
-AddrInfo::~AddrInfo() { freeaddrinfo(info); }
-
-AddrInfo AddrInfo::Get(const char *addr, const char *port) {
-  struct addrinfo hints;
-  memset(&hints, 0, sizeof(struct addrinfo));
-
-  hints.ai_family = AF_UNSPEC;      // IPv4 and IPv6
-  hints.ai_socktype = SOCK_STREAM;  // TCP socket
-  hints.ai_flags = AI_PASSIVE;
-
-  struct addrinfo *result;
-  auto status = getaddrinfo(addr, port, &hints, &result);
+AddrInfo::AddrInfo(const Endpoint &endpoint) : AddrInfo(endpoint.address, endpoint.port) {}
 
+AddrInfo::AddrInfo(const std::string &addr, uint16_t port) : info_{nullptr, nullptr} {
+  addrinfo hints{
+      .ai_flags = AI_PASSIVE,
+      .ai_family = AF_UNSPEC,     // IPv4 and IPv6
+      .ai_socktype = SOCK_STREAM  // TCP socket
+  };
+  addrinfo *info = nullptr;
+  auto status = getaddrinfo(addr.c_str(), std::to_string(port).c_str(), &hints, &info);
   if (status != 0) throw NetworkError(gai_strerror(status));
-
-  return AddrInfo(result);
+  info_ = std::unique_ptr<addrinfo, decltype(&freeaddrinfo)>(info, &freeaddrinfo);
 }
 
-AddrInfo::operator struct addrinfo *() { return info; }
+AddrInfo::Iterator::Iterator(addrinfo *p) noexcept : ptr_(p) {}
+
+AddrInfo::Iterator::reference AddrInfo::Iterator::operator*() const noexcept { return *ptr_; }
+
+AddrInfo::Iterator::pointer AddrInfo::Iterator::operator->() const noexcept { return ptr_; }
+
+// NOLINTNEXTLINE(cert-dcl21-cpp)
+AddrInfo::Iterator AddrInfo::Iterator::operator++(int) noexcept {
+  auto it = *this;
+  ++(*this);
+  return it;
+}
+AddrInfo::Iterator &AddrInfo::Iterator::operator++() noexcept {
+  ptr_ = ptr_->ai_next;
+  return *this;
+}
+
+bool operator==(const AddrInfo::Iterator &lhs, const AddrInfo::Iterator &rhs) noexcept { return lhs.ptr_ == rhs.ptr_; };
+
+bool operator!=(const AddrInfo::Iterator &lhs, const AddrInfo::Iterator &rhs) noexcept { return !(lhs == rhs); };
+
+void swap(AddrInfo::Iterator &lhs, AddrInfo::Iterator &rhs) noexcept { std::swap(lhs.ptr_, rhs.ptr_); };
+
 }  // namespace memgraph::io::network

src/io/network/addrinfo.hpp

@@ -11,6 +11,14 @@
 
 #pragma once
 
+#include <netdb.h>
+
+#include <iterator>
+#include <memory>
+#include <string>
+
+#include "io/network/endpoint.hpp"
+
 namespace memgraph::io::network {
 
 /**
@@ -18,16 +26,38 @@ namespace memgraph::io::network {
  * see: man 3 getaddrinfo
  */
 class AddrInfo {
-  explicit AddrInfo(struct addrinfo *info);
-
  public:
-  ~AddrInfo();
+  struct Iterator {
+    using iterator_category = std::forward_iterator_tag;
+    using value_type = addrinfo;
+    using difference_type = std::ptrdiff_t;
+    using pointer = addrinfo *;
+    using reference = addrinfo &;
 
-  static AddrInfo Get(const char *addr, const char *port);
+    Iterator() = default;
+    Iterator(const Iterator &) = default;
+    explicit Iterator(addrinfo *p) noexcept;
+    Iterator &operator=(const Iterator &) = default;
+    reference operator*() const noexcept;
+    pointer operator->() const noexcept;
+    Iterator operator++(int) noexcept;
+    Iterator &operator++() noexcept;
 
-  operator struct addrinfo *();
+    friend bool operator==(const Iterator &lhs, const Iterator &rhs) noexcept;
+    friend bool operator!=(const Iterator &lhs, const Iterator &rhs) noexcept;
+    friend void swap(Iterator &lhs, Iterator &rhs) noexcept;
+
+   private:
+    addrinfo *ptr_{nullptr};
+  };
+
+  AddrInfo(const std::string &addr, uint16_t port);
+  explicit AddrInfo(const Endpoint &endpoint);
+
+  auto begin() const noexcept { return Iterator(info_.get()); }
+  auto end() const noexcept { return Iterator{nullptr}; }
 
  private:
-  struct addrinfo *info;
+  std::unique_ptr<addrinfo, void (*)(addrinfo *)> info_;
 };
 }  // namespace memgraph::io::network

src/io/network/socket.cpp

@@ -9,39 +9,25 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
-#include "io/network/socket.hpp"
-
-#include <cstdio>
-#include <cstring>
-#include <iostream>
-#include <stdexcept>
-
 #include <arpa/inet.h>
-#include <errno.h>
 #include <fcntl.h>
-#include <netdb.h>
-#include <netinet/in.h>
 #include <netinet/tcp.h>
 #include <poll.h>
-#include <sys/epoll.h>
-#include <sys/socket.h>
-#include <sys/types.h>
-#include <unistd.h>
 
 #include "io/network/addrinfo.hpp"
+#include "io/network/socket.hpp"
 #include "utils/likely.hpp"
 #include "utils/logging.hpp"
 
 namespace memgraph::io::network {
 
-Socket::Socket(Socket &&other) {
-  socket_ = other.socket_;
-  endpoint_ = std::move(other.endpoint_);
+Socket::Socket(Socket &&other) noexcept : socket_(other.socket_), endpoint_(std::move(other.endpoint_)) {
   other.socket_ = -1;
 }
 
-Socket &Socket::operator=(Socket &&other) {
+Socket &Socket::operator=(Socket &&other) noexcept {
   if (this != &other) {
+    if (socket_ != -1) close(socket_);
     socket_ = other.socket_;
     endpoint_ = std::move(other.endpoint_);
     other.socket_ = -1;
@@ -49,9 +35,8 @@ Socket &Socket::operator=(Socket &&other) {
   return *this;
 }
 
-Socket::~Socket() {
-  if (socket_ == -1) return;
-  close(socket_);
+Socket::~Socket() noexcept {
+  if (socket_ != -1) close(socket_);
 }
 
 void Socket::Close() {
@@ -70,33 +55,27 @@ bool Socket::IsOpen() const { return socket_ != -1; }
 bool Socket::Connect(const Endpoint &endpoint) {
   if (socket_ != -1) return false;
 
-  auto info = AddrInfo::Get(endpoint.address.c_str(), std::to_string(endpoint.port).c_str());
-
-  for (struct addrinfo *it = info; it != nullptr; it = it->ai_next) {
-    int sfd = socket(it->ai_family, it->ai_socktype, it->ai_protocol);
+  for (const auto &it : AddrInfo{endpoint}) {
+    int sfd = socket(it.ai_family, it.ai_socktype, it.ai_protocol);
     if (sfd == -1) continue;
-    if (connect(sfd, it->ai_addr, it->ai_addrlen) == 0) {
+    if (connect(sfd, it.ai_addr, it.ai_addrlen) == 0) {
       socket_ = sfd;
       endpoint_ = endpoint;
       break;
-    } else {
-      // If the connect failed close the file descriptor to prevent file
-      // descriptors being leaked
-      close(sfd);
     }
+    // If the connect failed close the file descriptor to prevent file
+    // descriptors being leaked
+    close(sfd);
   }
 
-  if (socket_ == -1) return false;
-  return true;
+  return !(socket_ == -1);
 }
 
 bool Socket::Bind(const Endpoint &endpoint) {
   if (socket_ != -1) return false;
 
-  auto info = AddrInfo::Get(endpoint.address.c_str(), std::to_string(endpoint.port).c_str());
-
-  for (struct addrinfo *it = info; it != nullptr; it = it->ai_next) {
-    int sfd = socket(it->ai_family, it->ai_socktype, it->ai_protocol);
+  for (const auto &it : AddrInfo{endpoint}) {
+    int sfd = socket(it.ai_family, it.ai_socktype, it.ai_protocol);
     if (sfd == -1) continue;
 
     int on = 1;
@@ -107,14 +86,13 @@ bool Socket::Bind(const Endpoint &endpoint) {
       continue;
     }
 
-    if (bind(sfd, it->ai_addr, it->ai_addrlen) == 0) {
+    if (bind(sfd, it.ai_addr, it.ai_addrlen) == 0) {
       socket_ = sfd;
       break;
-    } else {
-      // If the bind failed close the file descriptor to prevent file
-      // descriptors being leaked
-      close(sfd);
     }
+    // If the bind failed close the file descriptor to prevent file
+    // descriptors being leaked
+    close(sfd);
   }
 
   if (socket_ == -1) return false;
@@ -122,7 +100,7 @@ bool Socket::Bind(const Endpoint &endpoint) {
   // detect bound port, used when the server binds to a random port
   struct sockaddr_in6 portdata;
   socklen_t portdatalen = sizeof(portdata);
-  if (getsockname(socket_, (struct sockaddr *)&portdata, &portdatalen) < 0) {
+  if (getsockname(socket_, reinterpret_cast<sockaddr *>(&portdata), &portdatalen) < 0) {
     // If the getsockname failed close the file descriptor to prevent file
     // descriptors being leaked
     close(socket_);
@@ -136,36 +114,35 @@ bool Socket::Bind(const Endpoint &endpoint) {
 }
 
 void Socket::SetNonBlocking() {
-  int flags = fcntl(socket_, F_GETFL, 0);
+  const unsigned flags = fcntl(socket_, F_GETFL);
+  constexpr unsigned o_nonblock = O_NONBLOCK;
   MG_ASSERT(flags != -1, "Can't get socket mode");
-  flags |= O_NONBLOCK;
-  MG_ASSERT(fcntl(socket_, F_SETFL, flags) != -1, "Can't set socket nonblocking");
+  MG_ASSERT(fcntl(socket_, F_SETFL, flags | o_nonblock) != -1, "Can't set socket nonblocking");
 }
 
 void Socket::SetKeepAlive() {
   int optval = 1;
-  socklen_t optlen = sizeof(optval);
-
-  MG_ASSERT(!setsockopt(socket_, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen), "Can't set socket keep alive");
+  MG_ASSERT(!setsockopt(socket_, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(optval)), "Can't set socket keep alive");
 
   optval = 20;  // wait 20s before sending keep-alive packets
-  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPIDLE, (void *)&optval, optlen), "Can't set socket keep alive");
+  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPIDLE, (void *)&optval, sizeof(optval)),
+            "Can't set socket keep alive");
 
   optval = 4;  // 4 keep-alive packets must fail to close
-  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPCNT, (void *)&optval, optlen), "Can't set socket keep alive");
+  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPCNT, (void *)&optval, sizeof(optval)), "Can't set socket keep alive");
 
   optval = 15;  // send keep-alive packets every 15s
-  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPINTVL, (void *)&optval, optlen), "Can't set socket keep alive");
+  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_KEEPINTVL, (void *)&optval, sizeof(optval)),
+            "Can't set socket keep alive");
 }
 
 void Socket::SetNoDelay() {
   int optval = 1;
-  socklen_t optlen = sizeof(optval);
-
-  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_NODELAY, (void *)&optval, optlen), "Can't set socket no delay");
+  MG_ASSERT(!setsockopt(socket_, SOL_TCP, TCP_NODELAY, (void *)&optval, sizeof(optval)), "Can't set socket no delay");
 }
 
-void Socket::SetTimeout(long sec, long usec) {
+// NOLINTNEXTLINE(readability-make-member-function-const)
+void Socket::SetTimeout(int64_t sec, int64_t usec) {
   struct timeval tv;
   tv.tv_sec = sec;
   tv.tv_usec = usec;
@@ -176,7 +153,7 @@ void Socket::SetTimeout(long sec, long usec) {
 }
 
 int Socket::ErrorStatus() const {
-  int optval;
+  int optval = 0;
   socklen_t optlen = sizeof(optval);
   auto status = getsockopt(socket_, SOL_SOCKET, SO_ERROR, &optval, &optlen);
   MG_ASSERT(!status, "getsockopt failed");
@@ -189,21 +166,22 @@ std::optional<Socket> Socket::Accept() {
   sockaddr_storage addr;
   socklen_t addr_size = sizeof addr;
   char addr_decoded[INET6_ADDRSTRLEN];
-  void *addr_src;
-  unsigned short port;
 
-  int sfd = accept(socket_, (struct sockaddr *)&addr, &addr_size);
+  int sfd = accept(socket_, reinterpret_cast<sockaddr *>(&addr), &addr_size);
   if (UNLIKELY(sfd == -1)) return std::nullopt;
 
+  void *addr_src = nullptr;
+  uint16_t port = 0;
+
   if (addr.ss_family == AF_INET) {
-    addr_src = (void *)&(((sockaddr_in *)&addr)->sin_addr);
-    port = ntohs(((sockaddr_in *)&addr)->sin_port);
+    addr_src = &reinterpret_cast<sockaddr_in &>(addr).sin_addr;
+    port = ntohs(reinterpret_cast<sockaddr_in &>(addr).sin_port);
   } else {
-    addr_src = (void *)&(((sockaddr_in6 *)&addr)->sin6_addr);
-    port = ntohs(((sockaddr_in6 *)&addr)->sin6_port);
+    addr_src = &reinterpret_cast<sockaddr_in6 &>(addr).sin6_addr;
+    port = ntohs(reinterpret_cast<sockaddr_in6 &>(addr).sin6_port);
   }
 
-  inet_ntop(addr.ss_family, addr_src, addr_decoded, INET6_ADDRSTRLEN);
+  inet_ntop(addr.ss_family, addr_src, addr_decoded, sizeof(addr_decoded));
 
   Endpoint endpoint(addr_decoded, port);
 
@@ -213,9 +191,11 @@ std::optional<Socket> Socket::Accept() {
 bool Socket::Write(const uint8_t *data, size_t len, bool have_more) {
   // MSG_NOSIGNAL is here to disable raising a SIGPIPE signal when a
   // connection dies mid-write, the socket will only return an EPIPE error.
-  int flags = MSG_NOSIGNAL | (have_more ? MSG_MORE : 0);
+  constexpr unsigned msg_nosignal = MSG_NOSIGNAL;
+  constexpr unsigned msg_more = MSG_MORE;
+  const unsigned flags = msg_nosignal | (have_more ? msg_more : 0);
   while (len > 0) {
-    auto written = send(socket_, data, len, flags);
+    auto written = send(socket_, data, len, static_cast<int>(flags));
     if (written == -1) {
       if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
         // Terminal error, return failure.
@@ -253,7 +233,8 @@ bool Socket::WaitForReadyRead() {
   // event occurs.
   int ret = poll(&p, 1, -1);
   if (ret < 1) return false;
-  return p.revents & POLLIN;
+  constexpr unsigned pollin = POLLIN;
+  return static_cast<unsigned>(p.revents) & pollin;
 }
 
 bool Socket::WaitForReadyWrite() {
@@ -265,7 +246,8 @@ bool Socket::WaitForReadyWrite() {
   // event occurs.
   int ret = poll(&p, 1, -1);
   if (ret < 1) return false;
-  return p.revents & POLLOUT;
+  constexpr unsigned pollout = POLLOUT;
+  return static_cast<unsigned>(p.revents) & pollout;
 }
 
 }  // namespace memgraph::io::network

src/io/network/socket.hpp

@@ -27,12 +27,12 @@ namespace memgraph::io::network {
  */
 class Socket {
  public:
-  Socket() = default;
+  Socket() noexcept = default;
   Socket(const Socket &) = delete;
   Socket &operator=(const Socket &) = delete;
-  Socket(Socket &&);
-  Socket &operator=(Socket &&);
-  ~Socket();
+  Socket(Socket &&) noexcept;
+  Socket &operator=(Socket &&) noexcept;
+  ~Socket() noexcept;
 
   /**
    * Closes the socket if it is open.
@@ -118,7 +118,7 @@ class Socket {
    * @param sec timeout seconds value
    * @param usec timeout microseconds value
    */
-  void SetTimeout(long sec, long usec);
+  void SetTimeout(int64_t sec, int64_t usec);
 
   /**
    * Checks if there are any errors on a socket. Returns 0 if there are none.

src/memgraph.cpp

@@ -81,8 +81,8 @@
 #include "communication/bolt/v1/exceptions.hpp"
 #include "communication/bolt/v1/session.hpp"
 #include "communication/init.hpp"
-#include "communication/server.hpp"
-#include "communication/session.hpp"
+#include "communication/v2/server.hpp"
+#include "communication/v2/session.hpp"
 #include "glue/communication.hpp"
 
 #include "auth/auth.hpp"
@@ -260,7 +260,7 @@ DEFINE_uint64(
 
 namespace {
 using namespace std::literals;
-constexpr std::array isolation_level_mappings{
+inline constexpr std::array isolation_level_mappings{
     std::pair{"SNAPSHOT_ISOLATION"sv, memgraph::storage::IsolationLevel::SNAPSHOT_ISOLATION},
     std::pair{"READ_COMMITTED"sv, memgraph::storage::IsolationLevel::READ_COMMITTED},
     std::pair{"READ_UNCOMMITTED"sv, memgraph::storage::IsolationLevel::READ_UNCOMMITTED}};
@@ -348,7 +348,7 @@ DEFINE_bool(also_log_to_stderr, false, "Log messages go to stderr in addition to
 DEFINE_string(log_file, "", "Path to where the log should be stored.");
 
 namespace {
-constexpr std::array log_level_mappings{
+inline constexpr std::array log_level_mappings{
     std::pair{"TRACE"sv, spdlog::level::trace}, std::pair{"DEBUG"sv, spdlog::level::debug},
     std::pair{"INFO"sv, spdlog::level::info},   std::pair{"WARNING"sv, spdlog::level::warn},
     std::pair{"ERROR"sv, spdlog::level::err},   std::pair{"CRITICAL"sv, spdlog::level::critical}};
@@ -385,7 +385,7 @@ spdlog::level::level_enum ParseLogLevel() {
 }
 
 // 5 weeks * 7 days
-constexpr auto log_retention_count = 35;
+inline constexpr auto log_retention_count = 35;
 void CreateLoggerFromSink(const auto &sinks, const auto log_level) {
   auto logger = std::make_shared<spdlog::logger>("memgraph_log", sinks.begin(), sinks.end());
   logger->set_level(log_level);
@@ -456,7 +456,7 @@ struct SessionData {
 #endif
 };
 
-constexpr std::string_view default_user_role_regex = "[a-zA-Z0-9_.+-@]+";
+inline constexpr std::string_view default_user_role_regex = "[a-zA-Z0-9_.+-@]+";
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_string(auth_user_or_role_name_regex, default_user_role_regex.data(),
               "Set to the regular expression that each user or role name must fulfill.");
@@ -842,13 +842,14 @@ class AuthChecker final : public memgraph::query::AuthChecker {
   memgraph::utils::Synchronized<memgraph::auth::Auth, memgraph::utils::WritePrioritizedRWLock> *auth_;
 };
 
-class BoltSession final : public memgraph::communication::bolt::Session<memgraph::communication::InputStream,
-                                                                        memgraph::communication::OutputStream> {
+class BoltSession final : public memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
+                                                                        memgraph::communication::v2::OutputStream> {
  public:
-  BoltSession(SessionData *data, const memgraph::io::network::Endpoint &endpoint,
-              memgraph::communication::InputStream *input_stream, memgraph::communication::OutputStream *output_stream)
-      : memgraph::communication::bolt::Session<memgraph::communication::InputStream,
-                                               memgraph::communication::OutputStream>(input_stream, output_stream),
+  BoltSession(SessionData *data, const memgraph::communication::v2::ServerEndpoint &endpoint,
+              memgraph::communication::v2::InputStream *input_stream,
+              memgraph::communication::v2::OutputStream *output_stream)
+      : memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
+                                               memgraph::communication::v2::OutputStream>(input_stream, output_stream),
         db_(data->db),
         interpreter_(data->interpreter_context),
         auth_(data->auth),
@@ -858,8 +859,8 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
         endpoint_(endpoint) {
   }
 
-  using memgraph::communication::bolt::Session<memgraph::communication::InputStream,
-                                               memgraph::communication::OutputStream>::TEncoder;
+  using memgraph::communication::bolt::Session<memgraph::communication::v2::InputStream,
+                                               memgraph::communication::v2::OutputStream>::TEncoder;
 
   void BeginTransaction() override { interpreter_.BeginTransaction(); }
 
@@ -877,7 +878,8 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
     }
 #ifdef MG_ENTERPRISE
     if (memgraph::utils::license::global_license_checker.IsValidLicenseFast()) {
-      audit_log_->Record(endpoint_.address, user_ ? *username : "", query, memgraph::storage::PropertyValue(params_pv));
+      audit_log_->Record(endpoint_.address().to_string(), user_ ? *username : "", query,
+                         memgraph::storage::PropertyValue(params_pv));
     }
 #endif
     try {
@@ -996,10 +998,10 @@ class BoltSession final : public memgraph::communication::bolt::Session<memgraph
 #ifdef MG_ENTERPRISE
   memgraph::audit::Log *audit_log_;
 #endif
-  memgraph::io::network::Endpoint endpoint_;
+  memgraph::communication::v2::ServerEndpoint endpoint_;
 };
 
-using ServerT = memgraph::communication::Server<BoltSession, SessionData>;
+using ServerT = memgraph::communication::v2::Server<BoltSession, SessionData>;
 using memgraph::communication::ServerContext;
 
 // Needed to correctly handle memgraph destruction from a signal handler.
@@ -1241,8 +1243,10 @@ int main(int argc, char **argv) {
         memgraph::utils::MessageWithLink("Using non-secure Bolt connection (without SSL).", "https://memgr.ph/ssl"));
   }
 
-  ServerT server({FLAGS_bolt_address, static_cast<uint16_t>(FLAGS_bolt_port)}, &session_data, &context,
-                 FLAGS_bolt_session_inactivity_timeout, service_name, FLAGS_bolt_num_workers);
+  auto server_endpoint = memgraph::communication::v2::ServerEndpoint{
+      boost::asio::ip::address::from_string(FLAGS_bolt_address), static_cast<uint16_t>(FLAGS_bolt_port)};
+  ServerT server(server_endpoint, &session_data, &context, FLAGS_bolt_session_inactivity_timeout, service_name,
+                 FLAGS_bolt_num_workers);
 
   // Setup telemetry
   std::optional<memgraph::telemetry::Telemetry> telemetry;

src/query/constants.hpp

@@ -14,6 +14,6 @@
 #include <string>
 
 namespace memgraph::query {
-constexpr uint16_t kDefaultReplicationPort = 10000;
-constexpr auto *kDefaultReplicationServerIp = "0.0.0.0";
+inline constexpr uint16_t kDefaultReplicationPort = 10000;
+inline constexpr auto *kDefaultReplicationServerIp = "0.0.0.0";
 }  // namespace memgraph::query

src/query/frontend/ast/ast.lcp

@@ -852,7 +852,9 @@ cpp<#
         : arguments_(arguments),
           function_name_(function_name),
           function_(NameToFunction(function_name_)) {
-      DMG_ASSERT(function_, "Unexpected missing function: {}", function_name_);
+      if (!function_) {
+        throw SemanticException("Function '{}' doesn't exist.", function_name);
+      }
     }
     cpp<#)
   (:private
@@ -2629,5 +2631,39 @@ cpp<#
   (:serialize (:slk))
   (:clone))
 
+(lcp:define-class foreach (clause)
+  ((named_expression "NamedExpression *" :initval "nullptr" :scope :public
+          :slk-save #'slk-save-ast-pointer
+          :slk-load (slk-load-ast-pointer "Expression"))
+   (clauses "std::vector<Clause *>"
+          :scope :public
+          :slk-save #'slk-save-ast-vector
+          :slk-load (slk-load-ast-vector "Clause")))
+  (:public
+    #>cpp
+    Foreach() = default;
+
+    bool Accept(HierarchicalTreeVisitor &visitor) override {
+      if (visitor.PreVisit(*this)) {
+          named_expression_->Accept(visitor);
+          for (auto &clause : clauses_) {
+            clause->Accept(visitor);
+          }
+      }
+      return visitor.PostVisit(*this);
+    }
+    cpp<#)
+  (:protected
+    #>cpp
+    Foreach(NamedExpression *expression, std::vector<Clause *> clauses)
+        : named_expression_(expression), clauses_(clauses) {}
+    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

@@ -93,6 +93,7 @@ class CreateSnapshotQuery;
 class StreamQuery;
 class SettingQuery;
 class VersionQuery;
+class Foreach;
 
 using TreeCompositeVisitor = utils::CompositeVisitor<
     SingleQuery, CypherUnion, NamedExpression, OrOperator, XorOperator, AndOperator, NotOperator, AdditionOperator,
@@ -101,7 +102,7 @@ using TreeCompositeVisitor = utils::CompositeVisitor<
     ListSlicingOperator, IfOperator, UnaryPlusOperator, UnaryMinusOperator, IsNullOperator, ListLiteral, MapLiteral,
     PropertyLookup, LabelsTest, Aggregation, Function, Reduce, Coalesce, Extract, All, Single, Any, None, CallProcedure,
     Create, Match, Return, With, Pattern, NodeAtom, EdgeAtom, Delete, Where, SetProperty, SetProperties, SetLabels,
-    RemoveProperty, RemoveLabels, Merge, Unwind, RegexMatch, LoadCsv>;
+    RemoveProperty, RemoveLabels, Merge, Unwind, RegexMatch, LoadCsv, Foreach>;
 
 using TreeLeafVisitor = utils::LeafVisitor<Identifier, PrimitiveLiteral, ParameterLookup>;
 

src/query/frontend/ast/cypher_main_visitor.cpp

@@ -37,6 +37,7 @@
 #include "utils/exceptions.hpp"
 #include "utils/logging.hpp"
 #include "utils/string.hpp"
+#include "utils/typeinfo.hpp"
 
 namespace memgraph::query::frontend {
 
@@ -645,28 +646,28 @@ antlrcpp::Any CypherMainVisitor::visitKafkaCreateStreamConfig(MemgraphCypher::Ka
 
   if (ctx->TOPICS()) {
     ThrowIfExists(memory_, KafkaConfigKey::TOPICS);
-    constexpr auto topics_key = static_cast<uint8_t>(KafkaConfigKey::TOPICS);
+    static constexpr auto topics_key = static_cast<uint8_t>(KafkaConfigKey::TOPICS);
     GetTopicNames(memory_[topics_key], ctx->topicNames(), *this);
     return {};
   }
 
   if (ctx->CONSUMER_GROUP()) {
     ThrowIfExists(memory_, KafkaConfigKey::CONSUMER_GROUP);
-    constexpr auto consumer_group_key = static_cast<uint8_t>(KafkaConfigKey::CONSUMER_GROUP);
+    static constexpr auto consumer_group_key = static_cast<uint8_t>(KafkaConfigKey::CONSUMER_GROUP);
     memory_[consumer_group_key] = JoinSymbolicNamesWithDotsAndMinus(*this, *ctx->consumerGroup);
     return {};
   }
 
   if (ctx->CONFIGS()) {
     ThrowIfExists(memory_, KafkaConfigKey::CONFIGS);
-    constexpr auto configs_key = static_cast<uint8_t>(KafkaConfigKey::CONFIGS);
+    static constexpr auto configs_key = static_cast<uint8_t>(KafkaConfigKey::CONFIGS);
     memory_.emplace(configs_key, ctx->configsMap->accept(this).as<std::unordered_map<Expression *, Expression *>>());
     return {};
   }
 
   if (ctx->CREDENTIALS()) {
     ThrowIfExists(memory_, KafkaConfigKey::CREDENTIALS);
-    constexpr auto credentials_key = static_cast<uint8_t>(KafkaConfigKey::CREDENTIALS);
+    static constexpr auto credentials_key = static_cast<uint8_t>(KafkaConfigKey::CREDENTIALS);
     memory_.emplace(credentials_key,
                     ctx->credentialsMap->accept(this).as<std::unordered_map<Expression *, Expression *>>());
     return {};
@@ -956,7 +957,8 @@ antlrcpp::Any CypherMainVisitor::visitSingleQuery(MemgraphCypher::SingleQueryCon
                utils::IsSubtype(clause_type, SetProperty::kType) ||
                utils::IsSubtype(clause_type, SetProperties::kType) || utils::IsSubtype(clause_type, SetLabels::kType) ||
                utils::IsSubtype(clause_type, RemoveProperty::kType) ||
-               utils::IsSubtype(clause_type, RemoveLabels::kType) || utils::IsSubtype(clause_type, Merge::kType)) {
+               utils::IsSubtype(clause_type, RemoveLabels::kType) || utils::IsSubtype(clause_type, Merge::kType) ||
+               utils::IsSubtype(clause_type, Foreach::kType)) {
       if (has_return) {
         throw SemanticException("Update clause can't be used after RETURN.");
       }
@@ -1036,6 +1038,9 @@ antlrcpp::Any CypherMainVisitor::visitClause(MemgraphCypher::ClauseContext *ctx)
   if (ctx->loadCsv()) {
     return static_cast<Clause *>(ctx->loadCsv()->accept(this).as<LoadCsv *>());
   }
+  if (ctx->foreach ()) {
+    return static_cast<Clause *>(ctx->foreach ()->accept(this).as<Foreach *>());
+  }
   // TODO: implement other clauses.
   throw utils::NotYetImplemented("clause '{}'", ctx->getText());
   return 0;
@@ -2104,13 +2109,30 @@ antlrcpp::Any CypherMainVisitor::visitFunctionInvocation(MemgraphCypher::Functio
         storage_->Create<Aggregation>(expressions[1], expressions[0], Aggregation::Op::COLLECT_MAP));
   }
 
-  auto function = NameToFunction(function_name);
-  if (!function) throw SemanticException("Function '{}' doesn't exist.", function_name);
+  auto is_user_defined_function = [](const std::string &function_name) {
+    // Dots are present only in user-defined functions, since modules are case-sensitive, so must be user-defined
+    // functions. Builtin functions should be case insensitive.
+    return function_name.find('.') != std::string::npos;
+  };
+
+  // Don't cache queries which call user-defined functions. User-defined function's return
+  // types can vary depending on whether the module is reloaded, therefore the cache would
+  // be invalid.
+  if (is_user_defined_function(function_name)) {
+    query_info_.is_cacheable = false;
+  }
+
   return static_cast<Expression *>(storage_->Create<Function>(function_name, expressions));
 }
 
 antlrcpp::Any CypherMainVisitor::visitFunctionName(MemgraphCypher::FunctionNameContext *ctx) {
-  return utils::ToUpperCase(ctx->getText());
+  auto function_name = ctx->getText();
+  // Dots are present only in user-defined functions, since modules are case-sensitive, so must be user-defined
+  // functions. Builtin functions should be case insensitive.
+  if (function_name.find('.') != std::string::npos) {
+    return function_name;
+  }
+  return utils::ToUpperCase(function_name);
 }
 
 antlrcpp::Any CypherMainVisitor::visitDoubleLiteral(MemgraphCypher::DoubleLiteralContext *ctx) {
@@ -2283,6 +2305,37 @@ antlrcpp::Any CypherMainVisitor::visitFilterExpression(MemgraphCypher::FilterExp
   return 0;
 }
 
+antlrcpp::Any CypherMainVisitor::visitForeach(MemgraphCypher::ForeachContext *ctx) {
+  auto *for_each = storage_->Create<Foreach>();
+
+  auto *named_expr = storage_->Create<NamedExpression>();
+  named_expr->expression_ = ctx->expression()->accept(this);
+  named_expr->name_ = std::string(ctx->variable()->accept(this).as<std::string>());
+  for_each->named_expression_ = named_expr;
+
+  for (auto *update_clause_ctx : ctx->updateClause()) {
+    if (auto *set = update_clause_ctx->set(); set) {
+      auto set_items = visitSet(set).as<std::vector<Clause *>>();
+      std::copy(set_items.begin(), set_items.end(), std::back_inserter(for_each->clauses_));
+    } else if (auto *remove = update_clause_ctx->remove(); remove) {
+      auto remove_items = visitRemove(remove).as<std::vector<Clause *>>();
+      std::copy(remove_items.begin(), remove_items.end(), std::back_inserter(for_each->clauses_));
+    } else if (auto *merge = update_clause_ctx->merge(); merge) {
+      for_each->clauses_.push_back(visitMerge(merge).as<Merge *>());
+    } else if (auto *create = update_clause_ctx->create(); create) {
+      for_each->clauses_.push_back(visitCreate(create).as<Create *>());
+    } else if (auto *cypher_delete = update_clause_ctx->cypherDelete(); cypher_delete) {
+      for_each->clauses_.push_back(visitCypherDelete(cypher_delete).as<Delete *>());
+    } else {
+      auto *nested_for_each = update_clause_ctx->foreach ();
+      MG_ASSERT(nested_for_each != nullptr, "Unexpected clause in FOREACH");
+      for_each->clauses_.push_back(visitForeach(nested_for_each).as<Foreach *>());
+    }
+  }
+
+  return for_each;
+}
+
 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

@@ -844,6 +844,11 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
    */
   antlrcpp::Any visitFilterExpression(MemgraphCypher::FilterExpressionContext *) override;
 
+  /**
+   * @return Foreach*
+   */
+  antlrcpp::Any visitForeach(MemgraphCypher::ForeachContext *ctx) override;
+
  public:
   Query *query() { return query_; }
   const static std::string kAnonPrefix;

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

@@ -279,7 +279,7 @@ idInColl : variable IN expression ;
 
 functionInvocation : functionName '(' ( DISTINCT )? ( expression ( ',' expression )* )? ')' ;
 
-functionName : symbolicName ;
+functionName : symbolicName ( '.' symbolicName )* ;
 
 listComprehension : '[' filterExpression ( '|' expression )? ']' ;
 

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

@@ -47,6 +47,7 @@ memgraphCypherKeyword : cypherKeyword
                       | DUMP
                       | EXECUTE
                       | FOR
+                      | FOREACH
                       | FREE
                       | FROM
                       | GLOBAL
@@ -163,8 +164,19 @@ clause : cypherMatch
        | cypherReturn
        | callProcedure
        | loadCsv
+       | foreach
        ;
 
+updateClause : set
+             | remove
+             | create
+             | merge
+             | cypherDelete
+             | foreach
+             ;
+
+foreach :  FOREACH '(' variable IN expression '|' updateClause+  ')' ;
+
 streamQuery : checkStream
             | createStream
             | dropStream

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

@@ -54,6 +54,7 @@ DUMP                : D U M P ;
 DURABILITY          : D U R A B I L I T Y ;
 EXECUTE             : E X E C U T E ;
 FOR                 : F O R ;
+FOREACH             : F O R E A C H;
 FREE                : F R E E ;
 FREE_MEMORY         : F R E E UNDERSCORE M E M O R Y ;
 FROM                : F R O M ;

src/query/frontend/semantic/symbol_generator.cpp

@@ -15,7 +15,9 @@
 
 #include "query/frontend/semantic/symbol_generator.hpp"
 
+#include <algorithm>
 #include <optional>
+#include <ranges>
 #include <unordered_set>
 #include <variant>
 
@@ -39,35 +41,56 @@ std::unordered_map<std::string, Identifier *> GeneratePredefinedIdentifierMap(
 }  // namespace
 
 SymbolGenerator::SymbolGenerator(SymbolTable *symbol_table, const std::vector<Identifier *> &predefined_identifiers)
-    : symbol_table_(symbol_table), predefined_identifiers_{GeneratePredefinedIdentifierMap(predefined_identifiers)} {}
+    : symbol_table_(symbol_table),
+      predefined_identifiers_{GeneratePredefinedIdentifierMap(predefined_identifiers)},
+      scopes_(1, Scope()) {}
 
-auto SymbolGenerator::CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type, int token_position) {
-  auto symbol = symbol_table_->CreateSymbol(name, user_declared, type, token_position);
-  scope_.symbols[name] = symbol;
-  return symbol;
-}
-
-auto SymbolGenerator::GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type) {
-  auto search = scope_.symbols.find(name);
-  if (search != scope_.symbols.end()) {
-    auto symbol = search->second;
+std::optional<Symbol> SymbolGenerator::FindSymbolInScope(const std::string &name, const Scope &scope,
+                                                         Symbol::Type type) {
+  if (auto it = scope.symbols.find(name); it != scope.symbols.end()) {
+    const auto &symbol = it->second;
     // Unless we have `ANY` type, check that types match.
     if (type != Symbol::Type::ANY && symbol.type() != Symbol::Type::ANY && type != symbol.type()) {
       throw TypeMismatchError(name, Symbol::TypeToString(symbol.type()), Symbol::TypeToString(type));
     }
-    return search->second;
+    return symbol;
+  }
+  return std::nullopt;
+}
+
+auto SymbolGenerator::CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type, int token_position) {
+  auto symbol = symbol_table_->CreateSymbol(name, user_declared, type, token_position);
+  scopes_.back().symbols[name] = symbol;
+  return symbol;
+}
+
+auto SymbolGenerator::GetOrCreateSymbolLocalScope(const std::string &name, bool user_declared, Symbol::Type type) {
+  auto &scope = scopes_.back();
+  if (auto maybe_symbol = FindSymbolInScope(name, scope, type); maybe_symbol) {
+    return *maybe_symbol;
+  }
+  return CreateSymbol(name, user_declared, type);
+}
+
+auto SymbolGenerator::GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type) {
+  // NOLINTNEXTLINE
+  for (auto scope = scopes_.rbegin(); scope != scopes_.rend(); ++scope) {
+    if (auto maybe_symbol = FindSymbolInScope(name, *scope, type); maybe_symbol) {
+      return *maybe_symbol;
+    }
   }
   return CreateSymbol(name, user_declared, type);
 }
 
 void SymbolGenerator::VisitReturnBody(ReturnBody &body, Where *where) {
+  auto &scope = scopes_.back();
   for (auto &expr : body.named_expressions) {
     expr->Accept(*this);
   }
   std::vector<Symbol> user_symbols;
   if (body.all_identifiers) {
     // Carry over user symbols because '*' appeared.
-    for (auto sym_pair : scope_.symbols) {
+    for (const auto &sym_pair : scope.symbols) {
       if (!sym_pair.second.user_declared()) {
         continue;
       }
@@ -81,18 +104,18 @@ void SymbolGenerator::VisitReturnBody(ReturnBody &body, Where *where) {
   // declares only those established through named expressions. New declarations
   // must not be visible inside named expressions themselves.
   bool removed_old_names = false;
-  if ((!where && body.order_by.empty()) || scope_.has_aggregation) {
+  if ((!where && body.order_by.empty()) || scope.has_aggregation) {
     // WHERE and ORDER BY need to see both the old and new symbols, unless we
     // have an aggregation. Therefore, we can clear the symbols immediately if
     // there is neither ORDER BY nor WHERE, or we have an aggregation.
-    scope_.symbols.clear();
+    scope.symbols.clear();
     removed_old_names = true;
   }
   // Create symbols for named expressions.
   std::unordered_set<std::string> new_names;
   for (const auto &user_sym : user_symbols) {
     new_names.insert(user_sym.name());
-    scope_.symbols[user_sym.name()] = user_sym;
+    scope.symbols[user_sym.name()] = user_sym;
   }
   for (auto &named_expr : body.named_expressions) {
     const auto &name = named_expr->name_;
@@ -103,35 +126,35 @@ void SymbolGenerator::VisitReturnBody(ReturnBody &body, Where *where) {
     // new symbol would have a more specific type.
     named_expr->MapTo(CreateSymbol(name, true, Symbol::Type::ANY, named_expr->token_position_));
   }
-  scope_.in_order_by = true;
+  scope.in_order_by = true;
   for (const auto &order_pair : body.order_by) {
     order_pair.expression->Accept(*this);
   }
-  scope_.in_order_by = false;
+  scope.in_order_by = false;
   if (body.skip) {
-    scope_.in_skip = true;
+    scope.in_skip = true;
     body.skip->Accept(*this);
-    scope_.in_skip = false;
+    scope.in_skip = false;
   }
   if (body.limit) {
-    scope_.in_limit = true;
+    scope.in_limit = true;
     body.limit->Accept(*this);
-    scope_.in_limit = false;
+    scope.in_limit = false;
   }
   if (where) where->Accept(*this);
   if (!removed_old_names) {
     // We have an ORDER BY or WHERE, but no aggregation, which means we didn't
     // clear the old symbols, so do it now. We cannot just call clear, because
     // we've added new symbols.
-    for (auto sym_it = scope_.symbols.begin(); sym_it != scope_.symbols.end();) {
+    for (auto sym_it = scope.symbols.begin(); sym_it != scope.symbols.end();) {
       if (new_names.find(sym_it->first) == new_names.end()) {
-        sym_it = scope_.symbols.erase(sym_it);
+        sym_it = scope.symbols.erase(sym_it);
       } else {
         sym_it++;
       }
     }
   }
-  scope_.has_aggregation = false;
+  scopes_.back().has_aggregation = false;
 }
 
 // Query
@@ -145,7 +168,7 @@ bool SymbolGenerator::PreVisit(SingleQuery &) {
 // Union
 
 bool SymbolGenerator::PreVisit(CypherUnion &) {
-  scope_ = Scope();
+  scopes_.back() = Scope();
   return true;
 }
 
@@ -166,11 +189,11 @@ bool SymbolGenerator::PostVisit(CypherUnion &cypher_union) {
 // Clauses
 
 bool SymbolGenerator::PreVisit(Create &) {
-  scope_.in_create = true;
+  scopes_.back().in_create = true;
   return true;
 }
 bool SymbolGenerator::PostVisit(Create &) {
-  scope_.in_create = false;
+  scopes_.back().in_create = false;
   return true;
 }
 
@@ -183,7 +206,7 @@ bool SymbolGenerator::PreVisit(CallProcedure &call_proc) {
 
 bool SymbolGenerator::PostVisit(CallProcedure &call_proc) {
   for (auto *ident : call_proc.result_identifiers_) {
-    if (HasSymbol(ident->name_)) {
+    if (HasSymbolLocalScope(ident->name_)) {
       throw RedeclareVariableError(ident->name_);
     }
     ident->MapTo(CreateSymbol(ident->name_, true));
@@ -194,7 +217,7 @@ bool SymbolGenerator::PostVisit(CallProcedure &call_proc) {
 bool SymbolGenerator::PreVisit(LoadCsv &load_csv) { return false; }
 
 bool SymbolGenerator::PostVisit(LoadCsv &load_csv) {
-  if (HasSymbol(load_csv.row_var_->name_)) {
+  if (HasSymbolLocalScope(load_csv.row_var_->name_)) {
     throw RedeclareVariableError(load_csv.row_var_->name_);
   }
   load_csv.row_var_->MapTo(CreateSymbol(load_csv.row_var_->name_, true));
@@ -202,45 +225,47 @@ bool SymbolGenerator::PostVisit(LoadCsv &load_csv) {
 }
 
 bool SymbolGenerator::PreVisit(Return &ret) {
-  scope_.in_return = true;
+  auto &scope = scopes_.back();
+  scope.in_return = true;
   VisitReturnBody(ret.body_);
-  scope_.in_return = false;
+  scope.in_return = false;
   return false;  // We handled the traversal ourselves.
 }
 
 bool SymbolGenerator::PostVisit(Return &) {
-  for (const auto &name_symbol : scope_.symbols) curr_return_names_.insert(name_symbol.first);
+  for (const auto &name_symbol : scopes_.back().symbols) curr_return_names_.insert(name_symbol.first);
   return true;
 }
 
 bool SymbolGenerator::PreVisit(With &with) {
-  scope_.in_with = true;
+  auto &scope = scopes_.back();
+  scope.in_with = true;
   VisitReturnBody(with.body_, with.where_);
-  scope_.in_with = false;
+  scope.in_with = false;
   return false;  // We handled the traversal ourselves.
 }
 
 bool SymbolGenerator::PreVisit(Where &) {
-  scope_.in_where = true;
+  scopes_.back().in_where = true;
   return true;
 }
 bool SymbolGenerator::PostVisit(Where &) {
-  scope_.in_where = false;
+  scopes_.back().in_where = false;
   return true;
 }
 
 bool SymbolGenerator::PreVisit(Merge &) {
-  scope_.in_merge = true;
+  scopes_.back().in_merge = true;
   return true;
 }
 bool SymbolGenerator::PostVisit(Merge &) {
-  scope_.in_merge = false;
+  scopes_.back().in_merge = false;
   return true;
 }
 
 bool SymbolGenerator::PostVisit(Unwind &unwind) {
   const auto &name = unwind.named_expression_->name_;
-  if (HasSymbol(name)) {
+  if (HasSymbolLocalScope(name)) {
     throw RedeclareVariableError(name);
   }
   unwind.named_expression_->MapTo(CreateSymbol(name, true));
@@ -248,55 +273,70 @@ bool SymbolGenerator::PostVisit(Unwind &unwind) {
 }
 
 bool SymbolGenerator::PreVisit(Match &) {
-  scope_.in_match = true;
+  scopes_.back().in_match = true;
   return true;
 }
 bool SymbolGenerator::PostVisit(Match &) {
-  scope_.in_match = false;
+  auto &scope = scopes_.back();
+  scope.in_match = false;
   // Check variables in property maps after visiting Match, so that they can
   // reference symbols out of bind order.
-  for (auto &ident : scope_.identifiers_in_match) {
-    if (!HasSymbol(ident->name_) && !ConsumePredefinedIdentifier(ident->name_))
+  for (auto &ident : scope.identifiers_in_match) {
+    if (!HasSymbolLocalScope(ident->name_) && !ConsumePredefinedIdentifier(ident->name_))
       throw UnboundVariableError(ident->name_);
-    ident->MapTo(scope_.symbols[ident->name_]);
+    ident->MapTo(scope.symbols[ident->name_]);
   }
-  scope_.identifiers_in_match.clear();
+  scope.identifiers_in_match.clear();
+  return true;
+}
+
+bool SymbolGenerator::PreVisit(Foreach &for_each) {
+  const auto &name = for_each.named_expression_->name_;
+  scopes_.emplace_back(Scope());
+  scopes_.back().in_foreach = true;
+  for_each.named_expression_->MapTo(
+      CreateSymbol(name, true, Symbol::Type::ANY, for_each.named_expression_->token_position_));
+  return true;
+}
+bool SymbolGenerator::PostVisit([[maybe_unused]] Foreach &for_each) {
+  scopes_.pop_back();
   return true;
 }
 
 // Expressions
 
 SymbolGenerator::ReturnType SymbolGenerator::Visit(Identifier &ident) {
-  if (scope_.in_skip || scope_.in_limit) {
-    throw SemanticException("Variables are not allowed in {}.", scope_.in_skip ? "SKIP" : "LIMIT");
+  auto &scope = scopes_.back();
+  if (scope.in_skip || scope.in_limit) {
+    throw SemanticException("Variables are not allowed in {}.", scope.in_skip ? "SKIP" : "LIMIT");
   }
   Symbol symbol;
-  if (scope_.in_pattern && !(scope_.in_node_atom || scope_.visiting_edge)) {
+  if (scope.in_pattern && !(scope.in_node_atom || scope.visiting_edge)) {
     // If we are in the pattern, and outside of a node or an edge, the
     // identifier is the pattern name.
-    symbol = GetOrCreateSymbol(ident.name_, ident.user_declared_, Symbol::Type::PATH);
-  } else if (scope_.in_pattern && scope_.in_pattern_atom_identifier) {
+    symbol = GetOrCreateSymbolLocalScope(ident.name_, ident.user_declared_, Symbol::Type::PATH);
+  } else if (scope.in_pattern && scope.in_pattern_atom_identifier) {
     //  Patterns used to create nodes and edges cannot redeclare already
     //  established bindings. Declaration only happens in single node
     //  patterns and in edge patterns. OpenCypher example,
     //  `MATCH (n) CREATE (n)` should throw an error that `n` is already
     //  declared. While `MATCH (n) CREATE (n) -[:R]-> (n)` is allowed,
     //  since `n` now references the bound node instead of declaring it.
-    if ((scope_.in_create_node || scope_.in_create_edge) && HasSymbol(ident.name_)) {
+    if ((scope.in_create_node || scope.in_create_edge) && HasSymbolLocalScope(ident.name_)) {
       throw RedeclareVariableError(ident.name_);
     }
     auto type = Symbol::Type::VERTEX;
-    if (scope_.visiting_edge) {
+    if (scope.visiting_edge) {
       // Edge referencing is not allowed (like in Neo4j):
       // `MATCH (n) - [r] -> (n) - [r] -> (n) RETURN r` is not allowed.
-      if (HasSymbol(ident.name_)) {
+      if (HasSymbolLocalScope(ident.name_)) {
         throw RedeclareVariableError(ident.name_);
       }
-      type = scope_.visiting_edge->IsVariable() ? Symbol::Type::EDGE_LIST : Symbol::Type::EDGE;
+      type = scope.visiting_edge->IsVariable() ? Symbol::Type::EDGE_LIST : Symbol::Type::EDGE;
     }
-    symbol = GetOrCreateSymbol(ident.name_, ident.user_declared_, type);
-  } else if (scope_.in_pattern && !scope_.in_pattern_atom_identifier && scope_.in_match) {
-    if (scope_.in_edge_range && scope_.visiting_edge->identifier_->name_ == ident.name_) {
+    symbol = GetOrCreateSymbolLocalScope(ident.name_, ident.user_declared_, type);
+  } else if (scope.in_pattern && !scope.in_pattern_atom_identifier && scope.in_match) {
+    if (scope.in_edge_range && scope.visiting_edge->identifier_->name_ == ident.name_) {
       // Prevent variable path bounds to reference the identifier which is bound
       // by the variable path itself.
       throw UnboundVariableError(ident.name_);
@@ -304,30 +344,30 @@ SymbolGenerator::ReturnType SymbolGenerator::Visit(Identifier &ident) {
     // Variables in property maps or bounds of variable length path during MATCH
     // can reference symbols bound later in the same MATCH. We collect them
     // here, so that they can be checked after visiting Match.
-    scope_.identifiers_in_match.emplace_back(&ident);
+    scope.identifiers_in_match.emplace_back(&ident);
   } else {
     // Everything else references a bound symbol.
     if (!HasSymbol(ident.name_) && !ConsumePredefinedIdentifier(ident.name_)) throw UnboundVariableError(ident.name_);
-    symbol = scope_.symbols[ident.name_];
+    symbol = GetOrCreateSymbol(ident.name_, ident.user_declared_, Symbol::Type::ANY);
   }
   ident.MapTo(symbol);
   return true;
 }
 
 bool SymbolGenerator::PreVisit(Aggregation &aggr) {
+  auto &scope = scopes_.back();
   // Check if the aggregation can be used in this context. This check should
   // probably move to a separate phase, which checks if the query is well
   // formed.
-  if ((!scope_.in_return && !scope_.in_with) || scope_.in_order_by || scope_.in_skip || scope_.in_limit ||
-      scope_.in_where) {
+  if ((!scope.in_return && !scope.in_with) || scope.in_order_by || scope.in_skip || scope.in_limit || scope.in_where) {
     throw SemanticException("Aggregation functions are only allowed in WITH and RETURN.");
   }
-  if (scope_.in_aggregation) {
+  if (scope.in_aggregation) {
     throw SemanticException(
         "Using aggregation functions inside aggregation functions is not "
         "allowed.");
   }
-  if (scope_.num_if_operators) {
+  if (scope.num_if_operators) {
     // Neo allows aggregations here and produces very interesting behaviors.
     // To simplify implementation at this moment we decided to completely
     // disallow aggregations inside of the CASE.
@@ -341,23 +381,23 @@ bool SymbolGenerator::PreVisit(Aggregation &aggr) {
   // Currently, we only have aggregation operators which return numbers.
   auto aggr_name = Aggregation::OpToString(aggr.op_) + std::to_string(aggr.symbol_pos_);
   aggr.MapTo(CreateSymbol(aggr_name, false, Symbol::Type::NUMBER));
-  scope_.in_aggregation = true;
-  scope_.has_aggregation = true;
+  scope.in_aggregation = true;
+  scope.has_aggregation = true;
   return true;
 }
 
 bool SymbolGenerator::PostVisit(Aggregation &) {
-  scope_.in_aggregation = false;
+  scopes_.back().in_aggregation = false;
   return true;
 }
 
 bool SymbolGenerator::PreVisit(IfOperator &) {
-  ++scope_.num_if_operators;
+  ++scopes_.back().num_if_operators;
   return true;
 }
 
 bool SymbolGenerator::PostVisit(IfOperator &) {
-  --scope_.num_if_operators;
+  --scopes_.back().num_if_operators;
   return true;
 }
 
@@ -401,33 +441,36 @@ bool SymbolGenerator::PreVisit(Extract &extract) {
 // Pattern and its subparts.
 
 bool SymbolGenerator::PreVisit(Pattern &pattern) {
-  scope_.in_pattern = true;
-  if ((scope_.in_create || scope_.in_merge) && pattern.atoms_.size() == 1U) {
+  auto &scope = scopes_.back();
+  scope.in_pattern = true;
+  if ((scope.in_create || scope.in_merge) && pattern.atoms_.size() == 1U) {
     MG_ASSERT(utils::IsSubtype(*pattern.atoms_[0], NodeAtom::kType), "Expected a single NodeAtom in Pattern");
-    scope_.in_create_node = true;
+    scope.in_create_node = true;
   }
   return true;
 }
 
 bool SymbolGenerator::PostVisit(Pattern &) {
-  scope_.in_pattern = false;
-  scope_.in_create_node = false;
+  auto &scope = scopes_.back();
+  scope.in_pattern = false;
+  scope.in_create_node = false;
   return true;
 }
 
 bool SymbolGenerator::PreVisit(NodeAtom &node_atom) {
-  auto check_node_semantic = [&node_atom, this](const bool props_or_labels) {
+  auto &scope = scopes_.back();
+  auto check_node_semantic = [&node_atom, &scope, this](const bool props_or_labels) {
     const auto &node_name = node_atom.identifier_->name_;
-    if ((scope_.in_create || scope_.in_merge) && props_or_labels && HasSymbol(node_name)) {
+    if ((scope.in_create || scope.in_merge) && props_or_labels && HasSymbolLocalScope(node_name)) {
       throw SemanticException("Cannot create node '" + node_name +
                               "' with labels or properties, because it is already declared.");
     }
-    scope_.in_pattern_atom_identifier = true;
+    scope.in_pattern_atom_identifier = true;
     node_atom.identifier_->Accept(*this);
-    scope_.in_pattern_atom_identifier = false;
+    scope.in_pattern_atom_identifier = false;
   };
 
-  scope_.in_node_atom = true;
+  scope.in_node_atom = true;
   if (auto *properties = std::get_if<std::unordered_map<PropertyIx, Expression *>>(&node_atom.properties_)) {
     bool props_or_labels = !properties->empty() || !node_atom.labels_.empty();
 
@@ -447,20 +490,21 @@ bool SymbolGenerator::PreVisit(NodeAtom &node_atom) {
 }
 
 bool SymbolGenerator::PostVisit(NodeAtom &) {
-  scope_.in_node_atom = false;
+  scopes_.back().in_node_atom = false;
   return true;
 }
 
 bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
-  scope_.visiting_edge = &edge_atom;
-  if (scope_.in_create || scope_.in_merge) {
-    scope_.in_create_edge = true;
+  auto &scope = scopes_.back();
+  scope.visiting_edge = &edge_atom;
+  if (scope.in_create || scope.in_merge) {
+    scope.in_create_edge = true;
     if (edge_atom.edge_types_.size() != 1U) {
       throw SemanticException(
           "A single relationship type must be specified "
           "when creating an edge.");
     }
-    if (scope_.in_create &&  // Merge allows bidirectionality
+    if (scope.in_create &&  // Merge allows bidirectionality
         edge_atom.direction_ == EdgeAtom::Direction::BOTH) {
       throw SemanticException(
           "Bidirectional relationship are not supported "
@@ -480,15 +524,15 @@ bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
     std::get<ParameterLookup *>(edge_atom.properties_)->Accept(*this);
   }
   if (edge_atom.IsVariable()) {
-    scope_.in_edge_range = true;
+    scope.in_edge_range = true;
     if (edge_atom.lower_bound_) {
       edge_atom.lower_bound_->Accept(*this);
     }
     if (edge_atom.upper_bound_) {
       edge_atom.upper_bound_->Accept(*this);
     }
-    scope_.in_edge_range = false;
-    scope_.in_pattern = false;
+    scope.in_edge_range = false;
+    scope.in_pattern = false;
     if (edge_atom.filter_lambda_.expression) {
       VisitWithIdentifiers(edge_atom.filter_lambda_.expression,
                            {edge_atom.filter_lambda_.inner_edge, edge_atom.filter_lambda_.inner_node});
@@ -505,34 +549,36 @@ bool SymbolGenerator::PreVisit(EdgeAtom &edge_atom) {
       VisitWithIdentifiers(edge_atom.weight_lambda_.expression,
                            {edge_atom.weight_lambda_.inner_edge, edge_atom.weight_lambda_.inner_node});
     }
-    scope_.in_pattern = true;
+    scope.in_pattern = true;
   }
-  scope_.in_pattern_atom_identifier = true;
+  scope.in_pattern_atom_identifier = true;
   edge_atom.identifier_->Accept(*this);
-  scope_.in_pattern_atom_identifier = false;
+  scope.in_pattern_atom_identifier = false;
   if (edge_atom.total_weight_) {
-    if (HasSymbol(edge_atom.total_weight_->name_)) {
+    if (HasSymbolLocalScope(edge_atom.total_weight_->name_)) {
       throw RedeclareVariableError(edge_atom.total_weight_->name_);
     }
-    edge_atom.total_weight_->MapTo(GetOrCreateSymbol(edge_atom.total_weight_->name_,
-                                                     edge_atom.total_weight_->user_declared_, Symbol::Type::NUMBER));
+    edge_atom.total_weight_->MapTo(GetOrCreateSymbolLocalScope(
+        edge_atom.total_weight_->name_, edge_atom.total_weight_->user_declared_, Symbol::Type::NUMBER));
   }
   return false;
 }
 
 bool SymbolGenerator::PostVisit(EdgeAtom &) {
-  scope_.visiting_edge = nullptr;
-  scope_.in_create_edge = false;
+  auto &scope = scopes_.back();
+  scope.visiting_edge = nullptr;
+  scope.in_create_edge = false;
   return true;
 }
 
 void SymbolGenerator::VisitWithIdentifiers(Expression *expr, const std::vector<Identifier *> &identifiers) {
+  auto &scope = scopes_.back();
   std::vector<std::pair<std::optional<Symbol>, Identifier *>> prev_symbols;
   // Collect previous symbols if they exist.
   for (const auto &identifier : identifiers) {
     std::optional<Symbol> prev_symbol;
-    auto prev_symbol_it = scope_.symbols.find(identifier->name_);
-    if (prev_symbol_it != scope_.symbols.end()) {
+    auto prev_symbol_it = scope.symbols.find(identifier->name_);
+    if (prev_symbol_it != scope.symbols.end()) {
       prev_symbol = prev_symbol_it->second;
     }
     identifier->MapTo(CreateSymbol(identifier->name_, identifier->user_declared_));
@@ -545,14 +591,20 @@ void SymbolGenerator::VisitWithIdentifiers(Expression *expr, const std::vector<I
     const auto &prev_symbol = prev.first;
     const auto &identifier = prev.second;
     if (prev_symbol) {
-      scope_.symbols[identifier->name_] = *prev_symbol;
+      scope.symbols[identifier->name_] = *prev_symbol;
     } else {
-      scope_.symbols.erase(identifier->name_);
+      scope.symbols.erase(identifier->name_);
     }
   }
 }
 
-bool SymbolGenerator::HasSymbol(const std::string &name) { return scope_.symbols.find(name) != scope_.symbols.end(); }
+bool SymbolGenerator::HasSymbol(const std::string &name) const {
+  return std::ranges::any_of(scopes_, [&name](const auto &scope) { return scope.symbols.contains(name); });
+}
+
+bool SymbolGenerator::HasSymbolLocalScope(const std::string &name) const {
+  return scopes_.back().symbols.contains(name);
+}
 
 bool SymbolGenerator::ConsumePredefinedIdentifier(const std::string &name) {
   auto it = predefined_identifiers_.find(name);

src/query/frontend/semantic/symbol_generator.hpp

@@ -15,6 +15,9 @@
 
 #pragma once
 
+#include <optional>
+#include <vector>
+
 #include "query/exceptions.hpp"
 #include "query/frontend/ast/ast.hpp"
 #include "query/frontend/semantic/symbol_table.hpp"
@@ -59,6 +62,8 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
   bool PostVisit(Unwind &) override;
   bool PreVisit(Match &) override;
   bool PostVisit(Match &) override;
+  bool PreVisit(Foreach &) override;
+  bool PostVisit(Foreach &) override;
 
   // Expressions
   ReturnType Visit(Identifier &) override;
@@ -107,6 +112,7 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
     bool in_order_by{false};
     bool in_where{false};
     bool in_match{false};
+    bool in_foreach{false};
     // True when visiting a pattern atom (node or edge) identifier, which can be
     // reused or created in the pattern itself.
     bool in_pattern_atom_identifier{false};
@@ -125,7 +131,10 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
     int num_if_operators{0};
   };
 
-  bool HasSymbol(const std::string &name);
+  static std::optional<Symbol> FindSymbolInScope(const std::string &name, const Scope &scope, Symbol::Type type);
+
+  bool HasSymbol(const std::string &name) const;
+  bool HasSymbolLocalScope(const std::string &name) const;
 
   // @return true if it added a predefined identifier with that name
   bool ConsumePredefinedIdentifier(const std::string &name);
@@ -135,9 +144,10 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
   auto CreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY,
                     int token_position = -1);
 
+  auto GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY);
   // Returns the symbol by name. If the mapping already exists, checks if the
   // types match. Otherwise, returns a new symbol.
-  auto GetOrCreateSymbol(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY);
+  auto GetOrCreateSymbolLocalScope(const std::string &name, bool user_declared, Symbol::Type type = Symbol::Type::ANY);
 
   void VisitReturnBody(ReturnBody &body, Where *where = nullptr);
 
@@ -148,7 +158,7 @@ class SymbolGenerator : public HierarchicalTreeVisitor {
   // Identifiers which are injected from outside the query. Each identifier
   // is mapped by its name.
   std::unordered_map<std::string, Identifier *> predefined_identifiers_;
-  Scope scope_;
+  std::vector<Scope> scopes_;
   std::unordered_set<std::string> prev_return_names_;
   std::unordered_set<std::string> curr_return_names_;
 };

src/query/frontend/stripped_lexer_constants.hpp

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

src/query/interpret/awesome_memgraph_functions.cpp

@@ -22,6 +22,9 @@
 
 #include "query/db_accessor.hpp"
 #include "query/exceptions.hpp"
+#include "query/procedure/cypher_types.hpp"
+#include "query/procedure/mg_procedure_impl.hpp"
+#include "query/procedure/module.hpp"
 #include "query/typed_value.hpp"
 #include "utils/string.hpp"
 #include "utils/temporal.hpp"
@@ -307,9 +310,9 @@ void FType(const char *name, const TypedValue *args, int64_t nargs, int64_t pos
     }
     return;
   }
-  constexpr int64_t required_args = FTypeRequiredArgs<ArgType, ArgTypes...>();
-  constexpr int64_t optional_args = FTypeOptionalArgs<ArgType, ArgTypes...>();
-  constexpr int64_t total_args = required_args + optional_args;
+  static constexpr int64_t required_args = FTypeRequiredArgs<ArgType, ArgTypes...>();
+  static constexpr int64_t optional_args = FTypeOptionalArgs<ArgType, ArgTypes...>();
+  static constexpr int64_t total_args = required_args + optional_args;
   if constexpr (optional_args > 0) {
     if (nargs < required_args || nargs > total_args) {
       throw QueryRuntimeException("'{}' requires between {} and {} arguments.", name, required_args, total_args);
@@ -810,7 +813,7 @@ TypedValue StringMatchOperator(const TypedValue *args, int64_t nargs, const Func
 
 // Check if s1 starts with s2.
 struct StartsWithPredicate {
-  constexpr static const char *name = "startsWith";
+  static constexpr const char *name = "startsWith";
   bool operator()(const TypedValue::TString &s1, const TypedValue::TString &s2) const {
     if (s1.size() < s2.size()) return false;
     return std::equal(s2.begin(), s2.end(), s1.begin());
@@ -820,7 +823,7 @@ auto StartsWith = StringMatchOperator<StartsWithPredicate>;
 
 // Check if s1 ends with s2.
 struct EndsWithPredicate {
-  constexpr static const char *name = "endsWith";
+  static constexpr const char *name = "endsWith";
   bool operator()(const TypedValue::TString &s1, const TypedValue::TString &s2) const {
     if (s1.size() < s2.size()) return false;
     return std::equal(s2.rbegin(), s2.rend(), s1.rbegin());
@@ -830,7 +833,7 @@ auto EndsWith = StringMatchOperator<EndsWithPredicate>;
 
 // Check if s1 contains s2.
 struct ContainsPredicate {
-  constexpr static const char *name = "contains";
+  static constexpr const char *name = "contains";
   bool operator()(const TypedValue::TString &s1, const TypedValue::TString &s2) const {
     if (s1.size() < s2.size()) return false;
     return s1.find(s2) != std::string::npos;
@@ -1174,6 +1177,53 @@ TypedValue Duration(const TypedValue *args, int64_t nargs, const FunctionContext
   MapNumericParameters<Number>(parameter_mappings, args[0].ValueMap());
   return TypedValue(utils::Duration(duration_parameters), ctx.memory);
 }
+
+std::function<TypedValue(const TypedValue *, const int64_t, const FunctionContext &)> UserFunction(
+    const mgp_func &func, const std::string &fully_qualified_name) {
+  return [func, fully_qualified_name](const TypedValue *args, int64_t nargs, const FunctionContext &ctx) -> TypedValue {
+    /// Find function is called to aquire the lock on Module pointer while user-defined function is executed
+    const auto &maybe_found =
+        procedure::FindFunction(procedure::gModuleRegistry, fully_qualified_name, utils::NewDeleteResource());
+    if (!maybe_found) {
+      throw QueryRuntimeException(
+          "Function '{}' has been unloaded. Please check query modules to confirm that function is loaded in Memgraph.",
+          fully_qualified_name);
+    }
+    /// Explicit extraction of module pointer, to clearly state that the lock is aquired.
+    // NOLINTNEXTLINE(clang-diagnostic-unused-variable)
+    const auto &module_ptr = (*maybe_found).first;
+
+    const auto &func_cb = func.cb;
+    mgp_memory memory{ctx.memory};
+    mgp_func_context functx{ctx.db_accessor, ctx.view};
+    auto graph = mgp_graph::NonWritableGraph(*ctx.db_accessor, ctx.view);
+
+    std::vector<TypedValue> args_list;
+    args_list.reserve(nargs);
+    for (std::size_t i = 0; i < nargs; ++i) {
+      args_list.emplace_back(args[i]);
+    }
+
+    auto function_argument_list = mgp_list(ctx.memory);
+    procedure::ConstructArguments(args_list, func, fully_qualified_name, function_argument_list, graph);
+
+    mgp_func_result maybe_res;
+    func_cb(&function_argument_list, &functx, &maybe_res, &memory);
+    if (maybe_res.error_msg) {
+      throw QueryRuntimeException(*maybe_res.error_msg);
+    }
+
+    if (!maybe_res.value) {
+      throw QueryRuntimeException(
+          "Function '{}' didn't set the result nor the error message. Please either set the result by using "
+          "mgp_func_result_set_value or the error by using mgp_func_result_set_error_msg.",
+          fully_qualified_name);
+    }
+
+    return {*(maybe_res.value), ctx.memory};
+  };
+}
+
 }  // namespace
 
 std::function<TypedValue(const TypedValue *, int64_t, const FunctionContext &ctx)> NameToFunction(
@@ -1259,6 +1309,14 @@ std::function<TypedValue(const TypedValue *, int64_t, const FunctionContext &ctx
   if (function_name == "LOCALDATETIME") return LocalDateTime;
   if (function_name == "DURATION") return Duration;
 
+  const auto &maybe_found =
+      procedure::FindFunction(procedure::gModuleRegistry, function_name, utils::NewDeleteResource());
+
+  if (maybe_found) {
+    const auto *func = (*maybe_found).second;
+    return UserFunction(*func, function_name);
+  }
+
   return nullptr;
 }
 

src/query/interpreter.cpp

@@ -554,7 +554,7 @@ std::vector<std::string> EvaluateTopicNames(ExpressionEvaluator &evaluator,
 Callback::CallbackFunction GetKafkaCreateCallback(StreamQuery *stream_query, ExpressionEvaluator &evaluator,
                                                   InterpreterContext *interpreter_context,
                                                   const std::string *username) {
-  constexpr std::string_view kDefaultConsumerGroup = "mg_consumer";
+  static constexpr std::string_view kDefaultConsumerGroup = "mg_consumer";
   std::string consumer_group{stream_query->consumer_group_.empty() ? kDefaultConsumerGroup
                                                                    : stream_query->consumer_group_};
 
@@ -899,7 +899,7 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
   // Set up temporary memory for a single Pull. Initial memory comes from the
   // stack. 256 KiB should fit on the stack and should be more than enough for a
   // single `Pull`.
-  constexpr size_t stack_size = 256 * 1024;
+  static constexpr size_t stack_size = 256UL * 1024UL;
   char stack_data[stack_size];
   utils::ResourceWithOutOfMemoryException resource_with_exception;
   utils::MonotonicBufferResource monotonic_memory(&stack_data[0], stack_size, &resource_with_exception);

src/query/interpreter.hpp

@@ -47,7 +47,7 @@ extern const Event FailedQuery;
 
 namespace memgraph::query {
 
-static constexpr size_t kExecutionMemoryBlockSize = 1U * 1024U * 1024U;
+inline constexpr size_t kExecutionMemoryBlockSize = 1UL * 1024UL * 1024UL;
 
 class AuthQueryHandler {
  public:

src/query/plan/cost_estimator.hpp

@@ -61,6 +61,7 @@ class CostEstimator : public HierarchicalLogicalOperatorVisitor {
     static constexpr double kFilter{1.5};
     static constexpr double kEdgeUniquenessFilter{1.5};
     static constexpr double kUnwind{1.3};
+    static constexpr double kForeach{1.0};
   };
 
   struct CardParam {
@@ -72,6 +73,7 @@ class CostEstimator : public HierarchicalLogicalOperatorVisitor {
 
   struct MiscParam {
     static constexpr double kUnwindNoLiteral{10.0};
+    static constexpr double kForeachNoLiteral{10.0};
   };
 
   using HierarchicalLogicalOperatorVisitor::PostVisit;
@@ -193,6 +195,23 @@ class CostEstimator : public HierarchicalLogicalOperatorVisitor {
     return true;
   }
 
+  bool PostVisit(Foreach &foreach) override {
+    // Foreach cost depends both on the number elements in the list that get unwound
+    // as well as the total clauses that get called for each unwounded element.
+    // First estimate cardinality and then increment the cost.
+
+    double foreach_elements{0};
+    if (auto *literal = utils::Downcast<query::ListLiteral>(foreach.expression_)) {
+      foreach_elements = literal->elements_.size();
+    } else {
+      foreach_elements = MiscParam::kForeachNoLiteral;
+    }
+
+    cardinality_ *= foreach_elements;
+    IncrementCost(CostParam::kForeach);
+    return true;
+  }
+
   bool Visit(Once &) override { return true; }
 
   auto cost() const { return cost_; }

src/query/plan/operator.cpp

@@ -45,6 +45,7 @@
 #include "utils/fnv.hpp"
 #include "utils/likely.hpp"
 #include "utils/logging.hpp"
+#include "utils/memory.hpp"
 #include "utils/pmr/unordered_map.hpp"
 #include "utils/pmr/unordered_set.hpp"
 #include "utils/pmr/vector.hpp"
@@ -105,6 +106,7 @@ extern const Event DistinctOperator;
 extern const Event UnionOperator;
 extern const Event CartesianOperator;
 extern const Event CallProcedureOperator;
+extern const Event ForeachOperator;
 }  // namespace EventCounter
 
 namespace memgraph::query::plan {
@@ -3703,46 +3705,12 @@ void CallCustomProcedure(const std::string_view &fully_qualified_procedure_name,
                 "containers aware of that");
   // Build and type check procedure arguments.
   mgp_list proc_args(memory);
-  proc_args.elems.reserve(args.size());
-  if (args.size() < proc.args.size() ||
-      // Rely on `||` short circuit so we can avoid potential overflow of
-      // proc.args.size() + proc.opt_args.size() by subtracting.
-      (args.size() - proc.args.size() > proc.opt_args.size())) {
-    if (proc.args.empty() && proc.opt_args.empty()) {
-      throw QueryRuntimeException("'{}' requires no arguments.", fully_qualified_procedure_name);
-    } else if (proc.opt_args.empty()) {
-      throw QueryRuntimeException("'{}' requires exactly {} {}.", fully_qualified_procedure_name, proc.args.size(),
-                                  proc.args.size() == 1U ? "argument" : "arguments");
-    } else {
-      throw QueryRuntimeException("'{}' requires between {} and {} arguments.", fully_qualified_procedure_name,
-                                  proc.args.size(), proc.args.size() + proc.opt_args.size());
-    }
-  }
-  for (size_t i = 0; i < args.size(); ++i) {
-    auto arg = args[i]->Accept(*evaluator);
-    std::string_view name;
-    const query::procedure::CypherType *type{nullptr};
-    if (proc.args.size() > i) {
-      name = proc.args[i].first;
-      type = proc.args[i].second;
-    } else {
-      MG_ASSERT(proc.opt_args.size() > i - proc.args.size());
-      name = std::get<0>(proc.opt_args[i - proc.args.size()]);
-      type = std::get<1>(proc.opt_args[i - proc.args.size()]);
-    }
-    if (!type->SatisfiesType(arg)) {
-      throw QueryRuntimeException("'{}' argument named '{}' at position {} must be of type {}.",
-                                  fully_qualified_procedure_name, name, i, type->GetPresentableName());
-    }
-    proc_args.elems.emplace_back(std::move(arg), &graph);
-  }
-  // Fill missing optional arguments with their default values.
-  MG_ASSERT(args.size() >= proc.args.size());
-  size_t passed_in_opt_args = args.size() - proc.args.size();
-  MG_ASSERT(passed_in_opt_args <= proc.opt_args.size());
-  for (size_t i = passed_in_opt_args; i < proc.opt_args.size(); ++i) {
-    proc_args.elems.emplace_back(std::get<2>(proc.opt_args[i]), &graph);
+  std::vector<TypedValue> args_list;
+  args_list.reserve(args.size());
+  for (auto *expression : args) {
+    args_list.emplace_back(expression->Accept(*evaluator));
   }
+  procedure::ConstructArguments(args_list, proc, fully_qualified_procedure_name, proc_args, graph);
   if (memory_limit) {
     SPDLOG_INFO("Running '{}' with memory limit of {}", fully_qualified_procedure_name,
                 utils::GetReadableSize(*memory_limit));
@@ -3830,7 +3798,7 @@ class CallProcedureCursor : public Cursor {
       // generator like procedures which yield a new result on each invocation.
       auto *memory = context.evaluation_context.memory;
       auto memory_limit = EvaluateMemoryLimit(&evaluator, self_->memory_limit_, self_->memory_scale_);
-      mgp_graph graph{context.db_accessor, graph_view, &context};
+      auto graph = mgp_graph::WritableGraph(*context.db_accessor, graph_view, context);
       CallCustomProcedure(self_->procedure_name_, *proc, self_->arguments_, graph, &evaluator, memory, memory_limit,
                           &result_);
 
@@ -4024,4 +3992,85 @@ UniqueCursorPtr LoadCsv::MakeCursor(utils::MemoryResource *mem) const {
   return MakeUniqueCursorPtr<LoadCsvCursor>(mem, this, mem);
 };
 
+class ForeachCursor : public Cursor {
+ public:
+  explicit ForeachCursor(const Foreach &foreach, utils::MemoryResource *mem)
+      : loop_variable_symbol_(foreach.loop_variable_symbol_),
+        input_(foreach.input_->MakeCursor(mem)),
+        updates_(foreach.update_clauses_->MakeCursor(mem)),
+        expression(foreach.expression_) {}
+
+  bool Pull(Frame &frame, ExecutionContext &context) override {
+    SCOPED_PROFILE_OP(op_name_);
+
+    if (!input_->Pull(frame, context)) {
+      return false;
+    }
+
+    ExpressionEvaluator evaluator(&frame, context.symbol_table, context.evaluation_context, context.db_accessor,
+                                  storage::View::NEW);
+    TypedValue expr_result = expression->Accept(evaluator);
+
+    if (expr_result.IsNull()) {
+      return true;
+    }
+
+    if (!expr_result.IsList()) {
+      throw QueryRuntimeException("FOREACH expression must resolve to a list, but got '{}'.", expr_result.type());
+    }
+
+    const auto &cache_ = expr_result.ValueList();
+    for (const auto &index : cache_) {
+      frame[loop_variable_symbol_] = index;
+      while (updates_->Pull(frame, context)) {
+      }
+      ResetUpdates();
+    }
+
+    return true;
+  }
+
+  void Shutdown() override { input_->Shutdown(); }
+
+  void ResetUpdates() { updates_->Reset(); }
+
+  void Reset() override {
+    input_->Reset();
+    ResetUpdates();
+  }
+
+ private:
+  const Symbol loop_variable_symbol_;
+  const UniqueCursorPtr input_;
+  const UniqueCursorPtr updates_;
+  Expression *expression;
+  const char *op_name_{"Foreach"};
+};
+
+Foreach::Foreach(std::shared_ptr<LogicalOperator> input, std::shared_ptr<LogicalOperator> updates, Expression *expr,
+                 Symbol loop_variable_symbol)
+    : input_(input ? std::move(input) : std::make_shared<Once>()),
+      update_clauses_(std::move(updates)),
+      expression_(expr),
+      loop_variable_symbol_(loop_variable_symbol) {}
+
+UniqueCursorPtr Foreach::MakeCursor(utils::MemoryResource *mem) const {
+  EventCounter::IncrementCounter(EventCounter::ForeachOperator);
+  return MakeUniqueCursorPtr<ForeachCursor>(mem, *this, mem);
+}
+
+std::vector<Symbol> Foreach::ModifiedSymbols(const SymbolTable &table) const {
+  auto symbols = input_->ModifiedSymbols(table);
+  symbols.emplace_back(loop_variable_symbol_);
+  return symbols;
+}
+
+bool Foreach::Accept(HierarchicalLogicalOperatorVisitor &visitor) {
+  if (visitor.PreVisit(*this)) {
+    input_->Accept(visitor);
+    update_clauses_->Accept(visitor);
+  }
+  return visitor.PostVisit(*this);
+}
+
 }  // namespace memgraph::query::plan

src/query/plan/operator.lcp

@@ -131,6 +131,7 @@ class Union;
 class Cartesian;
 class CallProcedure;
 class LoadCsv;
+class Foreach;
 
 using LogicalOperatorCompositeVisitor = utils::CompositeVisitor<
     Once, CreateNode, CreateExpand, ScanAll, ScanAllByLabel,
@@ -139,7 +140,7 @@ using LogicalOperatorCompositeVisitor = utils::CompositeVisitor<
     Expand, ExpandVariable, ConstructNamedPath, Filter, Produce, Delete,
     SetProperty, SetProperties, SetLabels, RemoveProperty, RemoveLabels,
     EdgeUniquenessFilter, Accumulate, Aggregate, Skip, Limit, OrderBy, Merge,
-    Optional, Unwind, Distinct, Union, Cartesian, CallProcedure, LoadCsv>;
+    Optional, Unwind, Distinct, Union, Cartesian, CallProcedure, LoadCsv, Foreach>;
 
 using LogicalOperatorLeafVisitor = utils::LeafVisitor<Once>;
 
@@ -2261,6 +2262,42 @@ at once. Instead, each call of the callback should return a single row of the ta
   (:serialize (:slk))
   (:clone))
 
+(lcp:define-class foreach (logical-operator)
+  ((input "std::shared_ptr<LogicalOperator>" :scope :public
+          :slk-save #'slk-save-operator-pointer
+          :slk-load #'slk-load-operator-pointer)
+   (update-clauses "std::shared_ptr<LogicalOperator>" :scope :public
+          :slk-save #'slk-save-operator-pointer
+          :slk-load #'slk-load-operator-pointer)
+   (expression "Expression *" :scope :public
+           :slk-save #'slk-save-ast-pointer
+           :slk-load (slk-load-ast-pointer "Expression"))
+   (loop-variable-symbol "Symbol" :scope :public))
+
+  (:documentation
+   "Iterates over a collection of elements and applies one or more update
+clauses.
+")
+  (:public
+   #>cpp
+   Foreach() = default;
+   Foreach(std::shared_ptr<LogicalOperator> input, 
+           std::shared_ptr<LogicalOperator> updates,
+           Expression *named_expr, 
+           Symbol loop_variable_symbol);
+
+   bool Accept(HierarchicalLogicalOperatorVisitor &visitor) override;
+   UniqueCursorPtr MakeCursor(utils::MemoryResource *) const override;
+   std::vector<Symbol> ModifiedSymbols(const SymbolTable &) const override;
+   bool HasSingleInput() const override { return true; }
+   std::shared_ptr<LogicalOperator> input() const override { return input_; }
+   void set_input(std::shared_ptr<LogicalOperator> input) override {
+     input_ = std::move(input);
+   }
+   cpp<#)
+  (:serialize (:slk))
+  (:clone))
+
 (lcp:pop-namespace) ;; plan
 (lcp:pop-namespace) ;; query
 (lcp:pop-namespace) ;; memgraph

src/query/plan/preprocess.cpp

@@ -17,8 +17,10 @@
 #include <variant>
 
 #include "query/exceptions.hpp"
+#include "query/frontend/ast/ast.hpp"
 #include "query/frontend/ast/ast_visitor.hpp"
 #include "query/plan/preprocess.hpp"
+#include "utils/typeinfo.hpp"
 
 namespace memgraph::query::plan {
 
@@ -526,6 +528,18 @@ void Filters::AnalyzeAndStoreFilter(Expression *expr, const SymbolTable &symbol_
   // as `expr1 < n.prop AND n.prop < expr2`.
 }
 
+static void ParseForeach(query::Foreach &foreach, SingleQueryPart &query_part, AstStorage &storage,
+                         SymbolTable &symbol_table) {
+  for (auto *clause : foreach.clauses_) {
+    if (auto *merge = utils::Downcast<query::Merge>(clause)) {
+      query_part.merge_matching.emplace_back(Matching{});
+      AddMatching({merge->pattern_}, nullptr, symbol_table, storage, query_part.merge_matching.back());
+    } else if (auto *nested = utils::Downcast<query::Foreach>(clause)) {
+      ParseForeach(*nested, query_part, storage, symbol_table);
+    }
+  }
+}
+
 // Converts a Query to multiple QueryParts. In the process new Ast nodes may be
 // created, e.g. filter expressions.
 std::vector<SingleQueryPart> CollectSingleQueryParts(SymbolTable &symbol_table, AstStorage &storage,
@@ -546,6 +560,8 @@ std::vector<SingleQueryPart> CollectSingleQueryParts(SymbolTable &symbol_table,
       if (auto *merge = utils::Downcast<query::Merge>(clause)) {
         query_part->merge_matching.emplace_back(Matching{});
         AddMatching({merge->pattern_}, nullptr, symbol_table, storage, query_part->merge_matching.back());
+      } else if (auto *foreach = utils::Downcast<query::Foreach>(clause)) {
+        ParseForeach(*foreach, *query_part, storage, symbol_table);
       } else if (utils::IsSubtype(*clause, With::kType) || utils::IsSubtype(*clause, query::Unwind::kType) ||
                  utils::IsSubtype(*clause, query::CallProcedure::kType) ||
                  utils::IsSubtype(*clause, query::LoadCsv::kType)) {

src/query/plan/preprocess.hpp

@@ -306,6 +306,10 @@ struct Matching {
 /// will produce the second `merge_matching` element. This way, if someone
 /// traverses `remaining_clauses`, the order of appearance of `Merge` clauses is
 /// in the same order as their respective `merge_matching` elements.
+/// An exception to the above rule is Foreach. Its update clauses will not be contained in
+/// the `remaining_clauses`, but rather inside the foreach itself. The order guarantee is not
+/// violated because the update clauses of the foreach are immediately processed in
+/// the `RuleBasedPlanner` as if as they were pushed into the `remaining_clauses`.
 struct SingleQueryPart {
   /// @brief All `MATCH` clauses merged into one @c Matching.
   Matching matching;
@@ -320,6 +324,10 @@ struct SingleQueryPart {
   ///
   /// Since @c Merge is contained in `remaining_clauses`, this vector contains
   /// matching in the same order as @c Merge appears.
+  //
+  /// Foreach @c does not violate this gurantee. However, update clauses are not stored
+  /// in the `remaining_clauses` but rather in the `Foreach` itself and are guranteed
+  /// to be processed in the same order by the semantics of the `RuleBasedPlanner`.
   std::vector<Matching> merge_matching{};
   /// @brief All the remaining clauses (without @c Match).
   std::vector<Clause *> remaining_clauses{};

src/query/plan/pretty_print.cpp

@@ -241,6 +241,12 @@ bool PlanPrinter::PreVisit(query::plan::Cartesian &op) {
   return false;
 }
 
+bool PlanPrinter::PreVisit(query::plan::Foreach &op) {
+  WithPrintLn([](auto &out) { out << "* Foreach"; });
+  Branch(*op.update_clauses_);
+  op.input_->Accept(*this);
+  return false;
+}
 #undef PRE_VISIT
 
 bool PlanPrinter::DefaultPreVisit() {
@@ -883,6 +889,21 @@ bool PlanToJsonVisitor::PreVisit(Cartesian &op) {
   output_ = std::move(self);
   return false;
 }
+bool PlanToJsonVisitor::PreVisit(Foreach &op) {
+  json self;
+  self["name"] = "Foreach";
+  self["loop_variable_symbol"] = ToJson(op.loop_variable_symbol_);
+  self["expression"] = ToJson(op.expression_);
+
+  op.input_->Accept(*this);
+  self["input"] = PopOutput();
+
+  op.update_clauses_->Accept(*this);
+  self["update_clauses"] = PopOutput();
+
+  output_ = std::move(self);
+  return false;
+}
 
 }  // namespace impl
 

src/query/plan/pretty_print.hpp

@@ -92,6 +92,7 @@ class PlanPrinter : public virtual HierarchicalLogicalOperatorVisitor {
   bool PreVisit(Unwind &) override;
   bool PreVisit(CallProcedure &) override;
   bool PreVisit(LoadCsv &) override;
+  bool PreVisit(Foreach &) override;
 
   bool Visit(Once &) override;
 
@@ -204,6 +205,7 @@ class PlanToJsonVisitor : public virtual HierarchicalLogicalOperatorVisitor {
   bool PreVisit(Union &) override;
 
   bool PreVisit(Unwind &) override;
+  bool PreVisit(Foreach &) override;
   bool PreVisit(CallProcedure &) override;
   bool PreVisit(LoadCsv &) override;
 

src/query/plan/read_write_type_checker.cpp

@@ -79,6 +79,11 @@ bool ReadWriteTypeChecker::PreVisit(CallProcedure &op) {
   return true;
 }
 
+bool ReadWriteTypeChecker::PreVisit([[maybe_unused]] Foreach &op) {
+  UpdateType(RWType::RW);
+  return false;
+}
+
 #undef PRE_VISIT
 
 bool ReadWriteTypeChecker::Visit(Once &op) { return false; }

src/query/plan/read_write_type_checker.hpp

@@ -84,6 +84,7 @@ class ReadWriteTypeChecker : public virtual HierarchicalLogicalOperatorVisitor {
 
   bool PreVisit(Unwind &) override;
   bool PreVisit(CallProcedure &) override;
+  bool PreVisit(Foreach &) override;
 
   bool Visit(Once &) override;
 

src/query/plan/rewrite/index_lookup.hpp

@@ -433,6 +433,16 @@ class IndexLookupRewriter final : public HierarchicalLogicalOperatorVisitor {
     return true;
   }
 
+  bool PreVisit(Foreach &op) override {
+    prev_ops_.push_back(&op);
+    return false;
+  }
+
+  bool PostVisit(Foreach &) override {
+    prev_ops_.pop_back();
+    return true;
+  }
+
   std::shared_ptr<LogicalOperator> new_root_;
 
  private:

src/query/plan/rule_based_planner.hpp

@@ -22,6 +22,7 @@
 #include "query/plan/operator.hpp"
 #include "query/plan/preprocess.hpp"
 #include "utils/logging.hpp"
+#include "utils/typeinfo.hpp"
 
 namespace memgraph::query::plan {
 
@@ -223,6 +224,10 @@ class RuleBasedPlanner {
           input_op =
               std::make_unique<plan::LoadCsv>(std::move(input_op), load_csv->file_, load_csv->with_header_,
                                               load_csv->ignore_bad_, load_csv->delimiter_, load_csv->quote_, row_sym);
+        } else if (auto *foreach = utils::Downcast<query::Foreach>(clause)) {
+          is_write = true;
+          input_op = HandleForeachClause(foreach, std::move(input_op), *context.symbol_table, context.bound_symbols,
+                                         query_part, merge_id);
         } else {
           throw utils::NotYetImplemented("clause '{}' conversion to operator(s)", clause->GetTypeInfo().name);
         }
@@ -530,6 +535,27 @@ class RuleBasedPlanner {
     }
     return std::make_unique<plan::Merge>(std::move(input_op), std::move(on_match), std::move(on_create));
   }
+
+  std::unique_ptr<LogicalOperator> HandleForeachClause(query::Foreach *foreach,
+                                                       std::unique_ptr<LogicalOperator> input_op,
+                                                       const SymbolTable &symbol_table,
+                                                       std::unordered_set<Symbol> &bound_symbols,
+                                                       const SingleQueryPart &query_part, uint64_t &merge_id) {
+    const auto &symbol = symbol_table.at(*foreach->named_expression_);
+    bound_symbols.insert(symbol);
+    std::unique_ptr<LogicalOperator> op = std::make_unique<plan::Once>();
+    for (auto *clause : foreach->clauses_) {
+      if (auto *nested_for_each = utils::Downcast<query::Foreach>(clause)) {
+        op = HandleForeachClause(nested_for_each, std::move(op), symbol_table, bound_symbols, query_part, merge_id);
+      } else if (auto *merge = utils::Downcast<query::Merge>(clause)) {
+        op = GenMerge(*merge, std::move(op), query_part.merge_matching[merge_id++]);
+      } else {
+        op = HandleWriteClause(clause, op, symbol_table, bound_symbols);
+      }
+    }
+    return std::make_unique<plan::Foreach>(std::move(input_op), std::move(op), foreach->named_expression_->expression_,
+                                           symbol);
+  }
 };
 
 }  // namespace memgraph::query::plan

src/query/plan/scoped_profile.hpp

@@ -71,9 +71,9 @@ class ScopedProfile {
 
  private:
   query::ExecutionContext *context_;
-  ProfilingStats *root_;
-  ProfilingStats *stats_;
-  unsigned long long start_time_;
+  ProfilingStats *root_{nullptr};
+  ProfilingStats *stats_{nullptr};
+  unsigned long long start_time_{0};
 };
 
 }  // namespace memgraph::query::plan

src/query/procedure/mg_procedure_impl.cpp

@@ -187,7 +187,10 @@ template <typename TFunc, typename... Args>
   return MGP_ERROR_NO_ERROR;
 }
 
-bool MgpGraphIsMutable(const mgp_graph &graph) noexcept { return graph.view == memgraph::storage::View::NEW; }
+// Graph mutations
+bool MgpGraphIsMutable(const mgp_graph &graph) noexcept {
+  return graph.view == memgraph::storage::View::NEW && graph.ctx != nullptr;
+}
 
 bool MgpVertexIsMutable(const mgp_vertex &vertex) { return MgpGraphIsMutable(*vertex.graph); }
 
@@ -289,6 +292,7 @@ mgp_value_type FromTypedValueType(memgraph::query::TypedValue::Type type) {
       return MGP_VALUE_TYPE_DURATION;
   }
 }
+}  // namespace
 
 memgraph::query::TypedValue ToTypedValue(const mgp_value &val, memgraph::utils::MemoryResource *memory) {
   switch (val.type) {
@@ -345,8 +349,6 @@ memgraph::query::TypedValue ToTypedValue(const mgp_value &val, memgraph::utils::
   }
 }
 
-}  // namespace
-
 mgp_value::mgp_value(memgraph::utils::MemoryResource *m) noexcept : type(MGP_VALUE_TYPE_NULL), memory(m) {}
 
 mgp_value::mgp_value(bool val, memgraph::utils::MemoryResource *m) noexcept
@@ -1451,6 +1453,14 @@ mgp_error mgp_result_record_insert(mgp_result_record *record, const char *field_
   });
 }
 
+mgp_error mgp_func_result_set_error_msg(mgp_func_result *res, const char *msg, mgp_memory *memory) {
+  return WrapExceptions([=] { res->error_msg.emplace(msg, memory->impl); });
+}
+
+mgp_error mgp_func_result_set_value(mgp_func_result *res, mgp_value *value, mgp_memory *memory) {
+  return WrapExceptions([=] { res->value = ToTypedValue(*value, memory->impl); });
+}
+
 /// Graph Constructs
 
 void mgp_properties_iterator_destroy(mgp_properties_iterator *it) { DeleteRawMgpObject(it); }
@@ -2382,31 +2392,53 @@ mgp_error mgp_module_add_write_procedure(mgp_module *module, const char *name, m
   return WrapExceptions([=] { return mgp_module_add_procedure(module, name, cb, {.is_write = true}); }, result);
 }
 
-mgp_error mgp_proc_add_arg(mgp_proc *proc, const char *name, mgp_type *type) {
-  return WrapExceptions([=] {
-    if (!IsValidIdentifierName(name)) {
-      throw std::invalid_argument{fmt::format("Invalid argument name for procedure '{}': {}", proc->name, name)};
+namespace {
+template <typename T>
+concept IsCallable = memgraph::utils::SameAsAnyOf<T, mgp_proc, mgp_func>;
+
+template <IsCallable TCall>
+mgp_error MgpAddArg(TCall &callable, const std::string &name, mgp_type &type) {
+  return WrapExceptions([&]() mutable {
+    static constexpr std::string_view type_name = std::invoke([]() constexpr {
+      if constexpr (std::is_same_v<TCall, mgp_proc>) {
+        return "procedure";
+      } else if constexpr (std::is_same_v<TCall, mgp_func>) {
+        return "function";
+      }
+    });
+
+    if (!IsValidIdentifierName(name.c_str())) {
+      throw std::invalid_argument{fmt::format("Invalid argument name for {} '{}': {}", type_name, callable.name, name)};
     }
-    if (!proc->opt_args.empty()) {
-      throw std::logic_error{fmt::format(
-          "Cannot add required argument '{}' to procedure '{}' after adding any optional one", name, proc->name)};
+    if (!callable.opt_args.empty()) {
+      throw std::logic_error{fmt::format("Cannot add required argument '{}' to {} '{}' after adding any optional one",
+                                         name, type_name, callable.name)};
     }
-    proc->args.emplace_back(name, type->impl.get());
+    callable.args.emplace_back(name, type.impl.get());
   });
 }
 
-mgp_error mgp_proc_add_opt_arg(mgp_proc *proc, const char *name, mgp_type *type, mgp_value *default_value) {
-  return WrapExceptions([=] {
-    if (!IsValidIdentifierName(name)) {
-      throw std::invalid_argument{fmt::format("Invalid argument name for procedure '{}': {}", proc->name, name)};
+template <IsCallable TCall>
+mgp_error MgpAddOptArg(TCall &callable, const std::string name, mgp_type &type, mgp_value &default_value) {
+  return WrapExceptions([&]() mutable {
+    static constexpr std::string_view type_name = std::invoke([]() constexpr {
+      if constexpr (std::is_same_v<TCall, mgp_proc>) {
+        return "procedure";
+      } else if constexpr (std::is_same_v<TCall, mgp_func>) {
+        return "function";
+      }
+    });
+
+    if (!IsValidIdentifierName(name.c_str())) {
+      throw std::invalid_argument{fmt::format("Invalid argument name for {} '{}': {}", type_name, callable.name, name)};
     }
-    switch (MgpValueGetType(*default_value)) {
+    switch (MgpValueGetType(default_value)) {
       case MGP_VALUE_TYPE_VERTEX:
       case MGP_VALUE_TYPE_EDGE:
       case MGP_VALUE_TYPE_PATH:
         // default_value must not be a graph element.
-        throw ValueConversionException{
-            "Default value of argument '{}' of procedure '{}' name must not be a graph element!", name, proc->name};
+        throw ValueConversionException{"Default value of argument '{}' of {} '{}' name must not be a graph element!",
+                                       name, type_name, callable.name};
       case MGP_VALUE_TYPE_NULL:
       case MGP_VALUE_TYPE_BOOL:
       case MGP_VALUE_TYPE_INT:
@@ -2421,16 +2453,32 @@ mgp_error mgp_proc_add_opt_arg(mgp_proc *proc, const char *name, mgp_type *type,
         break;
     }
     // Default value must be of required `type`.
-    if (!type->impl->SatisfiesType(*default_value)) {
-      throw std::logic_error{
-          fmt::format("The default value of argument '{}' for procedure '{}' doesn't satisfy type '{}'", name,
-                      proc->name, type->impl->GetPresentableName())};
+    if (!type.impl->SatisfiesType(default_value)) {
+      throw std::logic_error{fmt::format("The default value of argument '{}' for {} '{}' doesn't satisfy type '{}'",
+                                         name, type_name, callable.name, type.impl->GetPresentableName())};
     }
-    auto *memory = proc->opt_args.get_allocator().GetMemoryResource();
-    proc->opt_args.emplace_back(memgraph::utils::pmr::string(name, memory), type->impl.get(),
-                                ToTypedValue(*default_value, memory));
+    auto *memory = callable.opt_args.get_allocator().GetMemoryResource();
+    callable.opt_args.emplace_back(memgraph::utils::pmr::string(name, memory), type.impl.get(),
+                                   ToTypedValue(default_value, memory));
   });
 }
+}  // namespace
+
+mgp_error mgp_proc_add_arg(mgp_proc *proc, const char *name, mgp_type *type) {
+  return MgpAddArg(*proc, std::string(name), *type);
+}
+
+mgp_error mgp_proc_add_opt_arg(mgp_proc *proc, const char *name, mgp_type *type, mgp_value *default_value) {
+  return MgpAddOptArg(*proc, std::string(name), *type, *default_value);
+}
+
+mgp_error mgp_func_add_arg(mgp_func *func, const char *name, mgp_type *type) {
+  return MgpAddArg(*func, std::string(name), *type);
+}
+
+mgp_error mgp_func_add_opt_arg(mgp_func *func, const char *name, mgp_type *type, mgp_value *default_value) {
+  return MgpAddOptArg(*func, std::string(name), *type, *default_value);
+}
 
 namespace {
 
@@ -2545,6 +2593,22 @@ void PrintProcSignature(const mgp_proc &proc, std::ostream *stream) {
   (*stream) << ")";
 }
 
+void PrintFuncSignature(const mgp_func &func, std::ostream &stream) {
+  stream << func.name << "(";
+  utils::PrintIterable(stream, func.args, ", ", [](auto &stream, const auto &arg) {
+    stream << arg.first << " :: " << arg.second->GetPresentableName();
+  });
+  if (!func.args.empty() && !func.opt_args.empty()) {
+    stream << ", ";
+  }
+  utils::PrintIterable(stream, func.opt_args, ", ", [](auto &stream, const auto &arg) {
+    const auto &[name, type, default_val] = arg;
+    stream << name << " = ";
+    PrintValue(default_val, &stream) << " :: " << type->GetPresentableName();
+  });
+  stream << ")";
+}
+
 bool IsValidIdentifierName(const char *name) {
   if (!name) return false;
   std::regex regex("[_[:alpha:]][_[:alnum:]]*");
@@ -2716,3 +2780,19 @@ mgp_error mgp_module_add_transformation(mgp_module *module, const char *name, mg
     module->transformations.emplace(name, mgp_trans(name, cb, memory));
   });
 }
+
+mgp_error mgp_module_add_function(mgp_module *module, const char *name, mgp_func_cb cb, mgp_func **result) {
+  return WrapExceptions(
+      [=] {
+        if (!IsValidIdentifierName(name)) {
+          throw std::invalid_argument{fmt::format("Invalid function name: {}", name)};
+        }
+        if (module->functions.find(name) != module->functions.end()) {
+          throw std::logic_error{fmt::format("Function with similar name already exists '{}'", name)};
+        };
+        auto *memory = module->functions.get_allocator().GetMemoryResource();
+
+        return &module->functions.emplace(name, mgp_func(name, cb, memory)).first->second;
+      },
+      result);
+}

src/query/procedure/mg_procedure_impl.hpp

@@ -562,14 +562,36 @@ struct mgp_result {
   std::optional<memgraph::utils::pmr::string> error_msg;
 };
 
+struct mgp_func_result {
+  mgp_func_result() {}
+  /// Return Magic function result. If user forgets it, the error is raised
+  std::optional<memgraph::query::TypedValue> value;
+  /// Return Magic function result with potential error
+  std::optional<memgraph::utils::pmr::string> error_msg;
+};
+
 struct mgp_graph {
   memgraph::query::DbAccessor *impl;
   memgraph::storage::View view;
   // TODO: Merge `mgp_graph` and `mgp_memory` into a single `mgp_context`. The
   // `ctx` field is out of place here.
   memgraph::query::ExecutionContext *ctx;
+
+  static mgp_graph WritableGraph(memgraph::query::DbAccessor &acc, memgraph::storage::View view,
+                                 memgraph::query::ExecutionContext &ctx) {
+    return mgp_graph{&acc, view, &ctx};
+  }
+
+  static mgp_graph NonWritableGraph(memgraph::query::DbAccessor &acc, memgraph::storage::View view) {
+    return mgp_graph{&acc, view, nullptr};
+  }
 };
 
+// Prevents user to use ExecutionContext in writable callables
+struct mgp_func_context {
+  memgraph::query::DbAccessor *impl;
+  memgraph::storage::View view;
+};
 struct mgp_properties_iterator {
   using allocator_type = memgraph::utils::Allocator<mgp_properties_iterator>;
 
@@ -779,18 +801,69 @@ struct mgp_trans {
       results;
 };
 
+struct mgp_func {
+  using allocator_type = memgraph::utils::Allocator<mgp_func>;
+
+  /// @throw std::bad_alloc
+  /// @throw std::length_error
+  mgp_func(const char *name, mgp_func_cb cb, memgraph::utils::MemoryResource *memory)
+      : name(name, memory), cb(cb), args(memory), opt_args(memory) {}
+
+  /// @throw std::bad_alloc
+  /// @throw std::length_error
+  mgp_func(const char *name, std::function<void(mgp_list *, mgp_func_context *, mgp_func_result *, mgp_memory *)> cb,
+           memgraph::utils::MemoryResource *memory)
+      : name(name, memory), cb(cb), args(memory), opt_args(memory) {}
+
+  /// @throw std::bad_alloc
+  /// @throw std::length_error
+  mgp_func(const mgp_func &other, memgraph::utils::MemoryResource *memory)
+      : name(other.name, memory), cb(other.cb), args(other.args, memory), opt_args(other.opt_args, memory) {}
+
+  mgp_func(mgp_func &&other, memgraph::utils::MemoryResource *memory)
+      : name(std::move(other.name), memory),
+        cb(std::move(other.cb)),
+        args(std::move(other.args), memory),
+        opt_args(std::move(other.opt_args), memory) {}
+
+  mgp_func(const mgp_func &other) = default;
+  mgp_func(mgp_func &&other) = default;
+
+  mgp_func &operator=(const mgp_func &) = delete;
+  mgp_func &operator=(mgp_func &&) = delete;
+
+  ~mgp_func() = default;
+
+  /// Name of the function.
+  memgraph::utils::pmr::string name;
+  /// Entry-point for the function.
+  std::function<void(mgp_list *, mgp_func_context *, mgp_func_result *, mgp_memory *)> cb;
+  /// Required, positional arguments as a (name, type) pair.
+  memgraph::utils::pmr::vector<std::pair<memgraph::utils::pmr::string, const memgraph::query::procedure::CypherType *>>
+      args;
+  /// Optional positional arguments as a (name, type, default_value) tuple.
+  memgraph::utils::pmr::vector<std::tuple<memgraph::utils::pmr::string, const memgraph::query::procedure::CypherType *,
+                                          memgraph::query::TypedValue>>
+      opt_args;
+};
+
 mgp_error MgpTransAddFixedResult(mgp_trans *trans) noexcept;
 
 struct mgp_module {
   using allocator_type = memgraph::utils::Allocator<mgp_module>;
 
-  explicit mgp_module(memgraph::utils::MemoryResource *memory) : procedures(memory), transformations(memory) {}
+  explicit mgp_module(memgraph::utils::MemoryResource *memory)
+      : procedures(memory), transformations(memory), functions(memory) {}
 
   mgp_module(const mgp_module &other, memgraph::utils::MemoryResource *memory)
-      : procedures(other.procedures, memory), transformations(other.transformations, memory) {}
+      : procedures(other.procedures, memory),
+        transformations(other.transformations, memory),
+        functions(other.functions, memory) {}
 
   mgp_module(mgp_module &&other, memgraph::utils::MemoryResource *memory)
-      : procedures(std::move(other.procedures), memory), transformations(std::move(other.transformations), memory) {}
+      : procedures(std::move(other.procedures), memory),
+        transformations(std::move(other.transformations), memory),
+        functions(std::move(other.functions), memory) {}
 
   mgp_module(const mgp_module &) = default;
   mgp_module(mgp_module &&) = default;
@@ -802,6 +875,7 @@ struct mgp_module {
 
   memgraph::utils::pmr::map<memgraph::utils::pmr::string, mgp_proc> procedures;
   memgraph::utils::pmr::map<memgraph::utils::pmr::string, mgp_trans> transformations;
+  memgraph::utils::pmr::map<memgraph::utils::pmr::string, mgp_func> functions;
 };
 
 namespace memgraph::query::procedure {
@@ -811,6 +885,11 @@ namespace memgraph::query::procedure {
 /// @throw anything std::ostream::operator<< may throw.
 void PrintProcSignature(const mgp_proc &, std::ostream *);
 
+/// @throw std::bad_alloc
+/// @throw std::length_error
+/// @throw anything std::ostream::operator<< may throw.
+void PrintFuncSignature(const mgp_func &, std::ostream &);
+
 bool IsValidIdentifierName(const char *name);
 
 }  // namespace memgraph::query::procedure
@@ -839,3 +918,5 @@ struct mgp_messages {
 
   storage_type messages;
 };
+
+memgraph::query::TypedValue ToTypedValue(const mgp_value &val, memgraph::utils::MemoryResource *memory);

src/query/procedure/module.cpp

@@ -52,6 +52,8 @@ class BuiltinModule final : public Module {
 
   const std::map<std::string, mgp_trans, std::less<>> *Transformations() const override;
 
+  const std::map<std::string, mgp_func, std::less<>> *Functions() const override;
+
   void AddProcedure(std::string_view name, mgp_proc proc);
 
   void AddTransformation(std::string_view name, mgp_trans trans);
@@ -62,6 +64,7 @@ class BuiltinModule final : public Module {
   /// Registered procedures
   std::map<std::string, mgp_proc, std::less<>> procedures_;
   std::map<std::string, mgp_trans, std::less<>> transformations_;
+  std::map<std::string, mgp_func, std::less<>> functions_;
 };
 
 BuiltinModule::BuiltinModule() {}
@@ -75,6 +78,7 @@ const std::map<std::string, mgp_proc, std::less<>> *BuiltinModule::Procedures()
 const std::map<std::string, mgp_trans, std::less<>> *BuiltinModule::Transformations() const {
   return &transformations_;
 }
+const std::map<std::string, mgp_func, std::less<>> *BuiltinModule::Functions() const { return &functions_; }
 
 void BuiltinModule::AddProcedure(std::string_view name, mgp_proc proc) { procedures_.emplace(name, std::move(proc)); }
 
@@ -300,9 +304,85 @@ void RegisterMgTransformations(const std::map<std::string, std::unique_ptr<Modul
   module->AddProcedure("transformations", std::move(procedures));
 }
 
+void RegisterMgFunctions(
+    // We expect modules to be sorted by name.
+    const std::map<std::string, std::unique_ptr<Module>, std::less<>> *all_modules, BuiltinModule *module) {
+  auto functions_cb = [all_modules](mgp_list * /*args*/, mgp_graph * /*graph*/, mgp_result *result,
+                                    mgp_memory *memory) {
+    // Iterating over all_modules assumes that the standard mechanism of magic
+    // functions invocations takes the ModuleRegistry::lock_ with READ access.
+    for (const auto &[module_name, module] : *all_modules) {
+      // Return the results in sorted order by module and by function_name.
+      static_assert(std::is_same_v<decltype(module->Functions()), const std::map<std::string, mgp_func, std::less<>> *>,
+                    "Expected module magic functions to be sorted by name");
+
+      const auto path = module->Path();
+      const auto path_string = GetPathString(path);
+      const auto is_editable = IsFileEditable(path);
+
+      for (const auto &[func_name, func] : *module->Functions()) {
+        mgp_result_record *record{nullptr};
+
+        if (!TryOrSetError([&] { return mgp_result_new_record(result, &record); }, result)) {
+          return;
+        }
+
+        const auto path_value = GetStringValueOrSetError(path_string.c_str(), memory, result);
+        if (!path_value) {
+          return;
+        }
+
+        MgpUniquePtr<mgp_value> is_editable_value{nullptr, mgp_value_destroy};
+        if (!TryOrSetError([&] { return CreateMgpObject(is_editable_value, mgp_value_make_bool, is_editable, memory); },
+                           result)) {
+          return;
+        }
+
+        utils::pmr::string full_name(module_name, memory->impl);
+        full_name.append(1, '.');
+        full_name.append(func_name);
+        const auto name_value = GetStringValueOrSetError(full_name.c_str(), memory, result);
+        if (!name_value) {
+          return;
+        }
+
+        std::stringstream ss;
+        ss << module_name << ".";
+        PrintFuncSignature(func, ss);
+        const auto signature = ss.str();
+        const auto signature_value = GetStringValueOrSetError(signature.c_str(), memory, result);
+        if (!signature_value) {
+          return;
+        }
+
+        if (!InsertResultOrSetError(result, record, "name", name_value.get())) {
+          return;
+        }
+
+        if (!InsertResultOrSetError(result, record, "signature", signature_value.get())) {
+          return;
+        }
+
+        if (!InsertResultOrSetError(result, record, "path", path_value.get())) {
+          return;
+        }
+
+        if (!InsertResultOrSetError(result, record, "is_editable", is_editable_value.get())) {
+          return;
+        }
+      }
+    }
+  };
+  mgp_proc functions("functions", functions_cb, utils::NewDeleteResource());
+  MG_ASSERT(mgp_proc_add_result(&functions, "name", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
+  MG_ASSERT(mgp_proc_add_result(&functions, "signature", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
+  MG_ASSERT(mgp_proc_add_result(&functions, "path", Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
+  MG_ASSERT(mgp_proc_add_result(&functions, "is_editable", Call<mgp_type *>(mgp_type_bool)) == MGP_ERROR_NO_ERROR);
+  module->AddProcedure("functions", std::move(functions));
+}
 namespace {
 bool IsAllowedExtension(const auto &extension) {
-  constexpr std::array<std::string_view, 1> allowed_extensions{".py"};
+  static constexpr std::array<std::string_view, 1> allowed_extensions{".py"};
   return std::any_of(allowed_extensions.begin(), allowed_extensions.end(),
                      [&](const auto allowed_extension) { return allowed_extension == extension; });
 }
@@ -650,10 +730,10 @@ void RegisterMgDeleteModuleFile(ModuleRegistry *module_registry, utils::RWLock *
 // `mgp_module::transformations into `proc_map`. The return value of WithModuleRegistration
 // is the same as that of `fun`. Note, the return value need only be convertible to `bool`,
 // it does not have to be `bool` itself.
-template <class TProcMap, class TTransMap, class TFun>
-auto WithModuleRegistration(TProcMap *proc_map, TTransMap *trans_map, const TFun &fun) {
+template <class TProcMap, class TTransMap, class TFuncMap, class TFun>
+auto WithModuleRegistration(TProcMap *proc_map, TTransMap *trans_map, TFuncMap *func_map, const TFun &fun) {
   // We probably don't need more than 256KB for module initialization.
-  constexpr size_t stack_bytes = 256 * 1024;
+  static constexpr size_t stack_bytes = 256UL * 1024UL;
   unsigned char stack_memory[stack_bytes];
   utils::MonotonicBufferResource monotonic_memory(stack_memory, stack_bytes);
   mgp_memory memory{&monotonic_memory};
@@ -664,6 +744,8 @@ auto WithModuleRegistration(TProcMap *proc_map, TTransMap *trans_map, const TFun
     for (const auto &proc : module_def.procedures) proc_map->emplace(proc);
     // Copy transformations into resulting trans_map.
     for (const auto &trans : module_def.transformations) trans_map->emplace(trans);
+    // Copy functions into resulting func_map.
+    for (const auto &func : module_def.functions) func_map->emplace(func);
   }
   return res;
 }
@@ -687,6 +769,8 @@ class SharedLibraryModule final : public Module {
 
   const std::map<std::string, mgp_trans, std::less<>> *Transformations() const override;
 
+  const std::map<std::string, mgp_func, std::less<>> *Functions() const override;
+
   std::optional<std::filesystem::path> Path() const override { return file_path_; }
 
  private:
@@ -702,6 +786,8 @@ class SharedLibraryModule final : public Module {
   std::map<std::string, mgp_proc, std::less<>> procedures_;
   /// Registered transformations
   std::map<std::string, mgp_trans, std::less<>> transformations_;
+  /// Registered functions
+  std::map<std::string, mgp_func, std::less<>> functions_;
 };
 
 SharedLibraryModule::SharedLibraryModule() : handle_(nullptr) {}
@@ -746,8 +832,8 @@ bool SharedLibraryModule::Load(const std::filesystem::path &file_path) {
       return with_error(error);
     }
     for (auto &trans : module_def->transformations) {
-      const bool was_result_added = MgpTransAddFixedResult(&trans.second);
-      if (!was_result_added) {
+      const bool success = MGP_ERROR_NO_ERROR == MgpTransAddFixedResult(&trans.second);
+      if (!success) {
         const auto error =
             fmt::format("Unable to add result to transformation in module {}; add result failed", file_path);
         return with_error(error);
@@ -755,7 +841,7 @@ bool SharedLibraryModule::Load(const std::filesystem::path &file_path) {
     }
     return true;
   };
-  if (!WithModuleRegistration(&procedures_, &transformations_, module_cb)) {
+  if (!WithModuleRegistration(&procedures_, &transformations_, &functions_, module_cb)) {
     return false;
   }
   // Get optional mgp_shutdown_module
@@ -801,6 +887,13 @@ const std::map<std::string, mgp_trans, std::less<>> *SharedLibraryModule::Transf
   return &transformations_;
 }
 
+const std::map<std::string, mgp_func, std::less<>> *SharedLibraryModule::Functions() const {
+  MG_ASSERT(handle_,
+            "Attempting to access functions of a module that has not "
+            "been loaded...");
+  return &functions_;
+}
+
 class PythonModule final : public Module {
  public:
   PythonModule();
@@ -816,6 +909,7 @@ class PythonModule final : public Module {
 
   const std::map<std::string, mgp_proc, std::less<>> *Procedures() const override;
   const std::map<std::string, mgp_trans, std::less<>> *Transformations() const override;
+  const std::map<std::string, mgp_func, std::less<>> *Functions() const override;
   std::optional<std::filesystem::path> Path() const override { return file_path_; }
 
  private:
@@ -823,6 +917,7 @@ class PythonModule final : public Module {
   py::Object py_module_;
   std::map<std::string, mgp_proc, std::less<>> procedures_;
   std::map<std::string, mgp_trans, std::less<>> transformations_;
+  std::map<std::string, mgp_func, std::less<>> functions_;
 };
 
 PythonModule::PythonModule() {}
@@ -853,7 +948,7 @@ bool PythonModule::Load(const std::filesystem::path &file_path) {
     };
     return result;
   };
-  py_module_ = WithModuleRegistration(&procedures_, &transformations_, module_cb);
+  py_module_ = WithModuleRegistration(&procedures_, &transformations_, &functions_, module_cb);
   if (py_module_) {
     spdlog::info("Loaded module {}", file_path);
 
@@ -877,6 +972,7 @@ bool PythonModule::Close() {
   auto gil = py::EnsureGIL();
   procedures_.clear();
   transformations_.clear();
+  functions_.clear();
   // Delete the module from the `sys.modules` directory so that the module will
   // be properly imported if imported again.
   py::Object sys(PyImport_ImportModule("sys"));
@@ -906,6 +1002,13 @@ const std::map<std::string, mgp_trans, std::less<>> *PythonModule::Transformatio
             "not been loaded...");
   return &transformations_;
 }
+
+const std::map<std::string, mgp_func, std::less<>> *PythonModule::Functions() const {
+  MG_ASSERT(py_module_,
+            "Attempting to access functions of a module that has "
+            "not been loaded...");
+  return &functions_;
+}
 namespace {
 
 std::unique_ptr<Module> LoadModuleFromFile(const std::filesystem::path &path) {
@@ -954,6 +1057,7 @@ ModuleRegistry::ModuleRegistry() {
   auto module = std::make_unique<BuiltinModule>();
   RegisterMgProcedures(&modules_, module.get());
   RegisterMgTransformations(&modules_, module.get());
+  RegisterMgFunctions(&modules_, module.get());
   RegisterMgLoad(this, &lock_, module.get());
   RegisterMgGetModuleFiles(this, module.get());
   RegisterMgGetModuleFile(this, module.get());
@@ -1083,7 +1187,7 @@ std::optional<std::pair<std::string_view, std::string_view>> FindModuleNameAndPr
 }
 
 template <typename T>
-concept ModuleProperties = utils::SameAsAnyOf<T, mgp_proc, mgp_trans>;
+concept ModuleProperties = utils::SameAsAnyOf<T, mgp_proc, mgp_trans, mgp_func>;
 
 template <ModuleProperties T>
 std::optional<std::pair<ModulePtr, const T *>> MakePairIfPropFound(const ModuleRegistry &module_registry,
@@ -1092,8 +1196,10 @@ std::optional<std::pair<ModulePtr, const T *>> MakePairIfPropFound(const ModuleR
   auto prop_fun = [](auto &module) {
     if constexpr (std::is_same_v<T, mgp_proc>) {
       return module->Procedures();
-    } else {
+    } else if constexpr (std::is_same_v<T, mgp_trans>) {
       return module->Transformations();
+    } else if constexpr (std::is_same_v<T, mgp_func>) {
+      return module->Functions();
     }
   };
   auto result = FindModuleNameAndProp(module_registry, fully_qualified_name, memory);
@@ -1121,4 +1227,10 @@ std::optional<std::pair<ModulePtr, const mgp_trans *>> FindTransformation(
   return MakePairIfPropFound<mgp_trans>(module_registry, fully_qualified_transformation_name, memory);
 }
 
+std::optional<std::pair<ModulePtr, const mgp_func *>> FindFunction(const ModuleRegistry &module_registry,
+                                                                   std::string_view fully_qualified_function_name,
+                                                                   utils::MemoryResource *memory) {
+  return MakePairIfPropFound<mgp_func>(module_registry, fully_qualified_function_name, memory);
+}
+
 }  // namespace memgraph::query::procedure

src/query/procedure/module.hpp

@@ -21,6 +21,7 @@
 #include <string_view>
 #include <unordered_map>
 
+#include "query/procedure/cypher_types.hpp"
 #include "query/procedure/mg_procedure_impl.hpp"
 #include "utils/memory.hpp"
 #include "utils/rw_lock.hpp"
@@ -45,6 +46,8 @@ class Module {
   virtual const std::map<std::string, mgp_proc, std::less<>> *Procedures() const = 0;
   /// Returns registered transformations of this module
   virtual const std::map<std::string, mgp_trans, std::less<>> *Transformations() const = 0;
+  // /// Returns registered functions of this module
+  virtual const std::map<std::string, mgp_func, std::less<>> *Functions() const = 0;
 
   virtual std::optional<std::filesystem::path> Path() const = 0;
 };
@@ -147,4 +150,62 @@ std::optional<std::pair<procedure::ModulePtr, const mgp_proc *>> FindProcedure(
 std::optional<std::pair<procedure::ModulePtr, const mgp_trans *>> FindTransformation(
     const ModuleRegistry &module_registry, const std::string_view fully_qualified_transformation_name,
     utils::MemoryResource *memory);
+
+/// Return the ModulePtr and `mgp_func *` of the found function after resolving
+/// `fully_qualified_function_name` if found. If there is no such function
+/// std::nullopt is returned. `memory` is used for temporary allocations
+/// inside this function. ModulePtr must be kept alive to make sure it won't be unloaded.
+std::optional<std::pair<procedure::ModulePtr, const mgp_func *>> FindFunction(
+    const ModuleRegistry &module_registry, const std::string_view fully_qualified_function_name,
+    utils::MemoryResource *memory);
+
+template <typename T>
+concept IsCallable = utils::SameAsAnyOf<T, mgp_proc, mgp_func>;
+
+template <IsCallable TCall>
+void ConstructArguments(const std::vector<TypedValue> &args, const TCall &callable,
+                        const std::string_view fully_qualified_name, mgp_list &args_list, mgp_graph &graph) {
+  const auto n_args = args.size();
+  const auto c_args_sz = callable.args.size();
+  const auto c_opt_args_sz = callable.opt_args.size();
+
+  if (n_args < c_args_sz || (n_args - c_args_sz > c_opt_args_sz)) {
+    if (callable.args.empty() && callable.opt_args.empty()) {
+      throw QueryRuntimeException("'{}' requires no arguments.", fully_qualified_name);
+    }
+
+    if (callable.opt_args.empty()) {
+      throw QueryRuntimeException("'{}' requires exactly {} {}.", fully_qualified_name, c_args_sz,
+                                  c_args_sz == 1U ? "argument" : "arguments");
+    }
+
+    throw QueryRuntimeException("'{}' requires between {} and {} arguments.", fully_qualified_name, c_args_sz,
+                                c_args_sz + c_opt_args_sz);
+  }
+  args_list.elems.reserve(n_args);
+
+  auto is_not_optional_arg = [c_args_sz](int i) { return c_args_sz > i; };
+  for (size_t i = 0; i < n_args; ++i) {
+    auto arg = args[i];
+    std::string_view name;
+    const query::procedure::CypherType *type;
+    if (is_not_optional_arg(i)) {
+      name = callable.args[i].first;
+      type = callable.args[i].second;
+    } else {
+      name = std::get<0>(callable.opt_args[i - c_args_sz]);
+      type = std::get<1>(callable.opt_args[i - c_args_sz]);
+    }
+    if (!type->SatisfiesType(arg)) {
+      throw QueryRuntimeException("'{}' argument named '{}' at position {} must be of type {}.", fully_qualified_name,
+                                  name, i, type->GetPresentableName());
+    }
+    args_list.elems.emplace_back(std::move(arg), &graph);
+  }
+  // Fill missing optional arguments with their default values.
+  const size_t passed_in_opt_args = n_args - c_args_sz;
+  for (size_t i = passed_in_opt_args; i < c_opt_args_sz; ++i) {
+    args_list.elems.emplace_back(std::get<2>(callable.opt_args[i]), &graph);
+  }
+}
 }  // namespace memgraph::query::procedure

src/query/procedure/py_module.cpp

@@ -447,62 +447,94 @@ PyObject *MakePyCypherType(mgp_type *type) {
 // clang-format off
 struct PyQueryProc {
   PyObject_HEAD
-  mgp_proc *proc;
+  mgp_proc *callable;
 };
 // clang-format on
 
-PyObject *PyQueryProcAddArg(PyQueryProc *self, PyObject *args) {
-  MG_ASSERT(self->proc);
+// clang-format off
+struct PyMagicFunc{
+  PyObject_HEAD
+  mgp_func *callable;
+};
+// clang-format on
+
+template <typename T>
+concept IsCallable = utils::SameAsAnyOf<T, PyQueryProc, PyMagicFunc>;
+
+template <IsCallable TCall>
+PyObject *PyCallableAddArg(TCall *self, PyObject *args) {
+  MG_ASSERT(self->callable);
   const char *name = nullptr;
   PyCypherType *py_type = nullptr;
   if (!PyArg_ParseTuple(args, "sO!", &name, &PyCypherTypeType, &py_type)) return nullptr;
   auto *type = py_type->type;
-  if (RaiseExceptionFromErrorCode(mgp_proc_add_arg(self->proc, name, type))) {
-    return nullptr;
+
+  if constexpr (std::is_same_v<TCall, PyQueryProc>) {
+    if (RaiseExceptionFromErrorCode(mgp_proc_add_arg(self->callable, name, type))) {
+      return nullptr;
+    }
+  } else if constexpr (std::is_same_v<TCall, PyMagicFunc>) {
+    if (RaiseExceptionFromErrorCode(mgp_func_add_arg(self->callable, name, type))) {
+      return nullptr;
+    }
   }
+
   Py_RETURN_NONE;
 }
 
-PyObject *PyQueryProcAddOptArg(PyQueryProc *self, PyObject *args) {
-  MG_ASSERT(self->proc);
+template <IsCallable TCall>
+PyObject *PyCallableAddOptArg(TCall *self, PyObject *args) {
+  MG_ASSERT(self->callable);
   const char *name = nullptr;
   PyCypherType *py_type = nullptr;
   PyObject *py_value = nullptr;
   if (!PyArg_ParseTuple(args, "sO!O", &name, &PyCypherTypeType, &py_type, &py_value)) return nullptr;
   auto *type = py_type->type;
-  mgp_memory memory{self->proc->opt_args.get_allocator().GetMemoryResource()};
+  mgp_memory memory{self->callable->opt_args.get_allocator().GetMemoryResource()};
   mgp_value *value = PyObjectToMgpValueWithPythonExceptions(py_value, &memory);
   if (value == nullptr) {
     return nullptr;
   }
-  if (RaiseExceptionFromErrorCode(mgp_proc_add_opt_arg(self->proc, name, type, value))) {
-    mgp_value_destroy(value);
-    return nullptr;
+  if constexpr (std::is_same_v<TCall, PyQueryProc>) {
+    if (RaiseExceptionFromErrorCode(mgp_proc_add_opt_arg(self->callable, name, type, value))) {
+      mgp_value_destroy(value);
+      return nullptr;
+    }
+  } else if constexpr (std::is_same_v<TCall, PyMagicFunc>) {
+    if (RaiseExceptionFromErrorCode(mgp_func_add_opt_arg(self->callable, name, type, value))) {
+      mgp_value_destroy(value);
+      return nullptr;
+    }
   }
+
   mgp_value_destroy(value);
   Py_RETURN_NONE;
 }
 
+PyObject *PyQueryProcAddArg(PyQueryProc *self, PyObject *args) { return PyCallableAddArg(self, args); }
+
+PyObject *PyQueryProcAddOptArg(PyQueryProc *self, PyObject *args) { return PyCallableAddOptArg(self, args); }
+
 PyObject *PyQueryProcAddResult(PyQueryProc *self, PyObject *args) {
-  MG_ASSERT(self->proc);
+  MG_ASSERT(self->callable);
   const char *name = nullptr;
   PyCypherType *py_type = nullptr;
   if (!PyArg_ParseTuple(args, "sO!", &name, &PyCypherTypeType, &py_type)) return nullptr;
 
   auto *type = reinterpret_cast<PyCypherType *>(py_type)->type;
-  if (RaiseExceptionFromErrorCode(mgp_proc_add_result(self->proc, name, type))) {
+  if (RaiseExceptionFromErrorCode(mgp_proc_add_result(self->callable, name, type))) {
     return nullptr;
   }
   Py_RETURN_NONE;
 }
 
 PyObject *PyQueryProcAddDeprecatedResult(PyQueryProc *self, PyObject *args) {
-  MG_ASSERT(self->proc);
+  MG_ASSERT(self->callable);
   const char *name = nullptr;
   PyCypherType *py_type = nullptr;
   if (!PyArg_ParseTuple(args, "sO!", &name, &PyCypherTypeType, &py_type)) return nullptr;
   auto *type = reinterpret_cast<PyCypherType *>(py_type)->type;
-  if (RaiseExceptionFromErrorCode(mgp_proc_add_deprecated_result(self->proc, name, type))) {
+  if (RaiseExceptionFromErrorCode(mgp_proc_add_deprecated_result(self->callable, name, type))) {
     return nullptr;
   }
   Py_RETURN_NONE;
@@ -532,6 +564,33 @@ static PyTypeObject PyQueryProcType = {
 };
 // clang-format on
 
+PyObject *PyMagicFuncAddArg(PyMagicFunc *self, PyObject *args) { return PyCallableAddArg(self, args); }
+
+PyObject *PyMagicFuncAddOptArg(PyMagicFunc *self, PyObject *args) { return PyCallableAddOptArg(self, args); }
+
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+static PyMethodDef PyMagicFuncMethods[] = {
+    {"__reduce__", reinterpret_cast<PyCFunction>(DisallowPickleAndCopy), METH_NOARGS, "__reduce__ is not supported"},
+    {"add_arg", reinterpret_cast<PyCFunction>(PyMagicFuncAddArg), METH_VARARGS,
+     "Add a required argument to a function."},
+    {"add_opt_arg", reinterpret_cast<PyCFunction>(PyMagicFuncAddOptArg), METH_VARARGS,
+     "Add an optional argument with a default value to a function."},
+    {nullptr},
+};
+
+// clang-format off
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+static PyTypeObject PyMagicFuncType = {
+    PyVarObject_HEAD_INIT(nullptr, 0)
+    .tp_name = "_mgp.Func",
+    .tp_basicsize = sizeof(PyMagicFunc),
+    // NOLINTNEXTLINE(hicpp-signed-bitwise)
+    .tp_flags = Py_TPFLAGS_DEFAULT,
+    .tp_doc = "Wraps struct mgp_func.",
+    .tp_methods = PyMagicFuncMethods,
+};
+// clang-format on
+
 // clang-format off
 struct PyQueryModule {
   PyObject_HEAD
@@ -796,7 +855,6 @@ py::Object MgpListToPyTuple(mgp_list *list, PyObject *py_graph) {
 }
 
 namespace {
-
 std::optional<py::ExceptionInfo> AddRecordFromPython(mgp_result *result, py::Object py_record) {
   py::Object py_mgp(PyImport_ImportModule("mgp"));
   if (!py_mgp) return py::FetchError();
@@ -870,6 +928,33 @@ std::optional<py::ExceptionInfo> AddMultipleRecordsFromPython(mgp_result *result
   return std::nullopt;
 }
 
+std::function<void()> PyObjectCleanup(py::Object &py_object) {
+  return [py_object]() {
+    // Run `gc.collect` (reference cycle-detection) explicitly, so that we are
+    // sure the procedure cleaned up everything it held references to. If the
+    // user stored a reference to one of our `_mgp` instances then the
+    // internally used `mgp_*` structs will stay unfreed and a memory leak
+    // will be reported at the end of the query execution.
+    py::Object gc(PyImport_ImportModule("gc"));
+    if (!gc) {
+      LOG_FATAL(py::FetchError().value());
+    }
+
+    if (!gc.CallMethod("collect")) {
+      LOG_FATAL(py::FetchError().value());
+    }
+
+    // After making sure all references from our side have been cleared,
+    // invalidate the `_mgp.Graph` object. If the user kept a reference to one
+    // of our `_mgp` instances then this will prevent them from using those
+    // objects (whose internal `mgp_*` pointers are now invalid and would cause
+    // a crash).
+    if (!py_object.CallMethod("invalidate")) {
+      LOG_FATAL(py::FetchError().value());
+    }
+  };
+}
+
 void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *graph, mgp_result *result,
                          mgp_memory *memory) {
   auto gil = py::EnsureGIL();
@@ -895,31 +980,6 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
     }
   };
 
-  auto cleanup = [](py::Object py_graph) {
-    // Run `gc.collect` (reference cycle-detection) explicitly, so that we are
-    // sure the procedure cleaned up everything it held references to. If the
-    // user stored a reference to one of our `_mgp` instances then the
-    // internally used `mgp_*` structs will stay unfreed and a memory leak
-    // will be reported at the end of the query execution.
-    py::Object gc(PyImport_ImportModule("gc"));
-    if (!gc) {
-      LOG_FATAL(py::FetchError().value());
-    }
-
-    if (!gc.CallMethod("collect")) {
-      LOG_FATAL(py::FetchError().value());
-    }
-
-    // After making sure all references from our side have been cleared,
-    // invalidate the `_mgp.Graph` object. If the user kept a reference to one
-    // of our `_mgp` instances then this will prevent them from using those
-    // objects (whose internal `mgp_*` pointers are now invalid and would cause
-    // a crash).
-    if (!py_graph.CallMethod("invalidate")) {
-      LOG_FATAL(py::FetchError().value());
-    }
-  };
-
   // It is *VERY IMPORTANT* to note that this code takes great care not to keep
   // any extra references to any `_mgp` instances (except for `_mgp.Graph`), so
   // as not to introduce extra reference counts and prevent their deallocation.
@@ -932,14 +992,9 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
   std::optional<std::string> maybe_msg;
   {
     py::Object py_graph(MakePyGraph(graph, memory));
+    utils::OnScopeExit clean_up(PyObjectCleanup(py_graph));
     if (py_graph) {
-      try {
-        maybe_msg = error_to_msg(call(py_graph));
-        cleanup(py_graph);
-      } catch (...) {
-        cleanup(py_graph);
-        throw;
-      }
+      maybe_msg = error_to_msg(call(py_graph));
     } else {
       maybe_msg = error_to_msg(py::FetchError());
     }
@@ -972,32 +1027,58 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
     return AddRecordFromPython(result, py_res);
   };
 
-  auto cleanup = [](py::Object py_graph, py::Object py_messages) {
-    // Run `gc.collect` (reference cycle-detection) explicitly, so that we are
-    // sure the procedure cleaned up everything it held references to. If the
-    // user stored a reference to one of our `_mgp` instances then the
-    // internally used `mgp_*` structs will stay unfreed and a memory leak
-    // will be reported at the end of the query execution.
-    py::Object gc(PyImport_ImportModule("gc"));
-    if (!gc) {
-      LOG_FATAL(py::FetchError().value());
-    }
+  // It is *VERY IMPORTANT* to note that this code takes great care not to keep
+  // any extra references to any `_mgp` instances (except for `_mgp.Graph`), so
+  // as not to introduce extra reference counts and prevent their deallocation.
+  // In particular, the `ExceptionInfo` object has a `traceback` field that
+  // contains references to the Python frames and their arguments, and therefore
+  // our `_mgp` instances as well. Within this code we ensure not to keep the
+  // `ExceptionInfo` object alive so that no extra reference counts are
+  // introduced. We only fetch the error message and immediately destroy the
+  // object.
+  std::optional<std::string> maybe_msg;
+  {
+    py::Object py_graph(MakePyGraph(graph, memory));
+    py::Object py_messages(MakePyMessages(msgs, memory));
 
-    if (!gc.CallMethod("collect")) {
-      LOG_FATAL(py::FetchError().value());
-    }
+    utils::OnScopeExit clean_up_graph(PyObjectCleanup(py_graph));
+    utils::OnScopeExit clean_up_messages(PyObjectCleanup(py_messages));
 
-    // After making sure all references from our side have been cleared,
-    // invalidate the `_mgp.Graph` object. If the user kept a reference to one
-    // of our `_mgp` instances then this will prevent them from using those
-    // objects (whose internal `mgp_*` pointers are now invalid and would cause
-    // a crash).
-    if (!py_graph.CallMethod("invalidate")) {
-      LOG_FATAL(py::FetchError().value());
+    if (py_graph && py_messages) {
+      maybe_msg = error_to_msg(call(py_graph, py_messages));
+    } else {
+      maybe_msg = error_to_msg(py::FetchError());
     }
-    if (!py_messages.CallMethod("invalidate")) {
-      LOG_FATAL(py::FetchError().value());
+  }
+
+  if (maybe_msg) {
+    static_cast<void>(mgp_result_set_error_msg(result, maybe_msg->c_str()));
+  }
+}
+
+void CallPythonFunction(const py::Object &py_cb, mgp_list *args, mgp_graph *graph, mgp_func_result *result,
+                        mgp_memory *memory) {
+  auto gil = py::EnsureGIL();
+
+  auto error_to_msg = [](const std::optional<py::ExceptionInfo> &exc_info) -> std::optional<std::string> {
+    if (!exc_info) return std::nullopt;
+    // Here we tell the traceback formatter to skip the first line of the
+    // traceback because that line will always be our wrapper function in our
+    // internal `mgp.py` file. With that line skipped, the user will always
+    // get only the relevant traceback that happened in his Python code.
+    return py::FormatException(*exc_info, /* skip_first_line = */ true);
+  };
+
+  auto call = [&](py::Object py_graph) -> utils::BasicResult<std::optional<py::ExceptionInfo>, mgp_value *> {
+    py::Object py_args(MgpListToPyTuple(args, py_graph.Ptr()));
+    if (!py_args) return {py::FetchError()};
+    auto py_res = py_cb.Call(py_graph, py_args);
+    if (!py_res) return {py::FetchError()};
+    mgp_value *ret_val = PyObjectToMgpValueWithPythonExceptions(py_res.Ptr(), memory);
+    if (ret_val == nullptr) {
+      return {py::FetchError()};
     }
+    return ret_val;
   };
 
   // It is *VERY IMPORTANT* to note that this code takes great care not to keep
@@ -1012,22 +1093,22 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
   std::optional<std::string> maybe_msg;
   {
     py::Object py_graph(MakePyGraph(graph, memory));
-    py::Object py_messages(MakePyMessages(msgs, memory));
-    if (py_graph && py_messages) {
-      try {
-        maybe_msg = error_to_msg(call(py_graph, py_messages));
-        cleanup(py_graph, py_messages);
-      } catch (...) {
-        cleanup(py_graph, py_messages);
-        throw;
+    utils::OnScopeExit clean_up(PyObjectCleanup(py_graph));
+    if (py_graph) {
+      auto maybe_result = call(py_graph);
+      if (!maybe_result.HasError()) {
+        static_cast<void>(mgp_func_result_set_value(result, maybe_result.GetValue(), memory));
+        return;
       }
+      maybe_msg = error_to_msg(maybe_result.GetError());
     } else {
       maybe_msg = error_to_msg(py::FetchError());
     }
   }
 
   if (maybe_msg) {
-    static_cast<void>(mgp_result_set_error_msg(result, maybe_msg->c_str()));
+    static_cast<void>(
+        mgp_func_result_set_error_msg(result, maybe_msg->c_str(), memory));  // No error fetching if this fails
   }
 }
 
@@ -1056,9 +1137,9 @@ PyObject *PyQueryModuleAddProcedure(PyQueryModule *self, PyObject *cb, bool is_w
     PyErr_SetString(PyExc_ValueError, "Already registered a procedure with the same name.");
     return nullptr;
   }
-  auto *py_proc = PyObject_New(PyQueryProc, &PyQueryProcType);
+  auto *py_proc = PyObject_New(PyQueryProc, &PyQueryProcType);  // NOLINT(cppcoreguidelines-pro-type-cstyle-cast)
   if (!py_proc) return nullptr;
-  py_proc->proc = &proc_it->second;
+  py_proc->callable = &proc_it->second;
   return reinterpret_cast<PyObject *>(py_proc);
 }
 }  // namespace
@@ -1100,6 +1181,39 @@ PyObject *PyQueryModuleAddTransformation(PyQueryModule *self, PyObject *cb) {
   Py_RETURN_NONE;
 }
 
+PyObject *PyQueryModuleAddFunction(PyQueryModule *self, PyObject *cb) {
+  MG_ASSERT(self->module);
+  if (!PyCallable_Check(cb)) {
+    PyErr_SetString(PyExc_TypeError, "Expected a callable object.");
+    return nullptr;
+  }
+  auto py_cb = py::Object::FromBorrow(cb);
+  py::Object py_name(py_cb.GetAttr("__name__"));
+  const auto *name = PyUnicode_AsUTF8(py_name.Ptr());
+  if (!name) return nullptr;
+  if (!IsValidIdentifierName(name)) {
+    PyErr_SetString(PyExc_ValueError, "Function name is not a valid identifier");
+    return nullptr;
+  }
+  auto *memory = self->module->functions.get_allocator().GetMemoryResource();
+  mgp_func func(
+      name,
+      [py_cb](mgp_list *args, mgp_func_context *func_ctx, mgp_func_result *result, mgp_memory *memory) {
+        auto graph = mgp_graph::NonWritableGraph(*(func_ctx->impl), func_ctx->view);
+        return CallPythonFunction(py_cb, args, &graph, result, memory);
+      },
+      memory);
+  const auto [func_it, did_insert] = self->module->functions.emplace(name, std::move(func));
+  if (!did_insert) {
+    PyErr_SetString(PyExc_ValueError, "Already registered a function with the same name.");
+    return nullptr;
+  }
+  auto *py_func = PyObject_New(PyMagicFunc, &PyMagicFuncType);  // NOLINT(cppcoreguidelines-pro-type-cstyle-cast)
+  if (!py_func) return nullptr;
+  py_func->callable = &func_it->second;
+  return reinterpret_cast<PyObject *>(py_func);
+}
+
 static PyMethodDef PyQueryModuleMethods[] = {
     {"__reduce__", reinterpret_cast<PyCFunction>(DisallowPickleAndCopy), METH_NOARGS, "__reduce__ is not supported"},
     {"add_read_procedure", reinterpret_cast<PyCFunction>(PyQueryModuleAddReadProcedure), METH_O,
@@ -1108,6 +1222,8 @@ static PyMethodDef PyQueryModuleMethods[] = {
      "Register a writeable procedure with this module."},
     {"add_transformation", reinterpret_cast<PyCFunction>(PyQueryModuleAddTransformation), METH_O,
      "Register a transformation with this module."},
+    {"add_function", reinterpret_cast<PyCFunction>(PyQueryModuleAddFunction), METH_O,
+     "Register a function with this module."},
     {nullptr},
 };
 
@@ -1980,6 +2096,7 @@ PyObject *PyInitMgpModule() {
   if (!register_type(&PyGraphType, "Graph")) return nullptr;
   if (!register_type(&PyEdgeType, "Edge")) return nullptr;
   if (!register_type(&PyQueryProcType, "Proc")) return nullptr;
+  if (!register_type(&PyMagicFuncType, "Func")) return nullptr;
   if (!register_type(&PyQueryModuleType, "Module")) return nullptr;
   if (!register_type(&PyVertexType, "Vertex")) return nullptr;
   if (!register_type(&PyPathType, "Path")) return nullptr;
@@ -2427,7 +2544,8 @@ mgp_value *PyObjectToMgpValue(PyObject *o, mgp_memory *memory) {
     }
     static_cast<void>(local_date_time.release());
   } else if (PyDelta_CheckExact(o)) {
-    constexpr int64_t microseconds_in_days = static_cast<std::chrono::microseconds>(std::chrono::days{1}).count();
+    static constexpr int64_t microseconds_in_days =
+        static_cast<std::chrono::microseconds>(std::chrono::days{1}).count();
     const auto days =
         PyDateTime_DELTA_GET_DAYS(o);  // NOLINT(cppcoreguidelines-pro-type-cstyle-cast,hicpp-signed-bitwise)
     auto microseconds =

src/query/stream/common.hpp

@@ -23,8 +23,8 @@
 
 namespace memgraph::query::stream {
 
-constexpr std::chrono::milliseconds kDefaultBatchInterval{100};
-constexpr int64_t kDefaultBatchSize{1000};
+inline constexpr std::chrono::milliseconds kDefaultBatchInterval{100};
+inline constexpr int64_t kDefaultBatchSize{1000};
 
 template <typename TMessage>
 using ConsumerFunction = std::function<void(const std::vector<TMessage> &)>;

src/query/stream/streams.cpp

@@ -42,7 +42,7 @@ extern const Event MessagesConsumed;
 
 namespace memgraph::query::stream {
 namespace {
-constexpr auto kExpectedTransformationResultSize = 2;
+inline constexpr auto kExpectedTransformationResultSize = 2;
 const utils::pmr::string query_param_name{"query", utils::NewDeleteResource()};
 const utils::pmr::string params_param_name{"parameters", utils::NewDeleteResource()};
 
@@ -172,9 +172,9 @@ void Streams::RegisterProcedures() {
 
 void Streams::RegisterKafkaProcedures() {
   {
-    constexpr std::string_view proc_name = "kafka_set_stream_offset";
-    auto set_stream_offset = [this, proc_name](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result,
-                                               mgp_memory * /*memory*/) {
+    static constexpr std::string_view proc_name = "kafka_set_stream_offset";
+    auto set_stream_offset = [this](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result,
+                                    mgp_memory * /*memory*/) {
       auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
       const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
       auto *arg_offset = procedure::Call<mgp_value *>(mgp_list_at, args, 1);
@@ -190,7 +190,7 @@ void Streams::RegisterKafkaProcedures() {
                                    "Unable to set procedure error message of procedure: {}", proc_name);
                        }
                      },
-                     [proc_name](auto && /*other*/) {
+                     [](auto && /*other*/) {
                        throw QueryRuntimeException("'{}' can be only used for Kafka stream sources", proc_name);
                      }},
                  it->second);
@@ -205,17 +205,15 @@ void Streams::RegisterKafkaProcedures() {
   }
 
   {
-    constexpr std::string_view proc_name = "kafka_stream_info";
+    static constexpr std::string_view proc_name = "kafka_stream_info";
 
-    constexpr std::string_view consumer_group_result_name = "consumer_group";
-    constexpr std::string_view topics_result_name = "topics";
-    constexpr std::string_view bootstrap_servers_result_name = "bootstrap_servers";
-    constexpr std::string_view configs_result_name = "configs";
-    constexpr std::string_view credentials_result_name = "credentials";
+    static constexpr std::string_view consumer_group_result_name = "consumer_group";
+    static constexpr std::string_view topics_result_name = "topics";
+    static constexpr std::string_view bootstrap_servers_result_name = "bootstrap_servers";
+    static constexpr std::string_view configs_result_name = "configs";
+    static constexpr std::string_view credentials_result_name = "credentials";
 
-    auto get_stream_info = [this, proc_name, consumer_group_result_name, topics_result_name,
-                            bootstrap_servers_result_name, configs_result_name, credentials_result_name](
-                               mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
+    auto get_stream_info = [this](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
       auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
       const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
       auto lock_ptr = streams_.Lock();
@@ -339,7 +337,7 @@ void Streams::RegisterKafkaProcedures() {
                   return;
                 }
               },
-              [proc_name](auto && /*other*/) {
+              [](auto && /*other*/) {
                 throw QueryRuntimeException("'{}' can be only used for Kafka stream sources", proc_name);
               }},
           it->second);
@@ -367,11 +365,10 @@ void Streams::RegisterKafkaProcedures() {
 
 void Streams::RegisterPulsarProcedures() {
   {
-    constexpr std::string_view proc_name = "pulsar_stream_info";
-    constexpr std::string_view service_url_result_name = "service_url";
-    constexpr std::string_view topics_result_name = "topics";
-    auto get_stream_info = [this, proc_name, service_url_result_name, topics_result_name](
-                               mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
+    static constexpr std::string_view proc_name = "pulsar_stream_info";
+    static constexpr std::string_view service_url_result_name = "service_url";
+    static constexpr std::string_view topics_result_name = "topics";
+    auto get_stream_info = [this](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
       auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
       const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
       auto lock_ptr = streams_.Lock();
@@ -427,7 +424,7 @@ void Streams::RegisterPulsarProcedures() {
                   return;
                 }
               },
-              [proc_name](auto && /*other*/) {
+              [](auto && /*other*/) {
                 throw QueryRuntimeException("'{}' can be only used for Pulsar stream sources", proc_name);
               }},
           it->second);

src/query/trigger.cpp

@@ -219,7 +219,7 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
   // Set up temporary memory for a single Pull. Initial memory comes from the
   // stack. 256 KiB should fit on the stack and should be more than enough for a
   // single `Pull`.
-  constexpr size_t stack_size = 256 * 1024;
+  static constexpr size_t stack_size = 256UL * 1024UL;
   char stack_data[stack_size];
 
   // We can throw on every query because a simple queries for deleting will use only
@@ -251,7 +251,7 @@ void Trigger::Execute(DbAccessor *dba, utils::MonotonicBufferResource *execution
 
 namespace {
 // When the format of the persisted trigger is changed, increase this version
-constexpr uint64_t kVersion{2};
+inline constexpr uint64_t kVersion{2};
 }  // namespace
 
 TriggerStore::TriggerStore(std::filesystem::path directory) : storage_{std::move(directory)} {}

src/query/typed_value.cpp

@@ -660,8 +660,8 @@ double ToDouble(const TypedValue &value) {
 
 namespace {
 bool IsTemporalType(const TypedValue::Type type) {
-  constexpr std::array temporal_types{TypedValue::Type::Date, TypedValue::Type::LocalTime,
-                                      TypedValue::Type::LocalDateTime, TypedValue::Type::Duration};
+  static constexpr std::array temporal_types{TypedValue::Type::Date, TypedValue::Type::LocalTime,
+                                             TypedValue::Type::LocalDateTime, TypedValue::Type::Duration};
   return std::any_of(temporal_types.begin(), temporal_types.end(),
                      [type](const auto temporal_type) { return temporal_type == type; });
 };

src/storage/v2/storage.cpp

@@ -49,7 +49,7 @@ namespace memgraph::storage {
 using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
 
 namespace {
-[[maybe_unused]] constexpr uint16_t kEpochHistoryRetention = 1000;
+inline constexpr uint16_t kEpochHistoryRetention = 1000;
 }  // namespace
 
 auto AdvanceToVisibleVertex(utils::SkipList<Vertex>::Iterator it, utils::SkipList<Vertex>::Iterator end,

src/utils/async_timer.cpp

@@ -25,7 +25,7 @@
 
 namespace {
 
-constexpr uint64_t kInvalidFlagId = 0U;
+inline constexpr uint64_t kInvalidFlagId = 0U;
 // std::numeric_limits<time_t>::max() cannot be represented precisely as a double, so the next smallest value is the
 // maximum number of seconds the timer can be used with
 const double max_seconds_as_double = std::nexttoward(std::numeric_limits<time_t>::max(), 0.0);
@@ -143,7 +143,7 @@ AsyncTimer::AsyncTimer(double seconds)
   MG_ASSERT(timer_create(CLOCK_MONOTONIC, &notification_settings, &timer_id_) == 0, "Couldn't create timer: ({}) {}",
             errno, strerror(errno));
 
-  constexpr auto kSecondsToNanos = 1000 * 1000 * 1000;
+  static constexpr auto kSecondsToNanos = 1000 * 1000 * 1000;
   // Casting will truncate down, but that's exactly what we want.
   const auto second_as_time_t = static_cast<time_t>(seconds);
   const auto remaining_nano_seconds = static_cast<time_t>((seconds - second_as_time_t) * kSecondsToNanos);

src/utils/base64.cpp

@@ -45,7 +45,7 @@ namespace {
 // two sets of base64 characters needs to be chosen.
 // They differ in their last two characters.
 //
-constexpr std::array base64_chars = {
+inline constexpr std::array base64_chars = {
     "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
     "abcdefghijklmnopqrstuvwxyz"
     "0123456789"

src/utils/event_counter.cpp

@@ -1,4 +1,4 @@
-// Copyright 2021 Memgraph Ltd.
+// 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
@@ -49,6 +49,7 @@
   M(UnionOperator, "Number of times Union operator was used.")                                             \
   M(CartesianOperator, "Number of times Cartesian operator was used.")                                     \
   M(CallProcedureOperator, "Number of times CallProcedure operator was used.")                             \
+  M(ForeachOperator, "Number of times Foreach operator was used.")                                         \
                                                                                                            \
   M(FailedQuery, "Number of times executing a query failed.")                                              \
   M(LabelIndexCreated, "Number of times a label index was created.")                                       \
@@ -65,7 +66,7 @@ namespace EventCounter {
 APPLY_FOR_EVENTS(M)
 #undef M
 
-constexpr Event END = __COUNTER__;
+inline constexpr Event END = __COUNTER__;
 
 // Initialize array for the global counter with all values set to 0
 Counter global_counters_array[END]{};

src/utils/file.hpp

@@ -68,7 +68,7 @@ bool RenamePath(const std::filesystem::path &src, const std::filesystem::path &d
 /// `write` for each of our (very small) logical reads/writes. Because of that,
 /// `read` or `write` is only called when the buffer is full and/or needs
 /// emptying.
-constexpr size_t kFileBufferSize = 262144;
+inline constexpr size_t kFileBufferSize = 262144;
 
 /// This class implements a file handler that is used to read binary files. It
 /// was developed because the C++ standard library has an awful API and makes

src/utils/fnv.hpp

@@ -67,8 +67,8 @@ struct FnvCollection {
 template <typename TA, typename TB, typename TAHash = std::hash<TA>, typename TBHash = std::hash<TB>>
 struct HashCombine {
   size_t operator()(const TA &a, const TB &b) const {
-    constexpr size_t fnv_prime = 1099511628211UL;
-    constexpr size_t fnv_offset = 14695981039346656037UL;
+    static constexpr size_t fnv_prime = 1099511628211UL;
+    static constexpr size_t fnv_offset = 14695981039346656037UL;
     size_t ret = fnv_offset;
     ret ^= TAHash()(a);
     ret *= fnv_prime;

src/utils/license.cpp

@@ -30,7 +30,7 @@
 namespace memgraph::utils::license {
 
 namespace {
-constexpr std::string_view license_key_prefix = "mglk-";
+inline constexpr std::string_view license_key_prefix = "mglk-";
 
 std::optional<License> GetLicense(const std::string &license_key) {
   if (license_key.empty()) {

src/utils/license.hpp

@@ -28,8 +28,8 @@ struct License {
   bool operator==(const License &) const = default;
 };
 
-constexpr std::string_view kEnterpriseLicenseSettingKey = "enterprise.license";
-constexpr std::string_view kOrganizationNameSettingKey = "organization.name";
+inline constexpr std::string_view kEnterpriseLicenseSettingKey = "enterprise.license";
+inline constexpr std::string_view kOrganizationNameSettingKey = "organization.name";
 
 enum class LicenseCheckError : uint8_t { INVALID_LICENSE_KEY_STRING, INVALID_ORGANIZATION_NAME, EXPIRED_LICENSE };
 

src/utils/math.hpp

@@ -24,7 +24,7 @@ static_assert(std::is_same_v<uint64_t, unsigned long>,
 /// This function computes the log2 function on integer types. It is faster than
 /// the cmath `log2` function because it doesn't use floating point values for
 /// calculation.
-constexpr inline uint64_t Log2(uint64_t val) {
+constexpr uint64_t Log2(uint64_t val) {
   // The `clz` function is undefined when the passed value is 0 and the value of
   // `log` is `-inf` so we special case it here.
   if (val == 0) return 0;
@@ -35,12 +35,12 @@ constexpr inline uint64_t Log2(uint64_t val) {
 }
 
 /// Return `true` if `val` is a power of 2.
-constexpr inline bool IsPow2(uint64_t val) noexcept { return val != 0ULL && (val & (val - 1ULL)) == 0ULL; }
+constexpr bool IsPow2(uint64_t val) noexcept { return val != 0ULL && (val & (val - 1ULL)) == 0ULL; }
 
 /// Return `val` if it is power of 2, otherwise get the next power of 2 value.
 /// If `val` is sufficiently large, the next power of 2 value may not fit into
 /// the result type and you will get a wrapped value to 1ULL.
-constexpr inline uint64_t Ceil2(uint64_t val) noexcept {
+constexpr uint64_t Ceil2(uint64_t val) noexcept {
   if (val == 0ULL || val == 1ULL) return 1ULL;
   return 1ULL << (Log2(val - 1ULL) + 1ULL);
 }
@@ -53,7 +53,7 @@ constexpr inline uint64_t Ceil2(uint64_t val) noexcept {
 ///     RoundUint64ToMultiple(5, 8) == 8
 ///     RoundUint64ToMultiple(8, 8) == 8
 ///     RoundUint64ToMultiple(9, 8) == 16
-constexpr inline std::optional<uint64_t> RoundUint64ToMultiple(uint64_t val, uint64_t multiple) noexcept {
+constexpr std::optional<uint64_t> RoundUint64ToMultiple(uint64_t val, uint64_t multiple) noexcept {
   if (multiple == 0) return std::nullopt;
   uint64_t numerator = val + multiple - 1;
   // Check for overflow.

src/utils/readable_size.cpp

@@ -19,8 +19,8 @@ namespace memgraph::utils {
 
 std::string GetReadableSize(double size) {
   // TODO (antonio2368): Add support for base 1000 (KB, GB, TB...)
-  constexpr std::array units = {"B", "KiB", "MiB", "GiB", "TiB"};
-  constexpr double delimiter = 1024;
+  static constexpr std::array units = {"B", "KiB", "MiB", "GiB", "TiB"};
+  static constexpr double delimiter = 1024;
 
   size_t i = 0;
   for (; i + 1 < units.size() && size >= delimiter; ++i) {

src/utils/small_vector.hpp

@@ -325,7 +325,7 @@ class SmallVectorTemplateBase<T, true> : public SmallVectorTemplateCommon<T> {
 };
 
 template <typename T>
-constexpr bool is_pod = std::is_standard_layout_v<T> &&std::is_trivial_v<T>;
+inline constexpr bool is_pod = std::is_standard_layout_v<T> &&std::is_trivial_v<T>;
 
 /// This class consists of common code factored out of the SmallVector class to
 /// reduce code duplication based on the SmallVector 'n' template parameter.

src/utils/temporal.cpp

@@ -90,7 +90,7 @@ LocalDateTime CurrentLocalDateTime() {
 }
 
 namespace {
-constexpr auto *kSupportedDateFormatsHelpMessage = R"help(
+inline constexpr auto *kSupportedDateFormatsHelpMessage = R"help(
 String representing the date should be in one of the following formats:
 
 - YYYY-MM-DD
@@ -112,7 +112,7 @@ std::pair<DateParameters, bool> ParseDateParameters(std::string_view date_string
   // https://en.wikipedia.org/wiki/ISO_8601#Dates
   // Date string with the '-' as separator are in the EXTENDED format,
   // otherwise they are in a BASIC format
-  constexpr std::array valid_sizes{
+  static constexpr std::array valid_sizes{
       10,  // YYYY-MM-DD
       8,   // YYYYMMDD
       7    // YYYY-MM
@@ -184,7 +184,7 @@ size_t DateHash::operator()(const Date &date) const {
 }
 
 namespace {
-constexpr auto *kSupportedTimeFormatsHelpMessage = R"help(
+inline constexpr auto *kSupportedTimeFormatsHelpMessage = R"help(
 String representing the time should be in one of the following formats:
 
 - [T]hh:mm:ss
@@ -388,7 +388,7 @@ size_t LocalTimeHash::operator()(const LocalTime &local_time) const {
 }
 
 namespace {
-constexpr auto *kSupportedLocalDateTimeFormatsHelpMessage = R"help(
+inline constexpr auto *kSupportedLocalDateTimeFormatsHelpMessage = R"help(
 String representing the LocalDateTime should be in one of the following formats:
 
 - YYYY-MM-DDThh:mm:ss
@@ -458,7 +458,7 @@ std::pair<DateParameters, LocalTimeParameters> ParseLocalDateTimeParameters(std:
 
 LocalDateTime::LocalDateTime(const int64_t microseconds) {
   auto chrono_microseconds = std::chrono::microseconds(microseconds);
-  constexpr int64_t one_day_in_microseconds = std::chrono::microseconds{std::chrono::days{1}}.count();
+  static constexpr int64_t one_day_in_microseconds = std::chrono::microseconds{std::chrono::days{1}}.count();
   if (microseconds < 0 && (microseconds % one_day_in_microseconds != 0)) {
     date = Date(microseconds - one_day_in_microseconds);
   } else {

src/utils/temporal.hpp

@@ -235,7 +235,8 @@ struct LocalTime {
     auto abs = [](auto value) { return (value >= 0) ? value : -value; };
     const auto lhs = local_time.MicrosecondsSinceEpoch();
     if (rhs < 0 && lhs < abs(rhs)) {
-      constexpr int64_t one_day_in_microseconds = chrono::duration_cast<chrono::microseconds>(chrono::days(1)).count();
+      static constexpr int64_t one_day_in_microseconds =
+          chrono::duration_cast<chrono::microseconds>(chrono::days(1)).count();
       rhs = one_day_in_microseconds + rhs;
     }
     auto result = chrono::microseconds(lhs + rhs);

src/utils/thread.cpp

@@ -18,7 +18,7 @@
 namespace memgraph::utils {
 
 void ThreadSetName(const std::string &name) {
-  constexpr auto max_name_length = GetMaxThreadNameSize();
+  static constexpr auto max_name_length = GetMaxThreadNameSize();
   MG_ASSERT(name.size() <= max_name_length, "Thread name '{}' is too long", max_name_length);
 
   if (prctl(PR_SET_NAME, name.c_str()) != 0) {

tests/benchmark/query/execution.cpp

@@ -499,4 +499,39 @@ BENCHMARK_TEMPLATE(Unwind, MonotonicBufferResource)
 
 BENCHMARK_TEMPLATE(Unwind, PoolResource)->Ranges({{4, 1U << 7U}, {512, 1U << 13U}})->Unit(benchmark::kMicrosecond);
 
+template <class TMemory>
+// NOLINTNEXTLINE(google-runtime-references)
+static void Foreach(benchmark::State &state) {
+  memgraph::query::AstStorage ast;
+  memgraph::storage::Storage db;
+  memgraph::query::SymbolTable symbol_table;
+  auto list_sym = symbol_table.CreateSymbol("list", false);
+  auto *list_expr = ast.Create<memgraph::query::Identifier>("list")->MapTo(list_sym);
+  auto out_sym = symbol_table.CreateSymbol("out", false);
+  auto create_node =
+      std::make_shared<memgraph::query::plan::CreateNode>(nullptr, memgraph::query::plan::NodeCreationInfo{});
+  auto foreach = std::make_shared<memgraph::query::plan::Foreach>(nullptr, std::move(create_node), list_expr, out_sym);
+
+  auto storage_dba = db.Access();
+  memgraph::query::DbAccessor dba(&storage_dba);
+  TMemory per_pull_memory;
+  memgraph::query::EvaluationContext evaluation_context{per_pull_memory.get()};
+  while (state.KeepRunning()) {
+    memgraph::query::ExecutionContext execution_context{&dba, symbol_table, evaluation_context};
+    TMemory memory;
+    memgraph::query::Frame frame(symbol_table.max_position(), memory.get());
+    frame[list_sym] = memgraph::query::TypedValue(std::vector<memgraph::query::TypedValue>(state.range(1)));
+    auto cursor = foreach->MakeCursor(memory.get());
+    while (cursor->Pull(frame, execution_context)) per_pull_memory.Reset();
+  }
+  state.SetItemsProcessed(state.iterations());
+}
+
+BENCHMARK_TEMPLATE(Foreach, PoolResource)->Ranges({{4, 1U << 7U}, {512, 1U << 13U}})->Unit(benchmark::kMicrosecond);
+BENCHMARK_TEMPLATE(Foreach, MonotonicBufferResource)
+    ->Ranges({{4, 1U << 7U}, {512, 1U << 13U}})
+    ->Unit(benchmark::kMicrosecond);
+
+BENCHMARK_TEMPLATE(Foreach, PoolResource)->Ranges({{4, 1U << 7U}, {512, 1U << 13U}})->Unit(benchmark::kMicrosecond);
+
 BENCHMARK_MAIN();

tests/benchmark/query/profile.cpp

@@ -409,7 +409,7 @@ namespace tree_storage {
 //
 // ProfilingStats
 
-constexpr size_t kMaxProfilingStatsChildren = 3;
+inline constexpr size_t kMaxProfilingStatsChildren = 3;
 struct ProfilingStatsStorage;
 
 struct ProfilingStats {

tests/concurrent/common.hpp

@@ -1,4 +1,4 @@
-// Copyright 2021 Memgraph Ltd.
+// 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
@@ -24,7 +24,7 @@
 // TODO: REFACTOR
 
 // Sets max number of threads that will be used in concurrent tests.
-constexpr int max_no_threads = 8;
+inline constexpr int max_no_threads = 8;
 
 using std::cout;
 using std::endl;

tests/concurrent/network_common.hpp

@@ -22,8 +22,8 @@
 
 #include "communication/server.hpp"
 
-static constexpr const int SIZE = 60000;
-static constexpr const int REPLY = 10;
+inline constexpr const int SIZE = 60000;
+inline constexpr const int REPLY = 10;
 
 using memgraph::io::network::Endpoint;
 using memgraph::io::network::Socket;

tests/concurrent/network_read_hang.cpp

@@ -24,7 +24,7 @@
 
 #include "communication/server.hpp"
 
-static constexpr const char interface[] = "127.0.0.1";
+inline constexpr const char interface[] = "127.0.0.1";
 
 using memgraph::io::network::Endpoint;
 using memgraph::io::network::Socket;

tests/concurrent/network_server.cpp

@@ -1,4 +1,4 @@
-// Copyright 2021 Memgraph Ltd.
+// 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
@@ -17,7 +17,7 @@
 
 #include "network_common.hpp"
 
-static constexpr const char interface[] = "127.0.0.1";
+inline constexpr const char interface[] = "127.0.0.1";
 
 unsigned char data[SIZE];
 

tests/concurrent/network_session_leak.cpp

@@ -1,4 +1,4 @@
-// Copyright 2021 Memgraph Ltd.
+// 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
@@ -18,7 +18,7 @@
 
 #include "network_common.hpp"
 
-static constexpr const char interface[] = "127.0.0.1";
+inline constexpr const char interface[] = "127.0.0.1";
 
 unsigned char data[SIZE];
 

tests/concurrent/spin_lock.cpp

@@ -37,7 +37,7 @@ void test_lock() {
 }
 
 int main() {
-  constexpr int N = 16;
+  static constexpr int N = 16;
   std::vector<std::thread> threads;
 
   for (int i = 0; i < N; ++i) threads.push_back(std::thread(test_lock));

tests/e2e/CMakeLists.txt

@@ -1,3 +1,11 @@
+# Set up C++ functions for e2e tests
+function(add_query_module target_name src)
+  add_library(${target_name} SHARED ${src})
+  SET_TARGET_PROPERTIES(${target_name} PROPERTIES PREFIX "")
+  target_include_directories(${target_name} PRIVATE ${CMAKE_SOURCE_DIR}/include)
+endfunction()
+
+
 function(copy_e2e_python_files TARGET_PREFIX FILE_NAME)
 add_custom_target(memgraph__e2e__${TARGET_PREFIX}__${FILE_NAME} ALL
         COMMAND ${CMAKE_COMMAND} -E copy
@@ -6,6 +14,14 @@ add_custom_target(memgraph__e2e__${TARGET_PREFIX}__${FILE_NAME} ALL
         DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME})
 endfunction()
 
+function(copy_e2e_cpp_files TARGET_PREFIX FILE_NAME)
+add_custom_target(memgraph__e2e__${TARGET_PREFIX}__${FILE_NAME} ALL
+        COMMAND ${CMAKE_COMMAND} -E copy
+                ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME}
+                ${CMAKE_CURRENT_BINARY_DIR}/${FILE_NAME}
+        DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/${FILE_NAME})
+endfunction()
+
 add_subdirectory(replication)
 add_subdirectory(memory)
 add_subdirectory(triggers)
@@ -13,6 +29,7 @@ add_subdirectory(isolation_levels)
 add_subdirectory(streams)
 add_subdirectory(temporal_types)
 add_subdirectory(write_procedures)
+add_subdirectory(magic_functions)
 add_subdirectory(module_file_manager)
 add_subdirectory(websocket)
 

tests/e2e/isolation_levels/isolation_levels.cpp

@@ -54,7 +54,7 @@ void TestSnapshotIsolation(std::unique_ptr<mg::Client> &client) {
   MG_ASSERT(client->BeginTransaction());
   MG_ASSERT(creator->BeginTransaction());
 
-  constexpr auto vertex_count = 10;
+  static constexpr auto vertex_count = 10;
   for (size_t i = 0; i < vertex_count; ++i) {
     MG_ASSERT(creator->Execute("CREATE ()"));
     creator->DiscardAll();
@@ -87,7 +87,7 @@ void TestReadCommitted(std::unique_ptr<mg::Client> &client) {
   MG_ASSERT(client->BeginTransaction());
   MG_ASSERT(creator->BeginTransaction());
 
-  constexpr auto vertex_count = 10;
+  static constexpr auto vertex_count = 10;
   for (size_t i = 0; i < vertex_count; ++i) {
     MG_ASSERT(creator->Execute("CREATE ()"));
     creator->DiscardAll();
@@ -119,7 +119,7 @@ void TestReadUncommitted(std::unique_ptr<mg::Client> &client) {
   MG_ASSERT(client->BeginTransaction());
   MG_ASSERT(creator->BeginTransaction());
 
-  constexpr auto vertex_count = 10;
+  static constexpr auto vertex_count = 10;
   for (size_t i = 1; i <= vertex_count; ++i) {
     MG_ASSERT(creator->Execute("CREATE ()"));
     creator->DiscardAll();
@@ -142,9 +142,9 @@ void TestReadUncommitted(std::unique_ptr<mg::Client> &client) {
   CleanDatabase();
 }
 
-constexpr std::array isolation_levels{std::pair{"SNAPSHOT ISOLATION", &TestSnapshotIsolation},
-                                      std::pair{"READ COMMITTED", &TestReadCommitted},
-                                      std::pair{"READ UNCOMMITTED", &TestReadUncommitted}};
+inline constexpr std::array isolation_levels{std::pair{"SNAPSHOT ISOLATION", &TestSnapshotIsolation},
+                                             std::pair{"READ COMMITTED", &TestReadCommitted},
+                                             std::pair{"READ UNCOMMITTED", &TestReadUncommitted}};
 
 void TestGlobalIsolationLevel() {
   spdlog::info("\n\n----Test global isolation levels----\n");

tests/e2e/magic_functions/CMakeLists.txt

@@ -0,0 +1,10 @@
+# Set up Python functions for e2e tests
+function(copy_magic_functions_e2e_python_files FILE_NAME)
+    copy_e2e_python_files(functions ${FILE_NAME})
+endfunction()
+
+copy_magic_functions_e2e_python_files(common.py)
+copy_magic_functions_e2e_python_files(conftest.py)
+copy_magic_functions_e2e_python_files(function_example.py)
+
+add_subdirectory(functions)

tests/e2e/magic_functions/common.py

@@ -0,0 +1,35 @@
+# 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(host="localhost", port=7687, **kwargs)
+    connection.autocommit = True
+    return connection
+
+
+def has_n_result_row(cursor: mgclient.Cursor, query: str, n: int):
+    results = execute_and_fetch_all(cursor, query)
+    return len(results) == n
+
+
+def has_one_result_row(cursor: mgclient.Cursor, query: str):
+    return has_n_result_row(cursor, query, 1)

tests/e2e/magic_functions/conftest.py

@@ -0,0 +1,22 @@
+# 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
+
+
+@pytest.fixture(autouse=True)
+def connection():
+    connection = connect()
+    yield connection
+    cursor = connection.cursor()
+    execute_and_fetch_all(cursor, "MATCH (n) DETACH DELETE n")

tests/e2e/magic_functions/function_example.py

@@ -0,0 +1,122 @@
+# 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 typing
+import mgclient
+import sys
+import pytest
+from common import execute_and_fetch_all, has_n_result_row
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_return_argument(connection, function_type):
+    cursor = connection.cursor()
+    execute_and_fetch_all(cursor, "CREATE (n:Label {id: 1});")
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 1)
+    result = execute_and_fetch_all(
+        cursor,
+        f"MATCH (n) RETURN {function_type}_read.return_function_argument(n) AS argument;",
+    )
+    vertex = result[0][0]
+    assert isinstance(vertex, mgclient.Node)
+    assert vertex.labels == set(["Label"])
+    assert vertex.properties == {"id": 1}
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_return_optional_argument(connection, function_type):
+    cursor = connection.cursor()
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 0)
+    result = execute_and_fetch_all(
+        cursor,
+        f"RETURN {function_type}_read.return_optional_argument(42) AS argument;",
+    )
+    result = result[0][0]
+    assert isinstance(result, int)
+    assert result == 42
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_return_optional_argument_no_arg(connection, function_type):
+    cursor = connection.cursor()
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 0)
+    result = execute_and_fetch_all(
+        cursor,
+        f"RETURN {function_type}_read.return_optional_argument() AS argument;",
+    )
+    result = result[0][0]
+    assert isinstance(result, int)
+    assert result == 42
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_add_two_numbers(connection, function_type):
+    cursor = connection.cursor()
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 0)
+    result = execute_and_fetch_all(
+        cursor,
+        f"RETURN {function_type}_read.add_two_numbers(1, 5) AS total;",
+    )
+    result_sum = result[0][0]
+    assert isinstance(result_sum, (float, int))
+    assert result_sum == 6
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_return_null(connection, function_type):
+    cursor = connection.cursor()
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 0)
+    result = execute_and_fetch_all(
+        cursor,
+        f"RETURN {function_type}_read.return_null() AS null;",
+    )
+    result_null = result[0][0]
+    assert result_null is None
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_too_many_arguments(connection, function_type):
+    cursor = connection.cursor()
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 0)
+    # Should raise too many arguments
+    with pytest.raises(mgclient.DatabaseError):
+        execute_and_fetch_all(
+            cursor,
+            f"RETURN {function_type}_read.return_null('parameter') AS null;",
+        )
+
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_try_to_write(connection, function_type):
+    cursor = connection.cursor()
+    execute_and_fetch_all(cursor, "CREATE (n:Label {id: 1});")
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 1)
+    # Should raise non mutable
+    with pytest.raises(mgclient.DatabaseError):
+        execute_and_fetch_all(
+            cursor,
+            f"MATCH (n) RETURN {function_type}_write.try_to_write(n, 'property', 1);",
+        )
+
+@pytest.mark.parametrize("function_type", ["py", "c"])
+def test_case_sensitivity(connection, function_type):
+    cursor = connection.cursor()
+    assert has_n_result_row(cursor, "MATCH (n) RETURN n", 0)
+    # Should raise function does not exist
+    with pytest.raises(mgclient.DatabaseError):
+        execute_and_fetch_all(
+            cursor,
+            f"RETURN {function_type}_read.ReTuRn_nUlL('parameter') AS null;",
+        )
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/magic_functions/functions/CMakeLists.txt

@@ -0,0 +1,5 @@
+copy_magic_functions_e2e_python_files(py_write.py)
+copy_magic_functions_e2e_python_files(py_read.py)
+
+add_query_module(c_read c_read.cpp)
+add_query_module(c_write c_write.cpp)

tests/e2e/magic_functions/functions/c_read.cpp

@@ -0,0 +1,178 @@
+// 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 <functional>
+#include <stdexcept>
+
+#include "mg_procedure.h"
+
+#include "utils/on_scope_exit.hpp"
+
+namespace {
+static void ReturnFunctionArgument(struct mgp_list *args, mgp_func_context *ctx, mgp_func_result *result,
+                                   struct mgp_memory *memory) {
+  mgp_value *value{nullptr};
+  auto err_code = mgp_list_at(args, 0, &value);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to fetch list!", memory);
+    return;
+  }
+
+  err_code = mgp_func_result_set_value(result, value, memory);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to construct return value!", memory);
+    return;
+  }
+}
+
+static void ReturnOptionalArgument(struct mgp_list *args, mgp_func_context *ctx, mgp_func_result *result,
+                                   struct mgp_memory *memory) {
+  mgp_value *value{nullptr};
+  auto err_code = mgp_list_at(args, 0, &value);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to fetch list!", memory);
+    return;
+  }
+
+  err_code = mgp_func_result_set_value(result, value, memory);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to construct return value!", memory);
+    return;
+  }
+}
+
+double GetElementFromArg(struct mgp_list *args, int index) {
+  mgp_value *value{nullptr};
+  if (mgp_list_at(args, index, &value) != MGP_ERROR_NO_ERROR) {
+    throw std::runtime_error("Error while argument fetching.");
+  }
+
+  double result;
+  int is_int;
+  mgp_value_is_int(value, &is_int);
+
+  if (is_int) {
+    int64_t result_int;
+    mgp_value_get_int(value, &result_int);
+    result = static_cast<double>(result_int);
+  } else {
+    mgp_value_get_double(value, &result);
+  }
+  return result;
+}
+
+static void AddTwoNumbers(struct mgp_list *args, mgp_func_context *ctx, mgp_func_result *result,
+                          struct mgp_memory *memory) {
+  double first = 0;
+  double second = 0;
+  try {
+    first = GetElementFromArg(args, 0);
+    second = GetElementFromArg(args, 1);
+  } catch (...) {
+    mgp_func_result_set_error_msg(result, "Unable to fetch the result!", memory);
+    return;
+  }
+
+  mgp_value *value{nullptr};
+  auto summation = first + second;
+  mgp_value_make_double(summation, memory, &value);
+  memgraph::utils::OnScopeExit delete_summation_value([&value] { mgp_value_destroy(value); });
+
+  auto err_code = mgp_func_result_set_value(result, value, memory);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to construct return value!", memory);
+  }
+}
+
+static void ReturnNull(struct mgp_list *args, mgp_func_context *ctx, mgp_func_result *result,
+                       struct mgp_memory *memory) {
+  mgp_value *value{nullptr};
+  mgp_value_make_null(memory, &value);
+  memgraph::utils::OnScopeExit delete_null([&value] { mgp_value_destroy(value); });
+
+  auto err_code = mgp_func_result_set_value(result, value, memory);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to fetch list!", memory);
+  }
+}
+}  // namespace
+
+// Each module needs to define mgp_init_module function.
+// Here you can register multiple functions/procedures your module supports.
+extern "C" int mgp_init_module(struct mgp_module *module, struct mgp_memory *memory) {
+  {
+    mgp_func *func{nullptr};
+    auto err_code = mgp_module_add_function(module, "return_function_argument", ReturnFunctionArgument, &func);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+
+    mgp_type *type_any{nullptr};
+    mgp_type_any(&type_any);
+    err_code = mgp_func_add_arg(func, "argument", type_any);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+  }
+
+  {
+    mgp_func *func{nullptr};
+    auto err_code = mgp_module_add_function(module, "return_optional_argument", ReturnOptionalArgument, &func);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+
+    mgp_value *default_value{nullptr};
+    mgp_value_make_int(42, memory, &default_value);
+    memgraph::utils::OnScopeExit delete_summation_value([&default_value] { mgp_value_destroy(default_value); });
+
+    mgp_type *type_int{nullptr};
+    mgp_type_int(&type_int);
+    err_code = mgp_func_add_opt_arg(func, "opt_argument", type_int, default_value);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+  }
+
+  {
+    mgp_func *func{nullptr};
+    auto err_code = mgp_module_add_function(module, "add_two_numbers", AddTwoNumbers, &func);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+
+    mgp_type *type_number{nullptr};
+    mgp_type_number(&type_number);
+    err_code = mgp_func_add_arg(func, "first", type_number);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+    err_code = mgp_func_add_arg(func, "second", type_number);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+  }
+
+  {
+    mgp_func *func{nullptr};
+    auto err_code = mgp_module_add_function(module, "return_null", ReturnNull, &func);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+  }
+
+  return 0;
+}
+
+// This is an optional function if you need to release any resources before the
+// module is unloaded. You will probably need this if you acquired some
+// resources in mgp_init_module.
+extern "C" int mgp_shutdown_module() { return 0; }

tests/e2e/magic_functions/functions/c_write.cpp

@@ -0,0 +1,80 @@
+// 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 "mg_procedure.h"
+
+static void TryToWrite(struct mgp_list *args, mgp_func_context *ctx, mgp_func_result *result,
+                       struct mgp_memory *memory) {
+  mgp_value *value{nullptr};
+  mgp_vertex *vertex{nullptr};
+  mgp_list_at(args, 0, &value);
+  mgp_value_get_vertex(value, &vertex);
+
+  const char *name;
+  mgp_list_at(args, 1, &value);
+  mgp_value_get_string(value, &name);
+
+  mgp_list_at(args, 2, &value);
+
+  // Setting a property should set an error
+  auto err_code = mgp_vertex_set_property(vertex, name, value);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Cannot set property in the function!", memory);
+    return;
+  }
+
+  err_code = mgp_func_result_set_value(result, value, memory);
+  if (err_code != MGP_ERROR_NO_ERROR) {
+    mgp_func_result_set_error_msg(result, "Failed to construct return value!", memory);
+    return;
+  }
+}
+
+// Each module needs to define mgp_init_module function.
+// Here you can register multiple functions/procedures your module supports.
+extern "C" int mgp_init_module(struct mgp_module *module, struct mgp_memory *memory) {
+  {
+    mgp_func *func{nullptr};
+    auto err_code = mgp_module_add_function(module, "try_to_write", TryToWrite, &func);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+
+    mgp_type *type_vertex{nullptr};
+    mgp_type_node(&type_vertex);
+    err_code = mgp_func_add_arg(func, "argument", type_vertex);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+
+    mgp_type *type_string{nullptr};
+    mgp_type_string(&type_string);
+    err_code = mgp_func_add_arg(func, "name", type_string);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+
+    mgp_type *any_type{nullptr};
+    mgp_type_any(&any_type);
+    mgp_type *nullable_type{nullptr};
+    mgp_type_nullable(any_type, &nullable_type);
+    err_code = mgp_func_add_arg(func, "value", nullable_type);
+    if (err_code != MGP_ERROR_NO_ERROR) {
+      return 1;
+    }
+  }
+  return 0;
+}
+
+// This is an optional function if you need to release any resources before the
+// module is unloaded. You will probably need this if you acquired some
+// resources in mgp_init_module.
+extern "C" int mgp_shutdown_module() { return 0; }

tests/e2e/magic_functions/functions/py_read.py

@@ -0,0 +1,32 @@
+# 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 mgp
+
+
+@mgp.function
+def return_function_argument(ctx: mgp.FuncCtx, argument: mgp.Any):
+    return argument
+
+
+@mgp.function
+def return_optional_argument(ctx: mgp.FuncCtx, opt_argument: mgp.Number = 42):
+    return opt_argument
+
+
+@mgp.function
+def add_two_numbers(ctx: mgp.FuncCtx, first: mgp.Number, second: mgp.Number):
+    return first + second
+
+
+@mgp.function
+def return_null(ctx: mgp.FuncCtx):
+    return None

tests/e2e/magic_functions/functions/py_write.py

@@ -0,0 +1,17 @@
+# 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 mgp
+
+
+@mgp.function
+def try_to_write(ctx: mgp.FuncCtx, argument: mgp.Vertex, name: str, value: mgp.Nullable[mgp.Any]):
+    argument.properties.set(name, value)