test: assert the client parses execution plans, and fix a plan parser bug

falkordb/execution_plan.py

@@ -307,11 +307,15 @@ def create_operation(args):
                 self.operations[child.name].append(child)
 
                 if current:
-                    current = stack.pop()
-                    current.append_child(child)
+                    # attach the sibling to the parent and keep the parent on
+                    # the stack, so any further sibling is attached to it too
+                    parent = stack.pop()
+                    parent.append_child(child)
+                    stack.append(parent)
+                else:
+                    stack.append(child)
                 current = child
                 i += 1
-                stack.append(child)
             elif op_level == level + 1:
                 # if the operation is child of the current operation
                 # add it as child and set as current operation

tests/plan_helpers.py

@@ -0,0 +1,82 @@
+"""Engine-agnostic assertions for execution plans.
+
+Which operations a query compiles into is the engine's business and it changes
+between engine versions. The client's job is to issue GRAPH.EXPLAIN /
+GRAPH.PROFILE and turn the reply into an operation tree, so that is what these
+helpers check: every line of the raw reply became exactly one operation, nested
+exactly as deep as that line was indented.
+"""
+
+from typing import Iterator, Optional, Tuple
+
+from falkordb.execution_plan import ExecutionPlan, Operation
+
+INDENT = "    "
+
+
+def iter_operations(op: Operation, depth: int = 0) -> Iterator[Tuple[int, Operation]]:
+    """Yields (depth, operation) for the whole tree, depth first."""
+
+    yield depth, op
+    for child in op.children:
+        yield from iter_operations(child, depth + 1)
+
+
+def assert_parsed_plan(
+    plan: ExecutionPlan,
+    min_operations: int = 1,
+    expect_args: bool = False,
+) -> None:
+    """Asserts the client parsed an execution plan reply correctly."""
+
+    root = plan.structured_plan
+    assert isinstance(root, Operation)
+
+    parsed = list(iter_operations(root))
+    lines = [line for line in plan.plan if line.strip()]
+
+    # every line of the raw reply became exactly one operation
+    assert len(parsed) == len(lines)
+    assert len(parsed) >= min_operations
+
+    for (depth, op), line in zip(parsed, lines):
+        # the operation sits as deep in the tree as its line was indented,
+        # which is what makes this a test of the parser rather than the engine
+        assert depth == (len(line) - len(line.lstrip())) // len(INDENT)
+
+        # indentation and the argument separator were stripped off the name
+        assert isinstance(op.name, str)
+        assert op.name == op.name.strip()
+        assert op.name != ""
+        assert "|" not in op.name
+        assert op.name == line.split("|")[0].strip()
+
+        assert op.args is None or isinstance(op.args, str)
+        assert isinstance(op.children, list)
+
+    if expect_args:
+        assert any(op.args for _, op in parsed)
+
+
+def assert_parsed_profile(
+    plan: ExecutionPlan,
+    min_operations: int = 1,
+    expect_args: bool = False,
+    records_produced: Optional[int] = None,
+) -> None:
+    """Asserts the client parsed a profile reply, statistics included."""
+
+    assert_parsed_plan(plan, min_operations, expect_args)
+
+    parsed = list(iter_operations(plan.structured_plan))
+    for _, op in parsed:
+        assert op.profile_stats is not None
+        assert isinstance(op.records_produced, int)
+        assert op.records_produced >= 0
+        assert isinstance(op.execution_time, float)
+        assert op.execution_time >= 0
+
+    if records_produced is not None:
+        # how many rows the query yields is a property of the query, not of the
+        # engine, so the client must report it whichever engine answered
+        assert max(op.records_produced for _, op in parsed) == records_produced

tests/test_async_explain.py

@@ -3,6 +3,8 @@
 
 from falkordb.asyncio import FalkorDB
 
+from .plan_helpers import assert_parsed_plan
+
 
 @pytest.mark.asyncio
 async def test_explain():
@@ -17,17 +19,9 @@ async def test_explain():
 
     plan = await g.explain("UNWIND range(0, 3) AS x RETURN x")
 
-    results_op = plan.structured_plan
-    assert results_op.name == "Results"
-    assert len(results_op.children) == 1
-
-    project_op = results_op.children[0]
-    assert project_op.name == "Project"
-    assert len(project_op.children) == 1
-
-    unwind_op = project_op.children[0]
-    assert unwind_op.name == "Unwind"
-    assert len(unwind_op.children) == 0
+    # which operations the query compiles into is up to the engine, the client
+    # is responsible for parsing whatever plan comes back
+    assert_parsed_plan(plan, min_operations=2)
 
     # close the connection pool
     await pool.aclose()
