feat(replays): add replay_id column to merged discover table

snuba/datasets/configuration/discover/storages/discover.yaml

@@ -88,6 +88,7 @@ schema:
         args: { schema_modifiers: [nullable], size: 64 },
       },
       { name: deleted, type: UInt, args: { size: 8 } },
+      { name: trace_id, type: UUID, args: { schema_modifiers: [nullable] } }
     ]
   local_table_name: discover_local
   dist_table_name: discover_dist
@@ -115,9 +116,7 @@ query_processors:
       column_name: tags
   - processor: UUIDColumnProcessor
     args:
-      columns: !!set
-        trace_id: null
-        event_id: null
+      columns: [event_id, trace_id, replay_id]
   - processor: HexIntColumnProcessor
     args:
       columns: !!set

snuba/datasets/configuration/events/storages/errors.yaml

@@ -303,10 +303,7 @@ query_processors:
   - processor: UserColumnProcessor
   - processor: UUIDColumnProcessor
     args:
-      columns: !!set
-        event_id: null
-        primary_hash: null
-        trace_id: null
+      columns: [event_id, primary_hash, trace_id, replay_id]
   - processor: HexIntColumnProcessor
     args:
       columns: !!set

snuba/datasets/configuration/events/storages/errors_ro.yaml

@@ -302,10 +302,7 @@ query_processors:
   - processor: UserColumnProcessor
   - processor: UUIDColumnProcessor
     args:
-      columns: !!set
-        primary_hash: null
-        event_id: null
-        trace_id: null
+      columns: [event_id, primary_hash, trace_id, replay_id]
   - processor: HexIntColumnProcessor
     args:
       columns: !!set

snuba/datasets/configuration/transactions/storages/transactions.yaml

@@ -157,6 +157,7 @@ schema:
       },
       { name: app_start_type, type: String },
       { name: profile_id, type: UUID, args: { schema_modifiers: [nullable] } },
+      { name: replay_id, type: UUID, args: { schema_modifiers: [nullable] } },
     ]
   local_table_name: transactions_local
   dist_table_name: transactions_dist
@@ -202,7 +203,7 @@ query_processors:
           trace.span_id: span_id
   - processor: UUIDColumnProcessor
     args:
-      columns: [event_id, trace_id, profile_id]
+      columns: [event_id, trace_id, profile_id, replay_id]
   - processor: HexIntColumnProcessor
     args:
       columns: [span_id]

snuba/migrations/group_loader.py

@@ -136,6 +136,7 @@ def get_migrations(self) -> Sequence[str]:
             "0005_discover_fix_transaction_name",
             "0006_discover_add_trace_id",
             "0007_discover_add_span_id",
+            "0008_discover_add_replay_id",
         ]
 
 

snuba/snuba_migrations/discover/0008_discover_add_replay_id.py

@@ -0,0 +1,56 @@
+from typing import Sequence
+
+from snuba.clickhouse.columns import UUID, Column
+from snuba.clusters.storage_sets import StorageSetKey
+from snuba.migrations import migration, operations
+from snuba.migrations.columns import MigrationModifiers as Modifiers
+from snuba.migrations.operations import OperationTarget
+
+
+class Migration(migration.ClickhouseNodeMigration):
+    """
+    Add replay_id to merge table
+    """
+
+    blocking = False
+
+    def forwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.AddColumn(
+                storage_set=StorageSetKey.DISCOVER,
+                table_name=table_name,
+                column=Column("replay_id", UUID(Modifiers(nullable=True))),
+                after="span_id",
+                target=target,
+            )
+            for table_name, target in [
+                ("discover_local", OperationTarget.LOCAL),
+                ("discover_dist", OperationTarget.DISTRIBUTED),
+            ]
+        ]
+
+    def backwards_ops(self) -> Sequence[operations.SqlOperation]:
+        return [
+            operations.DropColumn(
+                storage_set=StorageSetKey.DISCOVER,
+                table_name=table_name,
+                column_name="replay_id",
+                target=target,
+            )
+            for table_name, target in [
+                ("discover_dist", OperationTarget.DISTRIBUTED),
+                ("discover_local", OperationTarget.LOCAL),
+            ]
+        ]
+
+    def forwards_local(self) -> Sequence[operations.SqlOperation]:
+        return self.forwards_ops()
+
+    def backwards_local(self) -> Sequence[operations.SqlOperation]:
+        return self.backwards_ops()
+
+    def forwards_dist(self) -> Sequence[operations.SqlOperation]:
+        return self.forwards_ops()
+
+    def backwards_dist(self) -> Sequence[operations.SqlOperation]:
+        return self.backwards_ops()