first pass addition of _timestamp

posthog/clickhouse/test/__snapshots__/test_schema.ambr

@@ -337,7 +337,8 @@
       console_warn_count Int64,
       console_error_count Int64,
       size Int64,
-      event_count Int64
+      event_count Int64,
+      _timestamp DateTime
   ) ENGINE = Kafka('test.kafka.broker:9092', 'clickhouse_session_replay_events_test', 'group1', 'JSONEachRow')
 
   '
@@ -924,7 +925,8 @@
       console_warn_count Int64,
       console_error_count Int64,
       size Int64,
-      event_count Int64
+      event_count Int64,
+      _timestamp DateTime
   ) ENGINE = Kafka('kafka:9092', 'clickhouse_session_replay_events_test', 'group1', 'JSONEachRow')
 
   '
@@ -1353,7 +1355,8 @@
       -- this allows us to count the number of snapshot events received in a session
       -- often very useful in incidents or debugging
       -- because we batch events we expect message_count to be lower than event_count
-      event_count SimpleAggregateFunction(sum, Int64)
+      event_count SimpleAggregateFunction(sum, Int64),
+      _timestamp SimpleAggregateFunction(max, DateTime64(6, 'UTC'))
   ) ENGINE = Distributed('posthog', 'posthog_test', 'sharded_session_replay_events', sipHash64(distinct_id))
 
   '
@@ -1389,7 +1392,8 @@
   sum(size) as size,
   -- we can count the number of kafka messages instead of sending it explicitly
   count(*) as message_count,
-  sum(event_count) as event_count
+  sum(event_count) as event_count,
+  max(_timestamp) DateTime
   FROM posthog_test.kafka_session_replay_events
   group by session_id, team_id
 
@@ -1627,7 +1631,8 @@
       -- this allows us to count the number of snapshot events received in a session
       -- often very useful in incidents or debugging
       -- because we batch events we expect message_count to be lower than event_count
-      event_count SimpleAggregateFunction(sum, Int64)
+      event_count SimpleAggregateFunction(sum, Int64),
+      _timestamp SimpleAggregateFunction(max, DateTime64(6, 'UTC'))
   ) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/77f1df52-4b43-11e9-910f-b8ca3a9b9f3e_{shard}/posthog.session_replay_events', '{replica}')
 
       PARTITION BY toYYYYMM(min_first_timestamp)
@@ -2252,7 +2257,8 @@
       -- this allows us to count the number of snapshot events received in a session
       -- often very useful in incidents or debugging
       -- because we batch events we expect message_count to be lower than event_count
-      event_count SimpleAggregateFunction(sum, Int64)
+      event_count SimpleAggregateFunction(sum, Int64),
+      _timestamp SimpleAggregateFunction(max, DateTime64(6, 'UTC'))
   ) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/77f1df52-4b43-11e9-910f-b8ca3a9b9f3e_{shard}/posthog.session_replay_events', '{replica}')
 
       PARTITION BY toYYYYMM(min_first_timestamp)

posthog/models/session_replay_event/migrations_sql.py

@@ -70,7 +70,9 @@
 ALTER_SESSION_REPLAY_ADD_EVENT_COUNT_COLUMN = """
     ALTER TABLE {table_name} on CLUSTER '{cluster}'
         ADD COLUMN IF NOT EXISTS message_count SimpleAggregateFunction(sum, Int64),
-        ADD COLUMN IF NOT EXISTS event_count SimpleAggregateFunction(sum, Int64)
+        ADD COLUMN IF NOT EXISTS event_count SimpleAggregateFunction(sum, Int64),
+        -- fly by addition so that we can track lag in the data the same way as for other tables
+        ADD COLUMN IF NOT EXISTS _timestamp SimpleAggregateFunction(max, DateTime64(6, 'UTC'))
 """
 
 ADD_EVENT_COUNT_DISTRIBUTED_SESSION_REPLAY_EVENTS_TABLE_SQL = (

posthog/models/session_replay_event/sql.py

@@ -28,7 +28,8 @@
     console_warn_count Int64,
     console_error_count Int64,
     size Int64,
-    event_count Int64
+    event_count Int64,
+    _timestamp DateTime
 ) ENGINE = {engine}
 """
 
@@ -61,7 +62,8 @@
     -- this allows us to count the number of snapshot events received in a session
     -- often very useful in incidents or debugging
     -- because we batch events we expect message_count to be lower than event_count
-    event_count SimpleAggregateFunction(sum, Int64)
+    event_count SimpleAggregateFunction(sum, Int64),
+    _timestamp SimpleAggregateFunction(max, DateTime64(6, 'UTC'))
 ) ENGINE = {engine}
 """
 
@@ -128,7 +130,8 @@
 sum(size) as size,
 -- we can count the number of kafka messages instead of sending it explicitly
 count(*) as message_count,
-sum(event_count) as event_count
+sum(event_count) as event_count,
+max(_timestamp) DateTime
 FROM {database}.kafka_session_replay_events
 group by session_id, team_id
 """.format(