@@ -42,26 +36,7 @@ async def test_cartesian_product_explain():
     g = db.select_graph("async_explain")
     plan = await g.explain("MATCH (a), (b) RETURN *")
 
-    results_op = plan.structured_plan
-    assert results_op.name == "Results"
-    assert len(results_op.children) == 1
-
-    project_op = results_op.children[0]
-    assert project_op.name == "Project"
-    assert len(project_op.children) == 1
-
-    cp_op = project_op.children[0]
-    assert cp_op.name == "Cartesian Product"
-    assert len(cp_op.children) == 2
-
-    scan_a_op = cp_op.children[0]
-    scan_b_op = cp_op.children[1]
-
-    assert scan_a_op.name == "All Node Scan"
-    assert len(scan_a_op.children) == 0
-
-    assert scan_b_op.name == "All Node Scan"
-    assert len(scan_b_op.children) == 0
+    assert_parsed_plan(plan, min_operations=4, expect_args=True)
 
     # close the connection pool
     await pool.aclose()
@@ -81,37 +56,7 @@ async def test_merge():
         pass
     plan = await g.explain("MERGE (p1:person {age: 40}) MERGE (p2:person {age: 41})")
 
-    root = plan.structured_plan
-    assert root.name == "Merge"
-    assert len(root.children) == 3
-
-    merge_op = root.children[0]
-    assert merge_op.name == "Merge"
-    assert len(merge_op.children) == 2
-
-    index_scan_op = merge_op.children[0]
-    assert index_scan_op.name == "Node By Index Scan"
-    assert len(index_scan_op.children) == 0
-
-    merge_create_op = merge_op.children[1]
-    assert merge_create_op.name == "MergeCreate"
-    assert len(merge_create_op.children) == 0
-
-    index_scan_op = root.children[1]
-    assert index_scan_op.name == "Node By Index Scan"
-    assert len(index_scan_op.children) == 1
-
-    arg_op = index_scan_op.children[0]
-    assert arg_op.name == "Argument"
-    assert len(arg_op.children) == 0
-
-    merge_create_op = root.children[2]
-    assert merge_create_op.name == "MergeCreate"
-    assert len(merge_create_op.children) == 1
-
-    arg_op = merge_create_op.children[0]
-    assert arg_op.name == "Argument"
-    assert len(arg_op.children) == 0
+    assert_parsed_plan(plan, min_operations=4, expect_args=True)
 
     # close the connection pool
     await pool.aclose()

tests/test_async_graph.py

@@ -3,9 +3,11 @@
 from redis import ResponseError
 from redis.asyncio import BlockingConnectionPool
 
-from falkordb import Edge, Node, Operation, Path
+from falkordb import Edge, Node, Path
 from falkordb.asyncio import FalkorDB
 
+from .plan_helpers import assert_parsed_plan
+
 
 def quote_param_ref(key: str) -> str:
     """Mirror of the sync helper: render a Cypher parameter reference for
@@ -407,7 +409,7 @@ async def test_slowlog():
 
     await g.delete()
 
-    long_query = "UNWIND range (0, 200000) AS x RETURN max(x)"
+    long_query = "UNWIND range (0, 1000000) AS x RETURN max(x)"
     await g.query(long_query)
 
     results = await g.slowlog()
@@ -521,13 +523,10 @@ async def test_execution_plan():
         {"name": "Yehuda"},
     )
 
-    expected = (
-        "Results\n    Project\n        "
-        "Conditional Traverse | (t)->(r:Rider)\n"
-        "            Filter\n"
-        "                Node By Label Scan | (t:Team)"
-    )
-    assert str(result) == expected
+    # which operations the query compiles into is up to the engine, the client
+    # is responsible for parsing whatever plan comes back
+    assert_parsed_plan(result, min_operations=4, expect_args=True)
+    assert str(result)
 
     # close the connection pool
     await pool.aclose()
@@ -563,68 +562,10 @@ async def test_explain():
            RETURN r.name, t.name""",
         {"name": "Yamaha"},
     )
