Support Ubuntu 22.04 (#411 )

* Fix toolchain url * Add 22.04 os script * Update centos-9 stream * Update zlib * Upgrade dotnet version * Add libipt * Update toolchain extension * Add release container * Remove CentOS 8 * Enable arm packaging * Add centos 9 container * Ignore missing python virtualenv * Add virtualenvwrapper * Rearrange order * Install virtualenv without user * Run virtualenv via python * Add logrotate dependency * Bump up checkout action version
Update version and date
2022-06-23 14:09:00 +02:00 · 2022-06-15 23:06:38 +02:00 · 2022-06-15 12:16:11 +03:00 · 2022-06-10 11:26:24 +02:00 · 2022-06-09 22:19:43 +02:00 · 2022-06-09 18:55:28 +02:00
53 changed files with 1050 additions and 748 deletions
--- a/.clang-tidy
+++ b/.clang-tidy
@@ -88,4 +88,3 @@ CheckOptions:
  - key:             modernize-use-nullptr.NullMacros
    value:           'NULL'
 ...
-
--- a/.github/workflows/diff.yaml
+++ b/.github/workflows/diff.yaml
@@ -112,7 +112,7 @@ jobs:
          source /opt/toolchain-v4/activate

          # Restrict clang-tidy results only to the modified parts
-          git diff -U0 master... -- src ':!*.hpp' | ./tools/github/clang-tidy/clang-tidy-diff.py -p 1 -j $THREADS -path build | tee ./build/clang_tidy_output.txt
+          git diff -U0 master... -- src | ./tools/github/clang-tidy/clang-tidy-diff.py -p 1 -j $THREADS -path build | tee ./build/clang_tidy_output.txt

          # Fail if any warning is reported
          ! cat ./build/clang_tidy_output.txt | ./tools/github/clang-tidy/grep_error_lines.sh > /dev/null
--- a/.github/workflows/package_all.yaml
+++ b/.github/workflows/package_all.yaml
@@ -6,11 +6,11 @@ on: workflow_dispatch

 jobs:
  centos-7:
-    runs-on: [self-hosted, DockerMgBuild]
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -22,29 +22,29 @@ jobs:
          name: centos-7
          path: build/output/centos-7/memgraph*.rpm

-  centos-8:
-    runs-on: [self-hosted, DockerMgBuild]
+  centos-9:
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
        run: |
-          ./release/package/run.sh package centos-8
+          ./release/package/run.sh package centos-9
      - name: "Upload package"
        uses: actions/upload-artifact@v2
        with:
-          name: centos-8
-          path: build/output/centos-8/memgraph*.rpm
+          name: centos-9
+          path: build/output/centos-9/memgraph*.rpm

  debian-10:
-    runs-on: [self-hosted, DockerMgBuild]
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -57,11 +57,11 @@ jobs:
          path: build/output/debian-10/memgraph*.deb

  debian-11:
-    runs-on: [self-hosted, DockerMgBuild]
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -74,11 +74,11 @@ jobs:
          path: build/output/debian-11/memgraph*.deb

  docker:
-    runs-on: [self-hosted, DockerMgBuild]
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -93,11 +93,11 @@ jobs:
          path: build/output/docker/memgraph*.tar.gz

  ubuntu-1804:
-    runs-on: [self-hosted, DockerMgBuild]
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -110,11 +110,11 @@ jobs:
          path: build/output/ubuntu-18.04/memgraph*.deb

  ubuntu-2004:
-    runs-on: [self-hosted, DockerMgBuild]
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -126,12 +126,29 @@ jobs:
          name: ubuntu-2004
          path: build/output/ubuntu-20.04/memgraph*.deb

-  debian-11-platform:
-    runs-on: [self-hosted, DockerMgBuild]
+  ubuntu-2204:
+    runs-on: [self-hosted, DockerMgBuild, X64]
    timeout-minutes: 60
    steps:
      - name: "Set up repository"
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0 # Required because of release/get_version.py
+      - name: "Build package"
+        run: |
+          ./release/package/run.sh package ubuntu-22.04
+      - name: "Upload package"
+        uses: actions/upload-artifact@v2
+        with:
+          name: ubuntu-2204
+          path: build/output/ubuntu-22.04/memgraph*.deb
+
+  debian-11-platform:
+    runs-on: [self-hosted, DockerMgBuild, X64]
+    timeout-minutes: 60
+    steps:
+      - name: "Set up repository"
+        uses: actions/checkout@v3
        with:
          fetch-depth: 0 # Required because of release/get_version.py
      - name: "Build package"
@@ -142,3 +159,20 @@ jobs:
        with:
          name: debian-11-platform
          path: build/output/debian-11/memgraph*.deb
+
+  debian-11-arm:
+    runs-on: [self-hosted, DockerMgBuild, ARM64, strange]
+    timeout-minutes: 60
+    steps:
+      - name: "Set up repository"
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0 # Required because of release/get_version.py
+      - name: "Build package"
+        run: |
+          ./release/package/run.sh package debian-11-arm
+      - name: "Upload package"
+        uses: actions/upload-artifact@v2
+        with:
+          name: debian-11
+          path: build/output/debian-11/memgraph*.deb
--- a/CMakeLists.txt
+++ b/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 "")
+set(MEMGRAPH_OVERRIDE_VERSION "2.3.1")

 # Custom suffix that this version should have. The suffix can be any arbitrary
 # string. Primarily used when building a version for a specific customer.
