Allow clang compilers with different names

Rather than inspect for exactly binary name of "clang", check the
compiler ID instead.

.github/workflows/daily_benchmark.yaml

@@ -32,7 +32,7 @@ jobs:
 
           # Build only memgraph release binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=release ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=release ..
           make -j$THREADS
 
       - name: Run macro benchmarks

.github/workflows/diff.yaml

@@ -17,9 +17,9 @@ on:
       - "licenses/*"
 
 jobs:
-  release_build:
-    name: "Release build"
-    runs-on: [self-hosted, Linux, X64, gamora]
+  community_build:
+    name: "Community build"
+    runs-on: [self-hosted, Linux, X64, Diff]
     env:
       THREADS: 24
       MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
@@ -33,10 +33,202 @@ jobs:
           # branches and tags. (default: 1)
           fetch-depth: 0
 
-      - name: Check e2e service dependencies
+      - name: Build community binaries
         run: |
-          cd tests/e2e
-          ./dependency_check.sh
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Initialize dependencies.
+          ./init
+
+          # Build community binaries.
+          cd build
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=RelWithDebInfo -DMG_ENTERPRISE=OFF ..
+          make -j$THREADS
+
+      - name: Run unit tests
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Run unit tests.
+          cd build
+          ctest -R memgraph__unit --output-on-failure -j$THREADS
+
+  code_analysis:
+    name: "Code analysis"
+    runs-on: [self-hosted, Linux, X64, Diff]
+    env:
+      THREADS: 24
+      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
+      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
+
+    steps:
+      - name: Set up repository
+        uses: actions/checkout@v3
+        with:
+          # Number of commits to fetch. `0` indicates all history for all
+          # branches and tags. (default: 1)
+          fetch-depth: 0
+
+        # This is also needed if we want do to comparison against other branches
+        # See https://github.community/t/checkout-code-fails-when-it-runs-lerna-run-test-since-master/17920
+      - name: Fetch all history for all tags and branches
+        run: git fetch
+
+      - name: Initialize deps
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Initialize dependencies.
+          ./init
+
+      - name: Set base branch
+        if: ${{ github.event_name == 'pull_request' }}
+        run: |
+          echo "BASE_BRANCH=origin/${{ github.base_ref }}" >> $GITHUB_ENV
+
+      - name: Set base branch # if we manually dispatch or push to master
+        if: ${{ github.event_name != 'pull_request' }}
+        run: |
+          echo "BASE_BRANCH=origin/master" >> $GITHUB_ENV
+
+      - name: Python code analysis
+        run: |
+          CHANGED_FILES=$(git diff -U0 ${{ env.BASE_BRANCH }}... --name-only)
+          for file in ${CHANGED_FILES}; do
+            echo ${file}
+            if [[ ${file} == *.py ]]; then
+              python3 -m black --check --diff ${file}
+              python3 -m isort --profile black --check-only --diff ${file}
+            fi
+          done
+
+      - name: Build combined ASAN, UBSAN and coverage binaries
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          cd build
+          CC=clang CXX=clang++ cmake -DTEST_COVERAGE=ON -DASAN=ON -DUBSAN=ON ..
+          make -j$THREADS memgraph__unit
+
+      - name: Run unit tests
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Run unit tests. It is restricted to 2 threads intentionally, because higher concurrency makes the timing related tests unstable.
+          cd build
+          LSAN_OPTIONS=suppressions=$PWD/../tools/lsan.supp UBSAN_OPTIONS=halt_on_error=1 ctest -R memgraph__unit --output-on-failure -j2
+
+      - name: Compute code coverage
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Compute code coverage.
+          cd tools/github
+          ./coverage_convert
+
+          # Package code coverage.
+          cd generated
+          tar -czf code_coverage.tar.gz coverage.json html report.json summary.rmu
+
+      - name: Save code coverage
+        uses: actions/upload-artifact@v3
+        with:
+          name: "Code coverage"
+          path: tools/github/generated/code_coverage.tar.gz
+
+      - name: Run clang-tidy
+        run: |
+          source /opt/toolchain-v4/activate
+
+          # Restrict clang-tidy results only to the modified parts
+          git diff -U0 ${{ env.BASE_BRANCH }}... -- src | ./tools/github/clang-tidy/clang-tidy-diff.py -p 1 -j $THREADS -path build  -regex ".+\.cpp" | 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
+
+  debug_build:
+    name: "Debug build"
+    runs-on: [self-hosted, Linux, X64, Diff]
+    env:
+      THREADS: 24
+      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
+      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
+
+    steps:
+      - name: Set up repository
+        uses: actions/checkout@v3
+        with:
+          # Number of commits to fetch. `0` indicates all history for all
+          # branches and tags. (default: 1)
+          fetch-depth: 0
+
+      - name: Build debug binaries
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Initialize dependencies.
+          ./init
+
+          # Build debug binaries.
+          cd build
+          CC=clang CXX=clang++ cmake ..
+          make -j$THREADS
+
+      - name: Run leftover CTest tests
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Run leftover CTest tests (all except unit and benchmark tests).
+          cd build
+          ctest -E "(memgraph__unit|memgraph__benchmark)" --output-on-failure
+
+      - name: Run drivers tests
+        run: |
+          ./tests/drivers/run.sh
+
+      - name: Run integration tests
+        run: |
+          tests/integration/run.sh
+
+      - name: Run cppcheck and clang-format
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Run cppcheck and clang-format.
+          cd tools/github
+          ./cppcheck_and_clang_format diff
+
+      - name: Save cppcheck and clang-format errors
+        uses: actions/upload-artifact@v3
+        with:
+          name: "Code coverage"
+          path: tools/github/cppcheck_and_clang_format.txt
+
+  release_build:
+    name: "Release build"
+    runs-on: [self-hosted, Linux, X64, Diff]
+    env:
+      THREADS: 24
+      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
+      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
+
+    steps:
+      - name: Set up repository
+        uses: actions/checkout@v3
+        with:
+          # Number of commits to fetch. `0` indicates all history for all
+          # branches and tags. (default: 1)
+          fetch-depth: 0
+
 
       - name: Build release binaries
         run: |
@@ -48,7 +240,7 @@ jobs:
 
           # Build release binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=Release ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=Release ..
           make -j$THREADS
 
       - name: Run GQL Behave tests
@@ -75,6 +267,13 @@ jobs:
           cd build
           ctest -R memgraph__unit --output-on-failure -j$THREADS
 
+      - name: Ensure Kafka and Pulsar are up
+        run: |
+          cd tests/e2e/streams/kafka
+          docker-compose up -d
+          cd ../pulsar
+          docker-compose up -d
+
       - name: Run e2e tests
         run: |
           cd tests
@@ -83,6 +282,14 @@ jobs:
           cd e2e
           ./run.sh
 
+      - name: Ensure Kafka and Pulsar are down
+        if: always()
+        run: |
+          cd tests/e2e/streams/kafka
+          docker-compose down
+          cd ../pulsar
+          docker-compose down
+
       - name: Run stress test (plain)
         run: |
           cd tests/stress
@@ -129,3 +336,119 @@ jobs:
           path: |
             # multiple paths could be defined
             build/logs