-    expected = """\
-Results
-Distinct
-    Join
-        Project
-            Conditional Traverse | (t)->(r:Rider)
-                Filter
-                    Node By Label Scan | (t:Team)
-        Project
-            Conditional Traverse | (t)->(r:Rider)
-                Filter
-                    Node By Label Scan | (t:Team)"""
-    assert str(result).replace(" ", "").replace("\n", "") == expected.replace(
-        " ", ""
-    ).replace("\n", "")
-
-    expected = Operation("Results").append_child(
-        Operation("Distinct").append_child(
-            Operation("Join")
-            .append_child(
-                Operation("Project").append_child(
-                    Operation("Conditional Traverse", "(t)->(r:Rider)").append_child(
-                        Operation("Filter").append_child(
-                            Operation("Node By Label Scan", "(t:Team)")
-                        )
-                    )
-                )
-            )
-            .append_child(
-                Operation("Project").append_child(
-                    Operation("Conditional Traverse", "(t)->(r:Rider)").append_child(
-                        Operation("Filter").append_child(
-                            Operation("Node By Label Scan", "(t:Team)")
-                        )
-                    )
-                )
-            )
-        )
-    )
-
-    assert result.structured_plan == expected
+    assert_parsed_plan(result, min_operations=7, expect_args=True)
 
     result = await g.explain("MATCH (r:Rider), (t:Team) RETURN r.name, t.name")
-    expected = """\
-Results
-Project
-    Cartesian Product
-        Node By Label Scan | (r:Rider)
-        Node By Label Scan | (t:Team)"""
-    assert str(result).replace(" ", "").replace("\n", "") == expected.replace(
-        " ", ""
-    ).replace("\n", "")
-
-    expected = Operation("Results").append_child(
-        Operation("Project").append_child(
-            Operation("Cartesian Product")
-            .append_child(Operation("Node By Label Scan"))
-            .append_child(Operation("Node By Label Scan"))
-        )
-    )
-
-    assert result.structured_plan == expected
+    assert_parsed_plan(result, min_operations=4, expect_args=True)
 
     # close the connection pool
     await pool.aclose()

tests/test_async_profile.py

@@ -3,6 +3,8 @@
 
 from falkordb.asyncio import FalkorDB
 
+from .plan_helpers import assert_parsed_profile
+
 
 @pytest.mark.asyncio
 async def test_profile():
@@ -14,20 +16,9 @@ async def test_profile():
 
     plan = await g.profile("UNWIND range(0, 3) AS x RETURN x")
 
-    results_op = plan.structured_plan
-    assert results_op.name == "Results"
-    assert len(results_op.children) == 1
-    assert results_op.profile_stats.records_produced == 4
-
-    project_op = results_op.children[0]
-    assert project_op.name == "Project"
-    assert len(project_op.children) == 1
-    assert project_op.profile_stats.records_produced == 4
-
-    unwind_op = project_op.children[0]
-    assert unwind_op.name == "Unwind"
-    assert len(unwind_op.children) == 0
-    assert unwind_op.profile_stats.records_produced == 4
+    # which operations the query compiles into is up to the engine, the client
+    # is responsible for parsing the plan and its statistics
+    assert_parsed_profile(plan, min_operations=2, records_produced=4)
 
     # close the connection pool
     await pool.aclose()
@@ -43,31 +34,7 @@ async def test_cartesian_product_profile():
 
     plan = await g.profile("MATCH (a), (b) RETURN *")
 
-    results_op = plan.structured_plan
-    assert results_op.name == "Results"
-    assert len(results_op.children) == 1
-    assert results_op.profile_stats.records_produced == 0
-
-    project_op = results_op.children[0]
-    assert project_op.name == "Project"
-    assert len(project_op.children) == 1
-    assert project_op.profile_stats.records_produced == 0
-
-    cp_op = project_op.children[0]
-    assert cp_op.name == "Cartesian Product"
-    assert len(cp_op.children) == 2
-    assert cp_op.profile_stats.records_produced == 0
-
-    scan_a_op = cp_op.children[0]
-    scan_b_op = cp_op.children[1]
-
-    assert scan_a_op.name == "All Node Scan"
-    assert len(scan_a_op.children) == 0
-    assert scan_a_op.profile_stats.records_produced == 0
-
-    assert scan_b_op.name == "All Node Scan"
-    assert len(scan_b_op.children) == 0
-    assert scan_b_op.profile_stats.records_produced == 0
+    assert_parsed_profile(plan, min_operations=4, expect_args=True, records_produced=0)
 
     # close the connection pool
     await pool.aclose()