chore(recordings): Optimize recordings list query

posthog/api/session_recording.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any, List, cast
+from typing import Any, List, Type, cast
 
 import structlog
 from dateutil import parser
@@ -22,7 +22,7 @@
 from posthog.models.session_recording_event import SessionRecordingViewed
 from posthog.models.team.team import Team
 from posthog.permissions import ProjectMembershipNecessaryPermissions, TeamMemberAccessPermission
-from posthog.queries.session_recordings.session_recording_list import SessionRecordingList
+from posthog.queries.session_recordings.session_recording_list import SessionRecordingList, SessionRecordingListV2
 from posthog.queries.session_recordings.session_recording_properties import SessionRecordingProperties
 from posthog.rate_limit import ClickHouseBurstRateThrottle, ClickHouseSustainedRateThrottle
 from posthog.utils import format_query_params_absolute_url
@@ -238,7 +238,12 @@ def list_recordings(filter: SessionRecordingsFilter, request: request.Request, t
 
     if (all_session_ids and filter.session_ids) or not all_session_ids:
         # Only go to clickhouse if we still have remaining specified IDs or we are not specifying IDs
-        (ch_session_recordings, more_recordings_available) = SessionRecordingList(filter=filter, team=team).run()
+        session_recording_list_instance: Type[SessionRecordingList] = (
+            SessionRecordingListV2 if team.recordings_list_v2_query_enabled else SessionRecordingList
+        )
+        (ch_session_recordings, more_recordings_available) = session_recording_list_instance(
+            filter=filter, team=team
+        ).run()
         recordings_from_clickhouse = SessionRecording.get_or_build_from_clickhouse(team, ch_session_recordings)
         recordings = recordings + recordings_from_clickhouse
 

posthog/api/test/test_session_recordings.py

@@ -142,22 +142,23 @@ def test_get_session_recordings(self):
     @snapshot_postgres_queries
     def test_listing_recordings_is_not_nplus1_for_persons(self):
         # request once without counting queries to cache an ee.license lookup that makes results vary otherwise
-        self.client.get(f"/api/projects/{self.team.id}/session_recordings")
+        with freeze_time("2022-06-03T12:00:00.000Z"):
+            self.client.get(f"/api/projects/{self.team.id}/session_recordings")
 
-        base_time = (now() - relativedelta(days=1)).replace(microsecond=0)
-        num_queries = 10
+            base_time = (now() - relativedelta(days=1)).replace(microsecond=0)
+            num_queries = 9
 
-        self._person_with_snapshots(base_time=base_time, distinct_id="user", session_id="1")
-        with self.assertNumQueries(num_queries):
-            self.client.get(f"/api/projects/{self.team.id}/session_recordings")
+            self._person_with_snapshots(base_time=base_time, distinct_id="user", session_id="1")
+            with self.assertNumQueries(num_queries):
+                self.client.get(f"/api/projects/{self.team.id}/session_recordings")
 
-        self._person_with_snapshots(base_time=base_time, distinct_id="user2", session_id="2")
-        with self.assertNumQueries(num_queries):
-            self.client.get(f"/api/projects/{self.team.id}/session_recordings")
+            self._person_with_snapshots(base_time=base_time, distinct_id="user2", session_id="2")
+            with self.assertNumQueries(num_queries):
+                self.client.get(f"/api/projects/{self.team.id}/session_recordings")
 
-        self._person_with_snapshots(base_time=base_time, distinct_id="user3", session_id="3")
-        with self.assertNumQueries(num_queries):
-            self.client.get(f"/api/projects/{self.team.id}/session_recordings")
+            self._person_with_snapshots(base_time=base_time, distinct_id="user3", session_id="3")
+            with self.assertNumQueries(num_queries):
+                self.client.get(f"/api/projects/{self.team.id}/session_recordings")
 
     def _person_with_snapshots(self, base_time: datetime, distinct_id: str = "user", session_id: str = "1") -> None:
         Person.objects.create(

posthog/models/team/team.py

@@ -216,6 +216,26 @@ def _person_on_events_querying_enabled(self) -> bool:
         # on self-hosted, use the instance setting
         return get_instance_setting("PERSON_ON_EVENTS_ENABLED")
 
+    @property
+    def recordings_list_v2_query_enabled(self) -> bool:
+        if is_cloud():
+            return posthoganalytics.feature_enabled(
+                "recordings-list-v2-enabled",
+                str(self.uuid),
+                groups={"organization": str(self.organization_id)},
+                group_properties={
+                    "organization": {
+                        "id": str(self.organization_id),
+                        "created_at": self.organization.created_at,
+                        "name": self.organization.name,
+                    }
+                },
+                only_evaluate_locally=True,
+                send_feature_flag_events=False,
+            )
+
+        return False
+
     @property
     def strict_caching_enabled(self) -> bool:
         enabled_teams = get_list(get_instance_setting("STRICT_CACHING_TEAMS"))

posthog/queries/funnels/test/__snapshots__/test_funnel.ambr

@@ -684,8 +684,8 @@
                     HAVING argMax(is_deleted, version) = 0) AS pdi ON e.distinct_id = pdi.distinct_id
                  WHERE team_id = 2
                    AND event IN ['$autocapture', 'user signed up', '$autocapture']
-                   AND toTimeZone(timestamp, 'UTC') >= toDateTime('2023-02-20 00:00:00', 'UTC')
-                   AND toTimeZone(timestamp, 'UTC') <= toDateTime('2023-02-27 23:59:59', 'UTC')
+                   AND toTimeZone(timestamp, 'UTC') >= toDateTime('2023-02-21 00:00:00', 'UTC')
+                   AND toTimeZone(timestamp, 'UTC') <= toDateTime('2023-02-28 23:59:59', 'UTC')
                    AND (step_0 = 1
                         OR step_1 = 1) ))
            WHERE step_0 = 1 ))