--- a/environment/os/centos-8.sh
+++ b/environment/os/centos-8.sh
@@ -1,161 +0,0 @@
-#!/bin/bash
-
-set -Eeuo pipefail
-
-DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
-source "$DIR/../util.sh"
-
-TOOLCHAIN_BUILD_DEPS=(
-    coreutils-common gcc gcc-c++ make # generic build tools
-    wget # used for archive download
-    gnupg2 # used for archive signature verification
-    tar gzip bzip2 xz unzip # used for archive unpacking
-    zlib-devel # zlib library used for all builds
-    expat-devel libipt libipt-devel libbabeltrace-devel xz-devel python36-devel texinfo # for gdb
-    libcurl-devel # for cmake
-    curl # snappy
-    readline-devel # for cmake and llvm
-    libffi-devel libxml2-devel # for llvm
-    libedit-devel pcre-devel automake bison # for swig
-    file
-    openssl-devel
-    gmp-devel
-    gperf
-    patch
-)
-
-TOOLCHAIN_RUN_DEPS=(
-    make # generic build tools
-    tar gzip bzip2 xz # used for archive unpacking
-    zlib # zlib library used for all builds
-    expat libipt libbabeltrace xz-libs python36 # for gdb
-    readline # for cmake and llvm
-    libffi libxml2 # for llvm
-    openssl-devel
-)
-
-MEMGRAPH_BUILD_DEPS=(
-    git # source code control
-    make pkgconf-pkg-config # build system
-    curl wget # for downloading libs
-    libuuid-devel java-11-openjdk # required by antlr
-    readline-devel # for memgraph console
-    python36-devel # for query modules
-    openssl-devel
-    libseccomp-devel
-    python36 python3-virtualenv python3-pip nmap-ncat # for qa, macro_benchmark and stress tests
-    #
-    # IMPORTANT: python3-yaml does NOT exist on CentOS
-    # Install it manually using `pip3 install PyYAML`
-    #
-    PyYAML # Package name here does not correspond to the yum package!
-    libcurl-devel # mg-requests
-    rpm-build rpmlint # for RPM package building
-    doxygen graphviz # source documentation generators
-    which mono-complete dotnet-sdk-3.1 nodejs golang zip unzip java-11-openjdk-devel # for driver tests
-    sbcl # for custom Lisp C++ preprocessing
-    autoconf # for jemalloc code generation
-    libtool  # for protobuf code generation
-)
-
-list() {
-    echo "$1"
-}
-
-check() {
-    local missing=""
-    for pkg in $1; do
-        if [ "$pkg" == "PyYAML" ]; then
-            if ! python3 -c "import yaml" >/dev/null 2>/dev/null; then
-                missing="$pkg $missing"
-            fi
-            continue
-        fi
-        if ! yum list installed "$pkg" >/dev/null 2>/dev/null; then
-            missing="$pkg $missing"
-        fi
-    done
-    if [ "$missing" != "" ]; then
-        echo "MISSING PACKAGES: $missing"
-        exit 1
-    fi
-}
-
-install() {
-    cd "$DIR"
-    if [ "$EUID" -ne 0 ]; then
-        echo "Please run as root."
-        exit 1
-    fi
-    # 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
-    dnf install -y epel-release
-    dnf install -y 'dnf-command(config-manager)'
-    dnf config-manager --set-enabled powertools # Required to install texinfo.
-    dnf update -y
-    dnf install -y wget git python36 python3-pip
-    for pkg in $1; do
-        if [ "$pkg" == libipt ]; then
-            if ! dnf list installed libipt >/dev/null 2>/dev/null; then
-                dnf install -y http://repo.okay.com.mx/centos/8/x86_64/release/libipt-1.6.1-8.el8.x86_64.rpm
-            fi
-            continue
-        fi
-        if [ "$pkg" == libipt-devel ]; then
-            if ! yum list installed libipt-devel >/dev/null 2>/dev/null; then
-                dnf install -y http://repo.okay.com.mx/centos/8/x86_64/release/libipt-devel-1.6.1-8.el8.x86_64.rpm
-            fi
-            continue
-        fi
-        # Install GDB dependencies not present in the standard repos.
-        # https://bugs.centos.org/view.php?id=17068
-        # https://centos.pkgs.org
-	    # Since 2020, there is Babeltrace2 (https://babeltrace.org). Not used
-        # within GDB yet (an assumption).
-        # http://mirror.centos.org/centos/8/PowerTools/x86_64/os/Packages/libbabeltrace-devel-1.5.4-3.el8.x86_64.rpm not working
-        if [ "$pkg" == libbabeltrace-devel ]; then
-            if ! dnf list installed libbabeltrace-devel >/dev/null 2>/dev/null; then
-                dnf install -y https://rpmfind.net/linux/centos/8-stream/PowerTools/x86_64/os/Packages/libbabeltrace-devel-1.5.4-3.el8.x86_64.rpm
-            fi
-            continue
-        fi
-        if [ "$pkg" == sbcl ]; then
-            if ! dnf list installed cl-asdf >/dev/null 2>/dev/null; then
-                dnf install -y 	https://pkgs.dyn.su/el8/base/x86_64/cl-asdf-20101028-18.el8.noarch.rpm
-            fi
-            if ! dnf list installed common-lisp-controller >/dev/null 2>/dev/null; then
-                dnf install -y https://pkgs.dyn.su/el8/base/x86_64/common-lisp-controller-7.4-20.el8.noarch.rpm
-            fi
-            if ! dnf list installed sbcl >/dev/null 2>/dev/null; then
-                dnf install -y https://pkgs.dyn.su/el8/base/x86_64/sbcl-2.0.1-4.el8.x86_64.rpm
-            fi
-            continue
-        fi
-        if [ "$pkg" == dotnet-sdk-3.1 ]; then
-            if ! dnf list installed dotnet-sdk-3.1 >/dev/null 2>/dev/null; then
-                wget -nv https://packages.microsoft.com/config/centos/8/packages-microsoft-prod.rpm -O packages-microsoft-prod.rpm
-                rpm -Uvh https://packages.microsoft.com/config/centos/8/packages-microsoft-prod.rpm
-                dnf update -y
-                dnf install -y dotnet-sdk-3.1
-            fi
-            continue
-        fi
-        if [ "$pkg" == PyYAML ]; then
-            if [ -z ${SUDO_USER+x} ]; then # Running as root (e.g. Docker).
-                pip3 install --user PyYAML
-            else # Running using sudo.
-                sudo -H -u "$SUDO_USER" bash -c "pip3 install --user PyYAML"
-            fi
-            continue
-        fi
-        dnf install -y "$pkg"
-    done
-}
-
-deps=$2"[*]"
-"$1" "${!deps}"
--- a/environment/os/centos-9.sh
+++ b/environment/os/centos-9.sh
@@ -6,14 +6,12 @@ DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 source "$DIR/../util.sh"

 TOOLCHAIN_BUILD_DEPS=(
-    coreutils gcc gcc-c++ make # generic build tools
+    coreutils-common gcc gcc-c++ make # generic build tools
    wget # used for archive download
    gnupg2 # used for archive signature verification
    tar gzip bzip2 xz unzip # used for archive unpacking
    zlib-devel # zlib library used for all builds
-    expat-devel xz-devel python3-devel texinfo # for gdb
-    libcurl-devel # for cmake
-    curl # snappy
+    expat-devel xz-devel python3-devel texinfo libbabeltrace-devel # for gdb
    readline-devel # for cmake and llvm
    libffi-devel libxml2-devel # for llvm
    libedit-devel pcre-devel automake bison # for swig
@@ -21,6 +19,9 @@ TOOLCHAIN_BUILD_DEPS=(
    openssl-devel
    gmp-devel
    gperf
+    diffutils
+    libipt libipt-devel # intel
+    patch
 )

 TOOLCHAIN_RUN_DEPS=(
@@ -31,18 +32,19 @@ TOOLCHAIN_RUN_DEPS=(
    readline # for cmake and llvm
    libffi libxml2 # for llvm
    openssl-devel
+    perl # for openssl
 )

 MEMGRAPH_BUILD_DEPS=(
    git # source code control
    make pkgconf-pkg-config # build system
-    curl wget # for downloading libs
+    wget # for downloading libs
    libuuid-devel java-11-openjdk # required by antlr
    readline-devel # for memgraph console
    python3-devel # for query modules
    openssl-devel
    libseccomp-devel
-    python3 python3-virtualenv python3-pip nmap-ncat # for qa, macro_benchmark and stress tests
+    python3 python3-pip python3-virtualenv nmap-ncat # for qa, macro_benchmark and stress tests
    #
    # IMPORTANT: python3-yaml does NOT exist on CentOS
    # Install it manually using `pip3 install PyYAML`
@@ -73,12 +75,6 @@ check() {
        if [ "$pkg" == "python3-virtualenv" ]; then
            continue
        fi
-        if [ "$pkg" == sbcl ]; then
-            if ! sbcl --version &> /dev/null; then
-	        missing="$pkg $missing"
-            fi
-            continue
-        fi
        if ! yum list installed "$pkg" >/dev/null 2>/dev/null; then
            missing="$pkg $missing"
        fi
@@ -105,13 +101,37 @@ install() {
    yum update -y
    yum install -y wget git python3 python3-pip
    for pkg in $1; do
+        # Since there is no support for libipt-devel for CentOS 9 we install
+        # Fedoras version of same libs, they are the same version but released
+        # for different OS
+        # TODO Update when libipt-devel releases for CentOS 9
+        if [ "$pkg" == libipt ]; then
+            if ! dnf list installed libipt >/dev/null 2>/dev/null; then
+                dnf install -y http://repo.okay.com.mx/centos/8/x86_64/release/libipt-1.6.1-8.el8.x86_64.rpm
+            fi
+            continue
+        fi
+        if [ "$pkg" == libipt-devel ]; then
+            if ! dnf list installed libipt-devel >/dev/null 2>/dev/null; then
+                dnf install -y http://repo.okay.com.mx/centos/8/x86_64/release/libipt-devel-1.6.1-8.el8.x86_64.rpm
+            fi
+            continue
+        fi
+        if [ "$pkg" == libbabeltrace-devel ]; then
+            if ! dnf list installed libbabeltrace-devel >/dev/null 2>/dev/null; then
+                dnf install -y http://mirror.stream.centos.org/9-stream/CRB/x86_64/os/Packages/libbabeltrace-devel-1.5.8-10.el9.x86_64.rpm
+            fi
+            continue
+        fi
        if [ "$pkg" == sbcl ]; then
-            if ! sbcl --version &> /dev/null; then
-	        curl -s https://altushost-swe.dl.sourceforge.net/project/sbcl/sbcl/1.4.2/sbcl-1.4.2-arm64-linux-binary.tar.bz2 -o /tmp/sbcl-arm64.tar.bz2
-		tar xvjf /tmp/sbcl-arm64.tar.bz2 -C /tmp
-		pushd /tmp/sbcl-1.4.2-arm64-linux
-		INSTALL_ROOT=/usr/local sh install.sh
-		popd
+            if ! dnf list installed cl-asdf >/dev/null 2>/dev/null; then
+                dnf install -y 	https://pkgs.dyn.su/el8/base/x86_64/cl-asdf-20101028-18.el8.noarch.rpm
+            fi
+            if ! dnf list installed common-lisp-controller >/dev/null 2>/dev/null; then
+                dnf install -y https://pkgs.dyn.su/el8/base/x86_64/common-lisp-controller-7.4-20.el8.noarch.rpm
+            fi
+            if ! dnf list installed sbcl >/dev/null 2>/dev/null; then
+                dnf install -y https://pkgs.dyn.su/el8/base/x86_64/sbcl-2.0.1-4.el8.x86_64.rpm
            fi
            continue
        fi
@@ -125,9 +145,11 @@ install() {
        fi
        if [ "$pkg" == python3-virtualenv ]; then
            if [ -z ${SUDO_USER+x} ]; then # Running as root (e.g. Docker).
-                pip3 install --user virtualenv
+                pip3 install virtualenv
+                pip3 install virtualenvwrapper
            else # Running using sudo.
-                sudo -H -u "$SUDO_USER" bash -c "pip3 install --user virtualenv"
+                sudo -H -u "$SUDO_USER" bash -c "pip3 install virtualenv"
+                sudo -H -u "$SUDO_USER" bash -c "pip3 install virtualenvwrapper"
            fi
            continue
        fi
--- a/environment/os/debian-11-arm.sh
+++ b/environment/os/debian-11-arm.sh
@@ -87,15 +87,6 @@ EOF
    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
 }
--- a/environment/os/debian-11.sh
+++ b/environment/os/debian-11.sh
@@ -87,7 +87,7 @@ EOF
    fi
    apt install -y wget
    for pkg in $1; do
-        if [ "$pkg" == dotnet-sdk-3.1 ]; then
+        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
--- a/environment/os/ubuntu-22.04.sh
+++ b/environment/os/ubuntu-22.04.sh
@@ -0,0 +1,93 @@
+#!/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 libipt-dev libbabeltrace-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
+    curl # snappy
+    file
+    git # for thrift
+    libgmp-dev # for gdb
+    gperf # for proxygen
+    libssl-dev
+    libedit-dev libpcre3-dev automake bison # for swig
+)
+
+TOOLCHAIN_RUN_DEPS=(
+    make # generic build tools
+    tar gzip bzip2 xz-utils # used for archive unpacking
+    zlib1g # zlib library used for all builds
+    libexpat1 libipt2 libbabeltrace1 liblzma5 python3 # for gdb
+    libcurl4 # for cmake
+    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 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
+    dotnet-sdk-6.0 golang nodejs npm
+    autoconf # for jemalloc code generation
+    libtool  # for protobuf code generation
+)
+
+list() {
+    echo "$1"
+}
+
+check() {
+    check_all_dpkg "$1"
+}
+
+install() {
+    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-6.0 ]; then
+            if ! dpkg -s dotnet-sdk-6.0 2>/dev/null >/dev/null; then
+                wget -nv https://packages.microsoft.com/config/ubuntu/22.04/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-6.0
+            fi
+            continue
+        fi
+        apt install -y "$pkg"
+    done
+}
+
+deps=$2"[*]"
+"$1" "${!deps}"
--- a/environment/toolchain/v4.sh
+++ b/environment/toolchain/v4.sh
@@ -675,7 +675,7 @@ PROXYGEN_SHA256=5360a8ccdfb2f5a6c7b3eed331ec7ab0e2c792d579c6fff499c85c516c11fe14
 SNAPPY_SHA256=75c1fbb3d618dd3a0483bff0e26d0a92b495bbe5059c8b4f1c962b478b6e06e7
 SNAPPY_VERSION=1.1.9
 XZ_VERSION=5.2.5 # for LZMA
-ZLIB_VERSION=1.2.11
+ZLIB_VERSION=1.2.12
 ZSTD_VERSION=1.5.0
 WANGLE_SHA256=1002e9c32b6f4837f6a760016e3b3e22f3509880ef3eaad191c80dc92655f23f

@@ -1178,12 +1178,21 @@ popd

 # create toolchain archive
 if [ ! -f $NAME-binaries-$DISTRO.tar.gz ]; then
-    DISTRO_FULL_NAME=$DISTRO
-    if [ "$for_arm" = true ]; then
-        DISTRO_FULL_NAME="$DISTRO_FULL_NAME-aarch64"
+    DISTRO_FULL_NAME=${DISTRO}
+    if [[ "${DISTRO}" == centos* ]]; then
+        if [[ "$for_arm" = "true" ]]; then
+            DISTRO_FULL_NAME="$DISTRO_FULL_NAME-aarch64"
+        else
+            DISTRO_FULL_NAME="$DISTRO_FULL_NAME-x86_64"
+        fi
    else
-        DISTRO_FULL_NAME="$DISTRO_FULL_NAME-x86_64"
+        if [[ "$for_arm" = "true" ]]; then
+            DISTRO_FULL_NAME="$DISTRO_FULL_NAME-arm64"
+        else
+            DISTRO_FULL_NAME="$DISTRO_FULL_NAME-amd64"
+        fi
    fi
+
    tar --owner=root --group=root -cpvzf $NAME-binaries-$DISTRO_FULL_NAME.tar.gz -C /opt $NAME
 fi

--- a/4
+++ b/4
@@ -24,7 +24,7 @@ function setup_virtualenv () {
    fi

    # create new virtualenv
-    virtualenv -p python3 ve3 || exit 1
+    python3 -m virtualenv -p python3 ve3 || exit 1
    source ve3/bin/activate
    pip --timeout 1000 install -r requirements.txt || exit 1
    deactivate
@@ -66,7 +66,7 @@ fi

 DISTRO=$(operating_system)
 ARCHITECTURE=$(architecture)
-if [ "${ARCHITECTURE}" = "arm64" ]; then
+if [ "${ARCHITECTURE}" = "arm64" ] || [ "${ARCHITECTURE}" = "aarch64" ]; then
  OS_SCRIPT=$DIR/environment/os/$DISTRO-arm.sh
 else
  OS_SCRIPT=$DIR/environment/os/$DISTRO.sh
--- a/libs/.gitignore
+++ b/libs/.gitignore
@@ -4,5 +4,4 @@
 !cleanup.sh
 !CMakeLists.txt
 !__main.cpp
-!jemalloc.cmake
 !pulsar.patch
--- a/libs/jemalloc.cmake
+++ b/libs/jemalloc.cmake
@@ -1,55 +0,0 @@
-set(JEMALLOC_DIR "${LIB_DIR}/jemalloc")
-
-set(JEMALLOC_SRCS
-    ${JEMALLOC_DIR}/src/arena.c
-    ${JEMALLOC_DIR}/src/background_thread.c
-    ${JEMALLOC_DIR}/src/base.c
-    ${JEMALLOC_DIR}/src/bin.c
-    ${JEMALLOC_DIR}/src/bitmap.c
-    ${JEMALLOC_DIR}/src/ckh.c
-    ${JEMALLOC_DIR}/src/ctl.c
-    ${JEMALLOC_DIR}/src/div.c
-    ${JEMALLOC_DIR}/src/extent.c
-    ${JEMALLOC_DIR}/src/extent_dss.c
-    ${JEMALLOC_DIR}/src/extent_mmap.c
-    ${JEMALLOC_DIR}/src/hash.c
-    ${JEMALLOC_DIR}/src/hook.c
-    ${JEMALLOC_DIR}/src/jemalloc.c
-    ${JEMALLOC_DIR}/src/large.c
-    ${JEMALLOC_DIR}/src/log.c
-    ${JEMALLOC_DIR}/src/malloc_io.c
-    ${JEMALLOC_DIR}/src/mutex.c
-    ${JEMALLOC_DIR}/src/mutex_pool.c
-    ${JEMALLOC_DIR}/src/nstime.c
-    ${JEMALLOC_DIR}/src/pages.c
-    ${JEMALLOC_DIR}/src/prng.c
-    ${JEMALLOC_DIR}/src/prof.c
-    ${JEMALLOC_DIR}/src/rtree.c
-    ${JEMALLOC_DIR}/src/sc.c
-    ${JEMALLOC_DIR}/src/stats.c
-    ${JEMALLOC_DIR}/src/sz.c
-    ${JEMALLOC_DIR}/src/tcache.c
-    ${JEMALLOC_DIR}/src/test_hooks.c
-    ${JEMALLOC_DIR}/src/ticker.c
-    ${JEMALLOC_DIR}/src/tsd.c
-    ${JEMALLOC_DIR}/src/witness.c
-    ${JEMALLOC_DIR}/src/safety_check.c
-)
-
-add_library(jemalloc ${JEMALLOC_SRCS})
-target_include_directories(jemalloc PUBLIC "${JEMALLOC_DIR}/include")
-
-find_package(Threads REQUIRED)
-target_link_libraries(jemalloc PUBLIC Threads::Threads)
-
-target_compile_definitions(jemalloc PRIVATE -DJEMALLOC_NO_PRIVATE_NAMESPACE)
-
-if (CMAKE_BUILD_TYPE STREQUAL "DEBUG")
-    target_compile_definitions(jemalloc PRIVATE -DJEMALLOC_DEBUG=1 -DJEMALLOC_PROF=1)
-endif()
-
-target_compile_options(jemalloc PRIVATE -Wno-redundant-decls)
-# for RTLD_NEXT
-target_compile_definitions(jemalloc PRIVATE _GNU_SOURCE)
-
-set_property(TARGET jemalloc APPEND PROPERTY INTERFACE_COMPILE_DEFINITIONS USE_JEMALLOC=1)
--- a/licenses/BSL.txt
+++ b/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-27-04
+CHANGE DATE:            2026-16-06
 CHANGE LICENSE:         Apache License, Version 2.0

 For information about alternative licensing arrangements, please visit: https://memgraph.com/legal.
--- a/release/CMakeLists.txt
+++ b/release/CMakeLists.txt
@@ -67,7 +67,7 @@ It aims to deliver developers the speed, simplicity and scale required to build
 the next generation of applications driver by real-time connected data.")
 # Add `openssl` package to dependencies list. Used to generate SSL certificates.
 # We also depend on `python3` because we embed it in Memgraph.
-set(CPACK_RPM_PACKAGE_REQUIRES "openssl >= 1.0.0, curl >= 7.29.0, python3 >= 3.5.0, libstdc >= 6")
+set(CPACK_RPM_PACKAGE_REQUIRES "openssl >= 1.0.0, curl >= 7.29.0, python3 >= 3.5.0, libstdc >= 6, logrotate")

 # All variables must be set before including.
 include(CPack)
--- a/release/docker/memgraph_rpm.dockerfile
+++ b/release/docker/memgraph_rpm.dockerfile
@@ -1,31 +0,0 @@
-FROM dokken/centos-stream-9
-# NOTE: If you change the base distro update release/package as well.
-
-ARG release
-
-RUN yum update && yum install -y \
-    openssl libcurl libseccomp python3 python3-pip \
-    --nobest --allowerasing \
-  && rm -rf /tmp/* \
-  && yum clean all
-
-RUN pip3 install networkx==2.4 numpy==1.21.4 scipy==1.7.3
-
-COPY ${release} /
-
-# Install memgraph package
-RUN rpm -i ${release}
-
-# Memgraph listens for Bolt Protocol on this port by default.
-EXPOSE 7687
-# Snapshots and logging volumes
-VOLUME /var/log/memgraph
-VOLUME /var/lib/memgraph
-# Configuration volume
-VOLUME /etc/memgraph
-
-USER memgraph
-WORKDIR /usr/lib/memgraph
-
-ENTRYPOINT ["/usr/lib/memgraph/memgraph"]
-CMD [""]
--- a/release/package/centos-7/Dockerfile
+++ b/release/package/centos-7/Dockerfile
@@ -7,8 +7,8 @@ RUN yum -y update \
 # Do NOT be smart here and clean the cache because the container is used in the
 # stateful context.

-RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-centos-7.tar.gz \
-    -O ${TOOLCHAIN_VERSION}-binaries-centos-7.tar.gz \
-    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-centos-7.tar.gz -C /opt
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-centos-7-x86_64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-centos-7-x86_64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-centos-7-x86_64.tar.gz -C /opt

 ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/centos-8/Dockerfile
+++ b/release/package/centos-8/Dockerfile
@@ -1,14 +0,0 @@
-FROM centos:8
-
-ARG TOOLCHAIN_VERSION
-
-RUN dnf -y update \
-    && dnf install -y wget git
-# Do NOT be smart here and clean the cache because the container is used in the
-# stateful context.
-
-RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-centos-8.tar.gz \
-    -O ${TOOLCHAIN_VERSION}-binaries-centos-8.tar.gz \
-    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-centos-8.tar.gz -C /opt
-
-ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/centos-9/Dockerfile
+++ b/release/package/centos-9/Dockerfile
@@ -0,0 +1,14 @@
+FROM quay.io/centos/centos:stream9
+
+ARG TOOLCHAIN_VERSION
+
+RUN yum -y update \
+    && yum install -y wget git
+# Do NOT be smart here and clean the cache because the container is used in the
+# stateful context.
+
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-centos-9-x86_64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-centos-9-x86_64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-centos-9-x86_64.tar.gz -C /opt
+
+ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/debian-10/Dockerfile
+++ b/release/package/debian-10/Dockerfile
@@ -10,8 +10,8 @@ RUN apt update && apt install -y \
 # Do NOT be smart here and clean the cache because the container is used in the
 # stateful context.

-RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-debian-10.tar.gz \
-    -O ${TOOLCHAIN_VERSION}-binaries-debian-10.tar.gz \
-    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-debian-10.tar.gz -C /opt
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-debian-10-amd64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-debian-10-amd64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-debian-10-amd64.tar.gz -C /opt

 ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/debian-11-arm/Dockerfile
+++ b/release/package/debian-11-arm/Dockerfile
@@ -0,0 +1,17 @@
+FROM debian:11
+
+ARG TOOLCHAIN_VERSION
+
+# Stops tzdata interactive configuration.
+ENV DEBIAN_FRONTEND=noninteractive
+
+RUN apt update && apt install -y \
+    ca-certificates wget git
+# Do NOT be smart here and clean the cache because the container is used in the
+# stateful context.
+
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-debian-11-arm64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-debian-11-arm64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-debian-11-arm64.tar.gz -C /opt
+
+ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/debian-11/Dockerfile
+++ b/release/package/debian-11/Dockerfile
@@ -10,8 +10,8 @@ RUN apt update && apt install -y \
 # Do NOT be smart here and clean the cache because the container is used in the
 # stateful context.

-RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-debian-11.tar.gz \
-    -O ${TOOLCHAIN_VERSION}-binaries-debian-11.tar.gz \
-    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-debian-11.tar.gz -C /opt
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-debian-11-amd64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-debian-11-amd64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-debian-11-amd64.tar.gz -C /opt

 ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/docker-compose.yml
+++ b/release/package/docker-compose.yml
@@ -4,10 +4,10 @@ services:
    build:
      context: centos-7
    container_name: "mgbuild_centos-7"
-  mgbuild_centos-8:
+  mgbuild_centos-9:
    build:
-      context: centos-8
-    container_name: "mgbuild_centos-8"
+      context: centos-9
+    container_name: "mgbuild_centos-9"
  mgbuild_debian-10:
    build:
      context: debian-10
@@ -24,3 +24,7 @@ services:
    build:
      context: ubuntu-20.04
    container_name: "mgbuild_ubuntu-20.04"
+  mgbuild_ubuntu-22.04:
+    build:
+      context: ubuntu-22.04
+    container_name: "mgbuild_ubuntu-22.04"
--- a/release/package/run.sh
+++ b/release/package/run.sh
@@ -3,7 +3,7 @@
 set -Eeuo pipefail

 SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
-SUPPORTED_OS=(centos-7 centos-8 debian-10 debian-11 ubuntu-18.04 ubuntu-20.04)
+SUPPORTED_OS=(centos-7 centos-9 debian-10 debian-11 ubuntu-18.04 ubuntu-20.04 ubuntu-22.04 debian-11-arm)
 PROJECT_ROOT="$SCRIPT_DIR/../.."
 TOOLCHAIN_VERSION="toolchain-v4"
 ACTIVATE_TOOLCHAIN="source /opt/${TOOLCHAIN_VERSION}/activate"
@@ -67,14 +67,18 @@ make_package () {
    # environment/os/{os}.sh does not come within the toolchain package. When
    # migrating to the next version of toolchain do that, and remove the
    # TOOLCHAIN_RUN_DEPS installation from here.
-    echo "Installing dependencies..."
+    echo "Installing dependencies using '/memgraph/environment/os/$os.sh' script..."
    docker exec "$build_container" bash -c "/memgraph/environment/os/$os.sh install TOOLCHAIN_RUN_DEPS"
    docker exec "$build_container" bash -c "/memgraph/environment/os/$os.sh install MEMGRAPH_BUILD_DEPS"

    echo "Building targeted package..."
    docker exec "$build_container" bash -c "cd /memgraph && $ACTIVATE_TOOLCHAIN && ./init"
    docker exec "$build_container" bash -c "cd $container_build_dir && rm -rf ./*"
-    docker exec "$build_container" bash -c "cd $container_build_dir && $ACTIVATE_TOOLCHAIN && cmake -DCMAKE_BUILD_TYPE=release $telemetry_id_override_flag .."
+    if [[ "$os" == "debian-11-arm" ]]; then
+        docker exec "$build_container" bash -c "cd $container_build_dir && $ACTIVATE_TOOLCHAIN && cmake -DCMAKE_BUILD_TYPE=release -DMG_ARCH="ARM64" $telemetry_id_override_flag .."
+    else
+        docker exec "$build_container" bash -c "cd $container_build_dir && $ACTIVATE_TOOLCHAIN && cmake -DCMAKE_BUILD_TYPE=release $telemetry_id_override_flag .."
+    fi
    # ' is used instead of " because we need to run make within the allowed
    # container resources.
    # shellcheck disable=SC2016
--- a/release/package/ubuntu-18.04/Dockerfile
+++ b/release/package/ubuntu-18.04/Dockerfile
@@ -10,8 +10,8 @@ RUN apt update && apt install -y \
 # Do NOT be smart here and clean the cache because the container is used in the
 # stateful context.

-RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-ubuntu-18.04.tar.gz \
-    -O ${TOOLCHAIN_VERSION}-binaries-ubuntu-18.04.tar.gz \
-    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-ubuntu-18.04.tar.gz -C /opt
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-ubuntu-18.04-amd64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-ubuntu-18.04-amd64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-ubuntu-18.04-amd64.tar.gz -C /opt

 ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/ubuntu-20.04/Dockerfile
+++ b/release/package/ubuntu-20.04/Dockerfile
@@ -10,8 +10,8 @@ RUN apt update && apt install -y \
 # Do NOT be smart here and clean the cache because the container is used in the
 # stateful context.

-RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-ubuntu-20.04.tar.gz \
-    -O ${TOOLCHAIN_VERSION}-binaries-ubuntu-20.04.tar.gz \
-    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-ubuntu-20.04.tar.gz -C /opt
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-ubuntu-20.04-amd64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-ubuntu-20.04-amd64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-ubuntu-20.04-amd64.tar.gz -C /opt

 ENTRYPOINT ["sleep", "infinity"]
--- a/release/package/ubuntu-22.04/Dockerfile
+++ b/release/package/ubuntu-22.04/Dockerfile
@@ -0,0 +1,17 @@
+FROM ubuntu:22.04
+
+ARG TOOLCHAIN_VERSION
+
+# Stops tzdata interactive configuration.
+ENV DEBIAN_FRONTEND=noninteractive
+
+RUN apt update && apt install -y \
+    ca-certificates wget git
+# Do NOT be smart here and clean the cache because the container is used in the
+# stateful context.
+
+RUN wget -q https://s3-eu-west-1.amazonaws.com/deps.memgraph.io/${TOOLCHAIN_VERSION}/${TOOLCHAIN_VERSION}-binaries-ubuntu-22.04-amd64.tar.gz \
+    -O ${TOOLCHAIN_VERSION}-binaries-ubuntu-22.04-amd64.tar.gz \
+    && tar xzvf ${TOOLCHAIN_VERSION}-binaries-ubuntu-22.04-amd64.tar.gz -C /opt
+
+ENTRYPOINT ["sleep", "infinity"]
--- a/src/communication/v2/session.hpp
+++ b/src/communication/v2/session.hpp
@@ -105,9 +105,16 @@ class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TS
    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) {
+    ws_.set_option(boost::beast::websocket::stream_base::decorator([&req](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");
+
+      // We need to do this to support WASM clients, which explicitly send this flag
+      // in their upgrade request
+      // Neo4j client breaks when this flag is sent
+      if (const auto secondary_protocol = req.base().find(boost::beast::http::field::sec_websocket_protocol);
+          secondary_protocol != res.base().end() && secondary_protocol->value() == "binary") {
+        res.set(boost::beast::http::field::sec_websocket_protocol, "binary");
+      }
    }));
    ws_.binary(true);

@@ -162,7 +169,7 @@ class WebsocketSession : public std::enable_shared_from_this<WebsocketSession<TS
        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) {
+  void OnRead(const boost::system::error_code &ec, const size_t bytes_transferred) {
    // This indicates that the WebsocketSession was closed
    if (ec == boost::beast::websocket::error::closed) {
      return;
@@ -322,7 +329,8 @@ class Session final : public std::enable_shared_from_this<Session<TSession, TSes
      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());
+    spdlog::info("Accepted a connection from {}: {}:{}", service_name_, remote_endpoint_.address(),
+                 remote_endpoint_.port());
  }

  void DoRead() {
--- a/src/memgraph.cpp
+++ b/src/memgraph.cpp
@@ -252,6 +252,11 @@ DEFINE_double(query_execution_timeout_sec, 600,
              "Maximum allowed query execution time. Queries exceeding this "
              "limit will be aborted. Value of 0 means no limit.");

+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_uint64(replication_replica_check_frequency_sec, 1,
+              "The time duration between two replica checks/pings. If < 1, replicas will NOT be checked at all. NOTE: "
+              "The MAIN instance allocates a new thread for each REPLICA.");
+
 // NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_uint64(
    memory_limit, 0,
@@ -1216,6 +1221,7 @@ int main(int argc, char **argv) {
      &db,
      {.query = {.allow_load_csv = FLAGS_allow_load_csv},
       .execution_timeout_sec = FLAGS_query_execution_timeout_sec,
+       .replication_replica_check_frequency = std::chrono::seconds(FLAGS_replication_replica_check_frequency_sec),
       .default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
       .default_pulsar_service_url = FLAGS_pulsar_service_url,
       .stream_transaction_conflict_retries = FLAGS_stream_transaction_conflict_retries,
--- a/src/memory/CMakeLists.txt
+++ b/src/memory/CMakeLists.txt
@@ -9,4 +9,5 @@ target_link_libraries(mg-memory mg-utils fmt)

 if (ENABLE_JEMALLOC)
  target_link_libraries(mg-memory Jemalloc::Jemalloc)
+  target_compile_definitions(mg-memory PRIVATE USE_JEMALLOC=1)
 endif()
--- a/src/query/config.hpp
+++ b/src/query/config.hpp
@@ -21,6 +21,8 @@ struct InterpreterConfig {

  // The default execution timeout is 10 minutes.
  double execution_timeout_sec{600.0};
+  // The same as \ref memgraph::storage::replication::ReplicationClientConfig
+  std::chrono::seconds replication_replica_check_frequency{1};

  std::string default_kafka_bootstrap_servers;
  std::string default_pulsar_service_url;
--- a/src/query/interpreter.cpp
+++ b/src/query/interpreter.cpp
@@ -160,7 +160,8 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {

  /// @throw QueryRuntimeException if an error ocurred.
  void RegisterReplica(const std::string &name, const std::string &socket_address,
-                       const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) override {
+                       const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
+                       const std::chrono::seconds replica_check_frequency) override {
    if (db_->GetReplicationRole() == storage::ReplicationRole::REPLICA) {
      // replica can't register another replica
      throw QueryRuntimeException("Replica can't register another replica!");
@@ -182,8 +183,9 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
        io::network::Endpoint::ParseSocketOrIpAddress(socket_address, query::kDefaultReplicationPort);
    if (maybe_ip_and_port) {
      auto [ip, port] = *maybe_ip_and_port;
-      auto ret =
-          db_->RegisterReplica(name, {std::move(ip), port}, repl_mode, {.timeout = timeout, .ssl = std::nullopt});
+      auto ret = db_->RegisterReplica(
+          name, {std::move(ip), port}, repl_mode,
+          {.timeout = timeout, .replica_check_frequency = replica_check_frequency, .ssl = std::nullopt});
      if (ret.HasError()) {
        throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
      }
@@ -448,7 +450,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
      return callback;
    }
    case ReplicationQuery::Action::SHOW_REPLICATION_ROLE: {
-      callback.header = {"replication mode"};
+      callback.header = {"replication role"};
      callback.fn = [handler = ReplQueryHandler{interpreter_context->db}] {
        auto mode = handler.ShowReplicationRole();
        switch (mode) {
@@ -467,6 +469,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
      const auto &sync_mode = repl_query->sync_mode_;
      auto socket_address = repl_query->socket_address_->Accept(evaluator);
      auto timeout = EvaluateOptionalExpression(repl_query->timeout_, &evaluator);
+      const auto replica_check_frequency = interpreter_context->config.replication_replica_check_frequency;
      std::optional<double> maybe_timeout;
      if (timeout.IsDouble()) {
        maybe_timeout = timeout.ValueDouble();
@@ -474,8 +477,9 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
        maybe_timeout = static_cast<double>(timeout.ValueInt());
      }
      callback.fn = [handler = ReplQueryHandler{interpreter_context->db}, name, socket_address, sync_mode,
-                     maybe_timeout]() mutable {
-        handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout);
+                     maybe_timeout, replica_check_frequency]() mutable {
+        handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout,
+                                replica_check_frequency);
        return std::vector<std::vector<TypedValue>>();
      };
      notifications->emplace_back(SeverityLevel::INFO, NotificationCode::REGISTER_REPLICA,
@@ -512,7 +516,6 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
              typed_replica.emplace_back(TypedValue("async"));
              break;
          }
-          typed_replica.emplace_back(TypedValue(static_cast<int64_t>(replica.sync_mode)));
          if (replica.timeout) {
            typed_replica.emplace_back(TypedValue(*replica.timeout));
          } else {
@@ -726,7 +729,7 @@ Callback HandleStreamQuery(StreamQuery *stream_query, const Parameters &paramete
      return callback;
    }
    case StreamQuery::Action::CHECK_STREAM: {
-      callback.header = {"query", "parameters"};
+      callback.header = {"queries", "raw messages"};
      callback.fn = [interpreter_context, stream_name = stream_query->stream_name_,
                     timeout = GetOptionalValue<std::chrono::milliseconds>(stream_query->timeout_, evaluator),
                     batch_limit = GetOptionalValue<int64_t>(stream_query->batch_limit_, evaluator)]() mutable {
--- a/src/query/interpreter.hpp
+++ b/src/query/interpreter.hpp
@@ -137,7 +137,8 @@ class ReplicationQueryHandler {

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

  /// @throw QueryRuntimeException if an error ocurred.
  virtual void DropReplica(const std::string &replica_name) = 0;
--- a/src/query/stream/streams.cpp
+++ b/src/query/stream/streams.cpp
@@ -43,6 +43,7 @@ extern const Event MessagesConsumed;
 namespace memgraph::query::stream {
 namespace {
 inline constexpr auto kExpectedTransformationResultSize = 2;
+inline constexpr auto kCheckStreamResultSize = 2;
 const utils::pmr::string query_param_name{"query", utils::NewDeleteResource()};
 const utils::pmr::string params_param_name{"parameters", utils::NewDeleteResource()};

@@ -724,15 +725,27 @@ TransformationResult Streams::Check(const std::string &stream_name, std::optiona
          auto accessor = interpreter_context->db->Access();
          CallCustomTransformation(transformation_name, messages, result, accessor, *memory_resource, stream_name);

-          for (auto &row : result.rows) {
-            auto [query, parameters] = ExtractTransformationResult(row.values, transformation_name, stream_name);
-            std::vector<TypedValue> result_row;
-            result_row.reserve(kExpectedTransformationResultSize);
-            result_row.push_back(std::move(query));
-            result_row.push_back(std::move(parameters));
+          auto result_row = std::vector<TypedValue>();
+          result_row.reserve(kCheckStreamResultSize);

-            test_result.push_back(std::move(result_row));
-          }
+          auto queries_and_parameters = std::vector<TypedValue>(result.rows.size());
+          std::transform(
+              result.rows.cbegin(), result.rows.cend(), queries_and_parameters.begin(), [&](const auto &row) {
+                auto [query, parameters] = ExtractTransformationResult(row.values, transformation_name, stream_name);
+
+                return std::map<std::string, TypedValue>{{"query", std::move(query)},
+                                                         {"parameters", std::move(parameters)}};
+              });
+          result_row.emplace_back(std::move(queries_and_parameters));
+
+          auto messages_list = std::vector<TypedValue>(messages.size());
+          std::transform(messages.cbegin(), messages.cend(), messages_list.begin(), [](const auto &message) {
+            return std::string_view(message.Payload().data(), message.Payload().size());
+          });
+
+          result_row.emplace_back(std::move(messages_list));
+
+          test_result.emplace_back(std::move(result_row));
        };

        locked_stream_source->Check(timeout, batch_limit, consumer_function);
--- a/src/storage/v2/replication/config.hpp
+++ b/src/storage/v2/replication/config.hpp
@@ -16,6 +16,10 @@
 namespace memgraph::storage::replication {
 struct ReplicationClientConfig {
  std::optional<double> timeout;
+  // The default delay between main checking/pinging replicas is 1s because
+  // that seems like a reasonable timeframe in which main should notice a
+  // replica is down.
+  std::chrono::seconds replica_check_frequency{1};

  struct SSL {
    std::string key_file = "";
--- a/src/storage/v2/replication/replication_client.cpp
+++ b/src/storage/v2/replication/replication_client.cpp
@@ -41,12 +41,49 @@ Storage::ReplicationClient::ReplicationClient(std::string name, Storage *storage
  }

  rpc_client_.emplace(endpoint, &*rpc_context_);
-  TryInitializeClient();
+  TryInitializeClientSync();

  if (config.timeout && replica_state_ != replication::ReplicaState::INVALID) {
    timeout_.emplace(*config.timeout);
    timeout_dispatcher_.emplace();
  }
+
+  // Help the user to get the most accurate replica state possible.
+  if (config.replica_check_frequency > std::chrono::seconds(0)) {
+    replica_checker_.Run("Replica Checker", config.replica_check_frequency, [&] { FrequentCheck(); });
+  }
+}
+
+void Storage::ReplicationClient::TryInitializeClientAsync() {
+  thread_pool_.AddTask([this] {
+    rpc_client_->Abort();
+    this->TryInitializeClientSync();
+  });
+}
+
+void Storage::ReplicationClient::FrequentCheck() {
+  const auto is_success = std::invoke([this]() {
+    try {
+      auto stream{rpc_client_->Stream<replication::FrequentHeartbeatRpc>()};
+      const auto response = stream.AwaitResponse();
+      return response.success;
+    } catch (const rpc::RpcFailedException &) {
+      return false;
+    }
+  });
+  // States: READY, REPLICATING, RECOVERY, INVALID
+  // If success && ready, replicating, recovery -> stay the same because something good is going on.
+  // If success && INVALID -> [it's possible that replica came back to life] -> TryInitializeClient.
+  // If fail -> [replica is not reachable at all] -> INVALID state.
+  // NOTE: TryInitializeClient might return nothing if there is a branching point.
+  // NOTE: The early return pattern simplified the code, but the behavior should be as explained.
+  if (!is_success) {
+    replica_state_.store(replication::ReplicaState::INVALID);
+    return;
+  }
+  if (replica_state_.load() == replication::ReplicaState::INVALID) {
+    TryInitializeClientAsync();
+  }
 }

 /// @throws rpc::RpcFailedException
@@ -100,7 +137,7 @@ void Storage::ReplicationClient::InitializeClient() {
  }
 }

-void Storage::ReplicationClient::TryInitializeClient() {
+void Storage::ReplicationClient::TryInitializeClientSync() {
  try {
    InitializeClient();
  } catch (const rpc::RpcFailedException &) {
@@ -113,10 +150,7 @@ void Storage::ReplicationClient::TryInitializeClient() {

 void Storage::ReplicationClient::HandleRpcFailure() {
  spdlog::error(utils::MessageWithLink("Couldn't replicate data to {}.", name_, "https://memgr.ph/replication"));
-  thread_pool_.AddTask([this] {
-    rpc_client_->Abort();
-    this->TryInitializeClient();
-  });
+  TryInitializeClientAsync();
 }

 replication::SnapshotRes Storage::ReplicationClient::TransferSnapshot(const std::filesystem::path &path) {
--- a/src/storage/v2/replication/replication_client.hpp
+++ b/src/storage/v2/replication/replication_client.hpp
@@ -142,16 +142,14 @@ class Storage::ReplicationClient {

  std::vector<RecoveryStep> GetRecoverySteps(uint64_t replica_commit, utils::FileRetainer::FileLocker *file_locker);

+  void FrequentCheck();
  void InitializeClient();
-
-  void TryInitializeClient();
-
+  void TryInitializeClientSync();
+  void TryInitializeClientAsync();
  void HandleRpcFailure();

  std::string name_;
-
  Storage *storage_;
-
  std::optional<communication::ClientContext> rpc_context_;
  std::optional<rpc::Client> rpc_client_;

@@ -198,6 +196,8 @@ class Storage::ReplicationClient {
  //    to ignore concurrency problems inside the client.
  utils::ThreadPool thread_pool_{1};
  std::atomic<replication::ReplicaState> replica_state_{replication::ReplicaState::INVALID};
+
+  utils::Scheduler replica_checker_;
 };

 }  // namespace memgraph::storage
--- a/src/storage/v2/replication/replication_server.cpp
+++ b/src/storage/v2/replication/replication_server.cpp
@@ -60,6 +60,10 @@ Storage::ReplicationServer::ReplicationServer(Storage *storage, io::network::End
    spdlog::debug("Received HeartbeatRpc");
    this->HeartbeatHandler(req_reader, res_builder);
  });
+  rpc_server_->Register<replication::FrequentHeartbeatRpc>([](auto *req_reader, auto *res_builder) {
+    spdlog::debug("Received FrequentHeartbeatRpc");
+    FrequentHeartbeatHandler(req_reader, res_builder);
+  });
  rpc_server_->Register<replication::AppendDeltasRpc>([this](auto *req_reader, auto *res_builder) {
    spdlog::debug("Received AppendDeltasRpc");
    this->AppendDeltasHandler(req_reader, res_builder);
@@ -86,6 +90,13 @@ void Storage::ReplicationServer::HeartbeatHandler(slk::Reader *req_reader, slk::
  slk::Save(res, res_builder);
 }

+void Storage::ReplicationServer::FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
+  replication::FrequentHeartbeatReq req;
+  slk::Load(&req, req_reader);
+  replication::FrequentHeartbeatRes res{true};
+  slk::Save(res, res_builder);
+}
+
 void Storage::ReplicationServer::AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
  replication::AppendDeltasReq req;
  slk::Load(&req, req_reader);
--- a/src/storage/v2/replication/replication_server.hpp
+++ b/src/storage/v2/replication/replication_server.hpp
@@ -29,6 +29,7 @@ class Storage::ReplicationServer {
 private:
  // RPC handlers
  void HeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
+  static void FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
  void AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder);
  void SnapshotHandler(slk::Reader *req_reader, slk::Builder *res_builder);
  void WalFilesHandler(slk::Reader *req_reader, slk::Builder *res_builder);
--- a/src/storage/v2/replication/rpc.lcp
+++ b/src/storage/v2/replication/rpc.lcp
@@ -43,6 +43,12 @@ cpp<#
     (current-commit-timestamp :uint64_t)
     (epoch-id "std::string"))))

+;; FrequentHearthbeat is required because calling Heartbeat takes the storage lock.
+;; Configured by `replication_replica_check_delay`.
+(lcp:define-rpc frequent-heartbeat
+  (:request ())
+  (:response ((success :bool))))
+
 (lcp:define-rpc snapshot
  (:request ())
  (:response
--- a/tests/e2e/memgraph.py
+++ b/tests/e2e/memgraph.py
@@ -106,3 +106,10 @@ class MemgraphInstanceRunner:
        self.proc_mg.terminate()
        code = self.proc_mg.wait()
        assert code == 0, "The Memgraph process exited with non-zero!"
+
+    def kill(self):
+        if not self.is_running():
+            return
+        self.proc_mg.kill()
+        code = self.proc_mg.wait()
+        assert code == 9, "The killed Memgraph process exited with non-nine!"
--- a/tests/e2e/replication/CMakeLists.txt
+++ b/tests/e2e/replication/CMakeLists.txt
@@ -5,3 +5,7 @@ target_link_libraries(memgraph__e2e__replication__constraints gflags mgclient mg

 add_executable(memgraph__e2e__replication__read_write_benchmark read_write_benchmark.cpp)
 target_link_libraries(memgraph__e2e__replication__read_write_benchmark gflags json mgclient mg-utils mg-io Threads::Threads)
+
+copy_e2e_python_files(replication_show common.py)
+copy_e2e_python_files(replication_show conftest.py)
+copy_e2e_python_files(replication_show show.py)
--- a/tests/e2e/replication/common.py
+++ b/tests/e2e/replication/common.py
@@ -0,0 +1,26 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import mgclient
+import typing
+
+
+def execute_and_fetch_all(
+    cursor: mgclient.Cursor, query: str, params: dict = {}
+) -> typing.List[tuple]:
+    cursor.execute(query, params)
+    return cursor.fetchall()
+
+
+def connect(**kwargs) -> mgclient.Connection:
+    connection = mgclient.connect(**kwargs)
+    connection.autocommit = True
+    return connection
--- a/tests/e2e/replication/conftest.py
+++ b/tests/e2e/replication/conftest.py
@@ -0,0 +1,44 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import pytest
+
+from common import execute_and_fetch_all, connect
+
+
+# The fixture here is more complex because the connection has to be
+# parameterized based on the test parameters (info has to be available on both
+# sides).
+#
+# https://docs.pytest.org/en/latest/example/parametrize.html#indirect-parametrization
+# is not an elegant/feasible solution here.
+#
+# The solution was independently developed and then I stumbled upon the same
+# approach here https://stackoverflow.com/a/68286553/4888809 which I think is
+# optimal.
+@pytest.fixture(scope="function")
+def connection():
+    connection_holder = None
+    role_holder = None
+
+    def inner_connection(port, role):
+        nonlocal connection_holder, role_holder
+        connection_holder = connect(host="localhost", port=port)
+        role_holder = role
+        return connection_holder
+
+    yield inner_connection
+
+    # Only main instance can be cleaned up because replicas do NOT accept
+    # writes.
+    if role_holder == "main":
+        cursor = connection_holder.cursor()
+        execute_and_fetch_all(cursor, "MATCH (n) DETACH DELETE n;")
--- a/tests/e2e/replication/show.py
+++ b/tests/e2e/replication/show.py
@@ -0,0 +1,46 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import sys
+
+import pytest
+from common import execute_and_fetch_all
+
+
+@pytest.mark.parametrize(
+    "port, role",
+    [(7687, "main"), (7688, "replica"), (7689, "replica"), (7690, "replica")],
+)
+def test_show_replication_role(port, role, connection):
+    cursor = connection(port, role).cursor()
+    data = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;")
+    assert cursor.description[0].name == "replication role"
+    assert data[0][0] == role
+
+
+def test_show_replicas(connection):
+    cursor = connection(7687, "main").cursor()
+    actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
+
+    expected_column_names = {"name", "socket_address", "sync_mode", "timeout"}
+    actual_column_names = {x.name for x in cursor.description}
+    assert expected_column_names == actual_column_names
+
+    expected_data = {
+        ("replica_1", "127.0.0.1:10001", "sync", 0),
+        ("replica_2", "127.0.0.1:10002", "sync", 1.0),
+        ("replica_3", "127.0.0.1:10003", "async", None),
+    }
+    assert expected_data == actual_data
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))
--- a/tests/e2e/replication/workloads.yaml
+++ b/tests/e2e/replication/workloads.yaml
@@ -46,4 +46,31 @@ workloads:
    args: []
    <<: *template_cluster

-
+  - name: "Show"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["replication/show.py"]
+    cluster:
+      replica_1:
+        args: ["--bolt-port", "7688", "--log-level=TRACE"]
+        log_file: "replication-e2e-replica1.log"
+        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10001;"]
+        validation_queries: []
+      replica_2:
+        args: ["--bolt-port", "7689", "--log-level=TRACE"]
+        log_file: "replication-e2e-replica2.log"
+        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10002;"]
+        validation_queries: []
+      replica_3:
+        args: ["--bolt-port", "7690", "--log-level=TRACE"]
+        log_file: "replication-e2e-replica3.log"
+        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10003;"]
+        validation_queries: []
+      main:
+        args: ["--bolt-port", "7687", "--log-level=TRACE"]
+        log_file: "replication-e2e-main.log"
+        setup_queries: [
+          "REGISTER REPLICA replica_1 SYNC WITH TIMEOUT 0 TO '127.0.0.1:10001'",
+          "REGISTER REPLICA replica_2 SYNC WITH TIMEOUT 1 TO '127.0.0.1:10002'",
+          "REGISTER REPLICA replica_3 ASYNC TO '127.0.0.1:10003'"
+        ]
+        validation_queries: []
--- a/tests/e2e/streams/common.py
+++ b/tests/e2e/streams/common.py
@@ -12,7 +12,7 @@
 import mgclient
 import time

-from multiprocessing import Process, Value
+from multiprocessing import Manager, Process, Value

 # These are the indices of the different values in the result of SHOW STREAM
 # query
@@ -26,8 +26,10 @@ IS_RUNNING = 6

 # These are the indices of the query and parameters in the result of CHECK
 # STREAM query
-QUERY = 0
-PARAMS = 1
+QUERIES = 0
+RAWMESSAGES = 1
+PARAMETERS_LITERAL = "parameters"
+QUERY_LITERAL = "query"

 SIMPLE_MSG = b"message"

@@ -45,13 +47,13 @@ def connect(**kwargs):

 def timed_wait(fun):
    start_time = time.time()
-    seconds = 10
+    SECONDS = 10

    while True:
        current_time = time.time()
        elapsed_time = current_time - start_time

-        if elapsed_time > seconds:
+        if elapsed_time > SECONDS:
            return False

        if fun():
@@ -62,13 +64,13 @@ def timed_wait(fun):

 def check_one_result_row(cursor, query):
    start_time = time.time()
-    seconds = 10
+    SECONDS = 10

    while True:
        current_time = time.time()
        elapsed_time = current_time - start_time

-        if elapsed_time > seconds:
+        if elapsed_time > SECONDS:
            return False

        cursor.execute(query)
@@ -81,12 +83,10 @@ def check_one_result_row(cursor, query):


 def check_vertex_exists_with_properties(cursor, properties):
-    properties_string = ', '.join([f'{k}: {v}' for k, v in properties.items()])
+    properties_string = ", ".join([f"{k}: {v}" for k, v in properties.items()])
    assert check_one_result_row(
        cursor,
-        "MATCH (n: MESSAGE {"
-        f"{properties_string}"
-        "}) RETURN n",
+        f"MATCH (n: MESSAGE {{{properties_string}}}) RETURN n",
    )


@@ -129,28 +129,27 @@ def validate_info(actual_stream_info, expected_stream_info):
    for info, expected_info in zip(actual_stream_info, expected_stream_info):
        assert info == expected_info

+
 def check_stream_info(cursor, stream_name, expected_stream_info):
    stream_info = get_stream_info(cursor, stream_name)
    validate_info(stream_info, expected_stream_info)

+
 def kafka_check_vertex_exists_with_topic_and_payload(cursor, topic, payload_bytes):
-    decoded_payload = payload_bytes.decode('utf-8')
-    check_vertex_exists_with_properties(
-            cursor, {'topic': f'"{topic}"', 'payload': f'"{decoded_payload}"'})
+    decoded_payload = payload_bytes.decode("utf-8")
+    check_vertex_exists_with_properties(cursor, {"topic": f'"{topic}"', "payload": f'"{decoded_payload}"'})


-PULSAR_SERVICE_URL = 'pulsar://127.0.0.1:6650'
+PULSAR_SERVICE_URL = "pulsar://127.0.0.1:6650"
+

 def pulsar_default_namespace_topic(topic):
-    return f'persistent://public/default/{topic}'
+    return f"persistent://public/default/{topic}"


 def test_start_and_stop_during_check(
-        operation,
-        connection,
-        stream_creator,
-        message_sender,
-        already_stopped_error):
+    operation, connection, stream_creator, message_sender, already_stopped_error, batchSize
+):
    # This test is quite complex. The goal is to call START/STOP queries
    # while a CHECK query is waiting for its result. Because the Global
    # Interpreter Lock, running queries on multiple threads is not useful,
@@ -161,11 +160,9 @@ def test_start_and_stop_during_check(
    # synchronize between the different processes. Each value represents a
    # specific phase of the execution of the processes.
    assert operation in ["START", "STOP"]
+    assert batchSize == 1
    cursor = connection.cursor()
-    execute_and_fetch_all(
-        cursor,
-        stream_creator('test_stream')
-    )
+    execute_and_fetch_all(cursor, stream_creator("test_stream"))

    check_counter = Value("i", 0)
    check_result_len = Value("i", 0)
@@ -185,7 +182,9 @@ def test_start_and_stop_during_check(
        result = execute_and_fetch_all(cursor, "CHECK STREAM test_stream")
        result_len.value = len(result)
        counter.value = CHECK_AFTER_FETCHALL
-        if len(result) > 0 and "payload: 'message'" in result[0][QUERY]:
+        if (
+            len(result) > 0 and "payload: 'message'" in result[0][QUERIES][0][QUERY_LITERAL]
+        ):  # The 0 is only correct because batchSize is 1
            counter.value = CHECK_CORRECT_RESULT
        else:
            counter.value = CHECK_INCORRECT_RESULT
@@ -213,12 +212,8 @@ def test_start_and_stop_during_check(
        except Exception:
            counter.value = OP_UNEXPECTED_EXCEPTION

-    check_stream_proc = Process(
-        target=call_check, daemon=True, args=(check_counter, check_result_len)
-    )
-    operation_proc = Process(
-        target=call_operation, daemon=True, args=(operation_counter,)
-    )
+    check_stream_proc = Process(target=call_check, daemon=True, args=(check_counter, check_result_len))
+    operation_proc = Process(target=call_operation, daemon=True, args=(operation_counter,))

    try:
        check_stream_proc.start()
@@ -227,9 +222,7 @@ def test_start_and_stop_during_check(

        assert timed_wait(lambda: check_counter.value == CHECK_BEFORE_EXECUTE)
        assert timed_wait(lambda: get_is_running(cursor, "test_stream"))
-        assert check_counter.value == CHECK_BEFORE_EXECUTE, (
-            "SHOW STREAMS " "was blocked until the end of CHECK STREAM"
-        )
+        assert check_counter.value == CHECK_BEFORE_EXECUTE, "SHOW STREAMS " "was blocked until the end of CHECK STREAM"
        operation_proc.start()
        assert timed_wait(lambda: operation_counter.value == OP_BEFORE_EXECUTE)

@@ -255,31 +248,156 @@ def test_start_and_stop_during_check(
        if operation_proc.is_alive():
            operation_proc.terminate()

+
 def test_start_checked_stream_after_timeout(connection, stream_creator):
    cursor = connection.cursor()
-    execute_and_fetch_all(
-        cursor,
-        stream_creator('test_stream')
-    )
+    execute_and_fetch_all(cursor, stream_creator("test_stream"))

-    timeout_ms = 2000
+    TIMEOUT_MS = 2000

    def call_check():
-        execute_and_fetch_all(
-            connect().cursor(),
-            f"CHECK STREAM test_stream TIMEOUT {timeout_ms}")
+        execute_and_fetch_all(connect().cursor(), f"CHECK STREAM test_stream TIMEOUT {TIMEOUT_MS}")

    check_stream_proc = Process(target=call_check, daemon=True)

    start = time.time()
    check_stream_proc.start()
-    assert timed_wait(
-        lambda: get_is_running(
-            cursor, "test_stream"))
+    assert timed_wait(lambda: get_is_running(cursor, "test_stream"))
    start_stream(cursor, "test_stream")
    end = time.time()

-    assert (end - start) < 1.3 * \
-        timeout_ms, "The START STREAM was blocked too long"
+    assert (end - start) < 1.3 * TIMEOUT_MS, "The START STREAM was blocked too long"
    assert get_is_running(cursor, "test_stream")
    stop_stream(cursor, "test_stream")
+
+
+def test_check_stream_same_number_of_queries_than_messages(connection, stream_creator, message_sender):
+    BATCH_SIZE = 2
+    BATCH_LIMIT = 3
+    STREAM_NAME = "test_stream"
+    cursor = connection.cursor()
+    execute_and_fetch_all(cursor, stream_creator(STREAM_NAME, BATCH_SIZE))
+    time.sleep(2)
+
+    test_results = Manager().Namespace()
+
+    def check_stream(stream_name, batch_limit):
+        connection = connect()
+        cursor = connection.cursor()
+        test_results.value = execute_and_fetch_all(cursor, f"CHECK STREAM {stream_name} BATCH_LIMIT {batch_limit} ")
+
+    check_stream_proc = Process(target=check_stream, args=(STREAM_NAME, BATCH_LIMIT))
+    check_stream_proc.start()
+    time.sleep(2)
+
+    MESSAGES = [b"01", b"02", b"03", b"04", b"05", b"06"]
+    for message in MESSAGES:
+        message_sender(message)
+
+    check_stream_proc.join()
+
+    # # Transformation does not do any filtering and simply create queries as "Messages: {contentOfMessage}". Queries should be like:
+    # # -Batch 1: [{parameters: {"value": "Parameter: 01"}, query: "Message: 01"},
+    # #            {parameters: {"value": "Parameter: 02"}, query: "Message: 02"}]
+    # # -Batch 2: [{parameters: {"value": "Parameter: 03"}, query: "Message: 03"},
+    # #            {parameters: {"value": "Parameter: 04"}, query: "Message: 04"}]
+    # # -Batch 3: [{parameters: {"value": "Parameter: 05"}, query: "Message: 05"},
+    # #            {parameters: {"value": "Parameter: 06"}, query: "Message: 06"}]
+
+    assert len(test_results.value) == BATCH_LIMIT
+
+    expected_queries_and_raw_messages_1 = (
+        [  # queries
+            {PARAMETERS_LITERAL: {"value": "Parameter: 01"}, QUERY_LITERAL: "Message: 01"},
+            {PARAMETERS_LITERAL: {"value": "Parameter: 02"}, QUERY_LITERAL: "Message: 02"},
+        ],
+        ["01", "02"],  # raw message
+    )
+
+    expected_queries_and_raw_messages_2 = (
+        [  # queries
+            {PARAMETERS_LITERAL: {"value": "Parameter: 03"}, QUERY_LITERAL: "Message: 03"},
+            {PARAMETERS_LITERAL: {"value": "Parameter: 04"}, QUERY_LITERAL: "Message: 04"},
+        ],
+        ["03", "04"],  # raw message
+    )
+
+    expected_queries_and_raw_messages_3 = (
+        [  # queries
+            {PARAMETERS_LITERAL: {"value": "Parameter: 05"}, QUERY_LITERAL: "Message: 05"},
+            {PARAMETERS_LITERAL: {"value": "Parameter: 06"}, QUERY_LITERAL: "Message: 06"},
+        ],
+        ["05", "06"],  # raw message
+    )
+
+    assert expected_queries_and_raw_messages_1 == test_results.value[0]
+    assert expected_queries_and_raw_messages_2 == test_results.value[1]
+    assert expected_queries_and_raw_messages_3 == test_results.value[2]
+
+
+def test_check_stream_different_number_of_queries_than_messages(connection, stream_creator, message_sender):
+    BATCH_SIZE = 2
+    BATCH_LIMIT = 3
+    STREAM_NAME = "test_stream"
+    cursor = connection.cursor()
+    execute_and_fetch_all(cursor, stream_creator(STREAM_NAME, BATCH_SIZE))
+    time.sleep(2)
+
+    results = Manager().Namespace()
+
+    def check_stream(stream_name, batch_limit):
+        connection = connect()
+        cursor = connection.cursor()
+        results.value = execute_and_fetch_all(cursor, f"CHECK STREAM {stream_name} BATCH_LIMIT {batch_limit} ")
+
+    check_stream_proc = Process(target=check_stream, args=(STREAM_NAME, BATCH_LIMIT))
+    check_stream_proc.start()
+    time.sleep(2)
+
+    MESSAGES = [b"a_01", b"a_02", b"03", b"04", b"b_05", b"06"]
+    for message in MESSAGES:
+        message_sender(message)
+
+    check_stream_proc.join()
+
+    # Transformation does some filtering: if message contains "a", it is ignored.
+    # Transformation also has special rule to create query if message is "b": it create more queries.
+    #
+    # Queries should be like:
+    # -Batch 1: []
+    # -Batch 2: [{parameters: {"value": "Parameter: 03"}, query: "Message: 03"},
+    #            {parameters: {"value": "Parameter: 04"}, query: "Message: 04"}]
+    # -Batch 3: [{parameters: {"value": "Parameter: 05"}, query: "Message: 05"},
+    #            {parameters: {"value": "Parameter: extra_05"}, query: "Message: extra_05"}
+    #            {parameters: {"value": "Parameter: 06"}, query: "Message: 06"}]
+
+    assert len(results.value) == BATCH_LIMIT
+
+    expected_queries_and_raw_messages_1 = (
+        [],  # queries
+        ["a_01", "a_02"],  # raw message
+    )
+
+    expected_queries_and_raw_messages_2 = (
+        [  # queries
+            {PARAMETERS_LITERAL: {"value": "Parameter: 03"}, QUERY_LITERAL: "Message: 03"},
+            {PARAMETERS_LITERAL: {"value": "Parameter: 04"}, QUERY_LITERAL: "Message: 04"},
+        ],
+        ["03", "04"],  # raw message
+    )
+
+    expected_queries_and_raw_messages_3 = (
+        [  # queries
+            {PARAMETERS_LITERAL: {"value": "Parameter: b_05"}, QUERY_LITERAL: "Message: b_05"},
+            {
+                PARAMETERS_LITERAL: {"value": "Parameter: extra_b_05"},
+                QUERY_LITERAL: "Message: extra_b_05",
+            },
+            {PARAMETERS_LITERAL: {"value": "Parameter: 06"}, QUERY_LITERAL: "Message: 06"},
+        ],
+        ["b_05", "06"],  # raw message
+    )
+
+    assert expected_queries_and_raw_messages_1 == results.value[0]
+    assert expected_queries_and_raw_messages_2 == results.value[1]
+    assert expected_queries_and_raw_messages_3 == results.value[2]
--- a/tests/e2e/streams/kafka_streams_tests.py
+++ b/tests/e2e/streams/kafka_streams_tests.py
@@ -18,12 +18,10 @@ import time
 from multiprocessing import Process, Value
 import common

-TRANSFORMATIONS_TO_CHECK_C = [
-    "empty_transformation"]
+TRANSFORMATIONS_TO_CHECK_C = ["empty_transformation"]
+
+TRANSFORMATIONS_TO_CHECK_PY = ["kafka_transform.simple", "kafka_transform.with_parameters"]

-TRANSFORMATIONS_TO_CHECK_PY = [
-    "kafka_transform.simple",
-    "kafka_transform.with_parameters"]

@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
 def test_simple(kafka_producer, kafka_topics, connection, transformation):
@@ -31,9 +29,7 @@ def test_simple(kafka_producer, kafka_topics, connection, transformation):
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM test "
-        f"TOPICS {','.join(kafka_topics)} "
-        f"TRANSFORM {transformation}",
+        f"CREATE KAFKA STREAM test TOPICS {','.join(kafka_topics)} TRANSFORM {transformation}",
    )
    common.start_stream(cursor, "test")
    time.sleep(5)
@@ -42,16 +38,11 @@ def test_simple(kafka_producer, kafka_topics, connection, transformation):
        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=60)

    for topic in kafka_topics:
-        common.kafka_check_vertex_exists_with_topic_and_payload(
-            cursor, topic, common.SIMPLE_MSG)
+        common.kafka_check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
-def test_separate_consumers(
-        kafka_producer,
-        kafka_topics,
-        connection,
-        transformation):
+def test_separate_consumers(kafka_producer, kafka_topics, connection, transformation):
    assert len(kafka_topics) > 0
    cursor = connection.cursor()

@@ -61,9 +52,7 @@ def test_separate_consumers(
        stream_names.append(stream_name)
        common.execute_and_fetch_all(
            cursor,
-            f"CREATE KAFKA STREAM {stream_name} "
-            f"TOPICS {topic} "
-            f"TRANSFORM {transformation}",
+            f"CREATE KAFKA STREAM {stream_name} TOPICS {topic} TRANSFORM {transformation}",
        )

    for stream_name in stream_names:
@@ -75,12 +64,10 @@ def test_separate_consumers(
        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=60)

    for topic in kafka_topics:
-        common.kafka_check_vertex_exists_with_topic_and_payload(
-            cursor, topic, common.SIMPLE_MSG)
+        common.kafka_check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)


-def test_start_from_last_committed_offset(
-        kafka_producer, kafka_topics, connection):
+def test_start_from_last_committed_offset(kafka_producer, kafka_topics, connection):
    # This test creates a stream, consumes a message to have a committed
    # offset, then destroys the stream. A new message is sent before the
    # stream is recreated and then restarted. This simulates when Memgraph is
@@ -90,16 +77,15 @@ def test_start_from_last_committed_offset(
    assert len(kafka_topics) > 0
    cursor = connection.cursor()
    common.execute_and_fetch_all(
-        cursor, "CREATE KAFKA STREAM test "
-        f"TOPICS {kafka_topics[0]} "
-        "TRANSFORM kafka_transform.simple", )
+        cursor,
+        f"CREATE KAFKA STREAM test TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple",
+    )
    common.start_stream(cursor, "test")
    time.sleep(1)

    kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=60)

-    common.kafka_check_vertex_exists_with_topic_and_payload(
-        cursor, kafka_topics[0], common.SIMPLE_MSG)
+    common.kafka_check_vertex_exists_with_topic_and_payload(cursor, kafka_topics[0], common.SIMPLE_MSG)

    common.stop_stream(cursor, "test")
    common.drop_stream(cursor, "test")
@@ -111,36 +97,30 @@ def test_start_from_last_committed_offset(
    for message in messages:
        vertices_with_msg = common.execute_and_fetch_all(
            cursor,
-            "MATCH (n: MESSAGE {" f"payload: '{message.decode('utf-8')}'" "}) RETURN n",
+            f"MATCH (n: MESSAGE {{payload: '{message.decode('utf-8')}'}}) RETURN n",
        )

        assert len(vertices_with_msg) == 0

    common.execute_and_fetch_all(
-        cursor, "CREATE KAFKA STREAM test "
-        f"TOPICS {kafka_topics[0]} "
-        "TRANSFORM kafka_transform.simple", )
+        cursor,
+        f"CREATE KAFKA STREAM test TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple",
+    )
    common.start_stream(cursor, "test")

    for message in messages:
-        common.kafka_check_vertex_exists_with_topic_and_payload(
-            cursor, kafka_topics[0], message)
+        common.kafka_check_vertex_exists_with_topic_and_payload(cursor, kafka_topics[0], message)


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
-def test_check_stream(
-        kafka_producer,
-        kafka_topics,
-        connection,
-        transformation):
+def test_check_stream(kafka_producer, kafka_topics, connection, transformation):
    assert len(kafka_topics) > 0
+    BATCH_SIZE = 1
+    INDEX_OF_FIRST_BATCH = 0
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM test "
-        f"TOPICS {kafka_topics[0]} "
-        f"TRANSFORM {transformation} "
-        "BATCH_SIZE 1",
+        f"CREATE KAFKA STREAM test TOPICS {kafka_topics[0]} TRANSFORM {transformation} BATCH_SIZE {BATCH_SIZE}",
    )
    common.start_stream(cursor, "test")
    time.sleep(1)
@@ -153,24 +133,28 @@ def test_check_stream(
        kafka_producer.send(kafka_topics[0], message).get(timeout=60)

    def check_check_stream(batch_limit):
-        assert (
-            transformation == "kafka_transform.simple"
-            or transformation == "kafka_transform.with_parameters"
-        )
-        test_results = common.execute_and_fetch_all(
-            cursor, f"CHECK STREAM test BATCH_LIMIT {batch_limit}"
-        )
+        assert transformation == "kafka_transform.simple" or transformation == "kafka_transform.with_parameters"
+        test_results = common.execute_and_fetch_all(cursor, f"CHECK STREAM test BATCH_LIMIT {batch_limit}")
        assert len(test_results) == batch_limit

        for i in range(batch_limit):
            message_as_str = messages[i].decode("utf-8")
+            assert (
+                BATCH_SIZE == 1
+            )  # If batch size != 1, then the usage of INDEX_OF_FIRST_BATCH must change: the result will have a list of queries (pair<parameters,query>)
+
            if transformation == "kafka_transform.simple":
-                assert f"payload: '{message_as_str}'" in test_results[i][common.QUERY]
-                assert test_results[i][common.PARAMS] is None
+                assert (
+                    f"payload: '{message_as_str}'"
+                    in test_results[i][common.QUERIES][INDEX_OF_FIRST_BATCH][common.QUERY_LITERAL]
+                )
+                assert test_results[i][common.QUERIES][INDEX_OF_FIRST_BATCH][common.PARAMETERS_LITERAL] is None
            else:
-                assert f"payload: $payload" in test_results[i][
-                    common.QUERY] and f"topic: $topic" in test_results[i][common.QUERY]
-                parameters = test_results[i][common.PARAMS]
+                assert (
+                    f"payload: $payload" in test_results[i][common.QUERIES][INDEX_OF_FIRST_BATCH][common.QUERY_LITERAL]
+                    and f"topic: $topic" in test_results[i][common.QUERIES][INDEX_OF_FIRST_BATCH][common.QUERY_LITERAL]
+                )
+                parameters = test_results[i][common.QUERIES][INDEX_OF_FIRST_BATCH][common.PARAMETERS_LITERAL]
                # this is not a very sofisticated test, but checks if
                # timestamp has some kind of value
                assert parameters["timestamp"] > 1000000000000
@@ -183,8 +167,7 @@ def test_check_stream(
    common.start_stream(cursor, "test")

    for message in messages:
-        common.kafka_check_vertex_exists_with_topic_and_payload(
-            cursor, kafka_topics[0], message)
+        common.kafka_check_vertex_exists_with_topic_and_payload(cursor, kafka_topics[0], message)


 def test_show_streams(kafka_producer, kafka_topics, connection):
@@ -192,23 +175,15 @@ def test_show_streams(kafka_producer, kafka_topics, connection):
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM default_values "
-        f"TOPICS {kafka_topics[0]} "
-        f"TRANSFORM kafka_transform.simple "
-        f"BOOTSTRAP_SERVERS 'localhost:9092'",
+        f"CREATE KAFKA STREAM default_values TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple BOOTSTRAP_SERVERS 'localhost:9092'",
    )

    consumer_group = "my_special_consumer_group"
-    batch_interval = 42
-    batch_size = 3
+    BATCH_INTERVAL = 42
+    BATCH_SIZE = 3
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM complex_values "
-        f"TOPICS {','.join(kafka_topics)} "
-        f"TRANSFORM kafka_transform.with_parameters "
-        f"CONSUMER_GROUP {consumer_group} "
-        f"BATCH_INTERVAL {batch_interval} "
-        f"BATCH_SIZE {batch_size} ",
+        f"CREATE KAFKA STREAM complex_values TOPICS {','.join(kafka_topics)} TRANSFORM kafka_transform.with_parameters CONSUMER_GROUP {consumer_group} BATCH_INTERVAL {BATCH_INTERVAL} BATCH_SIZE {BATCH_SIZE} ",
    )

    assert len(common.execute_and_fetch_all(cursor, "SHOW STREAMS")) == 2
@@ -216,13 +191,7 @@ def test_show_streams(kafka_producer, kafka_topics, connection):
    common.check_stream_info(
        cursor,
        "default_values",
-        ("default_values",
-         "kafka",
-         100,
-         1000,
-         "kafka_transform.simple",
-         None,
-         False),
+        ("default_values", "kafka", 100, 1000, "kafka_transform.simple", None, False),
    )

    common.check_stream_info(
@@ -231,8 +200,8 @@ def test_show_streams(kafka_producer, kafka_topics, connection):
        (
            "complex_values",
            "kafka",
-            batch_interval,
-            batch_size,
+            BATCH_INTERVAL,
+            BATCH_SIZE,
            "kafka_transform.with_parameters",
            None,
            False,
@@ -241,15 +210,12 @@ def test_show_streams(kafka_producer, kafka_topics, connection):


@pytest.mark.parametrize("operation", ["START", "STOP"])
-def test_start_and_stop_during_check(
-        kafka_producer,
-        kafka_topics,
-        connection,
-        operation):
+def test_start_and_stop_during_check(kafka_producer, kafka_topics, connection, operation):
    assert len(kafka_topics) > 1
+    BATCH_SIZE = 1

    def stream_creator(stream_name):
-        return f"CREATE KAFKA STREAM {stream_name} TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple"
+        return f"CREATE KAFKA STREAM {stream_name} TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple BATCH_SIZE {BATCH_SIZE}"

    def message_sender(msg):
        kafka_producer.send(kafka_topics[0], msg).get(timeout=60)
@@ -259,7 +225,9 @@ def test_start_and_stop_during_check(
        connection,
        stream_creator,
        message_sender,
-        "Kafka consumer test_stream is already stopped")
+        "Kafka consumer test_stream is already stopped",
+        BATCH_SIZE,
+    )


 def test_check_already_started_stream(kafka_topics, connection):
@@ -268,9 +236,7 @@ def test_check_already_started_stream(kafka_topics, connection):

    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM started_stream "
-        f"TOPICS {kafka_topics[0]} "
-        f"TRANSFORM kafka_transform.simple",
+        f"CREATE KAFKA STREAM started_stream TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple",
    )
    common.start_stream(cursor, "started_stream")

@@ -289,41 +255,29 @@ def test_restart_after_error(kafka_producer, kafka_topics, connection):
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM test_stream "
-        f"TOPICS {kafka_topics[0]} "
-        f"TRANSFORM kafka_transform.query",
+        f"CREATE KAFKA STREAM test_stream TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.query",
    )

    common.start_stream(cursor, "test_stream")
    time.sleep(1)

    kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=60)
-    assert common.timed_wait(
-        lambda: not common.get_is_running(
-            cursor, "test_stream"))
+    assert common.timed_wait(lambda: not common.get_is_running(cursor, "test_stream"))

    common.start_stream(cursor, "test_stream")
    time.sleep(1)
    kafka_producer.send(kafka_topics[0], b"CREATE (n:VERTEX { id : 42 })")
-    assert common.check_one_result_row(
-        cursor, "MATCH (n:VERTEX { id : 42 }) RETURN n")
+    assert common.check_one_result_row(cursor, "MATCH (n:VERTEX { id : 42 }) RETURN n")


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
-def test_bootstrap_server(
-        kafka_producer,
-        kafka_topics,
-        connection,
-        transformation):
+def test_bootstrap_server(kafka_producer, kafka_topics, connection, transformation):
    assert len(kafka_topics) > 0
    cursor = connection.cursor()
-    local = "localhost:9092"
+    LOCAL = "localhost:9092"
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM test "
-        f"TOPICS {','.join(kafka_topics)} "
-        f"TRANSFORM {transformation} "
-        f"BOOTSTRAP_SERVERS '{local}'",
+        f"CREATE KAFKA STREAM test TOPICS {','.join(kafka_topics)} TRANSFORM {transformation} BOOTSTRAP_SERVERS '{LOCAL}'",
    )
    common.start_stream(cursor, "test")
    time.sleep(5)
@@ -332,25 +286,17 @@ def test_bootstrap_server(
        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=60)

    for topic in kafka_topics:
-        common.kafka_check_vertex_exists_with_topic_and_payload(
-            cursor, topic, common.SIMPLE_MSG)
+        common.kafka_check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
-def test_bootstrap_server_empty(
-        kafka_producer,
-        kafka_topics,
-        connection,
-        transformation):
+def test_bootstrap_server_empty(kafka_producer, kafka_topics, connection, transformation):
    assert len(kafka_topics) > 0
    cursor = connection.cursor()
    with pytest.raises(mgclient.DatabaseError):
        common.execute_and_fetch_all(
            cursor,
-            "CREATE KAFKA STREAM test "
-            f"TOPICS {','.join(kafka_topics)} "
-            f"TRANSFORM {transformation} "
-            "BOOTSTRAP_SERVERS ''",
+            f"CREATE KAFKA STREAM test TOPICS {','.join(kafka_topics)} TRANSFORM {transformation} BOOTSTRAP_SERVERS ''",
        )


@@ -360,10 +306,7 @@ def test_set_offset(kafka_producer, kafka_topics, connection, transformation):
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE KAFKA STREAM test "
-        f"TOPICS {kafka_topics[0]} "
-        f"TRANSFORM {transformation} "
-        "BATCH_SIZE 1",
+        f"CREATE KAFKA STREAM test TOPICS {kafka_topics[0]} TRANSFORM {transformation} BATCH_SIZE 1",
    )

    messages = [f"{i} message" for i in range(1, 21)]
@@ -377,27 +320,18 @@ def test_set_offset(kafka_producer, kafka_topics, connection, transformation):
        else:
            assert common.check_one_result_row(
                cursor,
-                (
-                    f"MATCH (n: MESSAGE {{payload: '{expected_msgs[-1]}'}})"
-                    "RETURN n"
-                ),
+                (f"MATCH (n: MESSAGE {{payload: '{expected_msgs[-1]}'}})" "RETURN n"),
            )
        common.stop_stream(cursor, "test")
-        res = common.execute_and_fetch_all(
-            cursor, "MATCH (n) RETURN n.payload"
-        )
+        res = common.execute_and_fetch_all(cursor, "MATCH (n) RETURN n.payload")
        return res

    def execute_set_offset_and_consume(id, expected_msgs):
-        common.execute_and_fetch_all(
-            cursor, f"CALL mg.kafka_set_stream_offset('test', {id})"
-        )
+        common.execute_and_fetch_all(cursor, f"CALL mg.kafka_set_stream_offset('test', {id})")
        return consume(expected_msgs)

    with pytest.raises(mgclient.DatabaseError):
-        res = common.execute_and_fetch_all(
-            cursor, "CALL mg.kafka_set_stream_offset('foo', 10)"
-        )
+        res = common.execute_and_fetch_all(cursor, "CALL mg.kafka_set_stream_offset('foo', 10)")

    def comparison_check(a, b):
        return a == str(b).strip("'(,)")
@@ -426,40 +360,60 @@ def test_set_offset(kafka_producer, kafka_topics, connection, transformation):

 def test_info_procedure(kafka_topics, connection):
    cursor = connection.cursor()
-    stream_name = 'test_stream'
-    configs = {"sasl.username": "michael.scott"}
-    local = "localhost:9092"
-    credentials = {"sasl.password": "S3cr3tP4ssw0rd"}
-    consumer_group = "ConsumerGr"
+    STREAM_NAME = "test_stream"
+    CONFIGS = {"sasl.username": "michael.scott"}
+    LOCAL = "localhost:9092"
+    CREDENTIALS = {"sasl.password": "S3cr3tP4ssw0rd"}
+    CONSUMER_GROUP = "ConsumerGr"
    common.execute_and_fetch_all(
        cursor,
-        f"CREATE KAFKA STREAM {stream_name} "
-        f"TOPICS {','.join(kafka_topics)} "
-        f"TRANSFORM pulsar_transform.simple "
-        f"CONSUMER_GROUP {consumer_group} "
-        f"BOOTSTRAP_SERVERS '{local}' "
-        f"CONFIGS {configs} "
-        f"CREDENTIALS {credentials}"
+        f"CREATE KAFKA STREAM {STREAM_NAME} TOPICS {','.join(kafka_topics)} TRANSFORM kafka_transform.simple CONSUMER_GROUP {CONSUMER_GROUP} BOOTSTRAP_SERVERS '{LOCAL}' CONFIGS {CONFIGS} CREDENTIALS {CREDENTIALS}",
    )

-    stream_info = common.execute_and_fetch_all(
-        cursor, f"CALL mg.kafka_stream_info('{stream_name}') YIELD *")
+    stream_info = common.execute_and_fetch_all(cursor, f"CALL mg.kafka_stream_info('{STREAM_NAME}') YIELD *")

-    reducted_credentials = {key: "<REDUCTED>" for
-                            key in credentials.keys()}
+    reducted_credentials = {key: "<REDUCTED>" for key in CREDENTIALS.keys()}

-    expected_stream_info = [
-        (local, configs, consumer_group, reducted_credentials, kafka_topics)]
+    expected_stream_info = [(LOCAL, CONFIGS, CONSUMER_GROUP, reducted_credentials, kafka_topics)]
    common.validate_info(stream_info, expected_stream_info)

-@pytest.mark.parametrize("transformation",TRANSFORMATIONS_TO_CHECK_C)
+
+@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_C)
 def test_load_c_transformations(connection, transformation):
    cursor = connection.cursor()
-    query = "CALL mg.transformations() YIELD * WITH name WHERE name STARTS WITH 'c_transformations." + transformation + "' RETURN name" 
-    result = common.execute_and_fetch_all(
-                 cursor, query)
+    query = f"CALL mg.transformations() YIELD * WITH name WHERE name STARTS WITH 'c_transformations.{transformation}' RETURN name"
+    result = common.execute_and_fetch_all(cursor, query)
    assert len(result) == 1
-    assert result[0][0] == "c_transformations." + transformation
-    
+    assert result[0][0] == f"c_transformations.{transformation}"
+
+
+def test_check_stream_same_number_of_queries_than_messages(kafka_producer, kafka_topics, connection):
+    assert len(kafka_topics) > 0
+
+    TRANSFORMATION = "common_transform.check_stream_no_filtering"
+
+    def stream_creator(stream_name, batch_size):
+        return f"CREATE KAFKA STREAM {stream_name} TOPICS {kafka_topics[0]} TRANSFORM {TRANSFORMATION} BATCH_INTERVAL 3000 BATCH_SIZE {batch_size}"
+
+    def message_sender(msg):
+        kafka_producer.send(kafka_topics[0], msg).get(timeout=60)
+
+    common.test_check_stream_same_number_of_queries_than_messages(connection, stream_creator, message_sender)
+
+
+def test_check_stream_different_number_of_queries_than_messages(kafka_producer, kafka_topics, connection):
+    assert len(kafka_topics) > 0
+
+    TRANSFORMATION = "common_transform.check_stream_with_filtering"
+
+    def stream_creator(stream_name, batch_size):
+        return f"CREATE KAFKA STREAM {stream_name} TOPICS {kafka_topics[0]} TRANSFORM {TRANSFORMATION} BATCH_INTERVAL 3000  BATCH_SIZE {batch_size}"
+
+    def message_sender(msg):
+        kafka_producer.send(kafka_topics[0], msg).get(timeout=60)
+
+    common.test_check_stream_different_number_of_queries_than_messages(connection, stream_creator, message_sender)
+
+
 if __name__ == "__main__":
    sys.exit(pytest.main([__file__, "-rA"]))
--- a/tests/e2e/streams/pulsar_streams_tests.py
+++ b/tests/e2e/streams/pulsar_streams_tests.py
@@ -18,17 +18,14 @@ import time
 from multiprocessing import Process, Value
 import common

-TRANSFORMATIONS_TO_CHECK = [
-    "pulsar_transform.simple",
-    "pulsar_transform.with_parameters"]
+TRANSFORMATIONS_TO_CHECK = ["pulsar_transform.simple", "pulsar_transform.with_parameters"]


 def check_vertex_exists_with_topic_and_payload(cursor, topic, payload_byte):
-    decoded_payload = payload_byte.decode('utf-8')
+    decoded_payload = payload_byte.decode("utf-8")
    common.check_vertex_exists_with_properties(
-        cursor, {
-            'topic': f'"{common.pulsar_default_namespace_topic(topic)}"',
-            'payload': f'"{decoded_payload}"'})
+        cursor, {"topic": f'"{common.pulsar_default_namespace_topic(topic)}"', "payload": f'"{decoded_payload}"'}
+    )


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK)
@@ -37,30 +34,23 @@ def test_simple(pulsar_client, pulsar_topics, connection, transformation):
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM test "
-        f"TOPICS '{','.join(pulsar_topics)}' "
-        f"TRANSFORM {transformation}",
+        f"CREATE PULSAR STREAM test TOPICS '{','.join(pulsar_topics)}' TRANSFORM {transformation}",
    )
    common.start_stream(cursor, "test")
    time.sleep(5)

    for topic in pulsar_topics:
        producer = pulsar_client.create_producer(
-            common.pulsar_default_namespace_topic(topic),
-            send_timeout_millis=60000)
+            common.pulsar_default_namespace_topic(topic), send_timeout_millis=60000
+        )
        producer.send(common.SIMPLE_MSG)

    for topic in pulsar_topics:
-        check_vertex_exists_with_topic_and_payload(
-            cursor, topic, common.SIMPLE_MSG)
+        check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK)
-def test_separate_consumers(
-        pulsar_client,
-        pulsar_topics,
-        connection,
-        transformation):
+def test_separate_consumers(pulsar_client, pulsar_topics, connection, transformation):
    assert len(pulsar_topics) > 0
    cursor = connection.cursor()

@@ -70,9 +60,7 @@ def test_separate_consumers(
        stream_names.append(stream_name)
        common.execute_and_fetch_all(
            cursor,
-            f"CREATE PULSAR STREAM {stream_name} "
-            f"TOPICS {topic} "
-            f"TRANSFORM {transformation}",
+            f"CREATE PULSAR STREAM {stream_name} TOPICS {topic} TRANSFORM {transformation}",
        )

    for stream_name in stream_names:
@@ -81,13 +69,11 @@ def test_separate_consumers(
    time.sleep(5)

    for topic in pulsar_topics:
-        producer = pulsar_client.create_producer(
-            topic, send_timeout_millis=60000)
+        producer = pulsar_client.create_producer(topic, send_timeout_millis=60000)
        producer.send(common.SIMPLE_MSG)

    for topic in pulsar_topics:
-        check_vertex_exists_with_topic_and_payload(
-            cursor, topic, common.SIMPLE_MSG)
+        check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)


 def test_start_from_latest_messages(pulsar_client, pulsar_topics, connection):
@@ -99,118 +85,112 @@ def test_start_from_latest_messages(pulsar_client, pulsar_topics, connection):
    assert len(pulsar_topics) > 0
    cursor = connection.cursor()
    common.execute_and_fetch_all(
-        cursor, "CREATE PULSAR STREAM test "
-        f"TOPICS {pulsar_topics[0]} "
-        "TRANSFORM pulsar_transform.simple", )
+        cursor,
+        f"CREATE PULSAR STREAM test TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple",
+    )
    common.start_stream(cursor, "test")
    time.sleep(1)

    def assert_message_not_consumed(message):
        vertices_with_msg = common.execute_and_fetch_all(
            cursor,
-            "MATCH (n: MESSAGE {" f"payload: '{message.decode('utf-8')}'" "}) RETURN n",
+            f"MATCH (n: MESSAGE {{payload: '{message.decode('utf-8')}'}}) RETURN n",
        )

        assert len(vertices_with_msg) == 0

    producer = pulsar_client.create_producer(
-        common.pulsar_default_namespace_topic(
-            pulsar_topics[0]), send_timeout_millis=60000)
+        common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
+    )
    producer.send(common.SIMPLE_MSG)

-    check_vertex_exists_with_topic_and_payload(
-        cursor, pulsar_topics[0], common.SIMPLE_MSG)
+    check_vertex_exists_with_topic_and_payload(cursor, pulsar_topics[0], common.SIMPLE_MSG)

    common.stop_stream(cursor, "test")

-    next_message = b"NEXT"
-    producer.send(next_message)
+    NEXT_MESSAGE = b"NEXT"
+    producer.send(NEXT_MESSAGE)

-    assert_message_not_consumed(next_message)
+    assert_message_not_consumed(NEXT_MESSAGE)

    common.start_stream(cursor, "test")
-    check_vertex_exists_with_topic_and_payload(
-        cursor, pulsar_topics[0], next_message)
+    check_vertex_exists_with_topic_and_payload(cursor, pulsar_topics[0], NEXT_MESSAGE)
    common.stop_stream(cursor, "test")

    common.drop_stream(cursor, "test")

-    lost_message = b"LOST"
-    valid_messages = [b"second message", b"third message"]
+    LOST_MESSAGE = b"LOST"
+    VALID_MESSAGES = [b"second message", b"third message"]

-    producer.send(lost_message)
+    producer.send(LOST_MESSAGE)

-    assert_message_not_consumed(lost_message)
+    assert_message_not_consumed(LOST_MESSAGE)

    common.execute_and_fetch_all(
-        cursor, "CREATE PULSAR STREAM test "
-        f"TOPICS {pulsar_topics[0]} "
-        "TRANSFORM pulsar_transform.simple", )
+        cursor,
+        f"CREATE PULSAR STREAM test TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple",
+    )

-    for message in valid_messages:
+    for message in VALID_MESSAGES:
        producer.send(message)
        assert_message_not_consumed(message)

    common.start_stream(cursor, "test")

-    assert_message_not_consumed(lost_message)
+    assert_message_not_consumed(LOST_MESSAGE)

-    for message in valid_messages:
-        check_vertex_exists_with_topic_and_payload(
-            cursor, pulsar_topics[0], message)
+    for message in VALID_MESSAGES:
+        check_vertex_exists_with_topic_and_payload(cursor, pulsar_topics[0], message)


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK)
-def test_check_stream(
-        pulsar_client,
-        pulsar_topics,
-        connection,
-        transformation):
+def test_check_stream(pulsar_client, pulsar_topics, connection, transformation):
    assert len(pulsar_topics) > 0
+    BATCH_SIZE = 1
+    INDEX_Of_FIRST_BATCH = 0
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM test "
-        f"TOPICS {pulsar_topics[0]} "
-        f"TRANSFORM {transformation} "
-        "BATCH_SIZE 1",
+        f"CREATE PULSAR STREAM test TOPICS {pulsar_topics[0]} TRANSFORM {transformation} BATCH_SIZE {BATCH_SIZE}",
    )
    common.start_stream(cursor, "test")
    time.sleep(1)

    producer = pulsar_client.create_producer(
-        common.pulsar_default_namespace_topic(
-            pulsar_topics[0]), send_timeout_millis=60000)
+        common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
+    )
    producer.send(common.SIMPLE_MSG)
-    check_vertex_exists_with_topic_and_payload(
-        cursor, pulsar_topics[0], common.SIMPLE_MSG)
+    check_vertex_exists_with_topic_and_payload(cursor, pulsar_topics[0], common.SIMPLE_MSG)
    common.stop_stream(cursor, "test")

-    messages = [b"first message", b"second message", b"third message"]
-    for message in messages:
+    MESSAGES = [b"first message", b"second message", b"third message"]
+    for message in MESSAGES:
        producer.send(message)

    def check_check_stream(batch_limit):
-        assert (
-            transformation == "pulsar_transform.simple"
-            or transformation == "pulsar_transform.with_parameters"
-        )
-        test_results = common.execute_and_fetch_all(
-            cursor, f"CHECK STREAM test BATCH_LIMIT {batch_limit}"
-        )
+        assert transformation == "pulsar_transform.simple" or transformation == "pulsar_transform.with_parameters"
+        test_results = common.execute_and_fetch_all(cursor, f"CHECK STREAM test BATCH_LIMIT {batch_limit}")
        assert len(test_results) == batch_limit

        for i in range(batch_limit):
-            message_as_str = messages[i].decode("utf-8")
+            message_as_str = MESSAGES[i].decode("utf-8")
+            assert (
+                BATCH_SIZE == 1
+            )  # If batch size != 1, then the usage of INDEX_Of_FIRST_BATCH must change: the result will have a list of queries (pair<parameters,query>)
+
            if transformation == "pulsar_transform.simple":
-                assert f"payload: '{message_as_str}'" in test_results[i][common.QUERY]
-                assert test_results[i][common.PARAMS] is None
+                assert (
+                    f"payload: '{message_as_str}'"
+                    in test_results[i][common.QUERIES][INDEX_Of_FIRST_BATCH][common.QUERY_LITERAL]
+                )
+                assert test_results[i][common.QUERIES][INDEX_Of_FIRST_BATCH][common.PARAMETERS_LITERAL] is None
            else:
-                assert f"payload: $payload" in test_results[i][
-                    common.QUERY] and f"topic: $topic" in test_results[i][common.QUERY]
-                parameters = test_results[i][common.PARAMS]
-                assert parameters["topic"] == common.pulsar_default_namespace_topic(
-                    pulsar_topics[0])
+                assert (
+                    f"payload: $payload" in test_results[i][common.QUERIES][INDEX_Of_FIRST_BATCH][common.QUERY_LITERAL]
+                    and f"topic: $topic" in test_results[i][common.QUERIES][INDEX_Of_FIRST_BATCH][common.QUERY_LITERAL]
+                )
+                parameters = test_results[i][common.QUERIES][INDEX_Of_FIRST_BATCH][common.PARAMETERS_LITERAL]
+                assert parameters["topic"] == common.pulsar_default_namespace_topic(pulsar_topics[0])
                assert parameters["payload"] == message_as_str

    check_check_stream(1)
@@ -218,45 +198,37 @@ def test_check_stream(
    check_check_stream(3)
    common.start_stream(cursor, "test")

-    for message in messages:
-        check_vertex_exists_with_topic_and_payload(
-            cursor, pulsar_topics[0], message)
+    for message in MESSAGES:
+        check_vertex_exists_with_topic_and_payload(cursor, pulsar_topics[0], message)


 def test_info_procedure(pulsar_client, pulsar_topics, connection):
    cursor = connection.cursor()
-    stream_name = 'test_stream'
+    STREAM_NAME = "test_stream"
    common.execute_and_fetch_all(
        cursor,
-        f"CREATE PULSAR STREAM {stream_name} "
-        f"TOPICS {','.join(pulsar_topics)} "
-        f"TRANSFORM pulsar_transform.simple ",
+        f"CREATE PULSAR STREAM {STREAM_NAME} TOPICS {','.join(pulsar_topics)} TRANSFORM pulsar_transform.simple ",
    )

-    stream_info = common.execute_and_fetch_all(cursor, f"CALL mg.pulsar_stream_info('{stream_name}') YIELD *")
+    stream_info = common.execute_and_fetch_all(cursor, f"CALL mg.pulsar_stream_info('{STREAM_NAME}') YIELD *")

    expected_stream_info = [(common.PULSAR_SERVICE_URL, pulsar_topics)]
    common.validate_info(stream_info, expected_stream_info)

+
 def test_show_streams(pulsar_client, pulsar_topics, connection):
    assert len(pulsar_topics) > 1
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM default_values "
-        f"TOPICS {pulsar_topics[0]} "
-        f"TRANSFORM pulsar_transform.simple ",
+        f"CREATE PULSAR STREAM default_values TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple ",
    )

-    batch_interval = 42
-    batch_size = 3
+    BATCH_INTERVAL = 42
+    BATCH_SIZE = 3
    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM complex_values "
-        f"TOPICS {','.join(pulsar_topics)} "
-        f"TRANSFORM pulsar_transform.with_parameters "
-        f"BATCH_INTERVAL {batch_interval} "
-        f"BATCH_SIZE {batch_size} ",
+        f"CREATE PULSAR STREAM complex_values TOPICS {','.join(pulsar_topics)} TRANSFORM pulsar_transform.with_parameters BATCH_INTERVAL {BATCH_INTERVAL} BATCH_SIZE {BATCH_SIZE} ",
    )

    assert len(common.execute_and_fetch_all(cursor, "SHOW STREAMS")) == 2
@@ -264,13 +236,7 @@ def test_show_streams(pulsar_client, pulsar_topics, connection):
    common.check_stream_info(
        cursor,
        "default_values",
-        ("default_values",
-         "pulsar",
-         100,
-         1000,
-         "pulsar_transform.simple",
-         None,
-         False),
+        ("default_values", "pulsar", 100, 1000, "pulsar_transform.simple", None, False),
    )

    common.check_stream_info(
@@ -279,8 +245,8 @@ def test_show_streams(pulsar_client, pulsar_topics, connection):
        (
            "complex_values",
            "pulsar",
-            batch_interval,
-            batch_size,
+            BATCH_INTERVAL,
+            BATCH_SIZE,
            "pulsar_transform.with_parameters",
            None,
            False,
@@ -289,19 +255,16 @@ def test_show_streams(pulsar_client, pulsar_topics, connection):


@pytest.mark.parametrize("operation", ["START", "STOP"])
-def test_start_and_stop_during_check(
-        pulsar_client,
-        pulsar_topics,
-        connection,
-        operation):
+def test_start_and_stop_during_check(pulsar_client, pulsar_topics, connection, operation):
    assert len(pulsar_topics) > 1
+    BATCH_SIZE = 1

    def stream_creator(stream_name):
-        return f"CREATE PULSAR STREAM {stream_name} TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple"
+        return f"CREATE PULSAR STREAM {stream_name} TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple BATCH_SIZE {BATCH_SIZE}"

    producer = pulsar_client.create_producer(
-        common.pulsar_default_namespace_topic(
-            pulsar_topics[0]), send_timeout_millis=60000)
+        common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
+    )

    def message_sender(msg):
        producer.send(msg)
@@ -311,7 +274,9 @@ def test_start_and_stop_during_check(
        connection,
        stream_creator,
        message_sender,
-        "Pulsar consumer test_stream is already stopped")
+        "Pulsar consumer test_stream is already stopped",
+        BATCH_SIZE,
+    )


 def test_check_already_started_stream(pulsar_topics, connection):
@@ -320,9 +285,7 @@ def test_check_already_started_stream(pulsar_topics, connection):

    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM started_stream "
-        f"TOPICS {pulsar_topics[0]} "
-        f"TRANSFORM pulsar_transform.simple",
+        f"CREATE PULSAR STREAM started_stream TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple",
    )
    common.start_stream(cursor, "started_stream")

@@ -333,6 +296,7 @@ def test_check_already_started_stream(pulsar_topics, connection):
 def test_start_checked_stream_after_timeout(pulsar_topics, connection):
    def stream_creator(stream_name):
        return f"CREATE PULSAR STREAM {stream_name} TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple"
+
    common.test_start_checked_stream_after_timeout(connection, stream_creator)


@@ -340,53 +304,80 @@ def test_restart_after_error(pulsar_client, pulsar_topics, connection):
    cursor = connection.cursor()
    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM test_stream "
-        f"TOPICS {pulsar_topics[0]} "
-        f"TRANSFORM pulsar_transform.query",
+        f"CREATE PULSAR STREAM test_stream TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.query",
    )

    common.start_stream(cursor, "test_stream")
    time.sleep(1)

    producer = pulsar_client.create_producer(
-        common.pulsar_default_namespace_topic(
-            pulsar_topics[0]), send_timeout_millis=60000)
+        common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
+    )
    producer.send(common.SIMPLE_MSG)
-    assert common.timed_wait(
-        lambda: not common.get_is_running(
-            cursor, "test_stream"))
+    assert common.timed_wait(lambda: not common.get_is_running(cursor, "test_stream"))

    common.start_stream(cursor, "test_stream")
    time.sleep(1)
    producer.send(b"CREATE (n:VERTEX { id : 42 })")
-    assert common.check_one_result_row(
-        cursor, "MATCH (n:VERTEX { id : 42 }) RETURN n")
+    assert common.check_one_result_row(cursor, "MATCH (n:VERTEX { id : 42 }) RETURN n")


@pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK)
 def test_service_url(pulsar_client, pulsar_topics, connection, transformation):
    assert len(pulsar_topics) > 0
    cursor = connection.cursor()
-    local = "pulsar://127.0.0.1:6650"
+    LOCAL = "pulsar://127.0.0.1:6650"
    common.execute_and_fetch_all(
        cursor,
-        "CREATE PULSAR STREAM test "
-        f"TOPICS {','.join(pulsar_topics)} "
-        f"TRANSFORM {transformation} "
-        f"SERVICE_URL '{local}'",
+        f"CREATE PULSAR STREAM test TOPICS {','.join(pulsar_topics)} TRANSFORM {transformation} SERVICE_URL '{LOCAL}'",
    )
    common.start_stream(cursor, "test")
    time.sleep(5)

    for topic in pulsar_topics:
        producer = pulsar_client.create_producer(
-            common.pulsar_default_namespace_topic(topic),
-            send_timeout_millis=60000)
+            common.pulsar_default_namespace_topic(topic), send_timeout_millis=60000
+        )
        producer.send(common.SIMPLE_MSG)

    for topic in pulsar_topics:
-        check_vertex_exists_with_topic_and_payload(
-            cursor, topic, common.SIMPLE_MSG)
+        check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)
+
+
+def test_check_stream_same_number_of_queries_than_messages(pulsar_client, pulsar_topics, connection):
+    assert len(pulsar_topics) > 0
+
+    TRANSFORMATION = "common_transform.check_stream_no_filtering"
+
+    def stream_creator(stream_name, batch_size):
+        return f"CREATE PULSAR STREAM {stream_name} TOPICS {pulsar_topics[0]} TRANSFORM {TRANSFORMATION} BATCH_INTERVAL 3000 BATCH_SIZE  {batch_size} "
+
+    producer = pulsar_client.create_producer(
+        common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
+    )
+
+    def message_sender(msg):
+        producer.send(msg)
+
+    common.test_check_stream_same_number_of_queries_than_messages(connection, stream_creator, message_sender)
+
+
+def test_check_stream_different_number_of_queries_than_messages(pulsar_client, pulsar_topics, connection):
+    assert len(pulsar_topics) > 0
+
+    TRANSFORMATION = "common_transform.check_stream_with_filtering"
+
+    def stream_creator(stream_name, batch_size):
+        return f"CREATE PULSAR STREAM {stream_name} TOPICS {pulsar_topics[0]} TRANSFORM {TRANSFORMATION} BATCH_INTERVAL 3000 BATCH_SIZE  {batch_size} "
+
+    producer = pulsar_client.create_producer(
+        common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
+    )
+
+    def message_sender(msg):
+        producer.send(msg)
+
+    common.test_check_stream_different_number_of_queries_than_messages(connection, stream_creator, message_sender)


 if __name__ == "__main__":
--- a/tests/e2e/streams/transformations/CMakeLists.txt
+++ b/tests/e2e/streams/transformations/CMakeLists.txt
@@ -1,3 +1,4 @@
 copy_streams_e2e_python_files(kafka_transform.py)
 copy_streams_e2e_python_files(pulsar_transform.py)
+copy_streams_e2e_python_files(common_transform.py)
 add_query_module(c_transformations c_transformations.cpp)
--- a/tests/e2e/streams/transformations/common_transform.py
+++ b/tests/e2e/streams/transformations/common_transform.py
@@ -0,0 +1,57 @@
+# Copyright 2021 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.transformation
+def check_stream_no_filtering(
+    context: mgp.TransCtx, messages: mgp.Messages
+) -> mgp.Record(query=str, parameters=mgp.Map):
+
+    result_queries = []
+
+    for i in range(0, messages.total_messages()):
+        message = messages.message_at(i)
+        payload_as_str = message.payload().decode("utf-8")
+        result_queries.append(
+            mgp.Record(query=f"Message: {payload_as_str}", parameters={"value": f"Parameter: {payload_as_str}"})
+        )
+
+    return result_queries
+
+
+@mgp.transformation
+def check_stream_with_filtering(
+    context: mgp.TransCtx, messages: mgp.Messages
+) -> mgp.Record(query=str, parameters=mgp.Map):
+
+    result_queries = []
+
+    for i in range(0, messages.total_messages()):
+        message = messages.message_at(i)
+        payload_as_str = message.payload().decode("utf-8")
+
+        if "a" in payload_as_str:
+            continue
+
+        result_queries.append(
+            mgp.Record(query=f"Message: {payload_as_str}", parameters={"value": f"Parameter: {payload_as_str}"})
+        )
+
+        if "b" in payload_as_str:
+            result_queries.append(
+                mgp.Record(
+                    query=f"Message: extra_{payload_as_str}", parameters={"value": f"Parameter: extra_{payload_as_str}"}
+                )
+            )
+
+    return result_queries
--- a/tests/e2e/streams/transformations/kafka_transform.py
+++ b/tests/e2e/streams/transformations/kafka_transform.py
@@ -13,9 +13,7 @@ import mgp


@mgp.transformation
-def simple(
-    context: mgp.TransCtx, messages: mgp.Messages
-) -> mgp.Record(query=str, parameters=mgp.Map):
+def simple(context: mgp.TransCtx, messages: mgp.Messages) -> mgp.Record(query=str, parameters=mgp.Map):

    result_queries = []

@@ -32,15 +30,15 @@ def simple(
                    offset: '{message.offset()}',
                    topic: '{message.topic_name()}'
                }})""",
-                parameters=None))
+                parameters=None,
+            )
+        )

    return result_queries


@mgp.transformation
-def with_parameters(
-    context: mgp.TransCtx, messages: mgp.Messages
-) -> mgp.Record(query=str, parameters=mgp.Map):
+def with_parameters(context: mgp.TransCtx, messages: mgp.Messages) -> mgp.Record(query=str, parameters=mgp.Map):

    result_queries = []

@@ -61,7 +59,10 @@ def with_parameters(
                    "timestamp": message.timestamp(),
                    "payload": payload_as_str,
                    "offset": message.offset(),
-                    "topic": message.topic_name()}))
+                    "topic": message.topic_name(),
+                },
+            )
+        )

    return result_queries

@@ -76,8 +77,6 @@ def query(
        message = messages.message_at(i)
        assert message.source_type() == mgp.SOURCE_TYPE_KAFKA
        payload_as_str = message.payload().decode("utf-8")
-        result_queries.append(
-            mgp.Record(query=payload_as_str, parameters=None)
-        )
+        result_queries.append(mgp.Record(query=payload_as_str, parameters=None))

    return result_queries
--- a/tests/e2e/streams/transformations/pulsar_transform.py
+++ b/tests/e2e/streams/transformations/pulsar_transform.py
@@ -13,9 +13,7 @@ import mgp


@mgp.transformation
-def simple(context: mgp.TransCtx,
-           messages: mgp.Messages
-           ) -> mgp.Record(query=str, parameters=mgp.Map):
+def simple(context: mgp.TransCtx, messages: mgp.Messages) -> mgp.Record(query=str, parameters=mgp.Map):

    result_queries = []

@@ -30,15 +28,15 @@ def simple(context: mgp.TransCtx,
                    payload: '{payload_as_str}',
                    topic: '{message.topic_name()}'
                }})""",
-                parameters=None))
+                parameters=None,
+            )
+        )

    return result_queries


@mgp.transformation
-def with_parameters(context: mgp.TransCtx,
-                    messages: mgp.Messages
-                    ) -> mgp.Record(query=str, parameters=mgp.Map):
+def with_parameters(context: mgp.TransCtx, messages: mgp.Messages) -> mgp.Record(query=str, parameters=mgp.Map):

    result_queries = []

@@ -53,23 +51,21 @@ def with_parameters(context: mgp.TransCtx,
                    payload: $payload,
                    topic: $topic
                })""",
-                parameters={
-                    "payload": payload_as_str,
-                    "topic": message.topic_name()}))
+                parameters={"payload": payload_as_str, "topic": message.topic_name()},
+            )
+        )

    return result_queries


@mgp.transformation
-def query(messages: mgp.Messages
-          ) -> mgp.Record(query=str, parameters=mgp.Nullable[mgp.Map]):
+def query(messages: mgp.Messages) -> mgp.Record(query=str, parameters=mgp.Nullable[mgp.Map]):
    result_queries = []

    for i in range(0, messages.total_messages()):
        message = messages.message_at(i)
        assert message.source_type() == mgp.SOURCE_TYPE_PULSAR
        payload_as_str = message.payload().decode("utf-8")
-        result_queries.append(mgp.Record(
-            query=payload_as_str, parameters=None))
+        result_queries.append(mgp.Record(query=payload_as_str, parameters=None))

    return result_queries
Author	SHA1	Message	Date
Jure Bajic	cb621e227b	Support Ubuntu 22.04 (#411 ) * Fix toolchain url * Add 22.04 os script * Update centos-9 stream * Update zlib * Upgrade dotnet version * Add libipt * Update toolchain extension * Add release container * Remove CentOS 8 * Enable arm packaging * Add centos 9 container * Ignore missing python virtualenv * Add virtualenvwrapper * Rearrange order * Install virtualenv without user * Run virtualenv via python * Add logrotate dependency * Bump up checkout action version	2022-06-23 14:09:00 +02:00
jbajic	9f6f747b42	Update version and date	2022-06-15 23:06:38 +02:00
Tyler Neely	1fb49c4865	Add an uncatchable kill method to the python MemgraphInstanceRunner for testing abrupt failures (#405 )	2022-06-15 12:16:11 +03:00
János Benjamin Antal	df1485aeec	Run clang-tidy also on our header files (#402 )	2022-06-10 11:26:24 +02:00
Jure Bajic	b2e1056389	Fix websocket connection (#403 ) * Fix websocket connection with other clients	2022-06-09 22:19:43 +02:00
Jeremy B	e4c9411e63	Correcting USE_JEMALLOC flag (#401 ) * Correcting USE_JEMALLOC flag * Removing references to jemalloc local copy in /libs	2022-06-09 18:55:28 +02:00
Jeremy B	a0bc1371dd	Adding "raw message" column to the result returned by CHECK STREAM query (#394 ) * Adding "raw message" column to the result return by CHECK STREAM query * Update way results of CHECK STREAM are built * Adapting CHECK STREAM integration tests (Pulsar/Kafka) to run with new result structure * Adding new tests covering the check stream functionality * Uppercase constants in stream tests * Reformat f-strings	2022-06-08 23:17:44 +02:00
Marko Budiselić	21ad5d4328	Fix SHOW REPLICATION ROLE and SHOW REPLICAS (#376 )	2022-05-20 20:17:59 -07:00
Marko Budiselić	8e3ab1ad0f	Add frequent replica ping (#380 )	2022-05-20 19:29:17 -07:00