Add show storage info testing

.github/pull_request_template.md

@@ -1,9 +1,13 @@
-PR checklist:
-- [ ] Check and update documentation if necessary
+[master < Epic] PR
+- [ ] Check, and update documentation if necessary
 - [ ] Write E2E tests
 - [ ] Compare the [benchmarking results](https://bench-graph.memgraph.com/) between the master branch and the Epic branch
 - [ ] Provide the full content or a guide for the final git message
-- [ ] Notify everyone downstream (platform, dx, infra, ...) if there is any breaking change and what to do about it
+
+[master < Task] PR
+- [ ] Check, and update documentation if necessary
+- [ ] Provide the full content or a guide for the final git message
+
 
 To keep docs changelog up to date, one more thing to do:
 - [ ] Write a release note here, including added/changed clauses

src/query/frontend/ast/cypher_main_visitor.cpp

@@ -1216,6 +1216,10 @@ antlrcpp::Any CypherMainVisitor::visitCallProcedure(MemgraphCypher::CallProcedur
       call_proc->memory_limit_ = memory_limit_info->first;
       call_proc->memory_scale_ = memory_limit_info->second;
     }
+  } else {
+    // Default to 100 MB
+    call_proc->memory_limit_ = storage_->Create<PrimitiveLiteral>(TypedValue(100));
+    call_proc->memory_scale_ = 1024U * 1024U;
   }
 
   const auto &maybe_found =
@@ -1236,13 +1240,11 @@ antlrcpp::Any CypherMainVisitor::visitCallProcedure(MemgraphCypher::CallProcedur
       throw SemanticException("There is no procedure named '{}'.", call_proc->procedure_name_);
     }
   }