posthog/queries/session_recordings/session_recording_list.py

@@ -5,6 +5,7 @@
 from posthog.constants import TREND_FILTER_TYPE_ACTIONS
 from posthog.models import Entity
 from posthog.models.action.util import format_entity_filter
+from posthog.models.filters.mixins.utils import cached_property
 from posthog.models.filters.session_recordings_filter import SessionRecordingsFilter
 from posthog.models.property.util import parse_prop_grouped_clauses
 from posthog.models.utils import PersonPropertiesMode
@@ -13,6 +14,8 @@
 
 
 class EventFiltersSQL(NamedTuple):
+    non_aggregate_select_condition_clause: str
+    aggregate_event_select_clause: str
     aggregate_select_clause: str
     aggregate_having_clause: str
     where_conditions: str
@@ -159,6 +162,7 @@ def _determine_should_join_persons(self) -> None:
     def _determine_should_join_events(self):
         return self._filter.entities and len(self._filter.entities) > 0
 
+    @cached_property
     def _get_properties_select_clause(self) -> str:
         clause = (
             f", events.elements_chain as elements_chain"
@@ -170,6 +174,7 @@ def _get_properties_select_clause(self) -> str:
         )
         return clause
 
+    @cached_property
     def _get_person_id_clause(self) -> Tuple[str, Dict[str, Any]]:
         person_id_clause = ""
         person_id_params = {}
@@ -180,6 +185,7 @@ def _get_person_id_clause(self) -> Tuple[str, Dict[str, Any]]:
 
     # We want to select events beyond the range of the recording to handle the case where
     # a recording spans the time boundaries
+    @cached_property
     def _get_events_timestamp_clause(self) -> Tuple[str, Dict[str, Any]]:
         timestamp_clause = ""
         timestamp_params = {}
@@ -191,6 +197,7 @@ def _get_events_timestamp_clause(self) -> Tuple[str, Dict[str, Any]]:
             timestamp_params["event_end_time"] = self._filter.date_to + timedelta(hours=12)
         return timestamp_clause, timestamp_params
 
+    @cached_property
     def _get_recording_start_time_clause(self) -> Tuple[str, Dict[str, Any]]:
         start_time_clause = ""
         start_time_params = {}
@@ -202,12 +209,14 @@ def _get_recording_start_time_clause(self) -> Tuple[str, Dict[str, Any]]:
             start_time_params["end_time"] = self._filter.date_to
         return start_time_clause, start_time_params
 
+    @cached_property
     def session_ids_clause(self) -> Tuple[str, Dict[str, Any]]:
         if self._filter.session_ids is None:
             return "", {}
 
         return "AND session_id in %(session_ids)s", {"session_ids": self._filter.session_ids}
 
+    @cached_property
     def _get_duration_clause(self) -> Tuple[str, Dict[str, Any]]:
         duration_clause = ""
         duration_params = {}
@@ -260,7 +269,10 @@ def format_event_filter(self, entity: Entity, prepend: str, team_id: int) -> Tup
 
         return filter_sql, params
 
