feat(upsampling) - Error Upsampling on Metric Alerts under ACI (#97734)

- Converted count() aggregation to upsampled_count() for Entity
Subscriptions on Events dataset for matching projects
- Prevented "upsampled_count()" aggregate from being used directly on
Detector creation

Author: yuvmen

src/sentry/snuba/entity_subscription.py

@@ -11,6 +11,7 @@
 from snuba_sdk import Column, Condition, Entity, Join, Op, Request
 
 from sentry import features
+from sentry.api.helpers.error_upsampling import are_any_projects_error_upsampled
 from sentry.constants import CRASH_RATE_ALERT_AGGREGATE_ALIAS
 from sentry.exceptions import InvalidQuerySubscription, UnsupportedQuerySubscription
 from sentry.models.environment import Environment
@@ -206,10 +207,20 @@ def build_query_builder(
             query_builder_cls = ErrorsQueryBuilder
             parser_config_overrides.update(PARSER_CONFIG_OVERRIDES)
 
+        # Conditionally upsample error counts for allowlisted projects
+        selected_aggregate = self.aggregate
+        if (
+            self.dataset == Dataset.Events
+            and isinstance(selected_aggregate, str)
+            and selected_aggregate.strip().lower() == "count()"
+            and are_any_projects_error_upsampled(project_ids)
+        ):
+            selected_aggregate = "upsampled_count()"
+
         return query_builder_cls(
             dataset=Dataset(self.dataset.value),
             query=query,
-            selected_columns=[self.aggregate],
+            selected_columns=[selected_aggregate],
             params=params,
             offset=None,
             limit=None,

src/sentry/snuba/snuba_query_validator.py

@@ -108,6 +108,18 @@ def __init__(self, *args, timeWindowSeconds=False, **kwargs):
         # TODO: only accept time_window in seconds once AlertRuleSerializer is removed
         self.time_window_seconds = timeWindowSeconds
 
+    def validate_aggregate(self, aggregate: str) -> str:
+        """
+        Reject upsampled_count() as user input. This function is reserved for internal use
+        and will be applied automatically when appropriate. Users should specify count().
+        """
+        if aggregate == "upsampled_count()":
+            raise serializers.ValidationError(
+                "upsampled_count() is not allowed as user input. Use count() instead - "
+                "it will be automatically converted to upsampled_count() when appropriate."
+            )
+        return aggregate
+
     def validate_query_type(self, value: int) -> SnubaQuery.Type:
         try:
             return SnubaQuery.Type(value)

tests/sentry/snuba/test_entity_subscriptions.py

@@ -391,6 +391,49 @@ def test_get_entity_subscription_for_events_dataset_with_join(self) -> None:
             Condition(Column("project_id", entity=g_entity), Op.IN, [self.project.id]),
         ]
 
+    def test_events_subscription_count_upsampling_toggle(self) -> None:
+        project = self.create_project(organization=self.organization)
+        sub = get_entity_subscription(
+            query_type=SnubaQuery.Type.ERROR,
+            dataset=Dataset.Events,
+            aggregate="count()",
+            time_window=60,
+        )
+
+        # Not allowlisted → expect plain count and no sample_weight
+        with self.options({"issues.client_error_sampling.project_allowlist": []}):
+            qb = sub.build_query_builder(query="", project_ids=[project.id], environment=None)
+            req = qb.get_snql_query()
+            assert len(req.query.select) == 1
+            func = req.query.select[0]
+            assert getattr(func, "alias", None) == "count"
+
+        # Allowlisted → expect full upsampled function structure
+        with self.options({"issues.client_error_sampling.project_allowlist": [project.id]}):
+            qb = sub.build_query_builder(query="", project_ids=[project.id], environment=None)
+            req = qb.get_snql_query()
+            assert len(req.query.select) == 1
+            func = req.query.select[0]
+            assert getattr(func, "alias", None) == "upsampled_count"
+            # Expect: toInt64(sum(ifNull(sample_weight, 1))) structure
+            assert isinstance(func, Function)
+            assert func.function == "toInt64"
+            assert len(func.parameters) == 1
+
+            outer_sum = func.parameters[0]
+            assert isinstance(outer_sum, Function)
+            assert outer_sum.function == "sum"
+            assert len(outer_sum.parameters) == 1
+
+            inner_ifnull = outer_sum.parameters[0]
+            assert isinstance(inner_ifnull, Function)
+            assert inner_ifnull.function == "ifNull"
+            assert len(inner_ifnull.parameters) == 2
+
+            weight_col = inner_ifnull.parameters[0]
+            assert isinstance(weight_col, Column)
+            assert weight_col.name == "sample_weight"
+
     def test_get_entity_subscription_for_eap_rpc_query(self) -> None:
         aggregate = "count(span.duration)"
         query = "span.op:http.client"

tests/sentry/workflow_engine/endpoints/test_organization_detector_index.py

@@ -652,6 +652,18 @@ def setUp(self) -> None:
             "workflowIds": [self.connected_workflow.id],
         }
 
+    def test_reject_upsampled_count_aggregate(self) -> None:
+        """Users should not be able to submit upsampled_count() directly in ACI."""
+        data = {**self.valid_data}
+        data["dataSource"] = {**self.valid_data["dataSource"], "aggregate": "upsampled_count()"}
+
+        response = self.get_error_response(
+            self.organization.slug,
+            **data,
+            status_code=400,
+        )
+        assert "upsampled_count() is not allowed as user input" in str(response.data)
+
     def test_missing_group_type(self) -> None:
         data = {**self.valid_data}
         del data["type"]