feat(unmerge): Additions to support hierarchical grouping (#1876)

Author: untitaker

snuba/datasets/errors_replacer.py

@@ -251,6 +251,7 @@ def process_message(self, message: ReplacementMessage) -> Optional[Replacement]:
             "start_delete_groups",
             "start_merge",
             "start_unmerge",
+            "start_unmerge_hierarchical",
             "start_delete_tag",
         ):
             return None
@@ -260,6 +261,10 @@ def process_message(self, message: ReplacementMessage) -> Optional[Replacement]:
             processed = process_merge(event, self.__all_columns)
         elif type_ == "end_unmerge":
             processed = process_unmerge(event, self.__all_columns, self.__state_name)
+        elif type_ == "end_unmerge_hierarchical":
+            processed = process_unmerge_hierarchical(
+                event, self.__all_columns, self.__state_name
+            )
         elif type_ == "end_delete_tag":
             processed = process_delete_tag(
                 event,
@@ -680,9 +685,9 @@ def process_unmerge(
     )
 
     where = """\
-        PREWHERE group_id = %(previous_group_id)s
-        WHERE project_id = %(project_id)s
-        AND primary_hash IN (%(hashes)s)
+        PREWHERE primary_hash IN (%(hashes)s)
+        WHERE group_id = %(previous_group_id)s
+        AND project_id = %(project_id)s
         AND received <= CAST('%(timestamp)s' AS DateTime)
         AND NOT deleted
     """
@@ -710,16 +715,103 @@ def process_unmerge(
         "previous_group_id": message["previous_group_id"],
         "project_id": message["project_id"],
         "timestamp": timestamp.strftime(DATETIME_FORMAT),
+        "hashes": ", ".join(_convert_hash(h, state_name) for h in hashes),
     }
 
+    query_time_flags = (NEEDS_FINAL, message["project_id"])
+
+    return LegacyReplacement(
+        count_query_template, insert_query_template, query_args, query_time_flags
+    )
+
+
+def _convert_hash(
+    hash: str, state_name: ReplacerState, convert_types: bool = False
+) -> str:
     if state_name == ReplacerState.ERRORS:
-        query_args["hashes"] = ", ".join(
-            ["'%s'" % str(uuid.UUID(_hashify(h))) for h in hashes]
-        )
+        if convert_types:
+            return "toUUID('%s')" % str(uuid.UUID(_hashify(hash)))
+        else:
+            return "'%s'" % str(uuid.UUID(_hashify(hash)))
     else:
-        query_args["hashes"] = ", ".join("'%s'" % _hashify(h) for h in hashes)
+        if convert_types:
+            return "toFixedString('%s', 32)" % _hashify(hash)
+        else:
+            return "'%s'" % _hashify(hash)
 
-    query_time_flags = (NEEDS_FINAL, message["project_id"])
+
+def process_unmerge_hierarchical(
+    message: Mapping[str, Any],
+    all_columns: Sequence[FlattenedColumn],
+    state_name: ReplacerState,
+) -> Optional[Replacement]:
+    all_column_names = [c.escaped for c in all_columns]
+    select_columns = map(
+        lambda i: i if i != "group_id" else str(message["new_group_id"]),
+        all_column_names,
+    )
+
+    try:
+        timestamp = datetime.strptime(
+            message["datetime"], settings.PAYLOAD_DATETIME_FORMAT
+        )
+
+        primary_hash = message["primary_hash"]
+        assert isinstance(primary_hash, str)
+
+        hierarchical_hash = message["hierarchical_hash"]
+        assert isinstance(hierarchical_hash, str)
+
+        uuid.UUID(primary_hash)
+        uuid.UUID(hierarchical_hash)
+    except Exception as exc:
+        # TODO(markus): We're sacrificing consistency over uptime as long as
+        # this is in development. At some point this piece of code should be
+        # stable enough to remove this.
+        logger.error("process_unmerge_hierarchical.failed", exc_info=exc)
+        return None
+
+    where = """\
+        PREWHERE primary_hash = %(primary_hash)s
+        WHERE group_id = %(previous_group_id)s
+        AND has(hierarchical_hashes, %(hierarchical_hash)s)
+        AND project_id = %(project_id)s
+        AND received <= CAST('%(timestamp)s' AS DateTime)
+        AND NOT deleted
+    """
+
+    count_query_template = (
+        """\
+        SELECT count()
+        FROM %(table_name)s FINAL
+    """
+        + where
+    )
+
+    insert_query_template = (
+        """\
+        INSERT INTO %(table_name)s (%(all_columns)s)
+        SELECT %(select_columns)s
+        FROM %(table_name)s FINAL
+    """
+        + where
+    )
+
+    query_args = {
+        "all_columns": ", ".join(all_column_names),
+        "select_columns": ", ".join(select_columns),
+        "previous_group_id": message["previous_group_id"],
+        "project_id": message["project_id"],
+        "timestamp": timestamp.strftime(DATETIME_FORMAT),
+        "primary_hash": _convert_hash(primary_hash, state_name),
+        "hierarchical_hash": _convert_hash(
+            hierarchical_hash, state_name, convert_types=True
+        ),
+    }
+
+    # Sentry is expected to send an `exclude_groups` message after unsplit is
+    # done, and we can live with data inconsistencies while this is ongoing.
+    query_time_flags = (None, message["project_id"])
 
     return LegacyReplacement(
         count_query_template, insert_query_template, query_args, query_time_flags

tests/datasets/test_errors_replacer.py

@@ -287,11 +287,11 @@ def test_unmerge_process(self) -> None:
 
         assert (
             re.sub("[\n ]+", " ", replacement.count_query_template).strip()
-            == "SELECT count() FROM %(table_name)s FINAL PREWHERE group_id = %(previous_group_id)s WHERE project_id = %(project_id)s AND primary_hash IN (%(hashes)s) AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+            == "SELECT count() FROM %(table_name)s FINAL PREWHERE primary_hash IN (%(hashes)s) WHERE group_id = %(previous_group_id)s AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
         )
         assert (
             re.sub("[\n ]+", " ", replacement.insert_query_template).strip()
-            == "INSERT INTO %(table_name)s (%(all_columns)s) SELECT %(select_columns)s FROM %(table_name)s FINAL PREWHERE group_id = %(previous_group_id)s WHERE project_id = %(project_id)s AND primary_hash IN (%(hashes)s) AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+            == "INSERT INTO %(table_name)s (%(all_columns)s) SELECT %(select_columns)s FROM %(table_name)s FINAL PREWHERE primary_hash IN (%(hashes)s) WHERE group_id = %(previous_group_id)s AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
         )
         assert replacement.query_args == {
             "all_columns": "project_id, timestamp, event_id, platform, environment, release, dist, ip_address_v4, ip_address_v6, user, user_id, user_name, user_email, sdk_name, sdk_version, http_method, http_referer, tags.key, tags.value, contexts.key, contexts.value, transaction_name, span_id, trace_id, partition, offset, message_timestamp, retention_days, deleted, group_id, primary_hash, hierarchical_hashes, received, message, title, culprit, level, location, version, type, exception_stacks.type, exception_stacks.value, exception_stacks.mechanism_type, exception_stacks.mechanism_handled, exception_frames.abs_path, exception_frames.colno, exception_frames.filename, exception_frames.function, exception_frames.lineno, exception_frames.in_app, exception_frames.package, exception_frames.module, exception_frames.stack_level, sdk_integrations, modules.name, modules.version",
@@ -307,6 +307,44 @@ def test_unmerge_process(self) -> None:
             self.project_id,
         )
 
+    def test_unmerge_hierarchical_process(self) -> None:
+        timestamp = datetime.now(tz=pytz.utc)
+
+        message = (
+            2,
+            "end_unmerge_hierarchical",
+            {
+                "project_id": self.project_id,
+                "previous_group_id": 1,
+                "new_group_id": 2,
+                "hierarchical_hash": "a" * 32,
+                "primary_hash": "b" * 32,
+                "datetime": timestamp.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            },
+        )
+
+        replacement = self.replacer.process_message(self._wrap(message))
+
+        assert (
+            re.sub("[\n ]+", " ", replacement.count_query_template).strip()
+            == "SELECT count() FROM %(table_name)s FINAL PREWHERE primary_hash = %(primary_hash)s WHERE group_id = %(previous_group_id)s AND has(hierarchical_hashes, %(hierarchical_hash)s) AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+        )
+        assert (
+            re.sub("[\n ]+", " ", replacement.insert_query_template).strip()
+            == "INSERT INTO %(table_name)s (%(all_columns)s) SELECT %(select_columns)s FROM %(table_name)s FINAL PREWHERE primary_hash = %(primary_hash)s WHERE group_id = %(previous_group_id)s AND has(hierarchical_hashes, %(hierarchical_hash)s) AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+        )
+        assert replacement.query_args == {
+            "all_columns": "project_id, timestamp, event_id, platform, environment, release, dist, ip_address_v4, ip_address_v6, user, user_id, user_name, user_email, sdk_name, sdk_version, http_method, http_referer, tags.key, tags.value, contexts.key, contexts.value, transaction_name, span_id, trace_id, partition, offset, message_timestamp, retention_days, deleted, group_id, primary_hash, hierarchical_hashes, received, message, title, culprit, level, location, version, type, exception_stacks.type, exception_stacks.value, exception_stacks.mechanism_type, exception_stacks.mechanism_handled, exception_frames.abs_path, exception_frames.colno, exception_frames.filename, exception_frames.function, exception_frames.lineno, exception_frames.in_app, exception_frames.package, exception_frames.module, exception_frames.stack_level, sdk_integrations, modules.name, modules.version",
+            "select_columns": "project_id, timestamp, event_id, platform, environment, release, dist, ip_address_v4, ip_address_v6, user, user_id, user_name, user_email, sdk_name, sdk_version, http_method, http_referer, tags.key, tags.value, contexts.key, contexts.value, transaction_name, span_id, trace_id, partition, offset, message_timestamp, retention_days, deleted, 2, primary_hash, hierarchical_hashes, received, message, title, culprit, level, location, version, type, exception_stacks.type, exception_stacks.value, exception_stacks.mechanism_type, exception_stacks.mechanism_handled, exception_frames.abs_path, exception_frames.colno, exception_frames.filename, exception_frames.function, exception_frames.lineno, exception_frames.in_app, exception_frames.package, exception_frames.module, exception_frames.stack_level, sdk_integrations, modules.name, modules.version",
+            "primary_hash": "'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'",
+            "hierarchical_hash": "toUUID('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa')",
+            "previous_group_id": 1,
+            "project_id": self.project_id,
+            "timestamp": timestamp.strftime(DATETIME_FORMAT),
+        }
+
+        assert replacement.query_time_flags == (None, self.project_id,)
+
     def test_delete_promoted_tag_process(self) -> None:
         timestamp = datetime.now(tz=pytz.utc)
         message = (

tests/test_replacer.py

@@ -164,11 +164,11 @@ def test_unmerge_process(self) -> None:
 
         assert (
             re.sub("[\n ]+", " ", replacement.count_query_template).strip()
-            == "SELECT count() FROM %(table_name)s FINAL PREWHERE group_id = %(previous_group_id)s WHERE project_id = %(project_id)s AND primary_hash IN (%(hashes)s) AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+            == "SELECT count() FROM %(table_name)s FINAL PREWHERE primary_hash IN (%(hashes)s) WHERE group_id = %(previous_group_id)s AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
         )
         assert (
             re.sub("[\n ]+", " ", replacement.insert_query_template).strip()
-            == "INSERT INTO %(table_name)s (%(all_columns)s) SELECT %(select_columns)s FROM %(table_name)s FINAL PREWHERE group_id = %(previous_group_id)s WHERE project_id = %(project_id)s AND primary_hash IN (%(hashes)s) AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+            == "INSERT INTO %(table_name)s (%(all_columns)s) SELECT %(select_columns)s FROM %(table_name)s FINAL PREWHERE primary_hash IN (%(hashes)s) WHERE group_id = %(previous_group_id)s AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
         )
         assert replacement.query_args == {
             "all_columns": "event_id, project_id, group_id, timestamp, deleted, retention_days, platform, message, primary_hash, hierarchical_hashes, received, search_message, title, location, user_id, username, email, ip_address, geo_country_code, geo_region, geo_city, sdk_name, sdk_version, type, version, offset, partition, message_timestamp, os_build, os_kernel_version, device_name, device_brand, device_locale, device_uuid, device_model_id, device_arch, device_battery_level, device_orientation, device_simulator, device_online, device_charging, level, logger, server_name, transaction, environment, `sentry:release`, `sentry:dist`, `sentry:user`, site, url, app_device, device, device_family, runtime, runtime_name, browser, browser_name, os, os_name, os_rooted, tags.key, tags.value, _tags_flattened, contexts.key, contexts.value, http_method, http_referer, exception_stacks.type, exception_stacks.value, exception_stacks.mechanism_type, exception_stacks.mechanism_handled, exception_frames.abs_path, exception_frames.filename, exception_frames.package, exception_frames.module, exception_frames.function, exception_frames.in_app, exception_frames.colno, exception_frames.lineno, exception_frames.stack_level, culprit, sdk_integrations, modules.name, modules.version",
@@ -183,6 +183,42 @@ def test_unmerge_process(self) -> None:
             self.project_id,
         )
 
+    def test_unmerge_hierarchical_process(self) -> None:
+        timestamp = datetime.now(tz=pytz.utc)
+        message = (
+            2,
+            "end_unmerge_hierarchical",
+            {
+                "project_id": self.project_id,
+                "previous_group_id": 1,
+                "new_group_id": 2,
+                "hierarchical_hash": "a" * 32,
+                "primary_hash": "b" * 32,
+                "datetime": timestamp.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            },
+        )
+
+        replacement = self.replacer.process_message(self._wrap(message))
+
+        assert (
+            re.sub("[\n ]+", " ", replacement.count_query_template).strip()
+            == "SELECT count() FROM %(table_name)s FINAL PREWHERE primary_hash = %(primary_hash)s WHERE group_id = %(previous_group_id)s AND has(hierarchical_hashes, %(hierarchical_hash)s) AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+        )
+        assert (
+            re.sub("[\n ]+", " ", replacement.insert_query_template).strip()
+            == "INSERT INTO %(table_name)s (%(all_columns)s) SELECT %(select_columns)s FROM %(table_name)s FINAL PREWHERE primary_hash = %(primary_hash)s WHERE group_id = %(previous_group_id)s AND has(hierarchical_hashes, %(hierarchical_hash)s) AND project_id = %(project_id)s AND received <= CAST('%(timestamp)s' AS DateTime) AND NOT deleted"
+        )
+        assert replacement.query_args == {
+            "all_columns": "event_id, project_id, group_id, timestamp, deleted, retention_days, platform, message, primary_hash, hierarchical_hashes, received, search_message, title, location, user_id, username, email, ip_address, geo_country_code, geo_region, geo_city, sdk_name, sdk_version, type, version, offset, partition, message_timestamp, os_build, os_kernel_version, device_name, device_brand, device_locale, device_uuid, device_model_id, device_arch, device_battery_level, device_orientation, device_simulator, device_online, device_charging, level, logger, server_name, transaction, environment, `sentry:release`, `sentry:dist`, `sentry:user`, site, url, app_device, device, device_family, runtime, runtime_name, browser, browser_name, os, os_name, os_rooted, tags.key, tags.value, _tags_flattened, contexts.key, contexts.value, http_method, http_referer, exception_stacks.type, exception_stacks.value, exception_stacks.mechanism_type, exception_stacks.mechanism_handled, exception_frames.abs_path, exception_frames.filename, exception_frames.package, exception_frames.module, exception_frames.function, exception_frames.in_app, exception_frames.colno, exception_frames.lineno, exception_frames.stack_level, culprit, sdk_integrations, modules.name, modules.version",
+            "select_columns": "event_id, project_id, 2, timestamp, deleted, retention_days, platform, message, primary_hash, hierarchical_hashes, received, search_message, title, location, user_id, username, email, ip_address, geo_country_code, geo_region, geo_city, sdk_name, sdk_version, type, version, offset, partition, message_timestamp, os_build, os_kernel_version, device_name, device_brand, device_locale, device_uuid, device_model_id, device_arch, device_battery_level, device_orientation, device_simulator, device_online, device_charging, level, logger, server_name, transaction, environment, `sentry:release`, `sentry:dist`, `sentry:user`, site, url, app_device, device, device_family, runtime, runtime_name, browser, browser_name, os, os_name, os_rooted, tags.key, tags.value, _tags_flattened, contexts.key, contexts.value, http_method, http_referer, exception_stacks.type, exception_stacks.value, exception_stacks.mechanism_type, exception_stacks.mechanism_handled, exception_frames.abs_path, exception_frames.filename, exception_frames.package, exception_frames.module, exception_frames.function, exception_frames.in_app, exception_frames.colno, exception_frames.lineno, exception_frames.stack_level, culprit, sdk_integrations, modules.name, modules.version",
+            "hierarchical_hash": "toFixedString('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', 32)",
+            "primary_hash": "'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'",
+            "previous_group_id": 1,
+            "project_id": self.project_id,
+            "timestamp": timestamp.strftime(DATETIME_FORMAT),
+        }
+        assert replacement.query_time_flags == (None, self.project_id,)
+
     def test_delete_promoted_tag_process(self) -> None:
         timestamp = datetime.now(tz=pytz.utc)
         message = (
@@ -373,6 +409,48 @@ def test_unmerge_insert(self) -> None:
 
         assert self._issue_count(self.project_id) == [{"count": 1, "group_id": 2}]
 
+    def test_unmerge_hierarchical_insert(self) -> None:
+        self.event["project_id"] = self.project_id
+        self.event["group_id"] = 1
+        self.event["primary_hash"] = "b" * 32
+        self.event["data"]["hierarchical_hashes"] = ["a" * 32]
+        write_unprocessed_events(self.storage, [self.event])
+
+        assert self._issue_count(self.project_id) == [{"count": 1, "group_id": 1}]
+
+        timestamp = datetime.now(tz=pytz.utc)
+
+        project_id = self.project_id
+
+        message: Message[KafkaPayload] = Message(
+            Partition(Topic("replacements"), 1),
+            42,
+            KafkaPayload(
+                None,
+                json.dumps(
+                    (
+                        2,
+                        "end_unmerge_hierarchical",
+                        {
+                            "project_id": project_id,
+                            "previous_group_id": 1,
+                            "new_group_id": 2,
+                            "hierarchical_hash": "a" * 32,
+                            "primary_hash": "b" * 32,
+                            "datetime": timestamp.strftime(PAYLOAD_DATETIME_FORMAT),
+                        },
+                    )
+                ).encode("utf-8"),
+                [],
+            ),
+            datetime.now(),
+        )
+
+        processed = self.replacer.process_message(message)
+        self.replacer.flush_batch([processed])
+
+        assert self._issue_count(self.project_id) == [{"count": 1, "group_id": 2}]
+
     def test_delete_tag_promoted_insert(self) -> None:
         self.event["project_id"] = self.project_id
         self.event["group_id"] = 1