feat(occurrences on eap): Implement tagstore EAP query for group list tag value (#111963)

Implements double reads of occurrences from EAP for
`get_group_list_tag_value` & `get_generic_group_list_tag_value` in
`src/sentry/tagstore/snuba/backend.py`.

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: shashjar

src/sentry/tagstore/snuba/backend.py

@@ -164,6 +164,32 @@ def _reasonable_release_tags_match(control: set[TagValue], experimental: set[Tag
     )
 
 
+def _reasonable_group_list_tag_value_match(
+    control: dict[int, GroupTagValue],
+    experimental: dict[int, GroupTagValue],
+) -> bool:
+    if not set(experimental.keys()).issubset(set(control.keys())):
+        return False
+    for group_id in experimental:
+        ctrl = control[group_id]
+        exp = experimental[group_id]
+        if exp.times_seen > ctrl.times_seen:
+            return False
+        if (
+            exp.first_seen is not None
+            and ctrl.first_seen is not None
+            and exp.first_seen < ctrl.first_seen
+        ):
+            return False
+        if (
+            exp.last_seen is not None
+            and ctrl.last_seen is not None
+            and exp.last_seen > ctrl.last_seen
+        ):
+            return False
+    return True
+
+
 class _OptimizeKwargs(TypedDict, total=False):
     turbo: bool
     sample: int
@@ -825,7 +851,7 @@ def get_group_list_tag_value(
         value,
         tenant_ids=None,
     ):
-        return self.__get_group_list_tag_value(
+        snuba_result = self.__get_group_list_tag_value(
             project_ids,
             group_id_list,
             environment_ids,
@@ -834,9 +860,38 @@ def get_group_list_tag_value(
             Dataset.Events,
             [DEFAULT_TYPE_CONDITION],
             [],
-            "tagstore.get_group_list_tag_value",
+            Referrer.TAGSTORE_GET_GROUP_LIST_TAG_VALUE.value,
             tenant_ids=tenant_ids,
         )
+        result = snuba_result
+
+        callsite = "SnubaTagStorage::get_group_list_tag_value"
+        if EAPOccurrencesComparator.should_check_experiment(callsite):
+            eap_result = self._eap_get_group_list_tag_value(
+                project_ids,
+                group_id_list,
+                environment_ids,
+                key,
+                value,
+                referrer=Referrer.TAGSTORE_GET_GROUP_LIST_TAG_VALUE.value,
+                occurrence_category=OccurrenceCategory.ERROR,
+            )
+            result = EAPOccurrencesComparator.check_and_choose(
+                control_data=snuba_result,
+                experimental_data=eap_result,
+                callsite=callsite,
+                is_experimental_data_a_null_result=len(eap_result) == 0,
+                reasonable_match_comparator=_reasonable_group_list_tag_value_match,
+                debug_context={
+                    "project_ids": list(project_ids),
+                    "group_ids": list(group_id_list),
+                    "environment_ids": list(environment_ids) if environment_ids else None,
+                    "key": key,
+                    "value": value,
+                },
+            )
+
+        return result
 
     def get_generic_group_list_tag_value(
         self,
@@ -890,12 +945,137 @@ def get_generic_group_list_tag_value(
             result_snql["data"], ["group_id"], ["times_seen", "first_seen", "last_seen"]
         )
 
-        return {
+        snuba_result = {
             group_id: GroupTagValue(
                 group_id=group_id, key=key, value=value, **fix_tag_value_data(data)
             )
             for group_id, data in nested_groups.items()
         }
+        result = snuba_result
+
+        callsite = "SnubaTagStorage::get_generic_group_list_tag_value"
+        if EAPOccurrencesComparator.should_check_experiment(callsite):
+            eap_result = self._eap_get_group_list_tag_value(
+                project_ids,
+                group_id_list,
+                environment_ids,
+                key,
+                value,
+                referrer=Referrer.TAGSTORE_GET_GENERIC_GROUP_LIST_TAG_VALUE.value,
+                occurrence_category=OccurrenceCategory.ISSUE_PLATFORM,
+            )
+            result = EAPOccurrencesComparator.check_and_choose(
+                control_data=snuba_result,
+                experimental_data=eap_result,
+                callsite=callsite,
+                is_experimental_data_a_null_result=len(eap_result) == 0,
+                reasonable_match_comparator=_reasonable_group_list_tag_value_match,
+                debug_context={
+                    "project_ids": list(project_ids),
+                    "group_ids": list(group_id_list),
+                    "environment_ids": list(environment_ids) if environment_ids else None,
+                    "key": key,
+                    "value": value,
+                },
+            )
+
+        return result
+
+    def _eap_get_group_list_tag_value(
+        self,
+        project_ids: Sequence[int],
+        group_id_list: Sequence[int],
+        environment_ids: Sequence[int] | None,
+        key: str,
+        value: str,
+        referrer: str,
+        occurrence_category: OccurrenceCategory,
+    ) -> dict[int, GroupTagValue]:
+        organization_id = get_organization_id_from_project_ids(project_ids)
+
+        now = datetime.now(tz=timezone.utc)
+        resolved_start = now - timedelta(days=90)
+        resolved_end = now
+
+        try:
+            organization = Organization.objects.get_from_cache(id=organization_id)
+        except Organization.DoesNotExist:
+            return {}
+
+        projects = list(Project.objects.filter(id__in=project_ids, organization_id=organization_id))
+        if not projects:
+            return {}
+
+        environments = (
+            list(Environment.objects.filter(id__in=environment_ids)) if environment_ids else []
+        )
+
+        group_id_filter = f"group_id:[{','.join(str(gid) for gid in group_id_list)}]"
+        tag_filter = build_escaped_term_filter(key, [str(value)])
+        query_string = f"{group_id_filter} {tag_filter}"
+
+        snuba_params = SnubaParams(
+            start=resolved_start,
+            end=resolved_end,
+            organization=organization,
+            projects=projects,
+            environments=environments,
+        )
+
+        try:
+            result = Occurrences.run_table_query(
+                params=snuba_params,
+                query_string=query_string,
+                selected_columns=["group_id", "count()", "min(timestamp)", "last_seen()"],
+                orderby=None,
+                offset=0,
+                limit=len(group_id_list),
+                referrer=referrer,
+                config=SearchResolverConfig(),
+                occurrence_category=occurrence_category,
+            )
+
+            output: dict[int, GroupTagValue] = {}
+            for row in result.get("data", []):
+                group_id = row.get("group_id")
+                if group_id is None:
+                    continue
+                first_seen_raw = row.get("min(timestamp)")
+                last_seen_raw = row.get("last_seen()")
+                output[int(group_id)] = GroupTagValue(
+                    group_id=int(group_id),
+                    key=key,
+                    value=value,
+                    times_seen=int(row.get("count()", 0)),
+                    first_seen=(
+                        parse_datetime(datetime_processor(first_seen_raw)).replace(
+                            tzinfo=timezone.utc
+                        )
+                        if first_seen_raw is not None
+                        else None
+                    ),
+                    last_seen=(
+                        parse_datetime(datetime_processor(last_seen_raw)).replace(
+                            tzinfo=timezone.utc
+                        )
+                        if last_seen_raw is not None
+                        else None
+                    ),
+                )
+            return output
+        except Exception:
+            logger.exception(
+                "EAP get_group_list_tag_value query failed",
+                extra={
+                    "organization_id": organization_id,
+                    "project_ids": list(project_ids),
+                    "group_ids": list(group_id_list),
+                    "key": key,
+                    "value": value,
+                    "occurrence_category": occurrence_category.value,
+                },
+            )
+            return {}
 
     def apply_group_filters(self, group: Group | None, filters: MutableMapping[str, Sequence[Any]]):
         dataset = Dataset.Events

tests/snuba/tagstore/test_tagstore_backend.py

@@ -1911,3 +1911,153 @@ def test_eap_and_snuba_release_tags_match(self) -> None:
         assert snuba_by_value["1.0"].last_seen == eap_by_value["1.0"].last_seen
         assert snuba_by_value["2.0"].first_seen == eap_by_value["2.0"].first_seen
         assert snuba_by_value["2.0"].last_seen == eap_by_value["2.0"].last_seen
+
+    @freeze_time(FROZEN_TIME)
+    def test_eap_and_snuba_group_list_tag_value_match_multiple_groups(self) -> None:
+        ts = (self.FROZEN_TIME - timedelta(minutes=5)).timestamp()
+        env = self.create_environment(project=self.project, name="production")
+
+        # Group A: 2 events in "production"
+        self.store_events_to_snuba_and_eap(
+            "group-a", count=1, timestamp=ts, extra_event_data={"environment": env.name}
+        )
+        events_a = self.store_events_to_snuba_and_eap(
+            "group-a", count=1, timestamp=ts, extra_event_data={"environment": env.name}
+        )
+        group_a = events_a[0].group
+        assert group_a is not None
+
+        # Group B: 1 event in "production"
+        events_b = self.store_events_to_snuba_and_eap(
+            "group-b", count=1, timestamp=ts, extra_event_data={"environment": env.name}
+        )
+        group_b = events_b[0].group
+        assert group_b is not None
+
+        group_ids = [group_a.id, group_b.id]
+
+        snuba_result = self.ts.get_group_list_tag_value(
+            [self.project.id],
+            group_ids,
+            [env.id],
+            "environment",
+            env.name,
+            tenant_ids={"referrer": "r", "organization_id": self.project.organization_id},
+        )
+        eap_result = self.ts._eap_get_group_list_tag_value(
+            [self.project.id],
+            group_ids,
+            [env.id],
+            "environment",
+            env.name,
+            referrer="tagstore.get_group_list_tag_value",
+            occurrence_category=OccurrenceCategory.ERROR,
+        )
+
+        assert set(snuba_result.keys()) == {group_a.id, group_b.id}
+        assert set(eap_result.keys()) == set(snuba_result.keys())
+        assert snuba_result[group_a.id].times_seen == eap_result[group_a.id].times_seen == 2
+        assert snuba_result[group_b.id].times_seen == eap_result[group_b.id].times_seen == 1
+        assert snuba_result[group_a.id].first_seen == eap_result[group_a.id].first_seen
+        assert snuba_result[group_a.id].last_seen == eap_result[group_a.id].last_seen
+        assert snuba_result[group_b.id].first_seen == eap_result[group_b.id].first_seen
+        assert snuba_result[group_b.id].last_seen == eap_result[group_b.id].last_seen
+
+    @freeze_time(FROZEN_TIME)
+    def test_eap_and_snuba_group_list_tag_value_with_environment_filter(self) -> None:
+        ts = (self.FROZEN_TIME - timedelta(minutes=5)).timestamp()
+        env_prod = self.create_environment(project=self.project, name="production")
+        env_staging = self.create_environment(project=self.project, name="staging")
+
+        # 2 events in "production", 1 event in "staging"
+        self.store_events_to_snuba_and_eap(
+            "group-a", count=1, timestamp=ts, extra_event_data={"environment": env_prod.name}
+        )
+        self.store_events_to_snuba_and_eap(
+            "group-a", count=1, timestamp=ts, extra_event_data={"environment": env_prod.name}
+        )
+        events = self.store_events_to_snuba_and_eap(
+            "group-a", count=1, timestamp=ts, extra_event_data={"environment": env_staging.name}
+        )
+        group_a = events[0].group
+        assert group_a is not None
+
+        # Filter by production only
+        snuba_result = self.ts.get_group_list_tag_value(
+            [self.project.id],
+            [group_a.id],
+            [env_prod.id],
+            "environment",
+            env_prod.name,
+            tenant_ids={"referrer": "r", "organization_id": self.project.organization_id},
+        )
+        eap_result = self.ts._eap_get_group_list_tag_value(
+            [self.project.id],
+            [group_a.id],
+            [env_prod.id],
+            "environment",
+            env_prod.name,
+            referrer="tagstore.get_group_list_tag_value",
+            occurrence_category=OccurrenceCategory.ERROR,
+        )
+
+        assert group_a.id in snuba_result
+        assert group_a.id in eap_result
+        assert snuba_result[group_a.id].times_seen == eap_result[group_a.id].times_seen == 2
+        assert snuba_result[group_a.id].first_seen == eap_result[group_a.id].first_seen
+        assert snuba_result[group_a.id].last_seen == eap_result[group_a.id].last_seen
+
+    @freeze_time(FROZEN_TIME)
+    def test_eap_group_list_tag_value_issue_platform(self) -> None:
+        ts = self.FROZEN_TIME - timedelta(minutes=5)
+        env = self.create_environment(project=self.project, name="production")
+
+        group = self.create_group(project=self.project)
+
+        eap_item_1 = self.create_eap_occurrence(
+            group_id=group.id,
+            timestamp=ts,
+            environment=env.name,
+            issue_occurrence_id=uuid4().hex,
+        )
+        eap_item_2 = self.create_eap_occurrence(
+            group_id=group.id,
+            timestamp=ts + timedelta(minutes=1),
+            environment=env.name,
+            issue_occurrence_id=uuid4().hex,
+        )
+        self.store_eap_items([eap_item_1, eap_item_2])
+
+        eap_result = self.ts._eap_get_group_list_tag_value(
+            [self.project.id],
+            [group.id],
+            [env.id],
+            "environment",
+            env.name,
+            referrer="tagstore.get_generic_group_list_tag_value",
+            occurrence_category=OccurrenceCategory.ISSUE_PLATFORM,
+        )
+
+        assert group.id in eap_result
+        assert eap_result[group.id].times_seen == 2
+        assert eap_result[group.id].key == "environment"
+        assert eap_result[group.id].value == env.name
+        assert eap_result[group.id].first_seen is not None
+        assert eap_result[group.id].last_seen is not None
+        assert eap_result[group.id].first_seen <= eap_result[group.id].last_seen
+
+    @freeze_time(FROZEN_TIME)
+    def test_eap_group_list_tag_value_empty_result(self) -> None:
+        group = self.create_group(project=self.project)
+
+        eap_result = self.ts._eap_get_group_list_tag_value(
+            [self.project.id],
+            [group.id],
+            None,
+            "environment",
+            "nonexistent",
+            referrer="tagstore.get_group_list_tag_value",
+            occurrence_category=OccurrenceCategory.ERROR,
+        )
+
+        assert eap_result == {}