Advanced User Controls and Proper Path Validation

ee/clickhouse/queries/paths/paths.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 from re import escape
 from typing import Dict, List, Literal, Optional, Tuple, Union, cast
 
@@ -7,7 +8,7 @@
 from ee.clickhouse.queries.funnels.funnel_persons import ClickhouseFunnelPersons
 from ee.clickhouse.queries.paths.path_event_query import PathEventQuery
 from ee.clickhouse.sql.paths.path import PATH_ARRAY_QUERY
-from posthog.constants import FUNNEL_PATH_BETWEEN_STEPS, LIMIT
+from posthog.constants import FUNNEL_PATH_BETWEEN_STEPS, LIMIT, PATH_EDGE_LIMIT
 from posthog.models import Filter, Team
 from posthog.models.filters.path_filter import PathFilter
 
@@ -38,6 +39,14 @@ def __init__(self, filter: PathFilter, team: Team, funnel_filter: Optional[Filte
         if not self._filter.limit:
             self._filter = self._filter.with_data({LIMIT: 100})
 
+        if self._filter.edge_limit is None and not (self._filter.start_point and self._filter.end_point):
+            # no edge restriction when both start and end points are defined
+            self._filter = self._filter.with_data({PATH_EDGE_LIMIT: 100})
+
+        if self._filter.max_edge_weight and self._filter.min_edge_weight:
+            if self._filter.max_edge_weight < self._filter.min_edge_weight:
+                raise ValidationError("Max Edge weight can't be lower than min edge weight")
+
         if self._filter.path_groupings:
             regex_groupings = []
             for grouping in self._filter.path_groupings:
@@ -47,11 +56,13 @@ def __init__(self, filter: PathFilter, team: Team, funnel_filter: Optional[Filte
                 regex_groupings.append(regex_grouping)
             self.params["regex_groupings"] = regex_groupings
 
-        # TODO: don't allow including $pageview and excluding $pageview at the same time
-
     def run(self, *args, **kwargs):
 
         results = self._exec_query()
+
+        if not self._filter.min_edge_weight and not self._filter.max_edge_weight:
+            results = self.validate_results(results)
+
         return self._format_results(results)
 
     def _format_results(self, results):
@@ -100,6 +111,11 @@ def get_path_query(self) -> str:
 
         paths_per_person_query = self.get_paths_per_person_query()
 
+        self.params["edge_limit"] = self._filter.edge_limit
+
+        edge_weight_filter, edge_weight_params = self.get_edge_weight_clause()
+        self.params.update(edge_weight_params)
+
         return f"""
             SELECT last_path_key as source_event,
                 path_key as target_event,
@@ -109,10 +125,11 @@ def get_path_query(self) -> str:
             WHERE source_event IS NOT NULL
             GROUP BY source_event,
                     target_event
+            {edge_weight_filter}
             ORDER BY event_count DESC,
                     source_event,
                     target_event
-            LIMIT 30
+            {'LIMIT %(edge_limit)s' if self._filter.edge_limit else ''}
         """
 
     def get_path_query_funnel_cte(self, funnel_filter: Filter):
@@ -141,6 +158,24 @@ def get_target_point_filter(self) -> str:
         else:
             return ""
 
+    def get_edge_weight_clause(self) -> Tuple[str, Dict]:
+        params: Dict[str, int] = {}
+
+        conditions = []
+
+        if self._filter.min_edge_weight:
+            params["min_edge_weight"] = self._filter.min_edge_weight
+            conditions.append("event_count >= %(min_edge_weight)s")
+
+        if self._filter.max_edge_weight:
+            params["max_edge_weight"] = self._filter.max_edge_weight
+            conditions.append("event_count <= %(max_edge_weight)s")
+
+        if conditions:
+            return f"HAVING {' AND '.join(conditions)}", params
+
+        return "", params
+
     def get_target_clause(self) -> Tuple[str, Dict]:
         params: Dict[str, Union[str, None]] = {"target_point": None, "secondary_target_point": None}
 
@@ -192,3 +227,33 @@ def get_filtered_path_ordering(self) -> Tuple[str, str]:
                 "arraySlice(filtered_path, 1, %(event_in_session_limit)s)",
                 "arraySlice(filtered_timings, 1, %(event_in_session_limit)s)",
             )
+
+    def validate_results(self, results):
+        # Query guarantees results list to be:
+        # 1. Directed, Acyclic Tree where each node has only 1 child
+        # 2. All start nodes beginning with 1_
+
+        seen = set()  # source nodes that've been traversed
+        edges = defaultdict(list)
+        validated_results = []
+        starting_nodes_stack = []
+
+        for result in results:
+            edges[result[0]].append(result[1])
+            if result[0].startswith("1_"):
+                # All nodes with 1_ are valid starting nodes
+                starting_nodes_stack.append(result[0])
+
+        while starting_nodes_stack:
+            current_node = starting_nodes_stack.pop()
+            seen.add(current_node)
+
+            for node in edges[current_node]:
+                if node not in seen:
+                    starting_nodes_stack.append(node)
+
+        for result in results:
+            if result[0] in seen:
+                validated_results.append(result)
+
+        return validated_results

ee/clickhouse/queries/paths/paths_persons.py

@@ -67,3 +67,7 @@ def _format_results(self, results):
         from posthog.api.person import PersonSerializer
 
         return PersonSerializer(people, many=True).data, len(results) > cast(int, self._filter.limit) - 1
+
+    def run(self, *args, **kwargs):
+        results = self._exec_query()
+        return self._format_results(results)