feat: cleverly query replays by logs (#22018)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

posthog/hogql/database/schema/session_replay_events.py

@@ -1,3 +1,5 @@
+from posthog.hogql.ast import SelectQuery
+from posthog.hogql.context import HogQLContext
 from posthog.hogql.database.models import (
     Table,
     StringDatabaseField,
@@ -9,11 +11,44 @@
     LazyTable,
     FieldOrTable,
 )
+from posthog.hogql.database.schema.log_entries import ReplayConsoleLogsLogEntriesTable
 from posthog.hogql.database.schema.person_distinct_ids import (
     PersonDistinctIdsTable,
     join_with_person_distinct_ids_table,
 )
 
+
+def join_with_console_logs_log_entries_table(
+    from_table: str,
+    to_table: str,
+    requested_fields: dict[str, list[str | int]],
+    context: HogQLContext,
+    node: SelectQuery,
+):
+    from posthog.hogql import ast
+
+    if "log_source_id" not in requested_fields:
+        requested_fields = {**requested_fields, "log_source_id": ["log_source_id"]}
+
+    select_query = SelectQuery(
+        select=[ast.Field(chain=chain) for chain in requested_fields.values()],
+        select_from=ast.JoinExpr(table=ast.Field(chain=["console_logs_log_entries"])),
+    )
+
+    join_expr = ast.JoinExpr(table=select_query)
+    join_expr.join_type = "LEFT JOIN"
+    join_expr.alias = to_table
+    join_expr.constraint = ast.JoinConstraint(
+        expr=ast.CompareOperation(
+            op=ast.CompareOperationOp.Eq,
+            left=ast.Field(chain=[from_table, "session_id"]),
+            right=ast.Field(chain=[to_table, "log_source_id"]),
+        )
+    )
+
+    return join_expr
+
+
 RAW_ONLY_FIELDS = ["min_first_timestamp", "max_last_timestamp"]
 
 SESSION_REPLAY_EVENTS_COMMON_FIELDS: dict[str, FieldOrTable] = {
@@ -38,6 +73,11 @@
         join_table=PersonDistinctIdsTable(),
         join_function=join_with_person_distinct_ids_table,
     ),
+    "console_logs": LazyJoin(
+        from_field=["session_id"],
+        join_table=ReplayConsoleLogsLogEntriesTable(),
+        join_function=join_with_console_logs_log_entries_table,
+    ),
     "person": FieldTraverser(chain=["pdi", "person"]),
     "person_id": FieldTraverser(chain=["pdi", "person_id"]),
 }

posthog/hogql/database/schema/test/test_session_replay_events.py

@@ -0,0 +1,74 @@
+from posthog.hogql import ast
+from posthog.hogql.parser import parse_select
+from posthog.hogql.query import execute_hogql_query
+from posthog.session_recordings.queries.test.session_replay_sql import produce_replay_summary
+from posthog.test.base import (
+    APIBaseTest,
+    ClickhouseTestMixin,
+)
+
+
+class TestFilterSessionReplaysByConsoleLogs(ClickhouseTestMixin, APIBaseTest):
+    def setUp(self):
+        super().setUp()
+
+        produce_replay_summary(
+            team_id=self.team.pk,
+            distinct_id="d1",
+            session_id="session_with_info_and_error_messages",
+            log_messages={
+                "info": ["This is an info message"],
+                "error": ["This is a generic message"],
+            },
+        )
+
+        produce_replay_summary(
+            team_id=self.team.pk,
+            distinct_id="d1",
+            session_id="session_with_only_info_messages",
+            log_messages={
+                "info": ["This is a generic message"],
+            },
+        )
+
+        produce_replay_summary(
+            team_id=self.team.pk, distinct_id="d1", session_id="session_with_no_log_messages", log_messages=None
+        )
+
+    def test_select_by_console_log_text(self):
+        response = execute_hogql_query(
+            parse_select(
+                "select distinct session_id from raw_session_replay_events where console_logs.message = {log_message} order by session_id asc",
+                placeholders={"log_message": ast.Constant(value="This is a generic message")},
+            ),
+            self.team,
+        )
+
+        assert response.results == [("session_with_info_and_error_messages",), ("session_with_only_info_messages",)]
+
+    def test_select_by_console_log_text_and_level(self):
+        response = execute_hogql_query(
+            parse_select(
+                "select distinct session_id from raw_session_replay_events where console_logs.message = {log_message} and console_logs.level = {log_level} order by session_id asc",
+                placeholders={
+                    "log_message": ast.Constant(value="This is a generic message"),
+                    "log_level": ast.Constant(value="error"),
+                },
+            ),
+            self.team,
+        )
+
+        assert response.results == [("session_with_info_and_error_messages",)]
+
+    def test_select_log_text(self):
+        response = execute_hogql_query(
+            parse_select(
+                "select distinct console_logs.message from raw_session_replay_events where console_logs.level = {log_level} order by session_id asc",
+                placeholders={
+                    "log_level": ast.Constant(value="info"),
+                },
+            ),
+            self.team,
+        )
+
+        assert response.results == [("This is an info message",), ("This is a generic message",)]

posthog/hogql/database/test/__snapshots__/test_database.ambr

@@ -439,6 +439,20 @@
                   "person"
               ]
           },
+          {
+              "key": "console_logs",
+              "type": "lazy_table",
+              "table": "console_logs_log_entries",
+              "fields": [
+                  "team_id",
+                  "log_source",
+                  "log_source_id",
+                  "instance_id",
+                  "timestamp",
+                  "message",
+                  "level"
+              ]
+          },
           {
               "key": "person",
               "type": "field_traverser",
@@ -804,6 +818,20 @@
                   "person"
               ]
           },
+          {
+              "key": "console_logs",
+              "type": "lazy_table",
+              "table": "console_logs_log_entries",
+              "fields": [
+                  "team_id",
+                  "log_source",
+                  "log_source_id",
+                  "instance_id",
+                  "timestamp",
+                  "message",
+                  "level"
+              ]
+          },
           {
               "key": "person",
               "type": "field_traverser",
@@ -1482,6 +1510,20 @@
                   "person"
               ]
           },
+          {
+              "key": "console_logs",
+              "type": "lazy_table",
+              "table": "console_logs_log_entries",
+              "fields": [
+                  "team_id",
+                  "log_source",
+                  "log_source_id",
+                  "instance_id",
+                  "timestamp",
+                  "message",
+                  "level"
+              ]
+          },
           {
               "key": "person",
               "type": "field_traverser",
@@ -1847,6 +1889,20 @@
                   "person"
               ]
           },
+          {
+              "key": "console_logs",
+              "type": "lazy_table",
+              "table": "console_logs_log_entries",
+              "fields": [
+                  "team_id",
+                  "log_source",
+                  "log_source_id",
+                  "instance_id",
+                  "timestamp",
+                  "message",
+                  "level"
+              ]
+          },
           {
               "key": "person",
               "type": "field_traverser",