fix(issues): Start processing new_group_first_seen in merge case (#7351)

This is part of an effort to fix the issue-search's sort-by-age feature.
As part of that feature, we are adding the "first_seen" time of a group
to each error. When we merge groups, we want to update the errors to
have the new group's "group_first_seen".

We've decided to do that by explicitly passing the primary group's
"first_seen" time along with the Kafka message. This PR starts ingesting
that field properly; a followup PR in the `sentry` repo will start
providing that field after this has deployed everywhere.

Author: thetruecpaul

requirements.txt

@@ -28,7 +28,7 @@ python-dateutil==2.8.2
 python-rapidjson==1.8
 redis==4.5.4
 sentry-arroyo==2.29.1
-sentry-kafka-schemas==2.0.2
+sentry-kafka-schemas==2.0.4
 sentry-protos==0.3.3
 sentry-redis-tools==0.5.0
 sentry-relay==0.9.5

snuba/replacers/errors_replacer.py

@@ -626,6 +626,7 @@ class MergeReplacement(Replacement):
     project_id: int
     previous_group_ids: Sequence[int]
     new_group_id: int
+    new_group_first_seen: datetime | None
     timestamp: datetime
 
     all_columns: Sequence[FlattenedColumn]
@@ -660,11 +661,22 @@ def parse_message(
             else:
                 SEEN_MERGE_TXN_CACHE.append(txn)
 
+        # new_group_first_seen was added to the message schema; keep this check
+        # for backwards compatibility.
+        raw_new_group_first_seen = message.data.get("new_group_first_seen")
+        if raw_new_group_first_seen:
+            new_group_first_seen = datetime.strptime(
+                raw_new_group_first_seen, settings.PAYLOAD_DATETIME_FORMAT
+            )
+        else:
+            new_group_first_seen = None
+
         return cls(
             project_id=project_id,
             previous_group_ids=previous_group_ids,
             new_group_id=message.data["new_group_id"],
             timestamp=timestamp,
+            new_group_first_seen=new_group_first_seen,
             all_columns=context.all_columns,
         )
 
@@ -690,9 +702,18 @@ def get_count_query(self, table_name: str) -> Optional[str]:
     def get_insert_query(self, table_name: str) -> Optional[str]:
         all_column_names = [c.escaped for c in self.all_columns]
         all_columns = ", ".join(all_column_names)
+        replacement_columns = {"group_id": str(self.new_group_id)}
+
+        if self.new_group_first_seen is not None:
+            group_first_seen_str = self.new_group_first_seen.strftime(DATETIME_FORMAT)
+            replacement_columns[
+                "group_first_seen"
+            ] = f"CAST('{group_first_seen_str}' AS DateTime)"
+
         select_columns = ", ".join(
             map(
-                lambda i: i if i != "group_id" else str(self.new_group_id),
+                # Get i from replacement_columns; default to i if no replacement.
+                lambda i: replacement_columns.get(i, i),
                 all_column_names,
             )
         )

tests/replacer/test_errors_replacer.py

@@ -0,0 +1,122 @@
+from datetime import datetime, timedelta
+from unittest import mock
+
+from sentry_kafka_schemas.schema_types.events_v1 import EndMergeMessageBody
+
+from snuba import settings
+from snuba.clickhouse import DATETIME_FORMAT as CLICKHOUSE_DATETIME_FORMAT
+from snuba.datasets.entities.entity_key import EntityKey
+from snuba.datasets.entities.factory import get_entity
+from snuba.datasets.schemas.tables import WritableTableSchema
+from snuba.processor import ReplacementType
+from snuba.replacers.errors_replacer import MergeReplacement, ReplacementContext
+from snuba.replacers.replacer_processor import (
+    ReplacementMessage,
+    ReplacementMessageMetadata,
+    ReplacerState,
+)
+from snuba.utils.schemas import FlattenedColumn
+
+
+def test_merge_replacement_with_first_seen() -> None:
+    now_dt = datetime.now()
+    first_seen_dt = now_dt - timedelta(days=1)
+    data: EndMergeMessageBody = {
+        "transaction_id": "",
+        "project_id": 100,
+        "previous_group_ids": [2, 3],
+        "new_group_id": 1,
+        "new_group_first_seen": first_seen_dt.strftime(
+            settings.PAYLOAD_DATETIME_FORMAT
+        ),
+        "datetime": now_dt.strftime(settings.PAYLOAD_DATETIME_FORMAT),
+    }
+
+    message = ReplacementMessage(
+        action_type=ReplacementType.END_MERGE,
+        data=data,
+        metadata=mock.Mock(ReplacementMessageMetadata),
+    )
+
+    columns = [
+        FlattenedColumn(None, c.name, c.type)
+        for c in (
+            get_entity(EntityKey.SEARCH_ISSUES)
+            .get_all_storages()[0]
+            .get_schema()
+            .get_columns()
+        ).columns
+        if c.name in {"organization_id", "group_id", "event_id", "group_first_seen"}
+    ]
+
+    context = ReplacementContext(
+        all_columns=columns,
+        required_columns=[],
+        state_name=ReplacerState.ERRORS,
+        tag_column_map={},
+        promoted_tags={},
+        schema=mock.Mock(WritableTableSchema),
+    )
+
+    mr = MergeReplacement.parse_message(message, context)
+    assert mr is not None
+    expected = f"""INSERT INTO table (organization_id, group_id, event_id, group_first_seen)
+            SELECT organization_id, 1, event_id, CAST('{first_seen_dt.strftime(CLICKHOUSE_DATETIME_FORMAT)}' AS DateTime)
+            FROM table FINAL
+                        PREWHERE group_id IN (2, 3)
+            WHERE project_id = 100
+            AND received <= CAST('{now_dt.strftime(CLICKHOUSE_DATETIME_FORMAT)}' AS DateTime)
+            AND NOT deleted"""
+    actual = mr.get_insert_query("table")
+    assert actual is not None
+    assert actual.strip() == expected
+
+
+def test_merge_replacement_without_first_seen() -> None:
+    now_dt = datetime.now()
+    data: EndMergeMessageBody = {
+        "transaction_id": "",
+        "project_id": 100,
+        "previous_group_ids": [2, 3],
+        "new_group_id": 1,
+        "datetime": now_dt.strftime(settings.PAYLOAD_DATETIME_FORMAT),
+    }
+
+    message = ReplacementMessage(
+        action_type=ReplacementType.END_MERGE,
+        data=data,
+        metadata=mock.Mock(ReplacementMessageMetadata),
+    )
+
+    columns = [
+        FlattenedColumn(None, c.name, c.type)
+        for c in (
+            get_entity(EntityKey.SEARCH_ISSUES)
+            .get_all_storages()[0]
+            .get_schema()
+            .get_columns()
+        ).columns
+        if c.name in {"organization_id", "group_id", "event_id", "group_first_seen"}
+    ]
+
+    context = ReplacementContext(
+        all_columns=columns,
+        required_columns=[],
+        state_name=ReplacerState.ERRORS,
+        tag_column_map={},
+        promoted_tags={},
+        schema=mock.Mock(WritableTableSchema),
+    )
+
+    mr = MergeReplacement.parse_message(message, context)
+    assert mr is not None
+    expected = f"""INSERT INTO table (organization_id, group_id, event_id, group_first_seen)
+            SELECT organization_id, 1, event_id, group_first_seen
+            FROM table FINAL
+                        PREWHERE group_id IN (2, 3)
+            WHERE project_id = 100
+            AND received <= CAST('{now_dt.strftime(CLICKHOUSE_DATETIME_FORMAT)}' AS DateTime)
+            AND NOT deleted"""
+    actual = mr.get_insert_query("table")
+    assert actual is not None
+    assert actual.strip() == expected