Add support for parallel runtime

doc/source/transactions.rst

@@ -51,7 +51,7 @@ Explicit Transactions
 Neomodel also supports  `explicit transactions <https://neo4j.com/docs/
 api/python-driver/current/transactions.html>`_ that are pre-designated as either *read* or *write*. 
 
-This is vital when using neomodel over a `Neo4J causal cluster <https://neo4j.com/docs/
+This is vital when using neomodel over a `Neo4j causal cluster <https://neo4j.com/docs/
 operations-manual/current/clustering/>`_ because internally, queries will be rerouted to different 
 servers depending on their designation. 
 
@@ -168,7 +168,7 @@ Impersonation
 
 *Neo4j Enterprise feature*
 
-Impersonation (`see Neo4j driver documentation <https://neo4j.com/docs/api/python-driver/current/api.html#impersonated-user-ref>``)
+Impersonation (`see Neo4j driver documentation <https://neo4j.com/docs/api/python-driver/current/api.html#impersonated-user-ref>`_)
 can be enabled via a context manager::
 
     from neomodel import db
@@ -197,4 +197,22 @@ This can be mixed with other context manager like transactions::
 
         @db.transaction()
         def func2():
-            ...
+            ...
+
+
+Parallel runtime
+----------------
+
+As of version 5.13, Neo4j *Enterprise Edition* supports parallel runtime for read transactions.
+
+To use it, you can simply use the `parallel_read_transaction` context manager::
+
+    from neomodel import db
+
+    with db.parallel_read_transaction:
+        # It works for both neomodel-generated and custom Cypher queries
+        parallel_count_1 = len(Coffee.nodes)
+        parallel_count_2 = db.cypher_query("MATCH (n:Coffee) RETURN count(n)")
+
+It is worth noting that the parallel runtime is only available for read transactions and that it is not enabled by default, because it is not always the fastest option. It is recommended to test it in your specific use case to see if it improves performance, and read the general considerations in the `Neo4j official documentation <https://neo4j.com/docs/cypher-manual/current/planning-and-tuning/runtimes/concepts/#runtimes-parallel-runtime-considerations>`_.
+

neomodel/async_/core.py

@@ -104,6 +104,7 @@ def __init__(self):
         self._database_version = None
         self._database_edition = None
         self.impersonated_user = None
+        self._parallel_runtime = False
 
     async def set_connection(self, url: str = None, driver: AsyncDriver = None):
         """
@@ -239,6 +240,10 @@ def write_transaction(self):
     def read_transaction(self):
         return AsyncTransactionProxy(self, access_mode="READ")
 
+    @property
+    def parallel_read_transaction(self):
+        return AsyncTransactionProxy(self, access_mode="READ", parallel_runtime=True)
+
     async def impersonate(self, user: str) -> "ImpersonationHandler":
         """All queries executed within this context manager will be executed as impersonated user
 
