feat: add a session recording metadata table

ee/clickhouse/migrations/0030_session_recording_metadata.py

@@ -0,0 +1,17 @@
+from infi.clickhouse_orm import migrations
+
+from posthog.models.session_recordings.sql import (
+    DISTRIBUTED_SESSION_RECORDINGS_TABLE_SQL,
+    KAFKA_SESSION_RECORDINGS_TABLE_SQL,
+    SESSION_RECORDINGS_TABLE_MV_SQL,
+    SESSION_RECORDINGS_TABLE_SQL,
+    WRITABLE_SESSION_RECORDINGS_TABLE_SQL,
+)
+
+operations = [
+    migrations.RunSQL(SESSION_RECORDINGS_TABLE_SQL()),
+    migrations.RunSQL(KAFKA_SESSION_RECORDINGS_TABLE_SQL()),
+    migrations.RunSQL(SESSION_RECORDINGS_TABLE_MV_SQL()),
+    migrations.RunSQL(WRITABLE_SESSION_RECORDINGS_TABLE_SQL()),
+    migrations.RunSQL(DISTRIBUTED_SESSION_RECORDINGS_TABLE_SQL()),
+]

plugin-server/src/config/kafka-topics.ts

@@ -11,6 +11,7 @@ export const KAFKA_PERSON = `${prefix}clickhouse_person${suffix}`
 export const KAFKA_PERSON_UNIQUE_ID = `${prefix}clickhouse_person_unique_id${suffix}`
 export const KAFKA_PERSON_DISTINCT_ID = `${prefix}clickhouse_person_distinct_id${suffix}`
 export const KAFKA_SESSION_RECORDING_EVENTS = `${prefix}clickhouse_session_recording_events${suffix}`
+export const KAFKA_SESSION_RECORDING_METADATA = `${prefix}clickhouse_session_recording_metadata${suffix}`
 export const KAFKA_EVENTS_PLUGIN_INGESTION = `${prefix}events_plugin_ingestion${suffix}`
 export const KAFKA_PLUGIN_LOG_ENTRIES = `${prefix}plugin_log_entries${suffix}`
 export const KAFKA_EVENTS_DEAD_LETTER_QUEUE = `${prefix}events_dead_letter_queue${suffix}`

posthog/kafka_client/topics.py

@@ -8,6 +8,7 @@
 KAFKA_PERSON_UNIQUE_ID = f"{KAFKA_PREFIX}clickhouse_person_unique_id{SUFFIX}"
 KAFKA_PERSON_DISTINCT_ID = f"{KAFKA_PREFIX}clickhouse_person_distinct_id{SUFFIX}"
 KAFKA_SESSION_RECORDING_EVENTS = f"{KAFKA_PREFIX}clickhouse_session_recording_events{SUFFIX}"
+KAFKA_SESSION_RECORDINGS = f"{KAFKA_PREFIX}clickhouse_session_recordings{SUFFIX}"
 KAFKA_PLUGIN_LOG_ENTRIES = f"{KAFKA_PREFIX}plugin_log_entries{SUFFIX}"
 KAFKA_DEAD_LETTER_QUEUE = f"{KAFKA_PREFIX}events_dead_letter_queue{SUFFIX}"
 KAFKA_GROUPS = f"{KAFKA_PREFIX}clickhouse_groups{SUFFIX}"

posthog/models/session_recordings/sql.py

@@ -0,0 +1,81 @@
+from django.conf import settings
+
+from posthog.kafka_client.topics import KAFKA_SESSION_RECORDINGS
+from posthog.clickhouse.kafka_engine import KAFKA_COLUMNS, kafka_engine
+from posthog.clickhouse.table_engines import Distributed, ReplacingMergeTree, ReplicationScheme
+
+SESSION_RECORDINGS_DATA_TABLE = "sharded_session_recordings"
+
+SESSION_RECORDINGS_TABLE_BASE_SQL = """
+CREATE TABLE IF NOT EXISTS {table_name} ON CLUSTER '{cluster}'
+(
+    team_id Int64,
+    distinct_id VARCHAR,
+    session_id VARCHAR,
+    session_start DateTime64(6, 'UTC'),
+    session_end DateTime64(6, 'UTC'),
+    duration Int64,
+    metadata VARCHAR,
+    snapshot_data_location VARCHAR
+    {extra_fields}
+) ENGINE = {engine}
+"""
+
+SESSION_RECORDINGS_DATA_TABLE_ENGINE = lambda: ReplacingMergeTree(
+    "session_recordings", ver="_timestamp", replication_scheme=ReplicationScheme.SHARDED
+)
+
+SESSION_RECORDINGS_TABLE_SQL = lambda: (
+    SESSION_RECORDINGS_TABLE_BASE_SQL
+    + """PARTITION BY toYYYYMMDD(session_start)
+ORDER BY (team_id, toStartOfHour(session_start), session_id)
+"""
+).format(
+    table_name=SESSION_RECORDINGS_DATA_TABLE,
+    cluster=settings.CLICKHOUSE_CLUSTER,
+    extra_fields=KAFKA_COLUMNS,
+    engine=SESSION_RECORDINGS_DATA_TABLE_ENGINE(),
+)
+
+KAFKA_SESSION_RECORDINGS_TABLE_SQL = lambda: SESSION_RECORDINGS_TABLE_BASE_SQL.format(
+    table_name="kafka_session_recordings",
+    cluster=settings.CLICKHOUSE_CLUSTER,
+    engine=kafka_engine(topic=KAFKA_SESSION_RECORDINGS),
+    extra_fields="",
+)
+
+SESSION_RECORDINGS_TABLE_MV_SQL = lambda: """
+CREATE MATERIALIZED VIEW session_recordings_mv ON CLUSTER '{cluster}'
+TO {database}.{target_table}
+AS SELECT
+team_id,
+distinct_id,
+session_id,
+session_start,
+session_end,
+duration,
+snapshot_data_location,
+_timestamp,
+_offset
+FROM {database}.kafka_session_recordings
+""".format(
+    target_table=("writable_session_recordings"),
+    cluster=settings.CLICKHOUSE_CLUSTER,
+    database=settings.CLICKHOUSE_DATABASE,
+)
+
+# This table is responsible for writing to sharded_session_recordings based on a sharding key.
+WRITABLE_SESSION_RECORDINGS_TABLE_SQL = lambda: SESSION_RECORDINGS_TABLE_BASE_SQL.format(
+    table_name="writable_session_recordings",
+    cluster=settings.CLICKHOUSE_CLUSTER,
+    engine=Distributed(data_table=SESSION_RECORDINGS_DATA_TABLE, sharding_key="sipHash64(session_id)"),
+    extra_fields=KAFKA_COLUMNS,
+)
+
+# This table is responsible for reading from session_recording_events on a cluster setting
+DISTRIBUTED_SESSION_RECORDINGS_TABLE_SQL = lambda: SESSION_RECORDINGS_TABLE_BASE_SQL.format(
+    table_name="session_recordings",
+    cluster=settings.CLICKHOUSE_CLUSTER,
+    engine=Distributed(data_table=SESSION_RECORDINGS_DATA_TABLE, sharding_key="sipHash64(session_id)"),
+    extra_fields=KAFKA_COLUMNS,
+)