profiled execution plan (#152)

* initial work on profiled execution plan

* address review

Author: AviAvni

redisgraph/execution_plan.py

@@ -1,18 +1,33 @@
+import re
+
+
+class ProfileStats:
+    """
+    ProfileStats, runtime execution statistics of operation.
+    """
+
+    def __init__(self, records_produced, execution_time):
+        self.records_produced = records_produced
+        self.execution_time = execution_time
+
+
 class Operation:
     """
     Operation, single operation within execution plan.
     """
 
-    def __init__(self, name, args=None):
+    def __init__(self, name, args=None, profile_stats=None):
         """
         Create a new operation.
 
         Args:
             name: string that represents the name of the operation
             args: operation arguments
+            profile_stats: profile statistics
         """
         self.name = name
         self.args = args
+        self.profile_stats = profile_stats
         self.children = []
 
     def append_child(self, child):
@@ -32,7 +47,7 @@ def __eq__(self, o: object) -> bool:
         return (self.name == o.name and self.args == o.args)
 
     def __str__(self) -> str:
-        args_str = "" if self.args is None else f" | {self.args}"
+        args_str = "" if self.args is None else " | " + self.args
         return f"{self.name}{args_str}"
 
 
@@ -131,21 +146,30 @@ def _operation_tree(self):
         stack = []
         current = None
 
+        def _create_operation(args):
+            profile_stats = None
+            name = args[0].strip()
+            args.pop(0)
+            if len(args) > 0 and "Records produced" in args[-1]:
+                records_produced = int(re.search("Records produced: (\\d+)", args[-1]).group(1))
+                execution_time = float(re.search("Execution time: (\\d+.\\d+) ms", args[-1]).group(1))
+                profile_stats = ProfileStats(records_produced, execution_time)
+                args.pop(-1)
+            return Operation(name, None if len(args) == 0 else args[0].strip(), profile_stats)
+
         # iterate plan operations
         while i < len(self.plan):
             current_op = self.plan[i]
             op_level = current_op.count("    ")
             if op_level == level:
                 # if the operation level equal to the current level
                 # set the current operation and move next
-                args = current_op.split("|")
-                current = Operation(args[0].strip(), None if len(args) == 1 else args[1].strip())
+                current = _create_operation(current_op.split("|"))
                 i += 1
             elif op_level == level + 1:
                 # if the operation is child of the current operation
                 # add it as child and set as current operation
-                args = current_op.split("|")
-                child = Operation(args[0].strip(), None if len(args) == 1 else args[1].strip())
+                child = _create_operation(current_op.split("|"))
                 current.append_child(child)
                 stack.append(current)
                 current = child

redisgraph/graph.py

@@ -219,7 +219,7 @@ def query(self, q, params=None, timeout=None, read_only=False):
     def execution_plan(self, query, params=None):
         """
         Get the execution plan for given query,
-        GRAPH.EXPLAIN returns an array of operations.
+        GRAPH.EXPLAIN returns ExecutionPlan object.
 
         Args:
             query: the query that will be executed
@@ -231,6 +231,21 @@ def execution_plan(self, query, params=None):
         plan = self.redis_con.execute_command("GRAPH.EXPLAIN", self.name, query)
         return ExecutionPlan(plan)
 
+    def profile(self, query, params=None):
+        """
+        Get the profield execution plan for given query,
+        GRAPH.PROFILE returns ExecutionPlan object.
+
+        Args:
+            query: the query that will be executed
+            params: query parameters
+        """
+        if params is not None:
+            query = self._build_params_header(params) + query
+
+        plan = self.redis_con.execute_command("GRAPH.PROFILE", self.name, query)
+        return ExecutionPlan(plan)
+
     def delete(self):
         """
         Deletes graph.

tests/functional/test_all.py

@@ -246,6 +246,7 @@ def test_cached_execution(self):
 
     def test_execution_plan(self):
         redis_graph = Graph('execution_plan', self.r)
+        # graph creation / population
         create_query = """CREATE
                           (:Rider {name:'Valentino Rossi'})-[:rides]->(:Team {name:'Yamaha'}),
                           (:Rider {name:'Dani Pedrosa'})-[:rides]->(:Team {name:'Honda'}),
@@ -254,11 +255,11 @@ def test_execution_plan(self):
 
         result = redis_graph.execution_plan("""MATCH (r:Rider)-[:rides]->(t:Team)
                                                WHERE t.name = $name
-                                               RETURN r.name, t.name, $params
+                                               RETURN r.name, t.name
                                                UNION
                                                MATCH (r:Rider)-[:rides]->(t:Team)
                                                WHERE t.name = $name
-                                               RETURN r.name, t.name, $params""", {'name': 'Yehuda'})
+                                               RETURN r.name, t.name""", {'name': 'Yamaha'})
         expected = '''\
 Results
     Distinct
@@ -290,6 +291,59 @@ def test_execution_plan(self):
 
         redis_graph.delete()
 
+    def test_profile(self):
+        redis_graph = Graph('profile', self.r)
+        # graph creation / population
+        create_query = """UNWIND range(1, 30) as x CREATE (:Person {id: x})"""
+        redis_graph.query(create_query)
+
+        plan = redis_graph.profile("""MATCH (p:Person)
+                                      WHERE p.id > 15
+                                      RETURN p""")
+
+        results = plan.structured_plan
+        self.assertEqual(results.name, "Results")
+        self.assertEqual(results.profile_stats.records_produced, 15)
+        self.assertGreater(results.profile_stats.execution_time, 0)
+
+        project = results.children[0]
+        self.assertEqual(project.name, "Project")
+        self.assertEqual(project.profile_stats.records_produced, 15)
+        self.assertGreater(project.profile_stats.execution_time, 0)
+
+        filter = project.children[0]
+        self.assertEqual(filter.name, "Filter")
+        self.assertEqual(filter.profile_stats.records_produced, 15)
+        self.assertGreater(filter.profile_stats.execution_time, 0)
+
+        node_by_label_scan = filter.children[0]
+        self.assertEqual(node_by_label_scan.name, "Node By Label Scan")
+        self.assertEqual(node_by_label_scan.profile_stats.records_produced, 30)
+        self.assertGreater(node_by_label_scan.profile_stats.execution_time, 0)
+
+        redis_graph.query("CREATE INDEX FOR (p:Person) ON (p.id)")
+
+        plan = redis_graph.profile("""MATCH (p:Person)
+                                      WHERE p.id > 15
+                                      RETURN p""")
+
+        results = plan.structured_plan
+        self.assertEqual(results.name, "Results")
+        self.assertEqual(results.profile_stats.records_produced, 15)
+        self.assertGreater(results.profile_stats.execution_time, 0)
+
+        project = results.children[0]
+        self.assertEqual(project.name, "Project")
+        self.assertEqual(project.profile_stats.records_produced, 15)
+        self.assertGreater(project.profile_stats.execution_time, 0)
+
+        node_by_index_scan = project.children[0]
+        self.assertEqual(node_by_index_scan.name, "Node By Index Scan")
+        self.assertEqual(node_by_index_scan.profile_stats.records_produced, 15)
+        self.assertGreater(node_by_index_scan.profile_stats.execution_time, 0)
+
+        redis_graph.delete()
+
     def test_query_timeout(self):
         redis_graph = Graph('timeout', self.r)
         # Build a sample graph with 1000 nodes.