+
+  release_jepsen_test:
+    name: "Release Jepsen Test"
+    runs-on: [self-hosted, Linux, X64, Debian10, JepsenControl]
+    #continue-on-error: true
+    env:
+      THREADS: 24
+      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
+      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
+
+    steps:
+      - name: Set up repository
+        uses: actions/checkout@v3
+        with:
+          # Number of commits to fetch. `0` indicates all history for all
+          # branches and tags. (default: 1)
+          fetch-depth: 0
+
+      - name: Build release binaries
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+          # Initialize dependencies.
+          ./init
+          # Build only memgraph release binarie.
+          cd build
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=RelWithDebInfo ..
+          make -j$THREADS memgraph
+
+      - name: Run Jepsen tests
+        run: |
+          cd tests/jepsen
+          ./run.sh test-all-individually --binary ../../build/memgraph --ignore-run-stdout-logs --ignore-run-stderr-logs
+
+      - name: Save Jepsen report
+        uses: actions/upload-artifact@v3
+        if: ${{ always() }}
+        with:
+          name: "Jepsen Report"
+          path: tests/jepsen/Jepsen.tar.gz
+
+  release_benchmarks:
+    name: "Release benchmarks"
+    runs-on: [self-hosted, Linux, X64, Diff, Gen7]
+    env:
+      THREADS: 24
+      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
+      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
+
+    steps:
+      - name: Set up repository
+        uses: actions/checkout@v3
+        with:
+          # Number of commits to fetch. `0` indicates all history for all
+          # branches and tags. (default: 1)
+          fetch-depth: 0
+
+      - name: Build release binaries
+        run: |
+          # Activate toolchain.
+          source /opt/toolchain-v4/activate
+
+          # Initialize dependencies.
+          ./init
+
+          # Build only memgraph release binaries.
+          cd build
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=release ..
+          make -j$THREADS
+
+      - name: Run macro benchmarks
+        run: |
+          cd tests/macro_benchmark
+          ./harness QuerySuite MemgraphRunner \
+            --groups aggregation 1000_create unwind_create dense_expand match \
+            --no-strict
+
+      - name: Get branch name (merge)
+        if: github.event_name != 'pull_request'
+        shell: bash
+        run: echo "BRANCH_NAME=$(echo ${GITHUB_REF#refs/heads/} | tr / -)" >> $GITHUB_ENV
+
+      - name: Get branch name (pull request)
+        if: github.event_name == 'pull_request'
+        shell: bash
+        run: echo "BRANCH_NAME=$(echo ${GITHUB_HEAD_REF} | tr / -)" >> $GITHUB_ENV
+
+      - name: Upload macro benchmark results
+        run: |
+          cd tools/bench-graph-client
+          virtualenv -p python3 ve3
+          source ve3/bin/activate
+          pip install -r requirements.txt
+          ./main.py --benchmark-name "macro_benchmark" \
+                    --benchmark-results "../../tests/macro_benchmark/.harness_summary" \
+                    --github-run-id "${{ github.run_id }}" \
+                    --github-run-number "${{ github.run_number }}" \
+                    --head-branch-name "${{ env.BRANCH_NAME }}"
+
+      # TODO (andi) No need for path flags and for --disk-storage and --in-memory-analytical
+      - name: Run mgbench
+        run: |
+          cd tests/mgbench
+          ./benchmark.py vendor-native --num-workers-for-benchmark 12 --export-results benchmark_result.json pokec/medium/*/*
+
+      - name: Upload mgbench results
+        run: |
+          cd tools/bench-graph-client
+          virtualenv -p python3 ve3
+          source ve3/bin/activate
+          pip install -r requirements.txt
+          ./main.py --benchmark-name "mgbench" \
+                    --benchmark-results "../../tests/mgbench/benchmark_result.json" \
+                    --github-run-id "${{ github.run_id }}" \
+                    --github-run-number "${{ github.run_number }}" \
+                    --head-branch-name "${{ env.BRANCH_NAME }}"

.github/workflows/full_clang_tidy.yaml

@@ -31,7 +31,7 @@ jobs:
           # Build debug binaries.
 
           cd build
-          cmake ..
+          CC=clang CXX=clang++ cmake ..
           make -j$THREADS
 
       - name: Run clang-tidy

.github/workflows/performance_benchmarks.yaml

@@ -30,7 +30,7 @@ jobs:
 
           # Build only memgraph release binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=release ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=release ..
           make -j$THREADS
 
       - name: Get branch name (merge)

.github/workflows/release_centos8.yaml

@@ -1,318 +0,0 @@
-name: Release CentOS 8
-
-on:
-  workflow_dispatch:
-    inputs:
-      build_type:
-        type: choice
-        description: "Memgraph Build type. Default value is Release."
-        default: 'Release'
-        options:
-          - Release
-          - RelWithDebInfo
-
-  schedule:
-    - cron: "0 22 * * *"
-
-jobs:
-  community_build:
-    name: "Community build"
-    runs-on: [self-hosted, Linux, X64, CentOS8]
-    env:
-      THREADS: 24
-      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
-      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
-    timeout-minutes: 960
-
-    steps:
-      - name: Set up repository
-        uses: actions/checkout@v3
-        with:
-          # Number of commits to fetch. `0` indicates all history for all
-          # branches and tags. (default: 1)
-          fetch-depth: 0
-
-      - name: Build community binaries
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Initialize dependencies.
-          ./init
-
-          # Set default build_type to Release
-          INPUT_BUILD_TYPE=${{ github.event.inputs.build_type }}
-          BUILD_TYPE=${INPUT_BUILD_TYPE:-"Release"}
-
-          # Build community binaries.
-          cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE -DMG_ENTERPRISE=OFF ..
-          make -j$THREADS
-
-      - name: Run unit tests
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run unit tests.
-          cd build
-          ctest -R memgraph__unit --output-on-failure
-
-  coverage_build:
-    name: "Coverage build"
-    runs-on: [self-hosted, Linux, X64, CentOS8]
-    env:
-      THREADS: 24
-      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
-      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
-
-    steps:
-      - name: Set up repository
-        uses: actions/checkout@v3
-        with:
-          # Number of commits to fetch. `0` indicates all history for all
-          # branches and tags. (default: 1)
-          fetch-depth: 0
-
-      - name: Build coverage binaries
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Initialize dependencies.
-          ./init
-
-          # Build coverage binaries.
-          cd build
-          cmake -DTEST_COVERAGE=ON ..
-          make -j$THREADS memgraph__unit
-
-      - name: Run unit tests
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run unit tests.
-          cd build
-          ctest -R memgraph__unit --output-on-failure
-
-      - name: Compute code coverage
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Compute code coverage.
-          cd tools/github
-          ./coverage_convert
-
-          # Package code coverage.
-          cd generated
-          tar -czf code_coverage.tar.gz coverage.json html report.json summary.rmu
-
-      - name: Save code coverage
-        uses: actions/upload-artifact@v3
-        with:
-          name: "Code coverage"
-          path: tools/github/generated/code_coverage.tar.gz
-
-  debug_build:
-    name: "Debug build"
-    runs-on: [self-hosted, Linux, X64, CentOS8]
-    env:
-      THREADS: 24
-      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
-      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
-
-    steps:
-      - name: Set up repository
-        uses: actions/checkout@v3
-        with:
-          # Number of commits to fetch. `0` indicates all history for all
-          # branches and tags. (default: 1)
-          fetch-depth: 0
-
-      - name: Build debug binaries
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Initialize dependencies.
-          ./init
-
-          # Build debug binaries.
-          cd build
-          cmake ..
-          make -j$THREADS
-
-      - name: Run leftover CTest tests
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run leftover CTest tests (all except unit and benchmark tests).
-          cd build
-          ctest -E "(memgraph__unit|memgraph__benchmark)" --output-on-failure
-
-      - name: Run drivers tests
-        run: |
-          ./tests/drivers/run.sh
-
-      - name: Run integration tests
-        run: |
-          tests/integration/run.sh
-
-      - name: Run cppcheck and clang-format
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run cppcheck and clang-format.
-          cd tools/github
-          ./cppcheck_and_clang_format diff
-
-      - name: Save cppcheck and clang-format errors
-        uses: actions/upload-artifact@v3
-        with:
-          name: "Code coverage"
-          path: tools/github/cppcheck_and_clang_format.txt
-
-  release_build:
-    name: "Release build"
-    runs-on: [self-hosted, Linux, X64, CentOS8]
-    env:
-      THREADS: 24
-      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
-      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
-    timeout-minutes: 960
-
-    steps:
-      - name: Set up repository
-        uses: actions/checkout@v3
-        with:
-          # Number of commits to fetch. `0` indicates all history for all
-          # branches and tags. (default: 1)
-          fetch-depth: 0
-
-      - name: Build release binaries
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Initialize dependencies.
-          ./init
-
-          # Set default build_type to Release
-          INPUT_BUILD_TYPE=${{ github.event.inputs.build_type }}
-          BUILD_TYPE=${INPUT_BUILD_TYPE:-"Release"}
-
-          # Build release binaries.
-          cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
-          make -j$THREADS
-
-      - name: Create enterprise RPM package
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          cd build
-
-          # create mgconsole
-          # we use the -B to force the build
-          make -j$THREADS -B mgconsole
-
-          # Create enterprise RPM package.
-          mkdir output && cd output
-          cpack -G RPM --config ../CPackConfig.cmake
-          rpmlint memgraph*.rpm
-
-      - name: Save enterprise RPM package
-        uses: actions/upload-artifact@v3
-        with:
-          name: "Enterprise RPM package"
-          path: build/output/memgraph*.rpm
-
-      - name: Run micro benchmark tests
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run micro benchmark tests.
-          cd build
-          # The `eval` benchmark needs a large stack limit.
-          ulimit -s 262144
-          ctest -R memgraph__benchmark -V
-
-      - name: Run macro benchmark tests
-        run: |
-          cd tests/macro_benchmark
-          ./harness QuerySuite MemgraphRunner \
-            --groups aggregation 1000_create unwind_create dense_expand match \
-            --no-strict
-
-      - name: Run parallel macro benchmark tests
-        run: |
-          cd tests/macro_benchmark
-          ./harness QueryParallelSuite MemgraphRunner \
-            --groups aggregation_parallel create_parallel bfs_parallel \
-            --num-database-workers 9 --num-clients-workers 30 \
-            --no-strict
-
-      - name: Run GQL Behave tests
-        run: |
-          cd tests
-          ./setup.sh /opt/toolchain-v4/activate
-          cd gql_behave
-          ./continuous_integration
-
-      - name: Save quality assurance status
-        uses: actions/upload-artifact@v3
-        with:
-          name: "GQL Behave Status"
-          path: |
-            tests/gql_behave/gql_behave_status.csv
-            tests/gql_behave/gql_behave_status.html
-
-      - name: Run unit tests
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run unit tests.
-          cd build
-          ctest -R memgraph__unit --output-on-failure
-
-      - name: Run e2e tests
-        run: |
-          cd tests
-          ./setup.sh /opt/toolchain-v4/activate
-          source ve3/bin/activate_e2e
-          cd e2e
-          ./run.sh
-
-      - name: Run stress test (plain)
-        run: |
-          cd tests/stress
-          ./continuous_integration
-
-      - name: Run stress test (SSL)
-        run: |
-          cd tests/stress
-          ./continuous_integration --use-ssl
-
-      - name: Run stress test (large)
-        run: |
-          cd tests/stress
-          ./continuous_integration --large-dataset
-
-      - name: Run durability test (plain)
-        run: |
-          cd tests/stress
-          source ve3/bin/activate
-          python3 durability --num-steps 5
-
-      - name: Run durability test (large)
-        run: |
-          cd tests/stress
-          source ve3/bin/activate
-          python3 durability --num-steps 20

.github/workflows/release_debian10.yaml

@@ -46,7 +46,7 @@ jobs:
 
           # Build community binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE -DMG_ENTERPRISE=OFF ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE -DMG_ENTERPRISE=OFF ..
           make -j$THREADS
 
       - name: Run unit tests
@@ -84,7 +84,7 @@ jobs:
 
           # Build coverage binaries.
           cd build
-          cmake -DTEST_COVERAGE=ON ..
+          CC=clang CXX=clang++ cmake -DTEST_COVERAGE=ON ..
           make -j$THREADS memgraph__unit
 
       - name: Run unit tests
@@ -141,7 +141,7 @@ jobs:
 
           # Build debug binaries.
           cd build
-          cmake ..
+          CC=clang CXX=clang++ cmake ..
           make -j$THREADS
 
       - name: Run leftover CTest tests
@@ -207,7 +207,7 @@ jobs:
 
           # Build release binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
           make -j$THREADS
 
       - name: Create enterprise DEB package
@@ -346,7 +346,7 @@ jobs:
 
           # Build only memgraph release binary.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
           make -j$THREADS memgraph
 
       - name: Run Jepsen tests

.github/workflows/release_ubuntu2004.yaml

@@ -46,7 +46,7 @@ jobs:
 
           # Build community binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE -DMG_ENTERPRISE=OFF ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE -DMG_ENTERPRISE=OFF ..
           make -j$THREADS
 
       - name: Run unit tests
@@ -84,7 +84,7 @@ jobs:
 
           # Build coverage binaries.
           cd build
-          cmake -DTEST_COVERAGE=ON ..
+          CC=clang CXX=clang++ cmake -DTEST_COVERAGE=ON ..
           make -j$THREADS memgraph__unit
 
       - name: Run unit tests
@@ -141,7 +141,7 @@ jobs:
 
           # Build debug binaries.
           cd build
-          cmake ..
+          CC=clang CXX=clang++ cmake ..
           make -j$THREADS
 
       - name: Run leftover CTest tests
@@ -207,7 +207,7 @@ jobs:
 
           # Build release binaries.
           cd build
-          cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
+          CC=clang CXX=clang++ cmake -DCMAKE_BUILD_TYPE=$BUILD_TYPE ..
           make -j$THREADS
 
       - name: Create enterprise DEB package

CMakeLists.txt

@@ -27,21 +27,14 @@ else ()
   message(STATUS "CCache: Not used")
 endif(CCACHE_FOUND AND USE_CCACHE)
 
-# choose a compiler
-# NOTE: must be choosen before use of project() or enable_language()
-find_program(CLANG_FOUND clang)
-find_program(CLANGXX_FOUND clang++)
-if (CLANG_FOUND AND CLANGXX_FOUND)
-  set(CMAKE_C_COMPILER ${CLANG_FOUND})
-  set(CMAKE_CXX_COMPILER ${CLANGXX_FOUND})
-else()
-  message(FATAL_ERROR "Couldn't find clang and/or clang++!")
-endif()
-
 # -----------------------------------------------------------------------------
 
 project(memgraph LANGUAGES C CXX)
 
+if (NOT CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
+    message(FATAL_ERROR "Clang compiler is required")
+endif()
+
 #TODO: upgrade to cmake 3.24 + CheckIPOSupported
 #cmake_policy(SET CMP0138 NEW)
 #include(CheckIPOSupported)

libs/setup.sh

@@ -259,7 +259,7 @@ repo_clone_try_double "${primary_urls[absl]}" "${secondary_urls[absl]}" "absl" "
 
 # jemalloc ea6b3e973b477b8061e0076bb257dbd7f3faa756
 JEMALLOC_COMMIT_VERSION="5.2.1"
-repo_clone_try_double "${secondary_urls[jemalloc]}" "${secondary_urls[jemalloc]}" "jemalloc" "$JEMALLOC_COMMIT_VERSION"
+repo_clone_try_double "${primary_urls[jemalloc]}" "${secondary_urls[jemalloc]}" "jemalloc" "$JEMALLOC_COMMIT_VERSION"
 
 # this is hack for cmake in libs to set path, and for FindJemalloc to use Jemalloc_INCLUDE_DIR
 pushd jemalloc

query_modules/schema.cpp

@@ -108,31 +108,83 @@ void Schema::ProcessPropertiesRel(mgp::Record &record, const std::string_view &t
   record.Insert(std::string(kReturnMandatory).c_str(), mandatory);
 }
 
+struct Property {
+  std::string name;
+  mgp::Value value;
+
+  Property(const std::string &name, mgp::Value &&value) : name(name), value(std::move(value)) {}
+};
+
+struct LabelsHash {
+  std::size_t operator()(const std::set<std::string> &set) const {
+    std::size_t seed = set.size();
+    for (const auto &i : set) {
+      seed ^= std::hash<std::string>{}(i) + 0x9e3779b9 + (seed << 6) + (seed >> 2);
+    }
+    return seed;
+  }
+};
+
+struct LabelsComparator {
+  bool operator()(const std::set<std::string> &lhs, const std::set<std::string> &rhs) const { return lhs == rhs; }
+};
+
+struct PropertyComparator {
+  bool operator()(const Property &lhs, const Property &rhs) const { return lhs.name < rhs.name; }
+};
+
+struct PropertyInfo {
+  std::set<Property, PropertyComparator> properties;
+  bool mandatory;
+};
+
 void Schema::NodeTypeProperties(mgp_list * /*args*/, mgp_graph *memgraph_graph, mgp_result *result,
                                 mgp_memory *memory) {
   mgp::MemoryDispatcherGuard guard{memory};
   const auto record_factory = mgp::RecordFactory(result);
   try {
-    const mgp::Graph graph = mgp::Graph(memgraph_graph);
-    for (auto node : graph.Nodes()) {
-      std::string type;
-      mgp::List labels = mgp::List();
+    std::unordered_map<std::set<std::string>, PropertyInfo, LabelsHash, LabelsComparator> node_types_properties;
+
+    for (auto node : mgp::Graph(memgraph_graph).Nodes()) {
+      std::set<std::string> labels_set = {};
       for (auto label : node.Labels()) {
-        labels.AppendExtend(mgp::Value(label));
-        type += ":`" + std::string(label) + "`";
+        labels_set.emplace(label);
+      }
+
+      if (node_types_properties.find(labels_set) == node_types_properties.end()) {
+        node_types_properties[labels_set] = PropertyInfo{std::set<Property, PropertyComparator>(), true};
       }
 
       if (node.Properties().empty()) {
-        auto record = record_factory.NewRecord();
-        ProcessPropertiesNode<std::string>(record, type, labels, "", "", false);
+        node_types_properties[labels_set].mandatory = false;  // if there is node with no property, it is not mandatory
         continue;
       }
 
+      auto &property_info = node_types_properties.at(labels_set);
       for (auto &[key, prop] : node.Properties()) {
-        auto property_type = mgp::List();
+        property_info.properties.emplace(key, std::move(prop));
+        if (property_info.mandatory) {
+          property_info.mandatory =
+              property_info.properties.size() == 1;  // if there is only one property, it is mandatory
+        }
+      }
+    }
+
+    for (auto &[labels, property_info] : node_types_properties) {
+      std::string label_type;
+      mgp::List labels_list = mgp::List();
+      for (auto const &label : labels) {
+        label_type += ":`" + std::string(label) + "`";
+        labels_list.AppendExtend(mgp::Value(label));
+      }
+      for (auto const &prop : property_info.properties) {
         auto record = record_factory.NewRecord();
-        property_type.AppendExtend(mgp::Value(TypeOf(prop.Type())));
-        ProcessPropertiesNode<mgp::List>(record, type, labels, key, property_type, true);
+        ProcessPropertiesNode(record, label_type, labels_list, prop.name, TypeOf(prop.value.Type()),
+                              property_info.mandatory);
+      }
+      if (property_info.properties.empty()) {
+        auto record = record_factory.NewRecord();
+        ProcessPropertiesNode<std::string>(record, label_type, labels_list, "", "", false);
       }
     }
 
@@ -144,23 +196,41 @@ void Schema::NodeTypeProperties(mgp_list * /*args*/, mgp_graph *memgraph_graph,
 
 void Schema::RelTypeProperties(mgp_list * /*args*/, mgp_graph *memgraph_graph, mgp_result *result, mgp_memory *memory) {
   mgp::MemoryDispatcherGuard guard{memory};
+
+  std::unordered_map<std::string, PropertyInfo> rel_types_properties;
   const auto record_factory = mgp::RecordFactory(result);
   try {
     const mgp::Graph graph = mgp::Graph(memgraph_graph);
-
     for (auto rel : graph.Relationships()) {
-      std::string type = ":`" + std::string(rel.Type()) + "`";
+      std::string rel_type = std::string(rel.Type());
+      if (rel_types_properties.find(rel_type) == rel_types_properties.end()) {
+        rel_types_properties[rel_type] = PropertyInfo{std::set<Property, PropertyComparator>(), true};
+      }
+
       if (rel.Properties().empty()) {
-        auto record = record_factory.NewRecord();
-        ProcessPropertiesRel<std::string>(record, type, "", "", false);
+        rel_types_properties[rel_type].mandatory = false;  // if there is rel with no property, it is not mandatory
         continue;
       }
 
+      auto &property_info = rel_types_properties.at(rel_type);
       for (auto &[key, prop] : rel.Properties()) {
-        auto property_type = mgp::List();
+        property_info.properties.emplace(key, std::move(prop));
+        if (property_info.mandatory) {
+          property_info.mandatory =
+              property_info.properties.size() == 1;  // if there is only one property, it is mandatory
+        }
+      }
+    }
+
+    for (auto &[type, property_info] : rel_types_properties) {
+      std::string type_str = ":`" + std::string(type) + "`";
+      for (auto const &prop : property_info.properties) {
         auto record = record_factory.NewRecord();
-        property_type.AppendExtend(mgp::Value(TypeOf(prop.Type())));
-        ProcessPropertiesRel<mgp::List>(record, type, key, property_type, true);
+        ProcessPropertiesRel(record, type_str, prop.name, TypeOf(prop.value.Type()), property_info.mandatory);
+      }
+      if (property_info.properties.empty()) {
+        auto record = record_factory.NewRecord();
+        ProcessPropertiesRel<std::string>(record, type_str, "", "", false);
       }
     }
 

src/CMakeLists.txt

@@ -53,7 +53,10 @@ set_target_properties(memgraph PROPERTIES
         OUTPUT_NAME "memgraph-${MEMGRAPH_VERSION}_${CMAKE_BUILD_TYPE}"
 
         # Output the executable in main binary dir.
-        RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
+        RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}
+
+        POSITION_INDEPENDENT_CODE ON
+        )
 
 # Create symlink to the built executable.
 add_custom_command(TARGET memgraph POST_BUILD

src/flags/general.cpp

@@ -66,6 +66,9 @@ DEFINE_bool(allow_load_csv, true, "Controls whether LOAD CSV clause is allowed i
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_VALIDATED_uint64(storage_gc_cycle_sec, 30, "Storage garbage collector interval (in seconds).",
                         FLAG_IN_RANGE(1, 24UL * 3600));
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_VALIDATED_uint64(storage_python_gc_cycle_sec, 180,
+                        "Storage python full garbage collection interval (in seconds).", FLAG_IN_RANGE(1, 24UL * 3600));
 // NOTE: The `storage_properties_on_edges` flag must be the same here and in
 // `mg_import_csv`. If you change it, make sure to change it there as well.
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

src/flags/general.hpp

@@ -52,6 +52,8 @@ DECLARE_bool(allow_load_csv);
 // Storage flags.
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DECLARE_uint64(storage_gc_cycle_sec);
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_uint64(storage_python_gc_cycle_sec);
 // NOTE: The `storage_properties_on_edges` flag must be the same here and in
 // `mg_import_csv`. If you change it, make sure to change it there as well.
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

src/memgraph.cpp

@@ -163,7 +163,7 @@ int main(int argc, char **argv) {
         // libstd.
         auto gil = memgraph::py::EnsureGIL();
         // NOLINTNEXTLINE(hicpp-signed-bitwise)
-        auto *flag = PyLong_FromLong(RTLD_NOW | RTLD_DEEPBIND);
+        auto *flag = PyLong_FromLong(RTLD_NOW);
         auto *setdl = PySys_GetObject("setdlopenflags");
         MG_ASSERT(setdl);
         auto *arg = PyTuple_New(1);
@@ -184,6 +184,10 @@ int main(int argc, char **argv) {
                                                    "https://memgr.ph/python"));
   }
 
+  memgraph::utils::Scheduler python_gc_scheduler;
+  python_gc_scheduler.Run("Python GC", std::chrono::seconds(FLAGS_storage_python_gc_cycle_sec),
+                          [] { memgraph::query::procedure::PyCollectGarbage(); });
+
   // Initialize the communication library.
   memgraph::communication::SSLInit sslInit;
 
@@ -318,6 +322,11 @@ int main(int argc, char **argv) {
                .durability_directory = FLAGS_data_directory + "/rocksdb_durability",
                .wal_directory = FLAGS_data_directory + "/rocksdb_wal"},
       .storage_mode = memgraph::flags::ParseStorageMode()};
+
+  memgraph::utils::Scheduler jemalloc_purge_scheduler;
+  jemalloc_purge_scheduler.Run("Jemalloc purge", std::chrono::seconds(FLAGS_storage_gc_cycle_sec),
+                               [] { memgraph::memory::PurgeUnusedMemory(); });
+
   if (FLAGS_storage_snapshot_interval_sec == 0) {
     if (FLAGS_storage_wal_enabled) {
       LOG_FATAL(

src/memory/global_memory_control.cpp

@@ -119,7 +119,7 @@ static bool my_commit(extent_hooks_t *extent_hooks, void *addr, size_t size, siz
 
   memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(length));
   if (GetQueriesMemoryControl().IsThreadTracked()) [[unlikely]] {
-    GetQueriesMemoryControl().TrackFreeOnCurrentThread(size);
+    GetQueriesMemoryControl().TrackAllocOnCurrentThread(size);
   }
 
   return false;

src/query/procedure/module.hpp

@@ -169,7 +169,7 @@ class ModuleRegistry final {
   // mentioned library will be first performed in the already existing binded
   // libraries and then the global namespace.
   // RTLD_DEEPBIND => https://linux.die.net/man/3/dlopen
-  SharedLibraryHandle libstd_handle{"libstdc++.so.6", RTLD_NOW | RTLD_LOCAL | RTLD_DEEPBIND, kLibstdcppWarning};
+  SharedLibraryHandle libstd_handle{"libstdc++.so.6", RTLD_NOW | RTLD_LOCAL, kLibstdcppWarning};
 #endif
   std::vector<std::filesystem::path> modules_dirs_;
   std::filesystem::path internal_module_dir_;

src/query/procedure/py_module.cpp

@@ -13,6 +13,7 @@
 
 #include <datetime.h>
 #include <methodobject.h>
+#include <objimpl.h>
 #include <pyerrors.h>
 #include <array>
 #include <optional>
@@ -57,7 +58,6 @@ PyObject *gMgpValueConversionError{nullptr};     // NOLINT(cppcoreguidelines-avo
 PyObject *gMgpSerializationError{nullptr};       // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
 PyObject *gMgpAuthorizationError{nullptr};       // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
 
-constexpr bool kStartGarbageCollection{true};
 constexpr auto kMicrosecondsInMillisecond{1000};
 constexpr auto kMicrosecondsInSecond{1000000};
 
@@ -867,6 +867,24 @@ py::Object MgpListToPyTuple(mgp_list *list, PyObject *py_graph) {
   return MgpListToPyTuple(list, reinterpret_cast<PyGraph *>(py_graph));
 }
 
+void PyCollectGarbage() {
+  if (!Py_IsInitialized() || _Py_IsFinalizing()) {
+    // Calling EnsureGIL will crash the program if this is true.
+    return;
+  }
+
+  auto gil = py::EnsureGIL();
+
+  py::Object gc(PyImport_ImportModule("gc"));
+  if (!gc) {
+    LOG_FATAL(py::FetchError().value());
+  }
+
+  if (!gc.CallMethod("collect")) {
+    LOG_FATAL(py::FetchError().value());
+  }
+}
+
 namespace {
 struct RecordFieldCache {
   PyObject *key;
@@ -1027,24 +1045,8 @@ std::optional<py::ExceptionInfo> AddMultipleBatchRecordsFromPython(mgp_result *r
   return std::nullopt;
 }
 
-std::function<void()> PyObjectCleanup(py::Object &py_object, bool start_gc) {
-  return [py_object, start_gc]() {
-    if (start_gc) {
-      // Run `gc.collect` (reference cycle-detection) explicitly, so that we are
-      // sure the procedure cleaned up everything it held references to. If the
-      // user stored a reference to one of our `_mgp` instances then the
-      // internally used `mgp_*` structs will stay unfreed and a memory leak
-      // will be reported at the end of the query execution.
-      py::Object gc(PyImport_ImportModule("gc"));
-      if (!gc) {
-        LOG_FATAL(py::FetchError().value());
-      }
-
-      if (!gc.CallMethod("collect")) {
-        LOG_FATAL(py::FetchError().value());
-      }
-    }
-
+std::function<void()> PyObjectCleanup(py::Object &py_object) {
+  return [py_object]() {
     // After making sure all references from our side have been cleared,
     // invalidate the `_mgp.Graph` object. If the user kept a reference to one
     // of our `_mgp` instances then this will prevent them from using those
@@ -1095,7 +1097,7 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
   std::optional<std::string> maybe_msg;
   {
     py::Object py_graph(MakePyGraph(graph, memory));
-    utils::OnScopeExit clean_up(PyObjectCleanup(py_graph, !is_batched));
+    utils::OnScopeExit clean_up(PyObjectCleanup(py_graph));
     if (py_graph) {
       maybe_msg = error_to_msg(call(py_graph));
     } else {
@@ -1110,22 +1112,11 @@ void CallPythonProcedure(const py::Object &py_cb, mgp_list *args, mgp_graph *gra
 
 void CallPythonCleanup(const py::Object &py_cleanup) {
   auto gil = py::EnsureGIL();
-
   auto py_res = py_cleanup.Call();
-
-  py::Object gc(PyImport_ImportModule("gc"));
-  if (!gc) {
-    LOG_FATAL(py::FetchError().value());
-  }
-
-  if (!gc.CallMethod("collect")) {
-    LOG_FATAL(py::FetchError().value());
-  }
 }
 
 void CallPythonInitializer(const py::Object &py_initializer, mgp_list *args, mgp_graph *graph, mgp_memory *memory) {
   auto gil = py::EnsureGIL();
-
   auto error_to_msg = [](const std::optional<py::ExceptionInfo> &exc_info) -> std::optional<std::string> {
     if (!exc_info) return std::nullopt;
     // Here we tell the traceback formatter to skip the first line of the
@@ -1146,7 +1137,7 @@ void CallPythonInitializer(const py::Object &py_initializer, mgp_list *args, mgp
   std::optional<std::string> maybe_msg;
   {
     py::Object py_graph(MakePyGraph(graph, memory));
-    utils::OnScopeExit clean_up_graph(PyObjectCleanup(py_graph, !kStartGarbageCollection));
+    utils::OnScopeExit clean_up_graph(PyObjectCleanup(py_graph));
     if (py_graph) {
       maybe_msg = error_to_msg(call(py_graph));
     } else {
@@ -1194,8 +1185,8 @@ void CallPythonTransformation(const py::Object &py_cb, mgp_messages *msgs, mgp_g
     py::Object py_graph(MakePyGraph(graph, memory));
     py::Object py_messages(MakePyMessages(msgs, memory));
 
-    utils::OnScopeExit clean_up_graph(PyObjectCleanup(py_graph, kStartGarbageCollection));
-    utils::OnScopeExit clean_up_messages(PyObjectCleanup(py_messages, kStartGarbageCollection));
+    utils::OnScopeExit clean_up_graph(PyObjectCleanup(py_graph));
+    utils::OnScopeExit clean_up_messages(PyObjectCleanup(py_messages));
 
     if (py_graph && py_messages) {
       maybe_msg = error_to_msg(call(py_graph, py_messages));
@@ -1264,7 +1255,7 @@ void CallPythonFunction(const py::Object &py_cb, mgp_list *args, mgp_graph *grap
   std::optional<std::string> maybe_msg;
   {
     py::Object py_graph(MakePyGraph(graph, memory));
-    utils::OnScopeExit clean_up(PyObjectCleanup(py_graph, kStartGarbageCollection));
+    utils::OnScopeExit clean_up(PyObjectCleanup(py_graph));
     if (py_graph) {
       auto maybe_result = call(py_graph);
       if (!maybe_result.HasError()) {

src/query/procedure/py_module.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 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
@@ -79,4 +79,7 @@ py::Object ImportPyModule(const char *, mgp_module *);
 /// Return nullptr and set appropriate Python exception on failure.
 py::Object ReloadPyModule(PyObject *, mgp_module *);
 
+/// Call full python circular reference garbage collection (all generations)
+void PyCollectGarbage();
+
 }  // namespace memgraph::query::procedure

src/storage/v2/edge_accessor.cpp

@@ -21,6 +21,7 @@
 #include "storage/v2/result.hpp"
 #include "storage/v2/storage.hpp"
 #include "storage/v2/vertex_accessor.hpp"
+#include "utils/atomic_memory_block.hpp"
 #include "utils/memory_tracker.hpp"
 
 namespace memgraph::storage {
@@ -126,24 +127,28 @@ Result<storage::PropertyValue> EdgeAccessor::SetProperty(PropertyId property, co
   if (!PrepareForWrite(transaction_, edge_.ptr)) return Error::SERIALIZATION_ERROR;
 
   if (edge_.ptr->deleted) return Error::DELETED_OBJECT;
-
-  auto current_value = edge_.ptr->properties.GetProperty(property);
-  // We could skip setting the value if the previous one is the same to the new
-  // one. This would save some memory as a delta would not be created as well as
-  // avoid copying the value. The reason we are not doing that is because the
-  // current code always follows the logical pattern of "create a delta" and
-  // "modify in-place". Additionally, the created delta will make other
-  // transactions get a SERIALIZATION_ERROR.
-
-  CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property, current_value);
-  edge_.ptr->properties.SetProperty(property, value);
+  using ReturnType = decltype(edge_.ptr->properties.GetProperty(property));
+  std::optional<ReturnType> current_value;
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [&current_value, &property, &value, transaction = transaction_, edge = edge_]() {
+        current_value.emplace(edge.ptr->properties.GetProperty(property));
+        // We could skip setting the value if the previous one is the same to the new
+        // one. This would save some memory as a delta would not be created as well as
+        // avoid copying the value. The reason we are not doing that is because the
+        // current code always follows the logical pattern of "create a delta" and
+        // "modify in-place". Additionally, the created delta will make other
+        // transactions get a SERIALIZATION_ERROR.
+        CreateAndLinkDelta(transaction, edge.ptr, Delta::SetPropertyTag(), property, *current_value);
+        edge.ptr->properties.SetProperty(property, value);
+      }};
+  std::invoke(atomic_memory_block);
 
   if (transaction_->IsDiskStorage()) {
     ModifiedEdgeInfo modified_edge(Delta::Action::SET_PROPERTY, from_vertex_->gid, to_vertex_->gid, edge_type_, edge_);
     transaction_->AddModifiedEdge(Gid(), modified_edge);
   }
 
-  return std::move(current_value);
+  return std::move(*current_value);
 }
 
 Result<bool> EdgeAccessor::InitProperties(const std::map<storage::PropertyId, storage::PropertyValue> &properties) {
@@ -157,9 +162,12 @@ Result<bool> EdgeAccessor::InitProperties(const std::map<storage::PropertyId, st
   if (edge_.ptr->deleted) return Error::DELETED_OBJECT;
 
   if (!edge_.ptr->properties.InitProperties(properties)) return false;
-  for (const auto &[property, _] : properties) {
-    CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property, PropertyValue());
-  }
+  utils::AtomicMemoryBlock atomic_memory_block{[&properties, transaction_ = transaction_, edge_ = edge_]() {
+    for (const auto &[property, _] : properties) {
+      CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property, PropertyValue());
+    }
+  }};
+  std::invoke(atomic_memory_block);
 
   return true;
 }
@@ -175,13 +183,18 @@ Result<std::vector<std::tuple<PropertyId, PropertyValue, PropertyValue>>> EdgeAc
 
   if (edge_.ptr->deleted) return Error::DELETED_OBJECT;
 
-  auto id_old_new_change = edge_.ptr->properties.UpdateProperties(properties);
+  using ReturnType = decltype(edge_.ptr->properties.UpdateProperties(properties));
+  std::optional<ReturnType> id_old_new_change;
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [transaction_ = transaction_, edge_ = edge_, &properties, &id_old_new_change]() {
+        id_old_new_change.emplace(edge_.ptr->properties.UpdateProperties(properties));
+        for (auto &[property, old_value, new_value] : *id_old_new_change) {
+          CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property, std::move(old_value));
+        }
+      }};
+  std::invoke(atomic_memory_block);
 
-  for (auto &[property, old_value, new_value] : id_old_new_change) {
-    CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property, std::move(old_value));
-  }
-
-  return id_old_new_change;
+  return id_old_new_change.has_value() ? std::move(id_old_new_change.value()) : ReturnType{};
 }
 
 Result<std::map<PropertyId, PropertyValue>> EdgeAccessor::ClearProperties() {
@@ -193,14 +206,19 @@ Result<std::map<PropertyId, PropertyValue>> EdgeAccessor::ClearProperties() {
 
   if (edge_.ptr->deleted) return Error::DELETED_OBJECT;
 
-  auto properties = edge_.ptr->properties.Properties();
-  for (const auto &property : properties) {
-    CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property.first, property.second);
-  }
+  using ReturnType = decltype(edge_.ptr->properties.Properties());
+  std::optional<ReturnType> properties;
+  utils::AtomicMemoryBlock atomic_memory_block{[&properties, transaction_ = transaction_, edge_ = edge_]() {
+    properties.emplace(edge_.ptr->properties.Properties());
+    for (const auto &property : *properties) {
+      CreateAndLinkDelta(transaction_, edge_.ptr, Delta::SetPropertyTag(), property.first, property.second);
+    }
 
-  edge_.ptr->properties.ClearProperties();
+    edge_.ptr->properties.ClearProperties();
+  }};
+  std::invoke(atomic_memory_block);
 
-  return std::move(properties);
+  return properties.has_value() ? std::move(properties.value()) : ReturnType{};
 }
 
 Result<PropertyValue> EdgeAccessor::GetProperty(PropertyId property, View view) const {

src/storage/v2/inmemory/storage.cpp

@@ -1,4 +1,4 @@
-// Copyright 2023 Memgraph Ltd.
+// Copyright 2024 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
@@ -26,6 +26,7 @@
 #include "storage/v2/inmemory/replication/recovery.hpp"
 #include "storage/v2/inmemory/unique_constraints.hpp"
 #include "storage/v2/property_value.hpp"
+#include "utils/atomic_memory_block.hpp"
 #include "utils/resource_lock.hpp"
 #include "utils/stat.hpp"
 
@@ -64,14 +65,14 @@ auto FindEdges(const View view, EdgeTypeId edge_type, const VertexAccessor *from
 
 using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
 
-InMemoryStorage::InMemoryStorage(Config config, StorageMode storage_mode)
-    : Storage(config, storage_mode),
+InMemoryStorage::InMemoryStorage(Config config)
+    : Storage(config, config.storage_mode),
       recovery_{config.durability.storage_directory / durability::kSnapshotDirectory,
                 config.durability.storage_directory / durability::kWalDirectory},
       lock_file_path_(config.durability.storage_directory / durability::kLockFile),
       uuid_(utils::GenerateUUID()),
       global_locker_(file_retainer_.AddLocker()) {
-  MG_ASSERT(storage_mode != StorageMode::ON_DISK_TRANSACTIONAL,
+  MG_ASSERT(config.storage_mode != StorageMode::ON_DISK_TRANSACTIONAL,
             "Invalid storage mode sent to InMemoryStorage constructor!");
   if (config_.durability.snapshot_wal_mode != Config::Durability::SnapshotWalMode::DISABLED ||
       config_.durability.snapshot_on_exit || config_.durability.recover_on_startup) {
@@ -144,7 +145,6 @@ InMemoryStorage::InMemoryStorage(Config config, StorageMode storage_mode)
     gc_runner_.Run("Storage GC", config_.gc.interval, [this] {
       this->FreeMemory(std::unique_lock<utils::ResourceLock>{main_lock_, std::defer_lock});
     });
-    gc_jemalloc_runner_.Run("Jemalloc GC", config_.gc.interval, [] { memory::PurgeUnusedMemory(); });
   }
   if (timestamp_ == kTimestampInitialId) {
     commit_log_.emplace();
@@ -153,12 +153,9 @@ InMemoryStorage::InMemoryStorage(Config config, StorageMode storage_mode)
   }
 }
 
-InMemoryStorage::InMemoryStorage(Config config) : InMemoryStorage(config, StorageMode::IN_MEMORY_TRANSACTIONAL) {}
-
 InMemoryStorage::~InMemoryStorage() {
   if (config_.gc.type == Config::Gc::Type::PERIODIC) {
     gc_runner_.Stop();
-    gc_jemalloc_runner_.Stop();
   }
   {
     // Stop replication (Stop all clients or stop the REPLICA server)
@@ -338,18 +335,22 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::CreateEdge(VertexAccesso
       delta->prev.Set(&*it);
     }
   }
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [this, edge, from_vertex = from_vertex, edge_type = edge_type, to_vertex = to_vertex]() {
+        CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), edge_type, to_vertex, edge);
+        from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge);
 
-  CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), edge_type, to_vertex, edge);
-  from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge);
+        CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), edge_type, from_vertex, edge);
+        to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge);
 
-  CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), edge_type, from_vertex, edge);
-  to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge);
+        transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
+        transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
 
-  transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
-  transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
+        // Increment edge count.
+        storage_->edge_count_.fetch_add(1, std::memory_order_acq_rel);
+      }};
 
-  // Increment edge count.
-  storage_->edge_count_.fetch_add(1, std::memory_order_acq_rel);
+  std::invoke(atomic_memory_block);
 
   return EdgeAccessor(edge, edge_type, from_vertex, to_vertex, storage_, &transaction_);
 }
@@ -433,18 +434,22 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::CreateEdgeEx(VertexAcces
       delta->prev.Set(&*it);
     }
   }
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [this, edge, from_vertex = from_vertex, edge_type = edge_type, to_vertex = to_vertex]() {
+        CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), edge_type, to_vertex, edge);
+        from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge);
 
-  CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), edge_type, to_vertex, edge);
-  from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge);
+        CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), edge_type, from_vertex, edge);
+        to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge);
 
-  CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), edge_type, from_vertex, edge);
-  to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge);
+        transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
+        transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
 