+    @cached_property
     def format_event_filters(self) -> EventFiltersSQL:
+        non_aggregate_select_condition_clause = ""
+        aggregate_event_select_clause = ""
         aggregate_select_clause = ""
         aggregate_having_clause = ""
         where_conditions = "AND event IN %(event_names)s"
@@ -289,7 +301,14 @@ def format_event_filters(self) -> EventFiltersSQL:
 
         params = {**params, "event_names": list(event_names_to_filter)}
 
-        return EventFiltersSQL(aggregate_select_clause, aggregate_having_clause, where_conditions, params)
+        return EventFiltersSQL(
+            non_aggregate_select_condition_clause,
+            aggregate_event_select_clause,
+            aggregate_select_clause,
+            aggregate_having_clause,
+            where_conditions,
+            params,
+        )
 
     def get_query(self) -> Tuple[str, Dict[str, Any]]:
         offset = self._filter.offset or 0
@@ -300,12 +319,11 @@ def get_query(self) -> Tuple[str, Dict[str, Any]]:
             self._filter.property_groups, person_id_joined_alias=f"{self.DISTINCT_ID_TABLE_ALIAS}.person_id"
         )
 
-        events_timestamp_clause, events_timestamp_params = self._get_events_timestamp_clause()
-        recording_start_time_clause, recording_start_time_params = self._get_recording_start_time_clause()
-        session_ids_clause, session_ids_params = self.session_ids_clause()
-        person_id_clause, person_id_params = self._get_person_id_clause()
-        duration_clause, duration_params = self._get_duration_clause()
-        properties_select_clause = self._get_properties_select_clause()
+        events_timestamp_clause, events_timestamp_params = self._get_events_timestamp_clause
+        recording_start_time_clause, recording_start_time_params = self._get_recording_start_time_clause
+        session_ids_clause, session_ids_params = self.session_ids_clause
+        person_id_clause, person_id_params = self._get_person_id_clause
+        duration_clause, duration_params = self._get_duration_clause
 
         core_recordings_query = self._core_session_recordings_query.format(
             recording_start_time_clause=recording_start_time_clause,
@@ -334,13 +352,9 @@ def get_query(self) -> Tuple[str, Dict[str, Any]]:
                 },
             )
 
-        event_filters = self.format_event_filters()
+        event_filters = self.format_event_filters
 
-        core_events_query = self._core_events_query.format(
-            properties_select_clause=properties_select_clause,
-            event_filter_where_conditions=event_filters.where_conditions,
-            events_timestamp_clause=events_timestamp_clause,
-        )
+        core_events_query, core_events_query_params = self._get_core_events_query()
 
         return (
             self._session_recordings_query_with_events.format(
@@ -363,9 +377,24 @@ def get_query(self) -> Tuple[str, Dict[str, Any]]:
                 **recording_start_time_params,
                 **event_filters.params,
                 **session_ids_params,
+                **core_events_query_params,
             },
         )
 
+    def _get_core_events_query(self) -> Tuple[str, Dict[str, Any]]:
+        params: Dict[str, Any] = {}
+        event_filters = self.format_event_filters
+        properties_select_clause = self._get_properties_select_clause
+        events_timestamp_clause, events_timestamp_params = self._get_events_timestamp_clause
+
+        core_events_query = self._core_events_query.format(
+            properties_select_clause=properties_select_clause,
+            event_filter_where_conditions=event_filters.where_conditions,
+            events_timestamp_clause=events_timestamp_clause,
+        )
+
+        return core_events_query, {**params, **events_timestamp_params}
+
     def _paginate_results(self, session_recordings) -> SessionRecordingQueryResult:
         more_recordings_available = False
         if len(session_recordings) > self.limit:
@@ -406,3 +435,143 @@ def run(self, *args, **kwargs) -> SessionRecordingQueryResult:
         query_results = sync_execute(query, {**query_params, **self._filter.hogql_context.values})
         session_recordings = self._data_to_return(query_results)
         return self._paginate_results(session_recordings)