@@ -454,7 +459,6 @@ async def cypher_query(
 
         :return: A tuple containing a list of results and a tuple of headers.
         """
-
         if self._active_transaction:
             # Use current session is a transaction is currently active
             results, meta = await self._run_cypher_query(
@@ -493,6 +497,8 @@ async def _run_cypher_query(
         try:
             # Retrieve the data
             start = time.time()
+            if self._parallel_runtime:
+                query = "CYPHER runtime=parallel " + query
             response: AsyncResult = await session.run(query, params)
             results, meta = [list(r.values()) async for r in response], response.keys()
             end = time.time()
@@ -598,6 +604,18 @@ async def edition_is_enterprise(self) -> bool:
         edition = await self.database_edition
         return edition == "enterprise"
 
+    @ensure_connection
+    async def parallel_runtime_available(self) -> bool:
+        """Returns true if the database supports parallel runtime
+
+        Returns:
+            bool: True if the database supports parallel runtime
+        """
+        return (
+            await self.version_is_higher_than("5.13")
+            and await self.edition_is_enterprise()
+        )
+
     async def change_neo4j_password(self, user, new_password):
         await self.cypher_query(f"ALTER USER {user} SET PASSWORD '{new_password}'")
 
@@ -1168,17 +1186,29 @@ async def install_all_labels(stdout=None):
 class AsyncTransactionProxy:
     bookmarks: Optional[Bookmarks] = None
 
-    def __init__(self, db: AsyncDatabase, access_mode=None):
+    def __init__(
+        self, db: AsyncDatabase, access_mode: str = None, parallel_runtime: bool = False
+    ):
         self.db = db
         self.access_mode = access_mode
+        self.parallel_runtime = parallel_runtime
 
     @ensure_connection
     async def __aenter__(self):
+        if self.parallel_runtime and not await self.db.parallel_runtime_available():
+            warnings.warn(
+                "Parallel runtime is only available in Neo4j Enterprise Edition 5.13 and above. "
+                "Reverting to default runtime.",
+                UserWarning,
+            )
+            self.parallel_runtime = False
+        self.db._parallel_runtime = self.parallel_runtime
         await self.db.begin(access_mode=self.access_mode, bookmarks=self.bookmarks)
         self.bookmarks = None
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
+        self.db._parallel_runtime = False
         if exc_value:
             await self.db.rollback()
 

neomodel/async_/match.py

@@ -1,6 +1,7 @@
 import inspect
 import re
 import string
+import warnings
 from dataclasses import dataclass
 from typing import Any, Dict, List
 from typing import Optional as TOptional
@@ -589,9 +590,11 @@ def build_traversal_from_path(
                 }
             else:
                 existing_rhs_name = subgraph[part][
-                    "rel_variable_name"
-                    if relation.get("relation_filtering")
-                    else "variable_name"
+                    (
+                        "rel_variable_name"
+                        if relation.get("relation_filtering")
+                        else "variable_name"
+                    )
                 ]
             if relation["include_in_return"] and not already_present:
                 self._additional_return(rel_ident)
@@ -973,7 +976,9 @@ async def _execute(self, lazy: bool = False, dict_output: bool = False):
                 ]
         query = self.build_query()
         results, prop_names = await adb.cypher_query(
-            query, self._query_params, resolve_objects=True
+            query,
+            self._query_params,
+            resolve_objects=True,
         )
         if dict_output:
             for item in results:

neomodel/sync_/core.py

@@ -104,6 +104,7 @@ def __init__(self):
         self._database_version = None
         self._database_edition = None
         self.impersonated_user = None
+        self._parallel_runtime = False
 
     def set_connection(self, url: str = None, driver: Driver = None):
         """
@@ -239,6 +240,10 @@ def write_transaction(self):
     def read_transaction(self):
         return TransactionProxy(self, access_mode="READ")
 
+    @property
+    def parallel_read_transaction(self):
+        return TransactionProxy(self, access_mode="READ", parallel_runtime=True)
+
     def impersonate(self, user: str) -> "ImpersonationHandler":
         """All queries executed within this context manager will be executed as impersonated user
 
@@ -452,7 +457,6 @@ def cypher_query(
 
         :return: A tuple containing a list of results and a tuple of headers.
         """
-
         if self._active_transaction:
             # Use current session is a transaction is currently active
             results, meta = self._run_cypher_query(
@@ -491,6 +495,8 @@ def _run_cypher_query(
         try:
             # Retrieve the data
             start = time.time()
+            if self._parallel_runtime:
+                query = "CYPHER runtime=parallel " + query
             response: Result = session.run(query, params)
             results, meta = [list(r.values()) for r in response], response.keys()
             end = time.time()
@@ -596,6 +602,15 @@ def edition_is_enterprise(self) -> bool:
         edition = self.database_edition
         return edition == "enterprise"
 
+    @ensure_connection
+    def parallel_runtime_available(self) -> bool:
+        """Returns true if the database supports parallel runtime
+
+        Returns:
+            bool: True if the database supports parallel runtime
+        """
+        return self.version_is_higher_than("5.13") and self.edition_is_enterprise()
+
     def change_neo4j_password(self, user, new_password):
         self.cypher_query(f"ALTER USER {user} SET PASSWORD '{new_password}'")
 
@@ -1162,17 +1177,29 @@ def install_all_labels(stdout=None):
 class TransactionProxy:
     bookmarks: Optional[Bookmarks] = None
 
-    def __init__(self, db: Database, access_mode=None):
+    def __init__(
+        self, db: Database, access_mode: str = None, parallel_runtime: bool = False
+    ):
         self.db = db
         self.access_mode = access_mode
+        self.parallel_runtime = parallel_runtime
 
     @ensure_connection
     def __enter__(self):
+        if self.parallel_runtime and not self.db.parallel_runtime_available():
+            warnings.warn(
+                "Parallel runtime is only available in Neo4j Enterprise Edition 5.13 and above. "
+                "Reverting to default runtime.",
+                UserWarning,
+            )
+            self.parallel_runtime = False
+        self.db._parallel_runtime = self.parallel_runtime
         self.db.begin(access_mode=self.access_mode, bookmarks=self.bookmarks)
         self.bookmarks = None
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
+        self.db._parallel_runtime = False
         if exc_value:
             self.db.rollback()
 

neomodel/sync_/match.py

@@ -1,6 +1,7 @@
 import inspect
 import re
 import string
+import warnings
 from dataclasses import dataclass
 from typing import Any, Dict, List
 from typing import Optional as TOptional
@@ -973,7 +974,9 @@ def _execute(self, lazy: bool = False, dict_output: bool = False):
                 ]
         query = self.build_query()
         results, prop_names = db.cypher_query(
-            query, self._query_params, resolve_objects=True
+            query,
+            self._query_params,
+            resolve_objects=True,
         )
         if dict_output:
             for item in results:

pyproject.toml

@@ -38,6 +38,7 @@ dev = [
     "pytest>=7.1",
     "pytest-asyncio",
     "pytest-cov>=4.0",
+    "pytest-mock",
     "pre-commit",
     "black",
     "isort",

requirements-dev.txt

@@ -5,6 +5,7 @@ unasync>=0.5.0
 pytest>=7.1
 pytest-asyncio>=0.19.0
 pytest-cov>=4.0
+pytest-mock
 pre-commit
 black
 isort

test/async_/test_match_api.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 from test._async_compat import mark_async_test
 
-from pytest import raises
+from pytest import raises, skip, warns
 
 from neomodel import (
     INCOMING,
@@ -31,7 +31,11 @@
     RawCypher,
     RelationNameResolver,
 )
-from neomodel.exceptions import MultipleNodesReturned, RelationshipClassNotDefined
+from neomodel.exceptions import (
+    FeatureNotSupported,
+    MultipleNodesReturned,
+    RelationshipClassNotDefined,
+)
 
 
 class SupplierRel(AsyncStructuredRel):
@@ -1113,3 +1117,58 @@ async def test_async_iterator():
 
         # assert that generator runs loop above
         assert counter == n
+
+
+def assert_last_query_startswith(mock_func, query) -> bool:
+    return mock_func.call_args_list[-1].args[0].startswith(query)
+
+
+@mark_async_test
+async def test_parallel_runtime(mocker):
+    if (
+        not await adb.version_is_higher_than("5.13")
+        or not await adb.edition_is_enterprise()
+    ):
+        skip("Only supported for Enterprise 5.13 and above.")
+
+    assert await adb.parallel_runtime_available()
+
+    # Parallel should be applied to custom Cypher query
+    async with adb.parallel_read_transaction:
+        # Mock transaction.run to access executed query
+        # Assert query starts with CYPHER runtime=parallel
+        assert adb._parallel_runtime == True
+        mock_transaction_run = mocker.patch("neo4j.AsyncTransaction.run")
+        await adb.cypher_query("MATCH (n:Coffee) RETURN n")
+        assert assert_last_query_startswith(
+            mock_transaction_run, "CYPHER runtime=parallel"
+        )
+    # Test exiting the context sets the parallel_runtime to False
+    assert adb._parallel_runtime == False
+
+    # Parallel should be applied to neomodel queries
+    async with adb.parallel_read_transaction:
+        mock_transaction_run_2 = mocker.patch("neo4j.AsyncTransaction.run")
+        await Coffee.nodes.all()
+        assert assert_last_query_startswith(
+            mock_transaction_run_2, "CYPHER runtime=parallel"
+        )
+
+
+@mark_async_test
+async def test_parallel_runtime_conflict(mocker):
+    if await adb.version_is_higher_than("5.13") and await adb.edition_is_enterprise():
+        skip("Test for unavailable parallel runtime.")
+
+    assert not await adb.parallel_runtime_available()
+    mock_transaction_run = mocker.patch("neo4j.AsyncTransaction.run")
+    with warns(
+        UserWarning,
+        match="Parallel runtime is only available in Neo4j Enterprise Edition 5.13",
+    ):
+        async with adb.parallel_read_transaction:
+            await Coffee.nodes.all()
+            assert not adb._parallel_runtime
+            assert not assert_last_query_startswith(
+                mock_transaction_run, "CYPHER runtime=parallel"
+            )