-  transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
-  transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
+        // Increment edge count.
+        storage_->edge_count_.fetch_add(1, std::memory_order_acq_rel);
+      }};
 
-  // Increment edge count.
-  storage_->edge_count_.fetch_add(1, std::memory_order_acq_rel);
+  std::invoke(atomic_memory_block);
 
   return EdgeAccessor(edge, edge_type, from_vertex, to_vertex, storage_, &transaction_);
 }
@@ -534,18 +539,22 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::EdgeSetFrom(EdgeAccessor
       return Error::DELETED_OBJECT;
     }
   }
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [this, edge_ref, old_from_vertex, new_from_vertex, edge_type, to_vertex]() {
+        CreateAndLinkDelta(&transaction_, old_from_vertex, Delta::AddOutEdgeTag(), edge_type, to_vertex, edge_ref);
+        CreateAndLinkDelta(&transaction_, to_vertex, Delta::AddInEdgeTag(), edge_type, old_from_vertex, edge_ref);
 
-  CreateAndLinkDelta(&transaction_, old_from_vertex, Delta::AddOutEdgeTag(), edge_type, to_vertex, edge_ref);
-  CreateAndLinkDelta(&transaction_, to_vertex, Delta::AddInEdgeTag(), edge_type, old_from_vertex, edge_ref);
+        CreateAndLinkDelta(&transaction_, new_from_vertex, Delta::RemoveOutEdgeTag(), edge_type, to_vertex, edge_ref);
+        new_from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge_ref);
+        CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), edge_type, new_from_vertex, edge_ref);
+        to_vertex->in_edges.emplace_back(edge_type, new_from_vertex, edge_ref);
 