-  if (maybe_found) {
-    call_proc->is_write_ = maybe_found->second->info.is_write;
-  }
+  call_proc->is_write_ = maybe_found->second->info.is_write;
 
   auto *yield_ctx = ctx->yieldProcedureResults();
   if (!yield_ctx) {
-    if ((maybe_found && !maybe_found->second->results.empty()) && !call_proc->void_procedure_) {
+    if (!maybe_found->second->results.empty() && !call_proc->void_procedure_) {
       throw SemanticException(
           "CALL without YIELD may only be used on procedures which do not "
           "return any result fields.");

src/query/plan/operator.cpp

@@ -3464,7 +3464,7 @@ class AggregateCursor : public Cursor {
     SCOPED_PROFILE_OP_BY_REF(self_);
 
     if (!pulled_all_input_) {
-      if (!ProcessAll(&frame, &context) && !self_.group_by_.empty()) return false;
+      if (!ProcessAll(&frame, &context) && self_.AreAllAggregationsForCollecting()) return false;
       pulled_all_input_ = true;
       aggregation_it_ = aggregation_.begin();
 
@@ -3824,6 +3824,12 @@ UniqueCursorPtr Aggregate::MakeCursor(utils::MemoryResource *mem) const {
   return MakeUniqueCursorPtr<AggregateCursor>(mem, *this, mem);
 }
 
+auto Aggregate::AreAllAggregationsForCollecting() const -> bool {
+  return std::all_of(aggregations_.begin(), aggregations_.end(), [](const auto &agg) {
+    return agg.op == Aggregation::Op::COLLECT_LIST || agg.op == Aggregation::Op::COLLECT_MAP;
+  });
+}
+
 Skip::Skip(const std::shared_ptr<LogicalOperator> &input, Expression *expression)
     : input_(input), expression_(expression) {}
 

src/query/plan/operator.hpp

@@ -1759,6 +1759,8 @@ class Aggregate : public memgraph::query::plan::LogicalOperator {
   Aggregate(const std::shared_ptr<LogicalOperator> &input, const std::vector<Element> &aggregations,
             const std::vector<Expression *> &group_by, const std::vector<Symbol> &remember);
 
+  auto AreAllAggregationsForCollecting() const -> bool;
+
   bool Accept(HierarchicalLogicalOperatorVisitor &visitor) override;
   UniqueCursorPtr MakeCursor(utils::MemoryResource *) const override;
   std::vector<Symbol> ModifiedSymbols(const SymbolTable &) const override;

src/storage/v2/inmemory/storage.cpp

@@ -835,14 +835,21 @@ void InMemoryStorage::InMemoryAccessor::Abort() {
           switch (current->action) {
             case Delta::Action::REMOVE_LABEL: {
               auto it = std::find(vertex->labels.begin(), vertex->labels.end(), current->label);
-              MG_ASSERT(it != vertex->labels.end(), "Invalid database state!");
+              // TODO(gitbuda): Figure out how to handle this case, assert fails on OOM.
+              if (it == vertex->labels.begin()) {
+                break;
+              }
+              // MG_ASSERT(it != vertex->labels.end(), "Invalid database state!");
               std::swap(*it, *vertex->labels.rbegin());
               vertex->labels.pop_back();
               break;
             }
             case Delta::Action::ADD_LABEL: {
               auto it = std::find(vertex->labels.begin(), vertex->labels.end(), current->label);
-              MG_ASSERT(it == vertex->labels.end(), "Invalid database state!");
+              if (it != vertex->labels.end()) {
+                break;
+              }
+              // MG_ASSERT(it == vertex->labels.end(), "Invalid database state!");
               vertex->labels.push_back(current->label);
               break;
             }
@@ -854,7 +861,10 @@ void InMemoryStorage::InMemoryAccessor::Abort() {
               std::tuple<EdgeTypeId, Vertex *, EdgeRef> link{current->vertex_edge.edge_type,
                                                              current->vertex_edge.vertex, current->vertex_edge.edge};
               auto it = std::find(vertex->in_edges.begin(), vertex->in_edges.end(), link);
-              MG_ASSERT(it == vertex->in_edges.end(), "Invalid database state!");
+              if (it != vertex->in_edges.end()) {
+                break;
+              }
+              // MG_ASSERT(it == vertex->in_edges.end(), "Invalid database state!");
               vertex->in_edges.push_back(link);
               break;
             }
@@ -862,7 +872,10 @@ void InMemoryStorage::InMemoryAccessor::Abort() {
               std::tuple<EdgeTypeId, Vertex *, EdgeRef> link{current->vertex_edge.edge_type,
                                                              current->vertex_edge.vertex, current->vertex_edge.edge};
               auto it = std::find(vertex->out_edges.begin(), vertex->out_edges.end(), link);
-              MG_ASSERT(it == vertex->out_edges.end(), "Invalid database state!");
+              if (it != vertex->out_edges.end()) {
+                break;
+              }
+              // MG_ASSERT(it == vertex->out_edges.end(), "Invalid database state!");
               vertex->out_edges.push_back(link);
               // Increment edge count. We only increment the count here because
               // the information in `ADD_IN_EDGE` and `Edge/RECREATE_OBJECT` is
@@ -875,7 +888,10 @@ void InMemoryStorage::InMemoryAccessor::Abort() {
               std::tuple<EdgeTypeId, Vertex *, EdgeRef> link{current->vertex_edge.edge_type,
                                                              current->vertex_edge.vertex, current->vertex_edge.edge};
               auto it = std::find(vertex->in_edges.begin(), vertex->in_edges.end(), link);
-              MG_ASSERT(it != vertex->in_edges.end(), "Invalid database state!");
+              if (it == vertex->in_edges.end()) {
+                break;
+              }
+              // MG_ASSERT(it != vertex->in_edges.end(), "Invalid database state!");
               std::swap(*it, *vertex->in_edges.rbegin());
               vertex->in_edges.pop_back();
               break;
@@ -884,7 +900,10 @@ void InMemoryStorage::InMemoryAccessor::Abort() {
               std::tuple<EdgeTypeId, Vertex *, EdgeRef> link{current->vertex_edge.edge_type,
                                                              current->vertex_edge.vertex, current->vertex_edge.edge};
               auto it = std::find(vertex->out_edges.begin(), vertex->out_edges.end(), link);
-              MG_ASSERT(it != vertex->out_edges.end(), "Invalid database state!");
+              if (it == vertex->out_edges.end()) {
+                break;
+              }
+              // MG_ASSERT(it != vertex->out_edges.end(), "Invalid database state!");
               std::swap(*it, *vertex->out_edges.rbegin());
               vertex->out_edges.pop_back();
               // Decrement edge count. We only decrement the count here because

src/storage/v2/storage.cpp

@@ -229,9 +229,10 @@ Storage::Accessor::DetachDelete(std::vector<VertexAccessor *> nodes, std::vector
   if (maybe_nodes_to_delete.HasError()) {
     return maybe_nodes_to_delete.GetError();
   }
-  const std::unordered_set<Vertex *> nodes_to_delete = *maybe_nodes_to_delete.GetValue();
+  const auto &nodes_to_delete = *maybe_nodes_to_delete.GetValue();
 
   // 2. Gather edges and corresponding node on the other end of the edge for the deletable nodes
+  // TODO(gitbuda): Somehow nodes_to_delete here is completely broken.
   EdgeInfoForDeletion edge_deletion_info = PrepareDeletableEdges(nodes_to_delete, edges, detach);
 
   // Detach nodes which need to be deleted

tests/e2e/garbage_collection/gc_periodic.py

@@ -49,7 +49,7 @@ def test_gc_periodic(connection):
     memory_pre_creation = get_memory(cursor)
     execute_and_fetch_all(cursor, "UNWIND range(1, 1000) AS index CREATE (:Node);")
     memory_after_creation = get_memory(cursor)
-    time.sleep(5)
+    time.sleep(2)
     memory_after_gc = get_memory(cursor)
 
     assert memory_after_gc < memory_pre_creation + (memory_after_creation - memory_pre_creation) / 4 * 3

tests/gql_behave/tests/memgraph_V1/features/aggregations.feature

@@ -425,4 +425,6 @@ Feature: Aggregations
       """
       MATCH (subnet:Subnet) WHERE FALSE WITH subnet, count(subnet.ip) as ips RETURN id(subnet) as id
       """
-      Then the result should be empty
+      Then the result should be:
+        | id |
+        | null  |

tests/gql_behave/tests/memgraph_V1_on_disk/features/aggregations.feature

@@ -425,4 +425,6 @@ Feature: Aggregations
       """
       MATCH (subnet:Subnet) WHERE FALSE WITH subnet, count(subnet.ip) as ips RETURN id(subnet) as id
       """
-      Then the result should be empty
+      Then the result should be:
+        | id |
+        | null  |

tests/stress/detach_delete.py

@@ -4,7 +4,7 @@
 # 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
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Busi ness 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

tests/stress/high_delta_transaction_abort.py

@@ -0,0 +1,236 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# 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 Busi ness 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.
+
+"""
+Large bipartite graph stress test.
+"""
+
+import atexit
+import logging
+import multiprocessing
+import time
+from argparse import Namespace as Args
+from dataclasses import dataclass
+from functools import wraps
+from typing import Any, Callable
+
+from common import (
+    OutputData,
+    SessionCache,
+    connection_argument_parser,
+    execute_till_success,
+)
+
+log = logging.getLogger(__name__)
+output_data = OutputData()
+
+DELETE_FUNCTION = "DELETE"
+MATCH_FUNCTION = "MATCH"
+SHOW_STORAGE_FUNCTION = "SHOW_STORAGE"
+
+
+atexit.register(SessionCache.cleanup)
+
+
+def parse_args() -> Args:
+    """
+    Parses user arguments
+
+    :return: parsed arguments
+    """
+    parser = connection_argument_parser()
+    parser.add_argument("--worker-count", type=int, default=4, help="Number of concurrent workers.")
+    parser.add_argument(
+        "--logging", default="INFO", choices=["INFO", "DEBUG", "WARNING", "ERROR"], help="Logging level"
+    )
+    parser.add_argument("--edge-size", type=int, default=2000000, help="Number of edges to create.")
+    parser.add_argument("--isolation-level", type=str, required=True, help="Database isolation level.")
+    parser.add_argument("--storage-mode", type=str, required=True, help="Database storage mode.")
+
+    return parser.parse_args()
+
+
+args = parse_args()
+
+
+@dataclass
+class Worker:
+    """
+    Class that performs a function defined in the `type` argument
+
+    Args:
+    type - either `CREATE` or `DELETE`, signifying the function that's going to be performed
+        by the worker
+    id - worker id
+    """
+
+    type: str
+    repetition_count: int
+    sleep_sec: int
+
+
+def timed_function(name) -> Callable:
+    """
+    Times performed function
+    """
+
+    def actual_decorator(func) -> Callable:
+        @wraps(func)
+        def timed_wrapper(*args, **kwargs) -> Any:
+            start_time = time.time()
+            result = func(*args, **kwargs)
+            end_time = time.time()
+            output_data.add_measurement(name, end_time - start_time)
+            return result
+
+        return timed_wrapper
+
+    return actual_decorator
+
+
+@timed_function("cleanup_time")
+def clean_database() -> None:
+    session = SessionCache.argument_session(args)
+    execute_till_success(session, "MATCH (n) DETACH DELETE n")
+
+
+def create_indices() -> None:
+    session = SessionCache.argument_session(args)
+    execute_till_success(session, "CREATE INDEX ON :Node")
+    execute_till_success(session, "CREATE INDEX ON :Node(prop)")
+    execute_till_success(session, "CREATE INDEX ON :Supernode")
+    execute_till_success(session, "CREATE INDEX ON :Supernode(prop)")
+
+
+def setup_database_mode() -> None:
+    session = SessionCache.argument_session(args)
+    execute_till_success(session, f"STORAGE MODE {args.storage_mode}")
+    execute_till_success(session, f"SET GLOBAL TRANSACTION ISOLATION LEVEL {args.isolation_level}")
+
+
+def execute_function(worker: Worker) -> Worker:
+    """
+    Executes the function based on the worker type
+    """
+    if worker.type == SHOW_STORAGE_FUNCTION:
+        run_show_storage(worker.repetition_count, worker.sleep_sec)
+        return worker
+
+    if worker.type == DELETE_FUNCTION:
+        run_deleter()
+        return worker
+
+    if worker.type == MATCH_FUNCTION:
+        run_matcher(worker.repetition_count, worker.sleep_sec)
+        return worker
+
+    raise Exception("Worker function not recognized, raising exception!")
+
+
+def create_data() -> None:
+    edge_size = args.edge_size
+    session = SessionCache.argument_session(args)
+    print("Creating supernode")
+    execute_till_success(session, "CREATE (n:Supernode {prop: 1});", 1)
+    for i in range(20):
+        print(f"Creating dataset {i}/20")
+        execute_till_success(
+            session, f"FOREACH (i in range(1, {int(edge_size / 20)}) | CREATE (n:Node {{prop: {i}}}));", 1
+        )
+        execute_till_success(session, f"MATCH (s:Supernode), (n:Node {{prop: {i}}}) CREATE (s)<-[:REL]-(n);", 1)
+
+
+def run_deleter() -> None:
+    """
+    Periodic deletion of an arbitrary subgraph in the graph
+    """
+    session = SessionCache.argument_session(args)
+    try:
+        print("Starting deleter")
+        execute_till_success(session, "MATCH (n)-[r]->(m) DELETE r RETURN count(r)", 1)
+        print("Deleter ended")
+    except Exception as e:
+        print("Deleter failed as it should with message: ")
+        print(e)
+
+
+def run_matcher(rep_count: int, sleep_sec: int) -> None:
+    """
+    Matching edges and returning the count.
+    """
+    session = SessionCache.argument_session(args)
+    print("Starting matcher")
+    try:
+        for i in range(rep_count):
+            execute_till_success(session, "MATCH (s:Supernode) RETURN COUNT(s)")
+            print(f"{i+1}. Executed matcher")
+            time.sleep(sleep_sec)
+    except Exception as e:
+        print("Matcher failed")
+        print(e)
+        raise
+
+    print("Matcher ended")
+
+
+def run_show_storage(rep_count: int, sleep_sec: int) -> None:
+    """
+    Running show storage info periodically to see if it always returns.
+    """
+    session = SessionCache.argument_session(args)
+    print("Starting show storage")
+    try:
+        for i in range(rep_count):
+            execute_till_success(session, "SHOW STORAGE INFO")
+            print(f"{i+1}. Executed show storage info")
+            time.sleep(sleep_sec)
+    except Exception as e:
+        print("Show storage failed")
+        print(e)
+        raise
+
+    print("Show storage ended")
+
+
+@timed_function("total_execution_time")
+def execution_handler() -> None:
+    clean_database()
+    log.info("Database is clean.")
+
+    setup_database_mode()
+    create_indices()
+    create_data()
+
+    sleep_sec = 0.5
+    rep_count = 1000
+
+    workers = [
+        Worker(DELETE_FUNCTION, rep_count, sleep_sec),
+        Worker(MATCH_FUNCTION, rep_count, sleep_sec),
+        Worker(SHOW_STORAGE_FUNCTION, rep_count, sleep_sec),
+    ]
+
+    with multiprocessing.Pool(processes=args.worker_count) as p:
+        for worker in p.map(execute_function, workers):
+            print(f"Worker {worker.type} finished!")
+
+    run_deleter()
+    run_matcher()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=args.logging)
+    execution_handler()
+    if args.logging in ["DEBUG", "INFO"]:
+        output_data.dump()

tests/stress/match_create_delete.py

@@ -0,0 +1,261 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# 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 Busi ness 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.
+
+"""
+Large bipartite graph stress test.
+"""
+
+import atexit
+import logging
+import multiprocessing
+import random
+import time
+from argparse import Namespace as Args
+from dataclasses import dataclass
+from functools import wraps
+from typing import Any, Callable, Tuple
+
+from common import (
+    OutputData,
+    SessionCache,
+    connection_argument_parser,
+    execute_till_success,
+)
+
+log = logging.getLogger(__name__)
+output_data = OutputData()
+
+NUMBER_NODES_IN_CHAIN = 4
+CREATE_FUNCTION = "CREATE"
+DELETE_FUNCTION = "DELETE"
+MATCH_FUNCTION = "MATCH"
+
+
+atexit.register(SessionCache.cleanup)
+
+
+def parse_args() -> Args:
+    """
+    Parses user arguments
+
+    :return: parsed arguments
+    """
+    parser = connection_argument_parser()
+    parser.add_argument("--worker-count", type=int, default=4, help="Number of concurrent workers.")
+    parser.add_argument(
+        "--logging", default="INFO", choices=["INFO", "DEBUG", "WARNING", "ERROR"], help="Logging level"
+    )
+    parser.add_argument("--repetition-count", type=int, default=1000, help="Number of times to perform the action")
+    parser.add_argument("--isolation-level", type=str, required=True, help="Database isolation level.")
+    parser.add_argument("--storage-mode", type=str, required=True, help="Database storage mode.")
+
+    return parser.parse_args()
+
+
+args = parse_args()
+
+
+@dataclass
+class Worker:
+    """
+    Class that performs a function defined in the `type` argument
+
+    Args:
+    type - either `CREATE` or `DELETE`, signifying the function that's going to be performed
+        by the worker
+    id - worker id
+    total_worker_cnt - total number of workers for reference
+    repetition_count - number of times to perform the worker action
+    sleep_sec - float for subsecond sleeping between two subsequent actions
+    """
+
+    type: str
+    id: int
+    total_worker_cnt: int
+    repetition_count: int
+    sleep_sec: float
+
+
+def timed_function(name) -> Callable:
+    """
+    Times performed function
+    """
+
+    def actual_decorator(func) -> Callable:
+        @wraps(func)
+        def timed_wrapper(*args, **kwargs) -> Any:
+            start_time = time.time()
+            result = func(*args, **kwargs)
+            end_time = time.time()
+            output_data.add_measurement(name, end_time - start_time)
+            return result
+
+        return timed_wrapper
+
+    return actual_decorator
+
+
+@timed_function("cleanup_time")
+def clean_database() -> None:
+    session = SessionCache.argument_session(args)
+    execute_till_success(session, "MATCH (n) DETACH DELETE n")
+
+
+def create_indices() -> None:
+    session = SessionCache.argument_session(args)
+    execute_till_success(session, "CREATE INDEX ON :Node")
+    execute_till_success(session, "CREATE INDEX ON :Node(prop)")
+    execute_till_success(session, "CREATE INDEX ON :Supernode")
+    execute_till_success(session, "CREATE INDEX ON :Supernode(prop)")
+
+
+def setup_database_mode() -> None:
+    session = SessionCache.argument_session(args)
+    execute_till_success(session, f"STORAGE MODE {args.storage_mode}")
+    execute_till_success(session, f"SET GLOBAL TRANSACTION ISOLATION LEVEL {args.isolation_level}")
+
+
+def execute_function(worker: Worker) -> Worker:
+    """
+    Executes the function based on the worker type
+    """
+    if worker.type == CREATE_FUNCTION:
+        run_writer(worker.total_worker_cnt, worker.repetition_count, worker.sleep_sec, worker.id)
+        return worker
+
+    if worker.type == DELETE_FUNCTION:
+        run_deleter(worker.total_worker_cnt, worker.repetition_count, worker.sleep_sec)
+        return worker
+
+    if worker.type == MATCH_FUNCTION:
+        run_matcher(worker.total_worker_cnt, worker.repetition_count, worker.sleep_sec)
+        return worker
+
+    raise Exception("Worker function not recognized, raising exception!")
+
+
+def run_writer(total_workers_cnt: int, repetition_count: int, sleep_sec: float, worker_id: int) -> int:
+    """
+    Creating nodes connected to a supernode
+    """
+    session = SessionCache.argument_session(args)
+
+    def create():
+        try:
+            number = random.randint(1, 1000000)
+            execute_till_success(
+                session,
+                f"FOREACH (i in range(1, 200000) | CREATE (n:Node {{prop: {number}}}))",
+            )
+            execute_till_success(
+                session,
+                f"CREATE (:Supernode {{prop: {number}}})",
+            )
+            execute_till_success(
+                session, f"MATCH (n:Node {{prop: {number}}}), (m:Supernode {{prop: {number}}}) CREATE (n)-[:TYPE]->(m)"
+            )
+        except Exception as ex:
+            pass
+
+    curr_repetition = 0
+
+    while curr_repetition < repetition_count:
+        create()
+        time.sleep(sleep_sec)
+        log.info(f"Worker {worker_id} created chain in iteration {curr_repetition}")
+        curr_repetition += 1
+
+
+def run_deleter(total_workers_cnt: int, repetition_count: int, sleep_sec: float) -> None:
+    """
+    Periodic deletion of an arbitrary subgraph in the graph
+    """
+    session = SessionCache.argument_session(args)
+
+    def delete_part_of_graph():
+        try:
+            maybe_prop = execute_till_success(session, "MATCH (n) WITH n.prop AS prop LIMIT 1 RETURN prop")[0]
+            if len(maybe_prop):
+                prop = maybe_prop[0]["prop"]
+                execute_till_success(session, f"MATCH (n {{prop: {prop}}})-[r]->() DELETE r")
+                execute_till_success(session, f"MATCH (n {{prop: {prop}}}) DETACH DELETE n")
+        except Exception as ex:
+            log.info(f"Worker failed to delete the chain with id {id}")
+            pass
+
+    curr_repetition = 0
+    while curr_repetition < repetition_count:
+        delete_part_of_graph()
+        time.sleep(sleep_sec)
+        curr_repetition += 1
+
+
+def run_matcher(total_workers_cnt: int, repetition_count: int, sleep_sec: float) -> None:
+    """
+    Matching edges and returning the count.
+    """
+    session = SessionCache.argument_session(args)
+
+    def delete_part_of_graph(id: int):
+        try:
+            execute_till_success(session, f"MATCH ()-[r]->(m) RETURN COUNT(r)")
+            log.info(f"Matching done {id}")
+            time.sleep(sleep_sec)
+        except Exception as ex:
+            log.info(f"Worker failed to match with id {id}")
+            time.sleep(sleep_sec)
+            pass
+
+    curr_repetition = 0
+    while curr_repetition < repetition_count:
+        random_part_of_graph = random.randint(0, total_workers_cnt - 1)
+        delete_part_of_graph(random_part_of_graph)
+        time.sleep(sleep_sec)
+        curr_repetition += 1
+
+
+@timed_function("total_execution_time")
+def execution_handler() -> None:
+    clean_database()
+    log.info("Database is clean.")
+
+    setup_database_mode()
+
+    create_indices()
+
+    rep_count = args.repetition_count
+    sleep_sec = 0.5
+
+    workers = [
+        Worker(CREATE_FUNCTION, 0, 3, rep_count, sleep_sec),
+        Worker(DELETE_FUNCTION, 1, 3, rep_count, sleep_sec),
+        Worker(MATCH_FUNCTION, 2, 3, rep_count, sleep_sec),
+        Worker(CREATE_FUNCTION, 3, 3, rep_count, sleep_sec),
+        Worker(DELETE_FUNCTION, 4, 3, rep_count, sleep_sec),
+        Worker(MATCH_FUNCTION, 5, 3, rep_count, sleep_sec),
+        Worker(CREATE_FUNCTION, 6, 3, rep_count, sleep_sec),
+        Worker(DELETE_FUNCTION, 7, 3, rep_count, sleep_sec),
+        Worker(MATCH_FUNCTION, 8, 3, rep_count, sleep_sec),
+    ]
+
+    with multiprocessing.Pool(processes=args.worker_count) as p:
+        for worker in p.map(execute_function, workers):
+            print(f"Worker {worker.type} finished!")
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=args.logging)
+    execution_handler()
+    if args.logging in ["DEBUG", "INFO"]:
+        output_data.dump()

tests/stress/test_config.py

@@ -89,6 +89,18 @@ SMALL_DATASET = [
         DatasetConstants.TIMEOUT: 5,
         DatasetConstants.MODE: [get_default_database_mode()],
     },
+    {
+        DatasetConstants.TEST: "match_create_delete.py",
+        DatasetConstants.OPTIONS: [],
+        DatasetConstants.TIMEOUT: 5,
+        DatasetConstants.MODE: [get_default_database_mode()],
+    },
+    {
+        DatasetConstants.TEST: "high_delta_transaction_abort.py",
+        DatasetConstants.OPTIONS: ["--edge-size", "2000000"],
+        DatasetConstants.TIMEOUT: 5,
+        DatasetConstants.MODE: [get_default_database_mode()],
+    },
     {
         DatasetConstants.TEST: "parser.cpp",
         DatasetConstants.OPTIONS: ["--per-worker-query-count", "1000"],
@@ -153,6 +165,18 @@ LARGE_DATASET = (
             DatasetConstants.TIMEOUT: 30,
             DatasetConstants.MODE: [get_default_database_mode()],
         },
+        {
+            DatasetConstants.TEST: "match_create_delete.py",
+            DatasetConstants.OPTIONS: ["--repetition-count", "3000000"],
+            DatasetConstants.TIMEOUT: 30,
+            DatasetConstants.MODE: [get_default_database_mode()],
+        },
+        {
+            DatasetConstants.TEST: "high_delta_transaction_abort.py",
+            DatasetConstants.OPTIONS: ["--edge-size", "2000000"],
+            DatasetConstants.TIMEOUT: 5,
+            DatasetConstants.MODE: [get_default_database_mode()],
+        },
     ]
     + [
         {

tests/unit/cypher_main_visitor.cpp

@@ -2833,6 +2833,18 @@ TEST_P(CypherMainVisitorTest, DumpDatabase) {
   ASSERT_TRUE(query);
 }
 
+namespace {
+template <class TAst>
+void CheckCallProcedureDefaultMemoryLimit(const TAst &ast, const CallProcedure &call_proc) {
+  // Should be 100 MB
+  auto *literal = dynamic_cast<PrimitiveLiteral *>(call_proc.memory_limit_);
+  ASSERT_TRUE(literal);
+  TypedValue value(literal->value_);
+  ASSERT_TRUE(TypedValue::BoolEqual{}(value, TypedValue(100)));
+  ASSERT_EQ(call_proc.memory_scale_, 1024 * 1024);
+}
+}  // namespace
+
 TEST_P(CypherMainVisitorTest, CallProcedureWithDotsInName) {
   AddProc(*mock_module_with_dots_in_name, "proc", {}, {"res"}, ProcedureType::WRITE);
   auto &ast_generator = *GetParam();
@@ -2856,6 +2868,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureWithDotsInName) {
   std::vector<std::string> expected_names{"res"};
   ASSERT_EQ(identifier_names, expected_names);
   ASSERT_EQ(identifier_names, call_proc->result_fields_);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureWithDashesInName) {
@@ -2881,6 +2894,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureWithDashesInName) {
   std::vector<std::string> expected_names{"res"};
   ASSERT_EQ(identifier_names, expected_names);
   ASSERT_EQ(identifier_names, call_proc->result_fields_);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureWithYieldSomeFields) {
@@ -2912,6 +2926,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureWithYieldSomeFields) {
     std::vector<std::string> expected_names{"fst", "field-with-dashes", "last_field"};
     ASSERT_EQ(identifier_names, expected_names);
     ASSERT_EQ(identifier_names, call_proc->result_fields_);
+    CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
   };
   check_proc(ProcedureType::READ);
   check_proc(ProcedureType::WRITE);
@@ -2944,6 +2959,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureWithYieldAliasedFields) {
   ASSERT_EQ(identifier_names, aliased_names);
   std::vector<std::string> field_names{"fst", "snd", "thrd"};
   ASSERT_EQ(call_proc->result_fields_, field_names);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureWithArguments) {
@@ -2970,6 +2986,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureWithArguments) {
   std::vector<std::string> expected_names{"res"};
   ASSERT_EQ(identifier_names, expected_names);
   ASSERT_EQ(identifier_names, call_proc->result_fields_);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureYieldAsterisk) {
@@ -2991,6 +3008,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureYieldAsterisk) {
   }
   ASSERT_THAT(identifier_names, UnorderedElementsAre("name", "signature", "is_write", "path", "is_editable"));
   ASSERT_EQ(identifier_names, call_proc->result_fields_);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureYieldAsteriskReturnAsterisk) {
@@ -3015,6 +3033,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureYieldAsteriskReturnAsterisk) {
   }
   ASSERT_THAT(identifier_names, UnorderedElementsAre("name", "signature", "is_write", "path", "is_editable"));
   ASSERT_EQ(identifier_names, call_proc->result_fields_);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureWithoutYield) {
@@ -3030,6 +3049,7 @@ TEST_P(CypherMainVisitorTest, CallProcedureWithoutYield) {
   ASSERT_TRUE(call_proc->arguments_.empty());
   ASSERT_TRUE(call_proc->result_fields_.empty());
   ASSERT_TRUE(call_proc->result_identifiers_.empty());
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 }
 
 TEST_P(CypherMainVisitorTest, CallProcedureWithMemoryLimitWithoutYield) {
@@ -3163,6 +3183,7 @@ void CheckParsedCallProcedure(const CypherQuery &query, Base &ast_generator,
   EXPECT_EQ(identifier_names, args_as_str);
   EXPECT_EQ(identifier_names, call_proc->result_fields_);
   ASSERT_EQ(call_proc->is_write_, type == ProcedureType::WRITE);
+  CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
 };
 }  // namespace
 
@@ -3556,6 +3577,7 @@ TEST_P(CypherMainVisitorTest, MemoryLimit) {
     auto *single_query = query->single_query_;
     ASSERT_EQ(single_query->clauses_.size(), 2U);
     auto *call_proc = dynamic_cast<CallProcedure *>(single_query->clauses_[0]);
+    CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
   }
 
   {
@@ -3616,6 +3638,7 @@ TEST_P(CypherMainVisitorTest, MemoryLimit) {
     auto *single_query = query->single_query_;
     ASSERT_EQ(single_query->clauses_.size(), 1U);
     auto *call_proc = dynamic_cast<CallProcedure *>(single_query->clauses_[0]);
+    CheckCallProcedureDefaultMemoryLimit(ast_generator, *call_proc);
   }
 }
 

tests/unit/query_plan_accumulate_aggregate.cpp

@@ -250,23 +250,30 @@ TYPED_TEST(QueryPlanAggregateOps, WithData) {
 TYPED_TEST(QueryPlanAggregateOps, WithoutDataWithGroupBy) {
   {
     auto results = this->AggregationResults(true, false, {Aggregation::Op::COUNT});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Int);
+    EXPECT_EQ(results[0][0].ValueInt(), 0);
   }
   {
     auto results = this->AggregationResults(true, false, {Aggregation::Op::SUM});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Int);
+    EXPECT_EQ(results[0][0].ValueInt(), 0);
   }
   {
     auto results = this->AggregationResults(true, false, {Aggregation::Op::AVG});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Null);
   }
   {
     auto results = this->AggregationResults(true, false, {Aggregation::Op::MIN});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Null);
   }
   {
     auto results = this->AggregationResults(true, false, {Aggregation::Op::MAX});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Null);
   }
   {
     auto results = this->AggregationResults(true, false, {Aggregation::Op::COLLECT_LIST});
@@ -659,23 +666,30 @@ TYPED_TEST(QueryPlanAggregateOps, WithDataDistinct) {
 TYPED_TEST(QueryPlanAggregateOps, WithoutDataWithDistinctAndWithGroupBy) {
   {
     auto results = this->AggregationResults(true, true, {Aggregation::Op::COUNT});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Int);
+    EXPECT_EQ(results[0][0].ValueInt(), 0);
   }
   {
     auto results = this->AggregationResults(true, true, {Aggregation::Op::SUM});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Int);
+    EXPECT_EQ(results[0][0].ValueInt(), 0);
   }
   {
     auto results = this->AggregationResults(true, true, {Aggregation::Op::AVG});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Null);
   }
   {
     auto results = this->AggregationResults(true, true, {Aggregation::Op::MIN});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Null);
   }
   {
     auto results = this->AggregationResults(true, true, {Aggregation::Op::MAX});
-    EXPECT_EQ(results.size(), 0);
+    EXPECT_EQ(results.size(), 1);
+    EXPECT_EQ(results[0][0].type(), TypedValue::Type::Null);
   }
   {
     auto results = this->AggregationResults(true, true, {Aggregation::Op::COLLECT_LIST});