Add show storage info testing

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/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()],
+        },
     ]
     + [
         {