-  CreateAndLinkDelta(&transaction_, new_from_vertex, Delta::RemoveOutEdgeTag(), edge_type, to_vertex, edge_ref);
-  new_from_vertex->out_edges.emplace_back(edge_type, to_vertex, edge_ref);
-  CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), edge_type, new_from_vertex, edge_ref);
-  to_vertex->in_edges.emplace_back(edge_type, new_from_vertex, edge_ref);
+        transaction_.manyDeltasCache.Invalidate(new_from_vertex, edge_type, EdgeDirection::OUT);
+        transaction_.manyDeltasCache.Invalidate(old_from_vertex, edge_type, EdgeDirection::OUT);
+        transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
+      }};
 
-  transaction_.manyDeltasCache.Invalidate(new_from_vertex, edge_type, EdgeDirection::OUT);
-  transaction_.manyDeltasCache.Invalidate(old_from_vertex, edge_type, EdgeDirection::OUT);
-  transaction_.manyDeltasCache.Invalidate(to_vertex, edge_type, EdgeDirection::IN);
+  std::invoke(atomic_memory_block);
 
   return EdgeAccessor(edge_ref, edge_type, new_from_vertex, to_vertex, storage_, &transaction_);
 }
@@ -636,17 +645,22 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::EdgeSetTo(EdgeAccessor *
     }
   }
 
-  CreateAndLinkDelta(&transaction_, from_vertex, Delta::AddOutEdgeTag(), edge_type, old_to_vertex, edge_ref);
-  CreateAndLinkDelta(&transaction_, old_to_vertex, Delta::AddInEdgeTag(), edge_type, from_vertex, edge_ref);
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [this, edge_ref, old_to_vertex, from_vertex, edge_type, new_to_vertex]() {
+        CreateAndLinkDelta(&transaction_, from_vertex, Delta::AddOutEdgeTag(), edge_type, old_to_vertex, edge_ref);
+        CreateAndLinkDelta(&transaction_, old_to_vertex, Delta::AddInEdgeTag(), edge_type, from_vertex, edge_ref);
 
-  CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), edge_type, new_to_vertex, edge_ref);
-  from_vertex->out_edges.emplace_back(edge_type, new_to_vertex, edge_ref);
-  CreateAndLinkDelta(&transaction_, new_to_vertex, Delta::RemoveInEdgeTag(), edge_type, from_vertex, edge_ref);
-  new_to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge_ref);
+        CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), edge_type, new_to_vertex, edge_ref);
+        from_vertex->out_edges.emplace_back(edge_type, new_to_vertex, edge_ref);
+        CreateAndLinkDelta(&transaction_, new_to_vertex, Delta::RemoveInEdgeTag(), edge_type, from_vertex, edge_ref);
+        new_to_vertex->in_edges.emplace_back(edge_type, from_vertex, edge_ref);
 
-  transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
-  transaction_.manyDeltasCache.Invalidate(old_to_vertex, edge_type, EdgeDirection::IN);
-  transaction_.manyDeltasCache.Invalidate(new_to_vertex, edge_type, EdgeDirection::IN);
+        transaction_.manyDeltasCache.Invalidate(from_vertex, edge_type, EdgeDirection::OUT);
+        transaction_.manyDeltasCache.Invalidate(old_to_vertex, edge_type, EdgeDirection::IN);
+        transaction_.manyDeltasCache.Invalidate(new_to_vertex, edge_type, EdgeDirection::IN);
+      }};
+
+  std::invoke(atomic_memory_block);
 
   return EdgeAccessor(edge_ref, edge_type, from_vertex, new_to_vertex, storage_, &transaction_);
 }
@@ -709,17 +723,21 @@ Result<EdgeAccessor> InMemoryStorage::InMemoryAccessor::EdgeChangeType(EdgeAcces
 
   MG_ASSERT((op1 && op2), "Invalid database state!");
 
-  // "deleting" old edge
-  CreateAndLinkDelta(&transaction_, from_vertex, Delta::AddOutEdgeTag(), edge_type, to_vertex, edge_ref);
-  CreateAndLinkDelta(&transaction_, to_vertex, Delta::AddInEdgeTag(), edge_type, from_vertex, edge_ref);
+  utils::AtomicMemoryBlock atomic_memory_block{[this, to_vertex, new_edge_type, edge_ref, from_vertex, edge_type]() {
+    // "deleting" old edge
+    CreateAndLinkDelta(&transaction_, from_vertex, Delta::AddOutEdgeTag(), edge_type, to_vertex, edge_ref);
+    CreateAndLinkDelta(&transaction_, to_vertex, Delta::AddInEdgeTag(), edge_type, from_vertex, edge_ref);
 
-  // "adding" new edge
-  CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), new_edge_type, to_vertex, edge_ref);
-  CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), new_edge_type, from_vertex, edge_ref);
+    // "adding" new edge
+    CreateAndLinkDelta(&transaction_, from_vertex, Delta::RemoveOutEdgeTag(), new_edge_type, to_vertex, edge_ref);
+    CreateAndLinkDelta(&transaction_, to_vertex, Delta::RemoveInEdgeTag(), new_edge_type, from_vertex, edge_ref);
 
-  // edge type is not used while invalidating cache so we can only call it once
-  transaction_.manyDeltasCache.Invalidate(from_vertex, new_edge_type, EdgeDirection::OUT);
-  transaction_.manyDeltasCache.Invalidate(to_vertex, new_edge_type, EdgeDirection::IN);
+    // edge type is not used while invalidating cache so we can only call it once
+    transaction_.manyDeltasCache.Invalidate(from_vertex, new_edge_type, EdgeDirection::OUT);
+    transaction_.manyDeltasCache.Invalidate(to_vertex, new_edge_type, EdgeDirection::IN);
+  }};
+
+  std::invoke(atomic_memory_block);
 
   return EdgeAccessor(edge_ref, new_edge_type, from_vertex, to_vertex, storage_, &transaction_);
 }

src/storage/v2/inmemory/storage.hpp

@@ -1,4 +1,4 @@
-// Copyright 2023 Memgraph Ltd.
+// Copyright 2024 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
@@ -60,7 +60,6 @@ class InMemoryStorage final : public Storage {
   /// @throw std::system_error
   /// @throw std::bad_alloc
   explicit InMemoryStorage(Config config = Config());
-  InMemoryStorage(Config config, StorageMode storage_mode);
 
   InMemoryStorage(const InMemoryStorage &) = delete;
   InMemoryStorage(InMemoryStorage &&) = delete;
@@ -427,7 +426,6 @@ class InMemoryStorage final : public Storage {
   std::optional<CommitLog> commit_log_;
 
   utils::Scheduler gc_runner_;
-  utils::Scheduler gc_jemalloc_runner_;
   std::mutex gc_lock_;
 
   using BondPmrLd = Bond<utils::pmr::list<Delta>>;

src/storage/v2/storage.cpp

@@ -17,6 +17,7 @@
 #include "storage/v2/storage.hpp"
 #include "storage/v2/transaction.hpp"
 #include "storage/v2/vertex_accessor.hpp"
+#include "utils/atomic_memory_block.hpp"
 #include "utils/event_counter.hpp"
 #include "utils/event_histogram.hpp"
 #include "utils/exceptions.hpp"
@@ -390,22 +391,29 @@ Result<std::optional<std::vector<EdgeAccessor>>> Storage::Accessor::ClearEdgesOn
       if (!PrepareForWrite(&transaction_, vertex_ptr)) return Error::SERIALIZATION_ERROR;
       MG_ASSERT(!vertex_ptr->deleted, "Invalid database state!");
 
-      attached_edges_to_vertex->pop_back();
-      if (storage_->config_.items.properties_on_edges) {
-        auto *edge_ptr = edge_ref.ptr;
-        MarkEdgeAsDeleted(edge_ptr);
-      }
+      // MarkEdgeAsDeleted allocates additional memory
+      // and CreateAndLinkDelta needs memory
+      utils::AtomicMemoryBlock atomic_memory_block{[&attached_edges_to_vertex, &deleted_edge_ids, &reverse_vertex_order,
+                                                    &vertex_ptr, &deleted_edges, deletion_delta = deletion_delta,
+                                                    edge_type = edge_type, opposing_vertex = opposing_vertex,
+                                                    edge_ref = edge_ref, this]() {
+        attached_edges_to_vertex->pop_back();
+        if (this->storage_->config_.items.properties_on_edges) {
+          auto *edge_ptr = edge_ref.ptr;
+          MarkEdgeAsDeleted(edge_ptr);
+        }
 
-      auto const edge_gid = storage_->config_.items.properties_on_edges ? edge_ref.ptr->gid : edge_ref.gid;
-      auto const [_, was_inserted] = deleted_edge_ids.insert(edge_gid);
-      bool const edge_cleared_from_both_directions = !was_inserted;
-      if (edge_cleared_from_both_directions) {
-        auto *from_vertex = reverse_vertex_order ? vertex_ptr : opposing_vertex;
-        auto *to_vertex = reverse_vertex_order ? opposing_vertex : vertex_ptr;
-        deleted_edges.emplace_back(edge_ref, edge_type, from_vertex, to_vertex, storage_, &transaction_, true);
-      }
-
-      CreateAndLinkDelta(&transaction_, vertex_ptr, deletion_delta, edge_type, opposing_vertex, edge_ref);
+        auto const edge_gid = storage_->config_.items.properties_on_edges ? edge_ref.ptr->gid : edge_ref.gid;
+        auto const [_, was_inserted] = deleted_edge_ids.insert(edge_gid);
+        bool const edge_cleared_from_both_directions = !was_inserted;
+        if (edge_cleared_from_both_directions) {
+          auto *from_vertex = reverse_vertex_order ? vertex_ptr : opposing_vertex;
+          auto *to_vertex = reverse_vertex_order ? opposing_vertex : vertex_ptr;
+          deleted_edges.emplace_back(edge_ref, edge_type, from_vertex, to_vertex, storage_, &transaction_, true);
+        }
+        CreateAndLinkDelta(&transaction_, vertex_ptr, deletion_delta, edge_type, opposing_vertex, edge_ref);
+      }};
+      std::invoke(atomic_memory_block);
     }
 
     return std::make_optional<ReturnType>();
@@ -449,30 +457,37 @@ Result<std::optional<std::vector<EdgeAccessor>>> Storage::Accessor::DetachRemain
           return !set_for_erasure.contains(edge_gid);
         });
 
-    for (auto it = mid; it != edges_attached_to_vertex->end(); it++) {
-      auto const &[edge_type, opposing_vertex, edge_ref] = *it;
-      std::unique_lock<utils::RWSpinLock> guard;
-      if (storage_->config_.items.properties_on_edges) {
-        auto edge_ptr = edge_ref.ptr;
-        guard = std::unique_lock{edge_ptr->lock};
-        // this can happen only if we marked edges for deletion with no nodes,
-        // so the method detaching nodes will not do anything
-        MarkEdgeAsDeleted(edge_ptr);
+    // Creating deltas and erasing edge only at the end -> we might have incomplete state as
+    // delta might cause OOM, so we don't remove edges from edges_attached_to_vertex
+    utils::AtomicMemoryBlock atomic_memory_block{[&mid, &edges_attached_to_vertex, &deleted_edges,
+                                                  &partially_detached_edge_ids, this, vertex_ptr, deletion_delta,
+                                                  reverse_vertex_order]() {
+      for (auto it = mid; it != edges_attached_to_vertex->end(); it++) {
+        auto const &[edge_type, opposing_vertex, edge_ref] = *it;
+        std::unique_lock<utils::RWSpinLock> guard;
+        if (storage_->config_.items.properties_on_edges) {
+          auto edge_ptr = edge_ref.ptr;
+          guard = std::unique_lock{edge_ptr->lock};
+          // this can happen only if we marked edges for deletion with no nodes,
+          // so the method detaching nodes will not do anything
+          MarkEdgeAsDeleted(edge_ptr);
+        }
+
+        CreateAndLinkDelta(&transaction_, vertex_ptr, deletion_delta, edge_type, opposing_vertex, edge_ref);
+
+        auto const edge_gid = storage_->config_.items.properties_on_edges ? edge_ref.ptr->gid : edge_ref.gid;
+        auto const [_, was_inserted] = partially_detached_edge_ids.insert(edge_gid);
+        bool const edge_cleared_from_both_directions = !was_inserted;
+        if (edge_cleared_from_both_directions) {
+          auto *from_vertex = reverse_vertex_order ? opposing_vertex : vertex_ptr;
+          auto *to_vertex = reverse_vertex_order ? vertex_ptr : opposing_vertex;
+          deleted_edges.emplace_back(edge_ref, edge_type, from_vertex, to_vertex, storage_, &transaction_, true);
+        }
       }
+      edges_attached_to_vertex->erase(mid, edges_attached_to_vertex->end());
+    }};
 
-      CreateAndLinkDelta(&transaction_, vertex_ptr, deletion_delta, edge_type, opposing_vertex, edge_ref);
-
-      auto const edge_gid = storage_->config_.items.properties_on_edges ? edge_ref.ptr->gid : edge_ref.gid;
-      auto const [_, was_inserted] = partially_detached_edge_ids.insert(edge_gid);
-      bool const edge_cleared_from_both_directions = !was_inserted;
-      if (edge_cleared_from_both_directions) {
-        auto *from_vertex = reverse_vertex_order ? opposing_vertex : vertex_ptr;
-        auto *to_vertex = reverse_vertex_order ? vertex_ptr : opposing_vertex;
-        deleted_edges.emplace_back(edge_ref, edge_type, from_vertex, to_vertex, storage_, &transaction_, true);
-      }
-    }
-
-    edges_attached_to_vertex->erase(mid, edges_attached_to_vertex->end());
+    std::invoke(atomic_memory_block);
 
     return std::make_optional<ReturnType>();
   };