+
+
+class SessionRecordingListV2(SessionRecordingList):
+
+    _core_events_query = """
+        SELECT
+            uuid,
+            distinct_id,
+            event,
+            team_id,
+            timestamp,
+            $session_id as session_id,
+            $window_id as window_id
+            {properties_select_clause}
+            {non_aggregate_select_condition_clause}
+
+        FROM events
+        WHERE
+            team_id = %(team_id)s
+            {event_filter_where_conditions}
+            {events_timestamp_clause}
+            AND notEmpty(session_id)
+    """
+
+    _core_events_query_grouped = """
+        SELECT
+            session_id,
+            distinct_id
+            {aggregate_event_select_clause}
+        FROM (
+            {ungrouped_core_events_query}
+        ) GROUP BY session_id, distinct_id
+        HAVING 1 = 1
+        {event_filter_aggregate_having_clause}
+    """
+
+    _session_recordings_query_with_events = """
+        SELECT
+            session_recordings.session_id,
+            start_time,
+            end_time,
+            click_count,
+            keypress_count,
+            urls,
+            duration,
+            session_recordings.distinct_id as distinct_id
+            {event_filter_aggregate_select_clause}
+        FROM (
+            {core_events_query}
+        ) AS events
+        JOIN (
+            {core_recordings_query}
+        ) AS session_recordings
+        ON session_recordings.session_id = events.session_id
+        {recording_person_query}
+        WHERE
+            session_recordings.distinct_id == events.distinct_id
+            {prop_filter_clause}
+            {person_id_clause}
+        ORDER BY start_time DESC
+        LIMIT %(limit)s OFFSET %(offset)s
+        """
+
+    def _get_core_events_query(self) -> Tuple[str, Dict[str, Any]]:
+        params: Dict[str, Any] = {}
+        event_filters = self.format_event_filters
+        properties_select_clause = self._get_properties_select_clause
+        events_timestamp_clause, events_timestamp_params = self._get_events_timestamp_clause
+
+        core_events_query = self._core_events_query.format(
+            properties_select_clause=properties_select_clause,
+            non_aggregate_select_condition_clause=event_filters.non_aggregate_select_condition_clause,
+            event_filter_where_conditions=event_filters.where_conditions,
+            events_timestamp_clause=events_timestamp_clause,
+        )
+
+        grouped_events_query = self._core_events_query_grouped.format(
+            aggregate_event_select_clause=event_filters.aggregate_event_select_clause,
+            ungrouped_core_events_query=core_events_query,
+            event_filter_aggregate_having_clause=event_filters.aggregate_having_clause,
+        )
+
+        return grouped_events_query, {**params, **events_timestamp_params}
+
+    @cached_property
+    def format_event_filters(self) -> EventFiltersSQL:
+        non_aggregate_select_condition_clause = ""
+        aggregate_event_select_clause = ""
+        aggregate_select_clause = ""
+        aggregate_having_clause = ""
+        where_conditions = "AND event IN %(event_names)s"
+        # Always include $pageview events so the start_url and end_url can be extracted
+        event_names_to_filter: List[Union[int, str]] = []
+
+        params: Dict = {}
+
+        for index, entity in enumerate(self._filter.entities):
+            if entity.type == TREND_FILTER_TYPE_ACTIONS:
+                action = entity.get_action()
+                event_names_to_filter.extend([ae for ae in action.get_step_events() if ae not in event_names_to_filter])
+            else:
+                if entity.id not in event_names_to_filter:
+                    event_names_to_filter.append(entity.id)
+
+            condition_sql, filter_params = self.format_event_filter(
+                entity, prepend=f"event_matcher_{index}", team_id=self._team_id
+            )
+            aggregate_event_select_clause += f"""
+            , groupUniqArrayIf(100)((timestamp, uuid, session_id, window_id), event_match_{index} = 1) AS matching_events_{index}
+            , sum(event_match_{index}) AS matches_{index}
+            """
+
+            aggregate_select_clause += f"""
+            , matches_{index}
+            , matching_events_{index}
+            """
+
+            non_aggregate_select_condition_clause += f"""
+            , if({condition_sql}, 1, 0) as event_match_{index}
+            """
+            aggregate_having_clause += f"\nAND matches_{index} > 0"
+            params = {**params, **filter_params}
+
+        params = {**params, "event_names": list(event_names_to_filter)}
+
+        return EventFiltersSQL(
+            non_aggregate_select_condition_clause,
+            aggregate_event_select_clause,
+            aggregate_select_clause,
+            aggregate_having_clause,
+            where_conditions,
+            params,
+        )
+
+    def run(self, *args, **kwargs) -> SessionRecordingQueryResult:
+        self._filter.hogql_context.using_person_on_events = True
+        query, query_params = self.get_query()
+        query_results = sync_execute(query, {**query_params, **self._filter.hogql_context.values})
+        session_recordings = self._data_to_return(query_results)
+        return self._paginate_results(session_recordings)