src/storage/v2/vertex_accessor.cpp

@@ -26,6 +26,7 @@
 #include "storage/v2/storage.hpp"
 #include "storage/v2/vertex_info_cache.hpp"
 #include "storage/v2/vertex_info_helpers.hpp"
+#include "utils/atomic_memory_block.hpp"
 #include "utils/logging.hpp"
 #include "utils/memory_tracker.hpp"
 #include "utils/variant_helpers.hpp"
@@ -107,8 +108,11 @@ Result<bool> VertexAccessor::AddLabel(LabelId label) {
   if (vertex_->deleted) return Error::DELETED_OBJECT;
   if (std::find(vertex_->labels.begin(), vertex_->labels.end(), label) != vertex_->labels.end()) return false;
 
-  CreateAndLinkDelta(transaction_, vertex_, Delta::RemoveLabelTag(), label);
-  vertex_->labels.push_back(label);
+  utils::AtomicMemoryBlock atomic_memory_block{[transaction = transaction_, vertex = vertex_, &label]() {
+    CreateAndLinkDelta(transaction, vertex, Delta::RemoveLabelTag(), label);
+    vertex->labels.push_back(label);
+  }};
+  std::invoke(atomic_memory_block);
 
   if (storage_->config_.items.enable_schema_metadata) {
     storage_->stored_node_labels_.try_insert(label);
@@ -136,9 +140,12 @@ Result<bool> VertexAccessor::RemoveLabel(LabelId label) {
   auto it = std::find(vertex_->labels.begin(), vertex_->labels.end(), label);
   if (it == vertex_->labels.end()) return false;
 
-  CreateAndLinkDelta(transaction_, vertex_, Delta::AddLabelTag(), label);
-  *it = vertex_->labels.back();
-  vertex_->labels.pop_back();
+  utils::AtomicMemoryBlock atomic_memory_block{[transaction = transaction_, vertex = vertex_, &label, &it]() {
+    CreateAndLinkDelta(transaction, vertex, Delta::AddLabelTag(), label);
+    *it = vertex->labels.back();
+    vertex->labels.pop_back();
+  }};
+  std::invoke(atomic_memory_block);
 
   /// TODO: some by pointers, some by reference => not good, make it better
   storage_->constraints_.unique_constraints_->UpdateOnRemoveLabel(label, *vertex_, transaction_->start_timestamp);
@@ -262,8 +269,12 @@ Result<PropertyValue> VertexAccessor::SetProperty(PropertyId property, const Pro
   // "modify in-place". Additionally, the created delta will make other
   // transactions get a SERIALIZATION_ERROR.
 
-  CreateAndLinkDelta(transaction_, vertex_, Delta::SetPropertyTag(), property, current_value);
-  vertex_->properties.SetProperty(property, value);
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [transaction = transaction_, vertex = vertex_, &value, &property, &current_value]() {
+        CreateAndLinkDelta(transaction, vertex, Delta::SetPropertyTag(), property, current_value);
+        vertex->properties.SetProperty(property, value);
+      }};
+  std::invoke(atomic_memory_block);
 
   if (!value.IsNull()) {
     transaction_->constraint_verification_info.AddedProperty(vertex_);
@@ -287,20 +298,28 @@ Result<bool> VertexAccessor::InitProperties(const std::map<storage::PropertyId,
   if (!PrepareForWrite(transaction_, vertex_)) return Error::SERIALIZATION_ERROR;
 
   if (vertex_->deleted) return Error::DELETED_OBJECT;
+  bool result{false};
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [&result, &properties, storage = storage_, transaction = transaction_, vertex = vertex_]() {
+        if (!vertex->properties.InitProperties(properties)) {
+          result = false;
+          return;
+        }
+        for (const auto &[property, value] : properties) {
+          CreateAndLinkDelta(transaction, vertex, Delta::SetPropertyTag(), property, PropertyValue());
+          storage->indices_.UpdateOnSetProperty(property, value, vertex, *transaction);
+          transaction->manyDeltasCache.Invalidate(vertex, property);
+          if (!value.IsNull()) {
+            transaction->constraint_verification_info.AddedProperty(vertex);
+          } else {
+            transaction->constraint_verification_info.RemovedProperty(vertex);
+          }
+        }
+        result = true;
+      }};
+  std::invoke(atomic_memory_block);
 
-  if (!vertex_->properties.InitProperties(properties)) return false;
-  for (const auto &[property, value] : properties) {
-    CreateAndLinkDelta(transaction_, vertex_, Delta::SetPropertyTag(), property, PropertyValue());
-    storage_->indices_.UpdateOnSetProperty(property, value, vertex_, *transaction_);
-    transaction_->manyDeltasCache.Invalidate(vertex_, property);
-    if (!value.IsNull()) {
-      transaction_->constraint_verification_info.AddedProperty(vertex_);
-    } else {
-      transaction_->constraint_verification_info.RemovedProperty(vertex_);
-    }
-  }
-
-  return true;
+  return result;
 }
 
 Result<std::vector<std::tuple<PropertyId, PropertyValue, PropertyValue>>> VertexAccessor::UpdateProperties(
@@ -316,20 +335,28 @@ Result<std::vector<std::tuple<PropertyId, PropertyValue, PropertyValue>>> Vertex
 
   if (vertex_->deleted) return Error::DELETED_OBJECT;
 
-  auto id_old_new_change = vertex_->properties.UpdateProperties(properties);
+  using ReturnType = decltype(vertex_->properties.UpdateProperties(properties));
+  std::optional<ReturnType> id_old_new_change;
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [storage = storage_, transaction = transaction_, vertex = vertex_, &properties, &id_old_new_change]() {
+        id_old_new_change.emplace(vertex->properties.UpdateProperties(properties));
+        if (!id_old_new_change.has_value()) {
+          return;
+        }
+        for (auto &[id, old_value, new_value] : *id_old_new_change) {
+          storage->indices_.UpdateOnSetProperty(id, new_value, vertex, *transaction);
+          CreateAndLinkDelta(transaction, vertex, Delta::SetPropertyTag(), id, std::move(old_value));
+          transaction->manyDeltasCache.Invalidate(vertex, id);
+          if (!new_value.IsNull()) {
+            transaction->constraint_verification_info.AddedProperty(vertex);
+          } else {
+            transaction->constraint_verification_info.RemovedProperty(vertex);
+          }
+        }
+      }};
+  std::invoke(atomic_memory_block);
 
-  for (auto &[id, old_value, new_value] : id_old_new_change) {
-    storage_->indices_.UpdateOnSetProperty(id, new_value, vertex_, *transaction_);
-    CreateAndLinkDelta(transaction_, vertex_, Delta::SetPropertyTag(), id, std::move(old_value));
-    transaction_->manyDeltasCache.Invalidate(vertex_, id);
-    if (!new_value.IsNull()) {
-      transaction_->constraint_verification_info.AddedProperty(vertex_);
-    } else {
-      transaction_->constraint_verification_info.RemovedProperty(vertex_);
-    }
-  }
-
-  return id_old_new_change;
+  return id_old_new_change.has_value() ? std::move(id_old_new_change.value()) : ReturnType{};
 }
 
 Result<std::map<PropertyId, PropertyValue>> VertexAccessor::ClearProperties() {
@@ -342,17 +369,25 @@ Result<std::map<PropertyId, PropertyValue>> VertexAccessor::ClearProperties() {
 
   if (vertex_->deleted) return Error::DELETED_OBJECT;
 
-  auto properties = vertex_->properties.Properties();
-  for (const auto &[property, value] : properties) {
-    CreateAndLinkDelta(transaction_, vertex_, Delta::SetPropertyTag(), property, value);
-    storage_->indices_.UpdateOnSetProperty(property, PropertyValue(), vertex_, *transaction_);
-    transaction_->constraint_verification_info.RemovedProperty(vertex_);
-    transaction_->manyDeltasCache.Invalidate(vertex_, property);
-  }
+  using ReturnType = decltype(vertex_->properties.Properties());
+  std::optional<ReturnType> properties;
+  utils::AtomicMemoryBlock atomic_memory_block{
+      [storage = storage_, transaction = transaction_, vertex = vertex_, &properties]() {
+        properties.emplace(vertex->properties.Properties());
+        if (!properties.has_value()) {
+          return;
+        }
+        for (const auto &[property, value] : *properties) {
+          CreateAndLinkDelta(transaction, vertex, Delta::SetPropertyTag(), property, value);
+          storage->indices_.UpdateOnSetProperty(property, PropertyValue(), vertex, *transaction);
+          transaction->constraint_verification_info.RemovedProperty(vertex);
+          transaction->manyDeltasCache.Invalidate(vertex, property);
+        }
+        vertex->properties.ClearProperties();
+      }};
+  std::invoke(atomic_memory_block);
 
-  vertex_->properties.ClearProperties();
-
-  return std::move(properties);
+  return properties.has_value() ? std::move(properties.value()) : ReturnType{};
 }
 
 Result<PropertyValue> VertexAccessor::GetProperty(PropertyId property, View view) const {

src/telemetry/telemetry.cpp

@@ -110,7 +110,13 @@ void Telemetry::CollectData(const std::string &event) {
   {
     std::lock_guard<std::mutex> guard(lock_);
     for (auto &collector : collectors_) {
-      data[collector.first] = collector.second();
+      try {
+        data[collector.first] = collector.second();
+      } catch (std::exception &e) {
+        spdlog::warn(fmt::format(
+            "Unknwon exception occured on in telemetry server {}, please contact support on https://memgr.ph/unknown ",
+            e.what()));
+      }
     }
   }
   if (event == "") {

src/utils/atomic_memory_block.hpp

@@ -0,0 +1,44 @@
+// Copyright 2023 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <functional>
+#include "utils/memory_tracker.hpp"
+
+namespace memgraph::utils {
+
+// Calls a function with out of memory exception blocker, checks memory allocation after block execution.
+// Use it in case you need block which will be executed atomically considering memory execution
+// but will check after block is executed if OOM exceptions needs to be thrown
+template <typename Callable>
+class [[nodiscard]] AtomicMemoryBlock {
+ public:
+  explicit AtomicMemoryBlock(Callable &&function) : function_{std::forward<Callable>(function)} {}
+  AtomicMemoryBlock(AtomicMemoryBlock const &) = delete;
+  AtomicMemoryBlock(AtomicMemoryBlock &&) = delete;
+  AtomicMemoryBlock &operator=(AtomicMemoryBlock const &) = delete;
+  AtomicMemoryBlock &operator=(AtomicMemoryBlock &&) = delete;
+  ~AtomicMemoryBlock() = default;
+
+  void operator()() {
+    {
+      utils::MemoryTracker::OutOfMemoryExceptionBlocker oom_blocker;
+      function_();
+    }
+    total_memory_tracker.DoCheck();
+  }
+
+ private:
+  Callable function_;
+};
+
+}  // namespace memgraph::utils

src/utils/memory_tracker.cpp

@@ -124,6 +124,19 @@ void MemoryTracker::Alloc(const int64_t size) {
   UpdatePeak(will_be);
 }
 
+void MemoryTracker::DoCheck() {
+  const auto current_hard_limit = hard_limit_.load(std::memory_order_relaxed);
+  const auto current_amount = amount_.load(std::memory_order_relaxed);
+  if (current_hard_limit && current_amount > current_hard_limit && MemoryTrackerCanThrow()) [[unlikely]] {
+    MemoryTracker::OutOfMemoryExceptionBlocker exception_blocker;
+    throw OutOfMemoryException(
+        fmt::format("Memory limit exceeded! Current "
+                    "use is {}, while the maximum allowed size for allocation is set to {}.",
+                    GetReadableSize(static_cast<double>(current_amount)),
+                    GetReadableSize(static_cast<double>(current_hard_limit))));
+  }
+}
+
 void MemoryTracker::Free(const int64_t size) { amount_.fetch_sub(size, std::memory_order_relaxed); }
 
 }  // namespace memgraph::utils

src/utils/memory_tracker.hpp

@@ -49,6 +49,7 @@ class MemoryTracker final {
 
   void Alloc(int64_t size);
   void Free(int64_t size);
+  void DoCheck();
 
   auto Amount() const { return amount_.load(std::memory_order_relaxed); }
 

src/utils/stat.hpp

@@ -25,17 +25,32 @@ namespace memgraph::utils {
 static constexpr int64_t VM_MAX_MAP_COUNT_DEFAULT{-1};
 
 /// Returns the number of bytes a directory is using on disk. If the given path
-/// isn't a directory, zero will be returned.
+/// isn't a directory, zero will be returned. If there are some files with
+/// wrong permission, it will be skipped
 template <bool IgnoreSymlink = true>
 inline uint64_t GetDirDiskUsage(const std::filesystem::path &path) {
   if (!std::filesystem::is_directory(path)) return 0;
 
+  if (!utils::HasReadAccess(path)) {
+    spdlog::warn(
+        "Skipping directory path on collecting directory disk usage '{}' because it is not readable, check file "
+        "ownership and read permissions!",
+        path);
+    return 0;
+  }
   uint64_t size = 0;
-  for (auto &p : std::filesystem::directory_iterator(path)) {
+  for (const auto &p : std::filesystem::directory_iterator(path)) {
     if (IgnoreSymlink && std::filesystem::is_symlink(p)) continue;
     if (std::filesystem::is_directory(p)) {
       size += GetDirDiskUsage(p);
     } else if (std::filesystem::is_regular_file(p)) {
+      if (!utils::HasReadAccess(p)) {
+        spdlog::warn(
+            "Skipping file path on collecting directory disk usage '{}' because it is not readable, check file "
+            "ownership and read permissions!",
+            p);
+        continue;
+      }
       size += std::filesystem::file_size(p);
     }
   }

tests/drivers/java/v5_8/run.sh

@@ -9,8 +9,8 @@ fi
 if [ -d "/usr/lib/jvm/java-17-openjdk-amd64" ]; then
   export JAVA_HOME="/usr/lib/jvm/java-17-openjdk-amd64"
 fi
-if [ -d "/opt/apache-maven-3.9.2" ]; then
-  export M2_HOME="/opt/apache-maven-3.9.2"
+if [ -d "/opt/apache-maven-3.9.3" ]; then
+  export M2_HOME="/opt/apache-maven-3.9.3"
 fi
 export PATH="$JAVA_HOME/bin:$M2_HOME/bin:$PATH"
 

tests/e2e/configuration/default_config.py

@@ -143,6 +143,7 @@ startup_config_dict = {
         "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.",
     ),
     "storage_gc_cycle_sec": ("30", "30", "Storage garbage collector interval (in seconds)."),
+    "storage_python_gc_cycle_sec": ("180", "180", "Storage python full garbage collection interval (in seconds)."),
     "storage_items_per_batch": (
         "1000000",
         "1000000",

tests/e2e/interactive_mg_runner.py

@@ -80,6 +80,8 @@ ACTIONS = {
     "quit": lambda _: sys.exit(1),
 }
 
+CLEANUP_DIRECTORIES_ON_EXIT = False
+
 log = logging.getLogger("memgraph.tests.e2e")
 
 
@@ -109,10 +111,11 @@ def _start_instance(name, args, log_file, setup_queries, use_ssl, procdir, data_
     assert not is_port_in_use(
         extract_bolt_port(args)
     ), "If this raises, you are trying to start an instance on a port already used by one already running instance."
-    mg_instance = MemgraphInstanceRunner(MEMGRAPH_BINARY, use_ssl)
-    MEMGRAPH_INSTANCES[name] = mg_instance
+
     log_file_path = os.path.join(BUILD_DIR, "logs", log_file)
     data_directory_path = os.path.join(BUILD_DIR, data_directory)
+    mg_instance = MemgraphInstanceRunner(MEMGRAPH_BINARY, use_ssl, {data_directory_path})
+    MEMGRAPH_INSTANCES[name] = mg_instance
     binary_args = args + ["--log-file", log_file_path] + ["--data-directory", data_directory_path]
 
     if len(procdir) != 0:
@@ -122,39 +125,43 @@ def _start_instance(name, args, log_file, setup_queries, use_ssl, procdir, data_
     assert mg_instance.is_running(), "An error occured after starting Memgraph instance: application stopped running."
 
 
-def stop_all():
+def stop_all(keep_directories=True):
     for mg_instance in MEMGRAPH_INSTANCES.values():
-        mg_instance.stop()
+        mg_instance.stop(keep_directories)
     MEMGRAPH_INSTANCES.clear()
 
 
-def stop_instance(context, name):
+def stop_instance(context, name, keep_directories=True):
     for key, _ in context.items():
         if key != name:
             continue
-        MEMGRAPH_INSTANCES[name].stop()
+        MEMGRAPH_INSTANCES[name].stop(keep_directories)
         MEMGRAPH_INSTANCES.pop(name)
 
 
-def stop(context, name):
+def stop(context, name, keep_directories=True):
     if name != "all":
-        stop_instance(context, name)
+        stop_instance(context, name, keep_directories)
         return
 
     stop_all()
 
 
-def kill(context, name):
+def kill(context, name, keep_directories=True):
     for key in context.keys():
         if key != name:
             continue
-        MEMGRAPH_INSTANCES[name].kill()
+        MEMGRAPH_INSTANCES[name].kill(keep_directories)
         MEMGRAPH_INSTANCES.pop(name)
 
 
+def cleanup_directories_on_exit(value=True):
+    CLEANUP_DIRECTORIES_ON_EXIT = value
+
+
 @atexit.register
 def cleanup():
-    stop_all()
+    stop_all(CLEANUP_DIRECTORIES_ON_EXIT)
 
 
 def start_instance(context, name, procdir):
@@ -184,8 +191,8 @@ def start_instance(context, name, procdir):
     assert len(mg_instances) == 1
 
 
-def start_all(context, procdir=""):
-    stop_all()
+def start_all(context, procdir="", keep_directories=True):
+    stop_all(keep_directories)
     for key, _ in context.items():
         start_instance(context, key, procdir)
 

tests/e2e/memgraph.py

@@ -11,6 +11,7 @@
 
 import copy
 import os
+import shutil
 import subprocess
 import sys
 import time
@@ -56,13 +57,14 @@ def replace_paths(path):
 
 
 class MemgraphInstanceRunner:
-    def __init__(self, binary_path=MEMGRAPH_BINARY, use_ssl=False):
+    def __init__(self, binary_path=MEMGRAPH_BINARY, use_ssl=False, delete_on_stop=None):
         self.host = "127.0.0.1"
         self.bolt_port = None
         self.binary_path = binary_path
         self.args = None
         self.proc_mg = None
         self.ssl = use_ssl
+        self.delete_on_stop = delete_on_stop
 
     def execute_setup_queries(self, setup_queries):
         if setup_queries is None:
@@ -128,7 +130,7 @@ class MemgraphInstanceRunner:
             return False
         return True
 
-    def stop(self):
+    def stop(self, keep_directories=False):
         if not self.is_running():
             return
 
@@ -140,9 +142,16 @@ class MemgraphInstanceRunner:
 
         time.sleep(1)
 
-    def kill(self):
+        if not keep_directories:
+            for folder in self.delete_on_stop or {}:
+                shutil.rmtree(folder)
+
+    def kill(self, keep_directories=False):
         if not self.is_running():
             return
         self.proc_mg.kill()
         code = self.proc_mg.wait()
+        if not keep_directories:
+            for folder in self.delete_on_stop or {}:
+                shutil.rmtree(folder)
         assert code == -9, "The killed Memgraph process exited with non-nine!"

tests/e2e/mg_utils.py

@@ -1,12 +1,11 @@
 import time
 
 
-def mg_sleep_and_assert(expected_value, function_to_retrieve_data, max_duration=20, time_between_attempt=0.05):
+def mg_sleep_and_assert(expected_value, function_to_retrieve_data, max_duration=20, time_between_attempt=0.2):
     result = function_to_retrieve_data()
     start_time = time.time()
     while result != expected_value:
-        current_time = time.time()
-        duration = current_time - start_time
+        duration = time.time() - start_time
         if duration > max_duration:
             assert (
                 False

tests/e2e/query_modules/schema_test.py

@@ -431,7 +431,7 @@ def test_node_type_properties1():
             f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
         )[0]
     )
-    assert (result) == [":`Activity`", ["Activity"], "location", ["String"], True]
+    assert (result) == [":`Activity`", ["Activity"], "location", "String", False]
 
     result = list(
         execute_and_fetch_all(
@@ -439,7 +439,7 @@ def test_node_type_properties1():
             f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
         )[1]
     )
-    assert (result) == [":`Activity`", ["Activity"], "name", ["String"], True]
+    assert (result) == [":`Activity`", ["Activity"], "name", "String", False]
 
     result = list(
         execute_and_fetch_all(
@@ -447,7 +447,7 @@ def test_node_type_properties1():
             f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
         )[2]
     )
-    assert (result) == [":`Dog`", ["Dog"], "name", ["String"], True]
+    assert (result) == [":`Dog`", ["Dog"], "name", "String", False]
 
     result = list(
         execute_and_fetch_all(
@@ -455,7 +455,81 @@ def test_node_type_properties1():
             f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
         )[3]
     )
-    assert (result) == [":`Dog`", ["Dog"], "owner", ["String"], True]
+    assert (result) == [":`Dog`", ["Dog"], "owner", "String", False]
+
+
+def test_node_type_properties2():
+    cursor = connect().cursor()
+    execute_and_fetch_all(
+        cursor,
+        """
+        CREATE (d:MyNode)
+        CREATE (n:MyNode)
+        """,
+    )
+    result = execute_and_fetch_all(
+        cursor,
+        f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
+    )
+    assert (list(result[0])) == [":`MyNode`", ["MyNode"], "", "", False]
+    assert (result.__len__()) == 1
+
+
+def test_node_type_properties3():
+    cursor = connect().cursor()
+    execute_and_fetch_all(
+        cursor,
+        """
+        CREATE (d:Dog {name: 'Rex', owner: 'Carl'})
+        CREATE (n:Dog)
+        """,
+    )
+    result = execute_and_fetch_all(
+        cursor,
+        f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
+    )
+
+    assert (list(result[0])) == [":`Dog`", ["Dog"], "name", "String", False]
+    assert (list(result[1])) == [":`Dog`", ["Dog"], "owner", "String", False]
+    assert (result.__len__()) == 2
+
+
+def test_node_type_properties4():
+    cursor = connect().cursor()
+    execute_and_fetch_all(
+        cursor,
+        """
+        CREATE (n:Label1:Label2 {property1: 'value1', property2: 'value2'})
+        CREATE (m:Label2:Label1 {property3: 'value3'})
+        """,
+    )
+    result = list(
+        execute_and_fetch_all(
+            cursor,
+            f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
+        )
+    )
+    assert (list(result[0])) == [":`Label1`:`Label2`", ["Label1", "Label2"], "property1", "String", False]
+    assert (list(result[1])) == [":`Label1`:`Label2`", ["Label1", "Label2"], "property2", "String", False]
+    assert (list(result[2])) == [":`Label1`:`Label2`", ["Label1", "Label2"], "property3", "String", False]
+    assert (result.__len__()) == 3
+
+
+def test_node_type_properties5():
+    cursor = connect().cursor()
+    execute_and_fetch_all(
+        cursor,
+        """
+        CREATE (d:Dog {name: 'Rex'})
+        """,
+    )
+    result = execute_and_fetch_all(
+        cursor,
+        f"CALL libschema.node_type_properties() YIELD nodeType, nodeLabels, propertyName, propertyTypes , mandatory RETURN nodeType, nodeLabels, propertyName, propertyTypes , mandatory ORDER BY propertyName, nodeLabels[0];",
+    )
+
+    assert (list(result[0])) == [":`Dog`", ["Dog"], "name", "String", True]
+    assert (result.__len__()) == 1
 
 
 def test_rel_type_properties1():
@@ -473,5 +547,38 @@ def test_rel_type_properties1():
     assert (result) == [":`LOVES`", "", "", False]
 
 
+def test_rel_type_properties2():
+    cursor = connect().cursor()
+    execute_and_fetch_all(
+        cursor,
+        """
+        CREATE (d:Dog {name: 'Rex', owner: 'Carl'})-[l:LOVES]->(a:Activity {name: 'Running', location: 'Zadar'})
+        CREATE (n:Dog {name: 'Simba', owner: 'Lucy'})-[j:LOVES {duration: 30}]->(b:Activity {name: 'Running', location: 'Zadar'})
+        """,
+    )
+    result = execute_and_fetch_all(
+        cursor,
+        f"CALL libschema.rel_type_properties() YIELD relType,propertyName, propertyTypes , mandatory RETURN relType, propertyName, propertyTypes , mandatory;",
+    )
+    assert (list(result[0])) == [":`LOVES`", "duration", "Int", False]
+    assert (result.__len__()) == 1
+
+
+def test_rel_type_properties3():
+    cursor = connect().cursor()
+    execute_and_fetch_all(
+        cursor,
+        """
+        CREATE (n:Dog {name: 'Simba', owner: 'Lucy'})-[j:LOVES {duration: 30}]->(b:Activity {name: 'Running', location: 'Zadar'})
+        """,
+    )
+    result = execute_and_fetch_all(
+        cursor,
+        f"CALL libschema.rel_type_properties() YIELD relType,propertyName, propertyTypes , mandatory RETURN relType, propertyName, propertyTypes , mandatory;",
+    )
+    assert (list(result[0])) == [":`LOVES`", "duration", "Int", True]
+    assert (result.__len__()) == 1
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/runner.py

@@ -53,8 +53,8 @@ def run(args):
 
         # Setup.
         @atexit.register
-        def cleanup():
-            interactive_mg_runner.stop_all()
+        def cleanup(keep_directories=True):
+            interactive_mg_runner.stop_all(keep_directories)
 
         if "pre_set_workload" in workload:
             binary = os.path.join(BUILD_DIR, workload["pre_set_workload"])
@@ -92,7 +92,7 @@ def run(args):
                     data = mg_instance.query(validation["query"], conn)[0][0]
                     assert data == validation["expected"]
                 conn.close()
-        cleanup()
+        cleanup(keep_directories=False)
         log.info("%s PASSED.", workload_name)
 
 

tests/e2e/streams/common.py

@@ -9,12 +9,12 @@
 # by the Apache License, Version 2.0, included in the file
 # licenses/APL.txt.
 
+import time
+from multiprocessing import Manager, Process, Value
+
 import mgclient
 import pytest
-import time
-
 from mg_utils import mg_sleep_and_assert
-from multiprocessing import Manager, Process, Value
 
 # These are the indices of the different values in the result of SHOW STREAM
 # query
@@ -103,15 +103,53 @@ def get_stream_info(cursor, stream_name):
 
 def get_is_running(cursor, stream_name):
     stream_info = get_stream_info(cursor, stream_name)
-
-    assert stream_info
+    assert stream_info is not None
     return stream_info[IS_RUNNING]
 
 
-def start_stream(cursor, stream_name):
-    execute_and_fetch_all(cursor, f"START STREAM {stream_name}")
+def create_stream(
+    cursor,
+    stream_name,
+    topics,
+    transformation,
+    consumer_group=None,
+    batch_interval=None,
+    batch_size=None,
+    bootstrap_servers=None,
+    configs=None,
+    credentials=None,
+):
+    query_str = f"CREATE KAFKA STREAM {stream_name} TOPICS {topics} TRANSFORM {transformation}"
+    if consumer_group is not None:
+        query_str += f" CONSUMER_GROUP {consumer_group}"
+    if batch_interval is not None:
+        query_str += f" BATCH_INTERVAL {batch_interval}"
+    if batch_size is not None:
+        query_str += f" BATCH_SIZE {batch_size}"
+    if bootstrap_servers is not None:
+        query_str += f" BOOTSTRAP_SERVERS {bootstrap_servers}"
+    if configs is not None:
+        query_str += f" CONFIGS {configs}"
+    if credentials is not None:
+        query_str += f" CREDENTIALS {credentials}"
+    execute_and_fetch_all(cursor, query_str)
 
+
+def start_stream(cursor, stream_name, sleep=True):
+    # Sleep is needed because although is_running returns True,
+    # the stream cannot accept messages yet
+    execute_and_fetch_all(cursor, f"START STREAM {stream_name}")
     assert get_is_running(cursor, stream_name)
+    if sleep:
+        time.sleep(5)
+
+
+def start_streams(cursor, stream_names):
+    # Start every stream but don't sleep after each creation
+    for stream_name in stream_names:
+        execute_and_fetch_all(cursor, f"START STREAM {stream_name}")
+        assert get_is_running(cursor, stream_name)
+    time.sleep(5)
 
 
 def start_stream_with_limit(cursor, stream_name, batch_limit, timeout=None):
@@ -123,13 +161,11 @@ def start_stream_with_limit(cursor, stream_name, batch_limit, timeout=None):
 
 def stop_stream(cursor, stream_name):
     execute_and_fetch_all(cursor, f"STOP STREAM {stream_name}")
-
     assert not get_is_running(cursor, stream_name)
 
 
 def drop_stream(cursor, stream_name):
     execute_and_fetch_all(cursor, f"DROP STREAM {stream_name}")
-
     assert get_stream_info(cursor, stream_name) is None
 
 
@@ -227,7 +263,7 @@ def test_start_and_stop_during_check(
     try:
         check_stream_proc.start()
 
-        time.sleep(0.5)
+        time.sleep(3)
 
         assert timed_wait(lambda: check_counter.value == CHECK_BEFORE_EXECUTE)
         assert timed_wait(lambda: get_is_running(cursor, "test_stream"))
@@ -260,25 +296,22 @@ def test_start_and_stop_during_check(
 
 def test_start_checked_stream_after_timeout(connection, stream_creator):
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor, stream_creator("test_stream"))
+    stream_name = "test_start_checked_stream_after_timeout"
+    execute_and_fetch_all(cursor, stream_creator(stream_name))
 
-    TIMEOUT_IN_MS = 2000
-    TIMEOUT_IN_SECONDS = TIMEOUT_IN_MS / 1000
+    timeout_in_ms = 2000
 
     def call_check():
-        execute_and_fetch_all(connect().cursor(), f"CHECK STREAM test_stream TIMEOUT {TIMEOUT_IN_MS}")
+        execute_and_fetch_all(connect().cursor(), f"CHECK STREAM {stream_name} TIMEOUT {timeout_in_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"))
-    start_stream(cursor, "test_stream")
-    end = time.time()
+    assert timed_wait(lambda: get_is_running(cursor, stream_name))
+    start_stream(cursor, stream_name)
 
-    assert (end - start) < 1.3 * TIMEOUT_IN_SECONDS, "The START STREAM was blocked too long"
-    assert get_is_running(cursor, "test_stream")
-    stop_stream(cursor, "test_stream")
+    assert get_is_running(cursor, stream_name)
+    stop_stream(cursor, stream_name)
 
 
 def test_check_stream_same_number_of_queries_than_messages(connection, stream_creator, message_sender):
@@ -313,7 +346,6 @@ def test_check_stream_same_number_of_queries_than_messages(connection, stream_cr
     # #            {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 = (
@@ -351,7 +383,7 @@ def test_check_stream_different_number_of_queries_than_messages(connection, stre
     STREAM_NAME = "test_stream"
     cursor = connection.cursor()
     execute_and_fetch_all(cursor, stream_creator(STREAM_NAME, BATCH_SIZE))
-    time.sleep(2)
+    time.sleep(3)
 
     results = Manager().Namespace()
 
@@ -413,30 +445,32 @@ def test_check_stream_different_number_of_queries_than_messages(connection, stre
     assert expected_queries_and_raw_messages_3 == results.value[2]
 
 
-def test_start_stream_with_batch_limit(connection, stream_creator, messages_sender):
-    STREAM_NAME = "test"
+def test_start_stream_with_batch_limit(connection, stream_name, stream_creator, messages_sender):
     BATCH_LIMIT = 5
+    TIMEOUT = 10000
 
     cursor = connection.cursor()
-    execute_and_fetch_all(cursor, stream_creator(STREAM_NAME))
+    execute_and_fetch_all(cursor, stream_creator())
+    results = execute_and_fetch_all(connection.cursor(), "SHOW STREAMS")
+    assert len(results) == 1
 
-    def start_new_stream_with_limit(stream_name, batch_limit):
+    def start_new_stream_with_limit():
         connection = connect()
         cursor = connection.cursor()
-        start_stream_with_limit(cursor, stream_name, batch_limit)
-
-    thread_stream_running = Process(target=start_new_stream_with_limit, daemon=True, args=(STREAM_NAME, BATCH_LIMIT))
-    thread_stream_running.start()
+        start_stream_with_limit(cursor, stream_name, BATCH_LIMIT, TIMEOUT)
 
     def is_running():
-        return get_is_running(cursor, STREAM_NAME)
+        return get_is_running(cursor, stream_name)
 
+    thread_stream_running = Process(target=start_new_stream_with_limit)
+    thread_stream_running.start()
+
+    execute_and_fetch_all(connection.cursor(), "SHOW STREAMS")
     assert mg_sleep_and_assert(True, is_running)
-
     messages_sender(BATCH_LIMIT - 1)
 
     # We have not sent enough batches to reach the limit. We check that the stream is still correctly running.
-    assert get_is_running(cursor, STREAM_NAME)
+    assert get_is_running(cursor, stream_name)
 
     # We send a last message to reach the batch_limit
     messages_sender(1)
@@ -549,7 +583,6 @@ def test_check_while_stream_with_batch_limit_running(connection, stream_creator,
     STREAM_NAME = "test_batch_limit_and_check"
     BATCH_LIMIT = 1
     TIMEOUT = 10000
-    TIMEOUT_IN_SECONDS = TIMEOUT / 1000
 
     cursor = connection.cursor()
     execute_and_fetch_all(cursor, stream_creator(STREAM_NAME))
@@ -625,7 +658,7 @@ def test_check_stream_with_batch_limit_with_invalid_batch_limit(connection, stre
 
     cursor = connection.cursor()
     execute_and_fetch_all(cursor, stream_creator(STREAM_NAME))
-    time.sleep(2)
+    time.sleep(3)
 
     # 1/ checking with batch_limit=-10
     batch_limit = -10

tests/e2e/streams/conftest.py

@@ -9,15 +9,14 @@
 # by the Apache License, Version 2.0, included in the file
 # licenses/APL.txt.
 
+import pulsar
 import pytest
+from common import NAME, PULSAR_SERVICE_URL, connect, execute_and_fetch_all
 from kafka import KafkaProducer
 from kafka.admin import KafkaAdminClient, NewTopic
 
-import pulsar
 import requests
 
-from common import NAME, connect, execute_and_fetch_all, PULSAR_SERVICE_URL
-
 # To run these test locally a running Kafka sever is necessery. The test tries
 # to connect on localhost:9092.
 
@@ -37,29 +36,22 @@ def connection():
 
 
 def get_topics(num):
-    return [f'topic_{i}' for i in range(num)]
+    return [f"topic_{i}" for i in range(num)]
 
 
 @pytest.fixture(scope="function")
 def kafka_topics():
-    admin_client = KafkaAdminClient(
-        bootstrap_servers="localhost:9092",
-        client_id="test")
+    admin_client = KafkaAdminClient(bootstrap_servers="localhost:29092", client_id="test")
     # The issue arises if we remove default kafka topics, e.g.
     # "__consumer_offsets"
-    previous_topics = [
-        topic for topic in admin_client.list_topics() if topic != "__consumer_offsets"]
+    previous_topics = [topic for topic in admin_client.list_topics() if topic != "__consumer_offsets"]
     if previous_topics:
         admin_client.delete_topics(topics=previous_topics, timeout_ms=5000)
 
     topics = get_topics(3)
     topics_to_create = []
     for topic in topics:
-        topics_to_create.append(
-            NewTopic(
-                name=topic,
-                num_partitions=1,
-                replication_factor=1))
+        topics_to_create.append(NewTopic(name=topic, num_partitions=1, replication_factor=1))
 
     admin_client.create_topics(new_topics=topics_to_create, timeout_ms=5000)
     yield topics
@@ -68,7 +60,7 @@ def kafka_topics():
 
 @pytest.fixture(scope="function")
 def kafka_producer():
-    yield KafkaProducer(bootstrap_servers="localhost:9092")
+    yield KafkaProducer(bootstrap_servers=["localhost:29092"], api_version_auto_timeout_ms=10000)
 
 
 @pytest.fixture(scope="function")
@@ -80,6 +72,5 @@ def pulsar_client():
 def pulsar_topics():
     topics = get_topics(3)
     for topic in topics:
-        requests.delete(
-            f'http://127.0.0.1:6652/admin/v2/persistent/public/default/{topic}?force=true')
+        requests.delete(f"http://localhost:6652/admin/v2/persistent/public/default/{topic}?force=true")
     yield topics

tests/e2e/streams/kafka.yml

@@ -1,20 +0,0 @@
-version: '3.7'
-services:
-  zookeeper:
-    image: 'bitnami/zookeeper:latest'
-    ports:
-      - '2181:2181'
-    environment:
-      - ALLOW_ANONYMOUS_LOGIN=yes
-  kafka:
-    image: 'bitnami/kafka:latest'
-    ports:
-      - '9092:9092'
-    environment:
-      - KAFKA_BROKER_ID=1
-      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
-      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
-      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
-      - ALLOW_PLAINTEXT_LISTENER=yes
-    depends_on:
-      - zookeeper

tests/e2e/streams/kafka/docker-compose.yml

@@ -0,0 +1,22 @@
+version: '3'
+services:
+  zookeeper:
+    image: 'bitnami/zookeeper:3.9.1'
+    ports:
+      - '2181:2181'
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+  kafka:
+    image: 'bitnami/kafka:3.6.1'
+    ports:
+      - '9092:9092'
+      - '29092:29092'
+    environment:
+      - KAFKA_BROKER_ID=1
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,PLAINTEXT_HOST://:29092
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092,PLAINTEXT_HOST://localhost:29092
+      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
+      - ALLOW_PLAINTEXT_LISTENER=yes
+    depends_on:
+      - zookeeper

tests/e2e/streams/kafka_streams_tests.py

@@ -12,31 +12,30 @@
 # licenses/APL.txt.
 
 import sys
-import pytest
-import mgclient
 import time
-from mg_utils import mg_sleep_and_assert
-from multiprocessing import Process, Value
+from multiprocessing import Process
+
 import common
+import mgclient
+import pytest
+from mg_utils import mg_sleep_and_assert
 
 TRANSFORMATIONS_TO_CHECK_C = ["c_transformations.empty_transformation"]
-
 TRANSFORMATIONS_TO_CHECK_PY = ["kafka_transform.simple", "kafka_transform.with_parameters"]
+KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT = 60
 
 
 @pytest.mark.parametrize("transformation", TRANSFORMATIONS_TO_CHECK_PY)
 def test_simple(kafka_producer, kafka_topics, connection, transformation):
     assert len(kafka_topics) > 0
+    stream_name = "test_simple_" + transformation.split(".")[1]
+
     cursor = connection.cursor()
-    common.execute_and_fetch_all(
-        cursor,
-        f"CREATE KAFKA STREAM test TOPICS {','.join(kafka_topics)} TRANSFORM {transformation}",
-    )
-    common.start_stream(cursor, "test")
-    time.sleep(5)
+    common.create_stream(cursor, stream_name, ",".join(kafka_topics), transformation)
+    common.start_stream(cursor, stream_name)
 
     for topic in kafka_topics:
-        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=60)
+        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     for topic in kafka_topics:
         common.kafka_check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)
@@ -47,22 +46,15 @@ def test_separate_consumers(kafka_producer, kafka_topics, connection, transforma
     assert len(kafka_topics) > 0
     cursor = connection.cursor()
 
-    stream_names = []
-    for topic in kafka_topics:
-        stream_name = "stream_" + topic
-        stream_names.append(stream_name)
-        common.execute_and_fetch_all(
-            cursor,
-            f"CREATE KAFKA STREAM {stream_name} TOPICS {topic} TRANSFORM {transformation}",
-        )
+    stream_names = ["stream_" + transformation.split(".")[1] + "_" + topic for topic in kafka_topics]
 
-    for stream_name in stream_names:
-        common.start_stream(cursor, stream_name)
+    for stream_name, topic in zip(stream_names, kafka_topics):
+        common.create_stream(cursor, stream_name, topic, transformation)
 
-    time.sleep(5)
+    common.start_streams(cursor, stream_names)
 
     for topic in kafka_topics:
-        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=60)
+        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     for topic in kafka_topics:
         common.kafka_check_vertex_exists_with_topic_and_payload(cursor, topic, common.SIMPLE_MSG)
@@ -77,23 +69,20 @@ def test_start_from_last_committed_offset(kafka_producer, kafka_topics, connecti
     # restarting Memgraph during a single workload cannot be done currently.
     assert len(kafka_topics) > 0
     cursor = connection.cursor()
-    common.execute_and_fetch_all(
-        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)
+    stream_name = "test_start_from_last_committed_offset"
+    common.create_stream(cursor, stream_name, kafka_topics[0], "kafka_transform.simple")
+    common.start_stream(cursor, stream_name)
 
+    kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
     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")
+    common.stop_stream(cursor, stream_name)
+    common.drop_stream(cursor, stream_name)
 
     messages = [b"second message", b"third message"]
     for message in messages:
-        kafka_producer.send(kafka_topics[0], message).get(timeout=60)
+        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     for message in messages:
         vertices_with_msg = common.execute_and_fetch_all(
@@ -103,11 +92,8 @@ def test_start_from_last_committed_offset(kafka_producer, kafka_topics, connecti
 
         assert len(vertices_with_msg) == 0
 
-    common.execute_and_fetch_all(
-        cursor,
-        f"CREATE KAFKA STREAM test TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple",
-    )
-    common.start_stream(cursor, "test")
+    common.create_stream(cursor, stream_name, kafka_topics[0], "kafka_transform.simple")
+    common.start_stream(cursor, stream_name)
 
     for message in messages:
         common.kafka_check_vertex_exists_with_topic_and_payload(cursor, kafka_topics[0], message)
@@ -118,32 +104,26 @@ 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,
-        f"CREATE KAFKA STREAM test TOPICS {kafka_topics[0]} TRANSFORM {transformation} BATCH_SIZE {BATCH_SIZE}",
-    )
-    common.start_stream(cursor, "test")
-    time.sleep(1)
+    stream_name = "test_check_stream_" + transformation.split(".")[1]
 
-    kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=60)
-    common.stop_stream(cursor, "test")
+    cursor = connection.cursor()
+
+    common.create_stream(cursor, stream_name, kafka_topics[0], transformation, batch_size=BATCH_SIZE)
+    common.start_stream(cursor, stream_name)
+    kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
+    common.stop_stream(cursor, stream_name)
 
     messages = [b"first message", b"second message", b"third message"]
     for message in messages:
-        kafka_producer.send(kafka_topics[0], message).get(timeout=60)
+        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     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}")
+        test_results = common.execute_and_fetch_all(cursor, f"CHECK STREAM {stream_name} 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 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}'"
@@ -165,41 +145,48 @@ def test_check_stream(kafka_producer, kafka_topics, connection, transformation):
     check_check_stream(1)
     check_check_stream(2)
     check_check_stream(3)
-    common.start_stream(cursor, "test")
+    common.start_stream(cursor, stream_name)
 
     for message in messages:
         common.kafka_check_vertex_exists_with_topic_and_payload(cursor, kafka_topics[0], message)
 
 
-def test_show_streams(kafka_producer, kafka_topics, connection):
+def test_show_streams(kafka_topics, connection):
     assert len(kafka_topics) > 1
     cursor = connection.cursor()
-    common.execute_and_fetch_all(
-        cursor,
-        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
-    common.execute_and_fetch_all(
+    default_values_stream = "default_values"
+    complex_values_stream = "complex_values"
+
+    common.create_stream(
+        cursor, default_values_stream, kafka_topics[0], "kafka_transform.simple", bootstrap_servers="'localhost:29092'"
+    )
+    common.create_stream(
         cursor,
-        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} ",
+        complex_values_stream,
+        ",".join(kafka_topics),
+        "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
 
     common.check_stream_info(
         cursor,
-        "default_values",
-        ("default_values", "kafka", 100, 1000, "kafka_transform.simple", None, False),
+        default_values_stream,
+        (default_values_stream, "kafka", 100, 1000, "kafka_transform.simple", None, False),
     )
 
     common.check_stream_info(
         cursor,
-        "complex_values",
+        complex_values_stream,
         (
-            "complex_values",
+            complex_values_stream,
             "kafka",
             BATCH_INTERVAL,
             BATCH_SIZE,
@@ -219,7 +206,7 @@ def test_start_and_stop_during_check(kafka_producer, kafka_topics, connection, o
         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)
+        kafka_producer.send(kafka_topics[0], msg).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     common.test_start_and_stop_during_check(
         operation,
@@ -235,14 +222,12 @@ def test_check_already_started_stream(kafka_topics, connection):
     assert len(kafka_topics) > 0
     cursor = connection.cursor()
 
-    common.execute_and_fetch_all(
-        cursor,
-        f"CREATE KAFKA STREAM started_stream TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple",
-    )
-    common.start_stream(cursor, "started_stream")
+    stream_name = "test_check_already_started_stream"
+    common.create_stream(cursor, stream_name, kafka_topics[0], "kafka_transform.simple")
+    common.start_stream(cursor, stream_name)
 
     with pytest.raises(mgclient.DatabaseError):
-        common.execute_and_fetch_all(cursor, "CHECK STREAM started_stream")
+        common.execute_and_fetch_all(cursor, f"CHECK STREAM {stream_name}")
 
 
 def test_start_checked_stream_after_timeout(kafka_topics, connection):
@@ -254,19 +239,14 @@ def test_start_checked_stream_after_timeout(kafka_topics, connection):
 
 def test_restart_after_error(kafka_producer, kafka_topics, connection):
     cursor = connection.cursor()
-    common.execute_and_fetch_all(
-        cursor,
-        f"CREATE KAFKA STREAM test_stream TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.query",
-    )
+    stream_name = "test_restart_after_error"
+    common.create_stream(cursor, stream_name, kafka_topics[0], "kafka_transform.query")
+    common.start_stream(cursor, stream_name)
 
-    common.start_stream(cursor, "test_stream")
-    time.sleep(1)
+    kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
+    assert common.timed_wait(lambda: not common.get_is_running(cursor, stream_name))
 
-    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"))
-
-    common.start_stream(cursor, "test_stream")
-    time.sleep(1)
+    common.start_stream(cursor, stream_name)
     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")
 
@@ -275,23 +255,21 @@ def test_restart_after_error(kafka_producer, kafka_topics, connection):
 def test_bootstrap_server(kafka_producer, kafka_topics, connection, transformation):
     assert len(kafka_topics) > 0
     cursor = connection.cursor()
-    LOCAL = "localhost:9092"
-    common.execute_and_fetch_all(
-        cursor,
-        f"CREATE KAFKA STREAM test TOPICS {','.join(kafka_topics)} TRANSFORM {transformation} BOOTSTRAP_SERVERS '{LOCAL}'",
-    )
-    common.start_stream(cursor, "test")
-    time.sleep(5)
+    local = "'localhost:29092'"
+    stream_name = "test_bootstrap_server_" + transformation.split(".")[1]
+
+    common.create_stream(cursor, stream_name, ",".join(kafka_topics), transformation, bootstrap_servers=local)
+    common.start_stream(cursor, stream_name)
 
     for topic in kafka_topics:
-        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=60)
+        kafka_producer.send(topic, common.SIMPLE_MSG).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     for topic in kafka_topics:
         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_topics, connection, transformation):
     assert len(kafka_topics) > 0
     cursor = connection.cursor()
     with pytest.raises(mgclient.DatabaseError):
@@ -312,7 +290,7 @@ def test_set_offset(kafka_producer, kafka_topics, connection, transformation):
 
     messages = [f"{i} message" for i in range(1, 21)]
     for message in messages:
-        kafka_producer.send(kafka_topics[0], message.encode()).get(timeout=60)
+        kafka_producer.send(kafka_topics[0], message.encode()).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     def consume(expected_msgs):
         common.start_stream(cursor, "test")
@@ -352,7 +330,7 @@ def test_set_offset(kafka_producer, kafka_topics, connection, transformation):
     res = execute_set_offset_and_consume(-2, [])
     assert len(res) == 0
     last_msg = "Final Message"
-    kafka_producer.send(kafka_topics[0], last_msg.encode()).get(timeout=60)
+    kafka_producer.send(kafka_topics[0], last_msg.encode()).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
     res = consume([last_msg])
     assert len(res) == 1
     assert comparison_check("Final Message", res[0])
@@ -361,21 +339,27 @@ 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"
-    common.execute_and_fetch_all(
+    stream_name = "test_stream"
+    configs = {"sasl.username": "michael.scott"}
+    local = "localhost:29092"
+    credentials = {"sasl.password": "S3cr3tP4ssw0rd"}
+    consumer_group = "ConsumerGr"
+
+    common.create_stream(
         cursor,
-        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_name,
+        ",".join(kafka_topics),
+        "kafka_transform.simple",
+        consumer_group=consumer_group,
+        bootstrap_servers=f"'{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)
 
 
@@ -398,7 +382,7 @@ def test_check_stream_same_number_of_queries_than_messages(kafka_producer, kafka
         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)
+        kafka_producer.send(kafka_topics[0], msg).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     common.test_check_stream_same_number_of_queries_than_messages(connection, stream_creator, message_sender)
 
@@ -409,30 +393,33 @@ def test_check_stream_different_number_of_queries_than_messages(kafka_producer,
     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}"
+        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)
+        kafka_producer.send(kafka_topics[0], msg).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     common.test_check_stream_different_number_of_queries_than_messages(connection, stream_creator, message_sender)
 
 
 def test_start_stream_with_batch_limit(kafka_producer, kafka_topics, connection):
     assert len(kafka_topics) > 0
+    STREAM_NAME = "test_start_stream_with_batch_limit"
 
-    def stream_creator(stream_name):
+    def stream_creator():
         return (
-            f"CREATE KAFKA STREAM {stream_name} TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple BATCH_SIZE 1"
+            f"CREATE KAFKA STREAM {STREAM_NAME} TOPICS {kafka_topics[0]} TRANSFORM kafka_transform.simple BATCH_SIZE 1"
         )
 
     def messages_sender(nof_messages):
-        for x in range(nof_messages):
-            kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(timeout=60)
+        for _ in range(nof_messages):
+            kafka_producer.send(kafka_topics[0], common.SIMPLE_MSG).get(
+                timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT
+            )
 
-    common.test_start_stream_with_batch_limit(connection, stream_creator, messages_sender)
+    common.test_start_stream_with_batch_limit(connection, STREAM_NAME, stream_creator, messages_sender)
 
 
-def test_start_stream_with_batch_limit_timeout(kafka_producer, kafka_topics, connection):
+def test_start_stream_with_batch_limit_timeout(kafka_topics, connection):
     assert len(kafka_topics) > 0
 
     def stream_creator(stream_name):
@@ -443,7 +430,7 @@ def test_start_stream_with_batch_limit_timeout(kafka_producer, kafka_topics, con
     common.test_start_stream_with_batch_limit_timeout(connection, stream_creator)
 
 
-def test_start_stream_with_batch_limit_reaching_timeout(kafka_producer, kafka_topics, connection):
+def test_start_stream_with_batch_limit_reaching_timeout(kafka_topics, connection):
     assert len(kafka_topics) > 0
 
     def stream_creator(stream_name, batch_size):
@@ -461,7 +448,7 @@ def test_start_stream_with_batch_limit_while_check_running(kafka_producer, kafka
         )
 
     def message_sender(message):
-        kafka_producer.send(kafka_topics[0], message).get(timeout=6000)
+        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     def setup_function(start_check_stream, cursor, stream_name, batch_limit, timeout):
         thread_stream_check = Process(target=start_check_stream, daemon=True, args=(stream_name, batch_limit, timeout))
@@ -488,12 +475,12 @@ def test_check_while_stream_with_batch_limit_running(kafka_producer, kafka_topic
         )
 
     def message_sender(message):
-        kafka_producer.send(kafka_topics[0], message).get(timeout=6000)
+        kafka_producer.send(kafka_topics[0], message).get(timeout=KAFKA_PRODUCER_SENDING_MSG_DEFAULT_TIMEOUT)
 
     common.test_check_while_stream_with_batch_limit_running(connection, stream_creator, message_sender)
 
 
-def test_start_stream_with_batch_limit_with_invalid_batch_limit(kafka_producer, kafka_topics, connection):
+def test_start_stream_with_batch_limit_with_invalid_batch_limit(kafka_topics, connection):
     assert len(kafka_topics) > 0
 
     def stream_creator(stream_name):
@@ -504,7 +491,7 @@ def test_start_stream_with_batch_limit_with_invalid_batch_limit(kafka_producer,
     common.test_start_stream_with_batch_limit_with_invalid_batch_limit(connection, stream_creator)
 
 
-def test_check_stream_with_batch_limit_with_invalid_batch_limit(kafka_producer, kafka_topics, connection):
+def test_check_stream_with_batch_limit_with_invalid_batch_limit(kafka_topics, connection):
     assert len(kafka_topics) > 0
 
     def stream_creator(stream_name):

tests/e2e/streams/pulsar.yml

@@ -1,8 +0,0 @@
-version: '3.7'
-services:
-  pulsar:
-    image: 'apachepulsar/pulsar:latest'
-    ports:
-      - '6652:8080'
-      - '6650:6650'
-    entrypoint: ['bin/pulsar', 'standalone']

tests/e2e/streams/pulsar/docker-compose.yml

@@ -0,0 +1,9 @@
+version: "3"
+
+services:
+  pulsar:
+    image: 'apachepulsar/pulsar:2.8.1'
+    ports:
+      - '6652:8080'
+      - '6650:6650'
+    entrypoint: [ 'bin/pulsar', 'standalone' ]

tests/e2e/streams/pulsar_streams_tests.py

@@ -12,11 +12,12 @@
 # licenses/APL.txt.
 
 import sys
-import pytest
-import mgclient
 import time
 from multiprocessing import Process, Value
+
 import common
+import mgclient
+import pytest
 
 TRANSFORMATIONS_TO_CHECK = ["pulsar_transform.simple", "pulsar_transform.with_parameters"]
 
@@ -80,12 +81,6 @@ def test_start_from_latest_messages(pulsar_client, pulsar_topics, connection):
     # inbetween should be lost. Additionally, we check that consumer continues from the correct message
     # after stopping and starting again.
     assert len(pulsar_topics) > 0
-    cursor = connection.cursor()
-    common.execute_and_fetch_all(
-        cursor,
-        f"CREATE PULSAR STREAM test TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple",
-    )
-    common.start_stream(cursor, "test")
 
     def assert_message_not_consumed(message):
         vertices_with_msg = common.execute_and_fetch_all(
@@ -95,6 +90,13 @@ def test_start_from_latest_messages(pulsar_client, pulsar_topics, connection):
 
         assert len(vertices_with_msg) == 0
 
+    cursor = connection.cursor()
+    common.execute_and_fetch_all(
+        cursor,
+        f"CREATE PULSAR STREAM test TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple",
+    )
+    common.start_stream(cursor, "test")
+
     producer = pulsar_client.create_producer(
         common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
     )
@@ -131,7 +133,7 @@ def test_start_from_latest_messages(pulsar_client, pulsar_topics, connection):
         producer.send(message)
         assert_message_not_consumed(message)
 
-    common.start_stream(cursor, "test")
+    common.start_stream(cursor, "test", sleep=False)
 
     assert_message_not_consumed(LOST_MESSAGE)
 
@@ -338,19 +340,20 @@ def test_service_url(pulsar_client, pulsar_topics, connection, transformation):
 
 def test_start_stream_with_batch_limit(pulsar_client, pulsar_topics, connection):
     assert len(pulsar_topics) > 1
+    STREAM_NAME = "test_start_stream_with_batch_limit"
 
-    def stream_creator(stream_name):
-        return f"CREATE PULSAR STREAM {stream_name} TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple BATCH_SIZE 1"
+    def stream_creator():
+        return f"CREATE PULSAR STREAM {STREAM_NAME} TOPICS {pulsar_topics[0]} TRANSFORM pulsar_transform.simple BATCH_SIZE 1"
 
     producer = pulsar_client.create_producer(
         common.pulsar_default_namespace_topic(pulsar_topics[0]), send_timeout_millis=60000
     )
 
     def messages_sender(nof_messages):
-        for x in range(nof_messages):
+        for _ in range(nof_messages):
             producer.send(common.SIMPLE_MSG)
 
-    common.test_start_stream_with_batch_limit(connection, stream_creator, messages_sender)
+    common.test_start_stream_with_batch_limit(connection, STREAM_NAME, stream_creator, messages_sender)
 
 
 def test_start_stream_with_batch_limit_timeout(pulsar_client, pulsar_topics, connection):
@@ -409,7 +412,7 @@ def test_check_stream_same_number_of_queries_than_messages(pulsar_client, pulsar
     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} "
+        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

tests/stress/memory_limit.py

@@ -173,39 +173,49 @@ def run_writer(repetition_count: int, sleep_sec: float, worker_id: int) -> int:
     """
     This writer creates lot of nodes on each write.
     Also it checks that query failed if memory limit is tried to be broken
+
+    Return:
+        True if write suceeded
+        False otherwise
     """
 
     session = SessionCache.argument_session(args)
 
-    def create() -> bool:
+    def try_create() -> bool:
         """
-        Returns True if done, False if needs to continue
+        Function tries to create until memory limit is reached
+        Return:
+            True if it can continue creating (OOM not reached)
+            False otherwise
         """
-        memory_tracker_data_before_start = get_tracker_data(session)
-        should_fail = memory_tracker_data_before_start >= 2048
-        failed = False
+        should_continue = True
         try:
             try_execute(
                 session,
                 f"FOREACH (i in range(1,10000) | CREATE (:Node {{prop:'big string or something like that'}}))",
             )
         except Exception as ex:
-            failed = True
             output = str(ex)
-            log.info("Exception in create", output)
-            assert "Memory limit exceeded!" in output
+            memory_over_2048_mb = False
+            memory_tracker_data_after_start = get_tracker_data(session)
+            if memory_tracker_data_after_start:
+                memory_over_2048_mb = memory_tracker_data_after_start >= 2048
+            log.info(
+                "Exception in create, exception output:",
+                output,
+                f"Worker {worker_id} started iteration {curr_repetition}, memory over 2048MB: {memory_over_2048_mb}",
+            )
+            has_oom_happend = "Memory limit exceeded!" in output and memory_over_2048_mb
+            should_continue = not has_oom_happend
 
-        if should_fail:
-            assert failed, "Query should have failed"
-            return False
-        return True
+        return should_continue
 
     curr_repetition = 0
 
     while curr_repetition < repetition_count:
         log.info(f"Worker {worker_id} started iteration {curr_repetition}")
 
-        should_continue = create()
+        should_continue = try_create()
 
         if not should_continue:
             return True
@@ -214,6 +224,7 @@ def run_writer(repetition_count: int, sleep_sec: float, worker_id: int) -> int:
         log.info(f"Worker {worker_id} created chain in iteration {curr_repetition}")
 
         curr_repetition += 1
+    return False
 
 
 def execute_function(worker: Worker) -> Worker:

tests/unit/dbms_handler.cpp

@@ -58,6 +58,7 @@ class TestEnvironment : public ::testing::Environment {
   void TearDown() override {
     ptr_.reset();
     auth.reset();
+    std::filesystem::remove_all(storage_directory);
   }
 
   static std::unique_ptr<memgraph::dbms::DbmsHandler> ptr_;

tests/unit/dbms_handler_community.cpp

@@ -58,6 +58,7 @@ class TestEnvironment : public ::testing::Environment {
   void TearDown() override {
     ptr_.reset();
     auth.reset();
+    std::filesystem::remove_all(storage_directory);
   }
 
   static std::unique_ptr<memgraph::dbms::DbmsHandler> ptr_;

tests/unit/interpreter.cpp

@@ -101,6 +101,8 @@ class InterpreterTest : public ::testing::Test {
       disk_test_utils::RemoveRocksDbDirs(testSuite);
       disk_test_utils::RemoveRocksDbDirs(testSuiteCsv);
     }
+
+    std::filesystem::remove_all(data_directory);
   }
 
   InterpreterFaker default_interpreter{&interpreter_context, db};

tests/unit/query_dump.cpp

@@ -700,6 +700,11 @@ TYPED_TEST(DumpTest, CheckStateVertexWithMultipleProperties) {
     config.disk = disk_test_utils::GenerateOnDiskConfig("query-dump-s1").disk;
     config.force_on_disk = true;
   }
+  auto on_exit_s1 = memgraph::utils::OnScopeExit{[&]() {
+    if constexpr (std::is_same_v<TypeParam, memgraph::storage::DiskStorage>) {
+      disk_test_utils::RemoveRocksDbDirs("query-dump-s1");
+    }
+  }};
   memgraph::replication::ReplicationState repl_state(ReplicationStateRootPath(config));
 
   memgraph::utils::Gatekeeper<memgraph::dbms::Database> db_gk(config, repl_state);
@@ -814,7 +819,11 @@ TYPED_TEST(DumpTest, CheckStateSimpleGraph) {
     config.disk = disk_test_utils::GenerateOnDiskConfig("query-dump-s2").disk;
     config.force_on_disk = true;
   }
-
+  auto on_exit_s2 = memgraph::utils::OnScopeExit{[&]() {
+    if constexpr (std::is_same_v<TypeParam, memgraph::storage::DiskStorage>) {
+      disk_test_utils::RemoveRocksDbDirs("query-dump-s2");
+    }
+  }};
   memgraph::replication::ReplicationState repl_state{ReplicationStateRootPath(config)};
   memgraph::utils::Gatekeeper<memgraph::dbms::Database> db_gk{config, repl_state};
   auto db_acc_opt = db_gk.access();

tests/unit/storage_v2_isolation_level.cpp

@@ -15,6 +15,7 @@
 #include "storage/v2/disk/storage.hpp"
 #include "storage/v2/inmemory/storage.hpp"
 #include "storage/v2/isolation_level.hpp"
+#include "utils/on_scope_exit.hpp"
 
 namespace {
 int64_t VerticesCount(memgraph::storage::Storage::Accessor *accessor) {
@@ -113,6 +114,7 @@ TEST_P(StorageIsolationLevelTest, VisibilityOnDiskStorage) {
 
   for (const auto override_isolation_level : isolation_levels) {
     std::unique_ptr<memgraph::storage::Storage> storage(new memgraph::storage::DiskStorage(config));
+    auto on_exit = memgraph::utils::OnScopeExit{[&]() { disk_test_utils::RemoveRocksDbDirs(testSuite); }};
     try {
       this->TestVisibility(storage, default_isolation_level, override_isolation_level);
     } catch (memgraph::utils::NotYetImplemented &) {
@@ -120,10 +122,8 @@ TEST_P(StorageIsolationLevelTest, VisibilityOnDiskStorage) {
           override_isolation_level != memgraph::storage::IsolationLevel::SNAPSHOT_ISOLATION) {
         continue;
       }
-      disk_test_utils::RemoveRocksDbDirs(testSuite);
       throw;
     }
-    disk_test_utils::RemoveRocksDbDirs(testSuite);
   }
 }
 

tests/unit/storage_v2_storage_mode.cpp

@@ -75,6 +75,8 @@ class StorageModeMultiTxTest : public ::testing::Test {
     return tmp;
   }();  // iile
 
+  void TearDown() override { std::filesystem::remove_all(data_directory); }
+
   memgraph::storage::Config config{.durability.storage_directory = data_directory,
                                    .disk.main_storage_directory = data_directory / "disk"};