From 7aaf391abe6b14a339b9c6c6fdbb3395298f0ade Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Thu, 3 Sep 2026 16:32:06 +0530 Subject: [PATCH 1/4] feat(experimentation): add conversion-over-time rows to experiment results --- api/experimentation/dataclasses.py | 35 +- api/experimentation/results_query.py | 68 +++- api/experimentation/services.py | 145 ++++++-- api/tests/unit/experimentation/test_models.py | 2 + .../unit/experimentation/test_services.py | 310 +++++++++++++++++- .../observability/_events-catalogue.md | 28 +- 6 files changed, 546 insertions(+), 42 deletions(-) diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 0c780a1529f8..699a27cc5c6b 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -1,4 +1,4 @@ -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime from core.dataclasses import AuthorData @@ -56,6 +56,26 @@ class ExposuresSummary: timeseries: ExposuresTimeseries +@dataclass(frozen=True) +class ConversionBucket: + metric_id: int + variant: str + bucket: datetime + converted_identities: int + + +@dataclass(frozen=True) +class ConversionsTimeseriesPoint: + bucket: str + converted_identities: dict[str, int] + + +@dataclass(frozen=True) +class ConversionsTimeseries: + granularity: ExposureGranularity + points: list[ConversionsTimeseriesPoint] + + @dataclass(frozen=True) class MetricSpec: metric_id: int @@ -68,11 +88,17 @@ class MetricSpec: class ResultsAggregates: """Sufficient statistics gathered from the warehouse for one experiment: the specs they were computed from, per-variant identity counts, and per - metric the per-variant ``VariantStats``. Bundled so the keys can't drift.""" + metric the per-variant ``VariantStats``. Bundled so the keys can't drift. + + The bucket rows feed the over-time charts and are left empty by callers + that only need the headline statistics.""" specs: list[MetricSpec] exposure_counts: dict[str, int] metric_stats: dict[int, dict[str, VariantStats]] + granularity: ExposureGranularity | None = None + exposure_buckets: list[ExposureBucket] = field(default_factory=list) + conversion_buckets: list[ConversionBucket] = field(default_factory=list) @dataclass(frozen=True) @@ -80,12 +106,17 @@ class MetricResult: metric_id: int variants: dict[str, VariantStats] inference: dict[str, Inference | None] + # Only occurrence metrics chart a conversion rate; None for the rest. + timeseries: ConversionsTimeseries | None = None @dataclass(frozen=True) class ResultsSummary: srm_p_value: float | None metrics: list[MetricResult] + # Denominator for the conversion charts, computed in the same run as the + # metrics so both sides of the rate share one as_of. + exposures_timeseries: ExposuresTimeseries | None = None @dataclass(frozen=True) diff --git a/api/experimentation/results_query.py b/api/experimentation/results_query.py index df6d7a191d8d..50f3a5849bd0 100644 --- a/api/experimentation/results_query.py +++ b/api/experimentation/results_query.py @@ -17,7 +17,7 @@ from dataclasses import dataclass from typing import Any -from experimentation.dataclasses import MetricSpec +from experimentation.dataclasses import ConversionBucket, MetricSpec from experimentation.models import MetricAggregation from experimentation.stats import VariantStats @@ -104,6 +104,18 @@ def outer_select(self) -> str: a = self._alias return f"sum({a}) AS {a}_sum, sum({a} * {a}) AS {a}_sum_squares" + @property + def conversion_alias(self) -> str: + return f"c{self.index}" + + def first_conversion_select(self) -> str: + """Per-identity timestamp of the first post-exposure conversion, NULL + when the identity never converted. Same attribution condition as + unit_select, so bucket totals add up to the metric's ``sum``.""" + return ( + f"minIfOrNull(m.timestamp, {self._condition()}) AS {self.conversion_alias}" + ) + def decode(self, n: int, row: Sequence[Any], index: dict[str, int]) -> VariantStats: """Read this slot's two columns (sum, sum_squares) from a row by name.""" return VariantStats( @@ -144,6 +156,46 @@ def build_query(self) -> str: GROUP BY variant""" ) + def build_conversions_query(self, *, bucket_function: str) -> str | None: + """Per variant and occurrence metric, how many identities first + converted in each time bucket. None when no occurrence metric is + attached, since there is nothing to chart.""" + slots = [ + s for s in self._slots if s.spec.aggregation == MetricAggregation.OCCURRENCE + ] + if not slots: + return None + + first_conversion_selects = ",\n ".join( + s.first_conversion_select() for s in slots + ) + indexes = ", ".join(str(s.index) for s in slots) + aliases = ", ".join(s.conversion_alias for s in slots) + + return ( + _EXPOSURES_CTE + + f""", +first_conversions AS ( + SELECT + e.variant AS variant, + {first_conversion_selects} + FROM exposures AS e +{_METRIC_JOIN} + WHERE e.quarantined = 0 + GROUP BY e.identifier, e.variant +) +SELECT + variant, + metric_index, + {bucket_function}(first_conversion, 'UTC') AS bucket, + count() AS converted_identities +FROM first_conversions +ARRAY JOIN [{indexes}] AS metric_index, [{aliases}] AS first_conversion +WHERE first_conversion IS NOT NULL +GROUP BY variant, metric_index, bucket +ORDER BY bucket""" + ) + def add_metric_params(self, params: dict[str, object]) -> None: """Add per-metric query parameters into an existing params dict.""" if not self._slots: @@ -152,6 +204,20 @@ def add_metric_params(self, params: dict[str, object]) -> None: for slot in self._slots: params[f"metric_{slot.index}_event"] = slot.spec.event + def decode_conversion_rows( + self, rows: Sequence[Sequence[Any]] + ) -> list[ConversionBucket]: + """Map conversions-query rows back to the metric behind each slot index.""" + return [ + ConversionBucket( + metric_id=self._slots[int(metric_index)].spec.metric_id, + variant=str(variant), + bucket=bucket, + converted_identities=int(converted_identities), + ) + for variant, metric_index, bucket, converted_identities in rows + ] + def decode_rows( self, rows: list[Any], column_names: Sequence[str] ) -> tuple[dict[str, int], dict[int, dict[str, VariantStats]]]: diff --git a/api/experimentation/services.py b/api/experimentation/services.py index 8496f9c3feaa..e339f47a92d8 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -35,6 +35,9 @@ WAREHOUSE_CONNECTION_FLAG, ) from experimentation.dataclasses import ( + ConversionBucket, + ConversionsTimeseries, + ConversionsTimeseriesPoint, ExposureBucket, ExposuresSummary, ExposuresTimeseries, @@ -347,27 +350,61 @@ def build_exposures_summary( excluded_identities=sum( b.first_exposed_identities for b in buckets if b.quarantined ), - timeseries=ExposuresTimeseries( - granularity=granularity, - points=_timeseries_points([b for b in buckets if not b.quarantined]), - ), + timeseries=_exposures_timeseries(buckets, granularity=granularity), ) -def _timeseries_points( +def _exposures_timeseries( buckets: Sequence[ExposureBucket], -) -> list[ExposuresTimeseriesPoint]: - new_identities_by_bucket: dict[datetime, dict[str, int]] = {} - for b in buckets: - new_identities_by_bucket.setdefault(b.bucket, {})[b.variant] = ( - b.first_exposed_identities - ) + *, + granularity: ExposureGranularity, +) -> ExposuresTimeseries: + return ExposuresTimeseries( + granularity=granularity, + points=[ + ExposuresTimeseriesPoint(bucket=bucket, new_identities=counts) + for bucket, counts in _counts_by_bucket( + (b.bucket, b.variant, b.first_exposed_identities) + for b in buckets + if not b.quarantined + ) + ], + ) + + +def _conversions_timeseries( + spec: MetricSpec, + aggregates: ResultsAggregates, +) -> ConversionsTimeseries | None: + if ( + aggregates.granularity is None + or spec.aggregation != MetricAggregation.OCCURRENCE + ): + return None + return ConversionsTimeseries( + granularity=aggregates.granularity, + points=[ + ConversionsTimeseriesPoint(bucket=bucket, converted_identities=counts) + for bucket, counts in _counts_by_bucket( + (b.bucket, b.variant, b.converted_identities) + for b in aggregates.conversion_buckets + if b.metric_id == spec.metric_id + ) + ], + ) + + +def _counts_by_bucket( + rows: typing.Iterable[tuple[datetime, str, int]], +) -> list[tuple[str, dict[str, int]]]: + """Group (bucket, variant, count) rows into one per-variant dict per bucket, + keyed by the bucket's ISO timestamp, in bucket order.""" + counts_by_bucket: dict[datetime, dict[str, int]] = {} + for bucket, variant, count in rows: + counts_by_bucket.setdefault(bucket, {})[variant] = count return [ - ExposuresTimeseriesPoint( - bucket=bucket_start.isoformat(), - new_identities=new_identities_by_bucket[bucket_start], - ) - for bucket_start in sorted(new_identities_by_bucket) + (bucket.isoformat(), counts_by_bucket[bucket]) + for bucket in sorted(counts_by_bucket) ] @@ -447,6 +484,37 @@ def get_metric_variant_stats( ) +def get_conversion_buckets( + *, + environment_key: str, + feature_name: str, + window_start: datetime, + window_end: datetime, + specs: Sequence[MetricSpec], + granularity: ExposureGranularity, +) -> list[ConversionBucket]: + """Per occurrence metric, variant and time bucket: identities whose first + post-exposure conversion landed in that bucket.""" + builder = ResultsQueryBuilder(specs) + query = builder.build_conversions_query( + bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] + ) + if query is None: + return [] + params: dict[str, object] = { + "environment_key": environment_key, + "exposure_event": EXPOSURE_EVENT_NAME, + "feature_name": feature_name, + "window_start": window_start, + "window_end": window_end, + } + builder.add_metric_params(params) + rows = _get_clickhouse_client( + send_receive_timeout=CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ).execute(query, params) + return builder.decode_conversion_rows(rows) + + def build_results_summary( aggregates: ResultsAggregates, *, @@ -470,9 +538,17 @@ def build_results_summary( inference=_metric_inference( spec, aggregates.metric_stats.get(spec.metric_id, {}) ), + timeseries=_conversions_timeseries(spec, aggregates), ) for spec in aggregates.specs ], + exposures_timeseries=( + _exposures_timeseries( + aggregates.exposure_buckets, granularity=aggregates.granularity + ) + if aggregates.granularity is not None + else None + ), ) @@ -482,15 +558,36 @@ def compute_results_summary( window_start: "datetime", window_end: "datetime", ) -> ResultsSummary: - """Gather an experiment's metric statistics from the warehouse and reduce - them to the stored results payload.""" + """Gather an experiment's metric statistics and chart rows from the + warehouse and reduce them to the stored results payload.""" specs = _experiment_metric_specs(experiment) - aggregates = get_metric_variant_stats( - environment_key=experiment.environment.api_key, - feature_name=experiment.feature.name, - window_start=window_start, - window_end=window_end, - specs=specs, + environment_key = experiment.environment.api_key + feature_name = experiment.feature.name + granularity = _select_exposure_granularity(window_start, window_end) + aggregates = replace( + get_metric_variant_stats( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + specs=specs, + ), + granularity=granularity, + exposure_buckets=get_exposure_buckets( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + granularity=granularity, + ), + conversion_buckets=get_conversion_buckets( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + specs=specs, + granularity=granularity, + ), ) return build_results_summary( aggregates, diff --git a/api/tests/unit/experimentation/test_models.py b/api/tests/unit/experimentation/test_models.py index 006fe325733d..db9115f9ba86 100644 --- a/api/tests/unit/experimentation/test_models.py +++ b/api/tests/unit/experimentation/test_models.py @@ -198,8 +198,10 @@ def test_experiment_results__record_refresh__stores_payload_and_clears_error( "control": {"n": 1000, "sum": 100.0, "sum_squares": 100.0} }, "inference": {}, + "timeseries": None, } ], + "exposures_timeseries": None, } assert results.as_of == as_of assert results.last_error_at is None diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index e318751dd05e..1daa7e8a004b 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -16,6 +16,9 @@ from environments.models import Environment from experimentation import services from experimentation.dataclasses import ( + ConversionBucket, + ConversionsTimeseries, + ConversionsTimeseriesPoint, ExposureBucket, ExposuresSummary, ExposuresTimeseries, @@ -38,7 +41,7 @@ WarehouseConnectionStatus, WarehouseType, ) -from experimentation.results_query import _MetricSlot +from experimentation.results_query import ResultsQueryBuilder, _MetricSlot from experimentation.services import ( annotate_warehouse_event_stats, verify_clickhouse_connection, @@ -1061,6 +1064,160 @@ def test_metric_slot_unit_select__unknown_aggregation__raises() -> None: _MetricSlot(spec=_spec(aggregation="median"), index=0).unit_select() +def test_build_conversions_query__mixed_slots__occurrence_slots_only() -> None: + # Given an occurrence metric, a sum metric and a second occurrence metric + builder = ResultsQueryBuilder( + [ + _spec(metric_id=7, event="purchase", aggregation="occurrence"), + _spec(metric_id=9, event="revenue", aggregation="sum"), + _spec(metric_id=11, event="signup", aggregation="occurrence"), + ] + ) + + # When + sql = builder.build_conversions_query(bucket_function="toStartOfDay") + + # Then each occurrence slot records the identity's first post-exposure + # conversion, with the same attribution condition as the results query + assert sql is not None + assert ( + "minIfOrNull(m.timestamp, m.event = %(metric_0_event)s" + " AND m.timestamp >= e.first_exposure) AS c0" in sql + ) + assert ( + "minIfOrNull(m.timestamp, m.event = %(metric_2_event)s" + " AND m.timestamp >= e.first_exposure) AS c2" in sql + ) + assert " AS c1" not in sql + # And conversions are counted per slot per UTC bucket, skipping identities + # that never converted and identities seen in more than one variant + assert "ARRAY JOIN [0, 2] AS metric_index, [c0, c2] AS first_conversion" in sql + assert "toStartOfDay(first_conversion, 'UTC') AS bucket" in sql + assert "WHERE first_conversion IS NOT NULL" in sql + assert "GROUP BY variant, metric_index, bucket" in sql + assert "WHERE e.quarantined = 0" in sql + + +def test_build_conversions_query__no_occurrence_slots__returns_none() -> None: + # Given only value metrics, which have no conversion rate to chart + builder = ResultsQueryBuilder( + [ + _spec(metric_id=9, event="revenue", aggregation="sum"), + _spec(metric_id=13, event="session", aggregation="mean"), + ] + ) + + # When / Then + assert builder.build_conversions_query(bucket_function="toStartOfDay") is None + + +def test_decode_conversion_rows__rows__maps_slot_index_to_metric_id() -> None: + # Given occurrence slots at index 0 and 2, and one warehouse row for each + builder = ResultsQueryBuilder( + [ + _spec(metric_id=7, event="purchase", aggregation="occurrence"), + _spec(metric_id=9, event="revenue", aggregation="sum"), + _spec(metric_id=11, event="signup", aggregation="occurrence"), + ] + ) + bucket = datetime(2026, 6, 1, tzinfo=timezone.utc) + rows = [("control", 0, bucket, 12), ("variant_a", 2, bucket, 3)] + + # When + buckets = builder.decode_conversion_rows(rows) + + # Then each row is attributed to the metric behind its slot index + assert buckets == [ + ConversionBucket( + metric_id=7, variant="control", bucket=bucket, converted_identities=12 + ), + ConversionBucket( + metric_id=11, variant="variant_a", bucket=bucket, converted_identities=3 + ), + ] + + +def test_get_conversion_buckets__day_granularity__queries_and_maps_rows( + mocker: MockerFixture, +) -> None: + # Given the warehouse returns one conversion row per variant for the + # occurrence metric in slot 0 + bucket = datetime(2026, 6, 1, tzinfo=timezone.utc) + mock_client = mocker.Mock() + mock_client.execute.return_value = [ + ("control", 0, bucket, 12), + ("variant_a", 0, bucket, 15), + ] + mock_get_client = mocker.patch( + "experimentation.services._get_clickhouse_client", + return_value=mock_client, + ) + specs = [ + _spec(metric_id=7, event="purchase", aggregation="occurrence"), + _spec(metric_id=9, event="revenue", aggregation="sum"), + ] + window_start = datetime(2026, 6, 1, tzinfo=timezone.utc) + window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) + + # When + result = services.get_conversion_buckets( + environment_key="env-key-123", + feature_name="my-feature", + window_start=window_start, + window_end=window_end, + specs=specs, + granularity="day", + ) + + # Then the rows are mapped to dataclasses + assert result == [ + ConversionBucket( + metric_id=7, variant="control", bucket=bucket, converted_identities=12 + ), + ConversionBucket( + metric_id=7, variant="variant_a", bucket=bucket, converted_identities=15 + ), + ] + # And the query buckets first conversions by UTC day, over the same window + # and metric events as the results query + sql, params = mock_client.execute.call_args.args + assert "toStartOfDay(first_conversion, 'UTC') AS bucket" in sql + assert params == { + "environment_key": "env-key-123", + "exposure_event": "$flag_exposure", + "feature_name": "my-feature", + "window_start": window_start, + "window_end": window_end, + "metric_events": ["purchase", "revenue"], + "metric_0_event": "purchase", + "metric_1_event": "revenue", + } + mock_get_client.assert_called_once_with( + send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, + ) + + +def test_get_conversion_buckets__no_occurrence_metrics__skips_query( + mocker: MockerFixture, +) -> None: + # Given only value metrics are attached + mock_get_client = mocker.patch("experimentation.services._get_clickhouse_client") + + # When + result = services.get_conversion_buckets( + environment_key="env-key-123", + feature_name="my-feature", + window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), + window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), + specs=[_spec(metric_id=9, event="revenue", aggregation="sum")], + granularity="day", + ) + + # Then nothing is charted and the warehouse is not queried + assert result == [] + mock_get_client.assert_not_called() + + def test_build_results_summary__healthy_arms__infers_each_treatment() -> None: # Given a 10% control and a 12% treatment, both well above the floor control = VariantStats(n=1000, sum=100.0, sum_squares=100.0) @@ -1261,6 +1418,110 @@ def test_build_results_summary__computed__serialises_to_wire_shape() -> None: "ci_high", "chance_to_win", } + # And a run without bucket rows still carries the chart keys, as null + assert payload["exposures_timeseries"] is None + assert payload["metrics"][0]["timeseries"] is None + + +def test_build_results_summary__exposure_rows__attaches_exposures_timeseries() -> None: + # Given exposure bucket rows arriving unordered, with a quarantined row + day_1 = datetime(2026, 6, 1, tzinfo=timezone.utc) + day_3 = datetime(2026, 6, 3, tzinfo=timezone.utc) + aggregates = ResultsAggregates( + specs=[], + exposure_counts={"control": 1000, "variant_a": 1000}, + metric_stats={}, + granularity="day", + exposure_buckets=[ + ExposureBucket("control", day_3, first_exposed_identities=400), + ExposureBucket("control", day_1, first_exposed_identities=600), + ExposureBucket("variant_a", day_1, first_exposed_identities=1000), + ExposureBucket("", day_1, first_exposed_identities=5, quarantined=True), + ], + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then the chart denominator is bucketed in order, quarantined identities + # left out + assert summary.exposures_timeseries == ExposuresTimeseries( + granularity="day", + points=[ + ExposuresTimeseriesPoint( + bucket=day_1.isoformat(), + new_identities={"control": 600, "variant_a": 1000}, + ), + ExposuresTimeseriesPoint( + bucket=day_3.isoformat(), + new_identities={"control": 400}, + ), + ], + ) + + +def test_build_results_summary__occurrence_metrics__attach_conversions() -> None: + # Given conversion rows for two occurrence metrics, arriving unordered + day_1 = datetime(2026, 6, 1, tzinfo=timezone.utc) + day_3 = datetime(2026, 6, 3, tzinfo=timezone.utc) + aggregates = ResultsAggregates( + specs=[ + _spec(metric_id=7, event="purchase"), + _spec(metric_id=11, event="signup"), + ], + exposure_counts={"control": 1000, "variant_a": 1000}, + metric_stats={}, + granularity="day", + conversion_buckets=[ + ConversionBucket(7, "variant_a", day_3, converted_identities=20), + ConversionBucket(7, "control", day_1, converted_identities=100), + ConversionBucket(7, "variant_a", day_1, converted_identities=100), + ConversionBucket(11, "control", day_1, converted_identities=7), + ], + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then each metric gets only its own rows, per variant, in bucket order + assert summary.metrics[0].timeseries == ConversionsTimeseries( + granularity="day", + points=[ + ConversionsTimeseriesPoint( + bucket=day_1.isoformat(), + converted_identities={"control": 100, "variant_a": 100}, + ), + ConversionsTimeseriesPoint( + bucket=day_3.isoformat(), + converted_identities={"variant_a": 20}, + ), + ], + ) + assert summary.metrics[1].timeseries == ConversionsTimeseries( + granularity="day", + points=[ + ConversionsTimeseriesPoint( + bucket=day_1.isoformat(), + converted_identities={"control": 7}, + ), + ], + ) + + +def test_build_results_summary__value_metric__timeseries_none() -> None: + # Given a sum metric in a run that gathered bucket rows + aggregates = ResultsAggregates( + specs=[_spec(metric_id=9, event="revenue", aggregation="sum")], + exposure_counts={"control": 1000}, + metric_stats={}, + granularity="day", + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then a value metric has no conversion rate to chart + assert summary.metrics[0].timeseries is None @pytest.mark.django_db @@ -1500,6 +1761,20 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( ) window_start = datetime(2026, 6, 1, tzinfo=timezone.utc) window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) + mock_exposures = mocker.patch( + "experimentation.services.get_exposure_buckets", + return_value=[ + ExposureBucket("control", window_start, first_exposed_identities=1000) + ], + ) + mock_conversions = mocker.patch( + "experimentation.services.get_conversion_buckets", + return_value=[ + ConversionBucket( + metric.id, "control", window_start, converted_identities=100 + ) + ], + ) # When summary = services.compute_results_summary( @@ -1521,6 +1796,39 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( assert summary.srm_p_value == pytest.approx(1.0) assert summary.metrics[0].metric_id == metric.id assert summary.metrics[0].inference["variant_a"] is not None + # And the chart rows are gathered over the same window, bucketed by day + # because the window is longer than 72 hours + mock_exposures.assert_called_once_with( + environment_key=environment.api_key, + feature_name=feature.name, + window_start=window_start, + window_end=window_end, + granularity="day", + ) + mock_conversions.assert_called_once_with( + environment_key=environment.api_key, + feature_name=feature.name, + window_start=window_start, + window_end=window_end, + specs=expected_specs, + granularity="day", + ) + assert summary.exposures_timeseries == ExposuresTimeseries( + granularity="day", + points=[ + ExposuresTimeseriesPoint( + bucket=window_start.isoformat(), new_identities={"control": 1000} + ) + ], + ) + assert summary.metrics[0].timeseries == ConversionsTimeseries( + granularity="day", + points=[ + ConversionsTimeseriesPoint( + bucket=window_start.isoformat(), converted_identities={"control": 100} + ) + ], + ) def test_apply_experiment_rollout__no_segment__creates_segment_and_override( diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 6045c4df06ce..2cd515ba005e 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -838,7 +838,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1190` + - `api/experimentation/services.py:1287` Attributes: - `environment.id` @@ -847,8 +847,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:267` - - `api/experimentation/services.py:1290` + - `api/experimentation/services.py:270` + - `api/experimentation/services.py:1387` Attributes: - `environment.id` @@ -858,7 +858,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1253` + - `api/experimentation/services.py:1350` Attributes: - `environment.id` @@ -867,7 +867,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:964` + - `api/experimentation/services.py:1061` Attributes: - `environment.id` @@ -876,7 +876,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1165` + - `api/experimentation/services.py:1262` Attributes: - `environment.id` @@ -886,7 +886,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1175` + - `api/experimentation/services.py:1272` Attributes: - `environment.id` @@ -895,7 +895,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1120` + - `api/experimentation/services.py:1217` Attributes: - `connection.id` @@ -906,7 +906,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:1009` + - `api/experimentation/services.py:1106` Attributes: - `connection.id` @@ -917,7 +917,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1130` + - `api/experimentation/services.py:1227` Attributes: - `connection.id` @@ -930,7 +930,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:1103` + - `api/experimentation/services.py:1200` Attributes: - `connection.id` @@ -941,7 +941,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:1038` + - `api/experimentation/services.py:1135` Attributes: - `connection.id` @@ -953,7 +953,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:560` + - `api/experimentation/services.py:657` Attributes: - `environment.id` @@ -963,7 +963,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:546` + - `api/experimentation/services.py:643` Attributes: - `environment.id` From 39414ddab8553ae824475c49aab2fa9a4da26ec7 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Thu, 3 Sep 2026 17:11:49 +0530 Subject: [PATCH 2/4] refactor(experimentation): gather results aggregates in one pass, key conversions by metric --- api/experimentation/dataclasses.py | 24 +- api/experimentation/results_query.py | 111 ++++++-- api/experimentation/services.py | 170 +++++------- api/tests/unit/experimentation/test_models.py | 6 +- .../unit/experimentation/test_services.py | 258 ++++++++++-------- api/tests/unit/experimentation/test_tasks.py | 2 + .../observability/_events-catalogue.md | 28 +- 7 files changed, 334 insertions(+), 265 deletions(-) diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 699a27cc5c6b..0a7bc3e96b87 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -1,4 +1,4 @@ -from dataclasses import dataclass, field +from dataclasses import dataclass from datetime import datetime from core.dataclasses import AuthorData @@ -58,7 +58,6 @@ class ExposuresSummary: @dataclass(frozen=True) class ConversionBucket: - metric_id: int variant: str bucket: datetime converted_identities: int @@ -86,19 +85,18 @@ class MetricSpec: @dataclass(frozen=True) class ResultsAggregates: - """Sufficient statistics gathered from the warehouse for one experiment: - the specs they were computed from, per-variant identity counts, and per - metric the per-variant ``VariantStats``. Bundled so the keys can't drift. - - The bucket rows feed the over-time charts and are left empty by callers - that only need the headline statistics.""" + """Everything one results refresh reads from the warehouse: the specs it + was computed from, per-variant identity counts, per metric the per-variant + ``VariantStats``, and the time-bucketed rows behind the over-time charts. + Bundled so the keys can't drift.""" specs: list[MetricSpec] exposure_counts: dict[str, int] metric_stats: dict[int, dict[str, VariantStats]] - granularity: ExposureGranularity | None = None - exposure_buckets: list[ExposureBucket] = field(default_factory=list) - conversion_buckets: list[ConversionBucket] = field(default_factory=list) + granularity: ExposureGranularity + exposure_buckets: list[ExposureBucket] + # Keyed by metric id, one entry per charted metric. + conversion_buckets: dict[int, list[ConversionBucket]] @dataclass(frozen=True) @@ -107,7 +105,7 @@ class MetricResult: variants: dict[str, VariantStats] inference: dict[str, Inference | None] # Only occurrence metrics chart a conversion rate; None for the rest. - timeseries: ConversionsTimeseries | None = None + conversions_timeseries: ConversionsTimeseries | None @dataclass(frozen=True) @@ -116,7 +114,7 @@ class ResultsSummary: metrics: list[MetricResult] # Denominator for the conversion charts, computed in the same run as the # metrics so both sides of the rate share one as_of. - exposures_timeseries: ExposuresTimeseries | None = None + exposures_timeseries: ExposuresTimeseries @dataclass(frozen=True) diff --git a/api/experimentation/results_query.py b/api/experimentation/results_query.py index 50f3a5849bd0..430dd99da4dc 100644 --- a/api/experimentation/results_query.py +++ b/api/experimentation/results_query.py @@ -15,8 +15,10 @@ from collections.abc import Sequence from dataclasses import dataclass +from datetime import datetime from typing import Any +from experimentation.constants import EXPOSURE_EVENT_NAME from experimentation.dataclasses import ConversionBucket, MetricSpec from experimentation.models import MetricAggregation from experimentation.stats import VariantStats @@ -24,7 +26,7 @@ _FLOAT_VALUE = "toFloat64OrZero(m.value)" # Events are delivered at-least-once, so dedup keeps duplicates from inflating -# counts. Shared by the exposures and results queries. +# counts. Shared by the exposures, results and conversions queries. _EXPOSURES_CTE = """ WITH exposures AS ( SELECT @@ -50,10 +52,33 @@ GROUP BY variant""" ) -_METRIC_JOIN = """ LEFT JOIN events AS m + +def exposure_window_params( + *, + environment_key: str, + feature_name: str, + window_start: datetime, + window_end: datetime, +) -> dict[str, object]: + """The parameters ``_EXPOSURES_CTE`` binds, for every query that starts + from it.""" + return { + "environment_key": environment_key, + "exposure_event": EXPOSURE_EVENT_NAME, + "feature_name": feature_name, + "window_start": window_start, + "window_end": window_end, + } + + +def _metric_join(events_param: str) -> str: + """Join each exposed identity to its metric events. ``events_param`` names + the bound list of event names, so a query can join only the events it + aggregates.""" + return f""" LEFT JOIN events AS m ON m.identifier = e.identifier AND m.environment_key = %(environment_key)s - AND m.event IN %(metric_events)s + AND m.event IN %({events_param})s AND m.timestamp >= %(window_start)s AND m.timestamp < %(window_end)s""" @@ -108,10 +133,15 @@ def outer_select(self) -> str: def conversion_alias(self) -> str: return f"c{self.index}" - def first_conversion_select(self) -> str: + def first_conversion_select(self) -> str | None: """Per-identity timestamp of the first post-exposure conversion, NULL when the identity never converted. Same attribution condition as - unit_select, so bucket totals add up to the metric's ``sum``.""" + unit_select, so bucket totals add up to the metric's ``sum``. + + None for value metrics: a count or sum accrues per event rather than + once per identity, so a first-conversion timestamp can't chart it.""" + if self.spec.aggregation != MetricAggregation.OCCURRENCE: + return None return ( f"minIfOrNull(m.timestamp, {self._condition()}) AS {self.conversion_alias}" ) @@ -146,7 +176,7 @@ def build_query(self) -> str: e.variant AS variant, {unit_selects} FROM exposures AS e -{_METRIC_JOIN} +{_metric_join("metric_events")} WHERE e.quarantined = 0 GROUP BY e.identifier, e.variant ) @@ -157,17 +187,15 @@ def build_query(self) -> str: ) def build_conversions_query(self, *, bucket_function: str) -> str | None: - """Per variant and occurrence metric, how many identities first - converted in each time bucket. None when no occurrence metric is - attached, since there is nothing to chart.""" - slots = [ - s for s in self._slots if s.spec.aggregation == MetricAggregation.OCCURRENCE - ] + """Per variant and charted metric, how many identities first converted + in each time bucket. None when no attached metric charts, since there + is nothing to query.""" + slots = self._charted_slots if not slots: return None first_conversion_selects = ",\n ".join( - s.first_conversion_select() for s in slots + select for s in slots if (select := s.first_conversion_select()) ) indexes = ", ".join(str(s.index) for s in slots) aliases = ", ".join(s.conversion_alias for s in slots) @@ -180,7 +208,7 @@ def build_conversions_query(self, *, bucket_function: str) -> str | None: e.variant AS variant, {first_conversion_selects} FROM exposures AS e -{_METRIC_JOIN} +{_metric_join("conversion_events")} WHERE e.quarantined = 0 GROUP BY e.identifier, e.variant ) @@ -196,27 +224,54 @@ def build_conversions_query(self, *, bucket_function: str) -> str | None: ORDER BY bucket""" ) - def add_metric_params(self, params: dict[str, object]) -> None: - """Add per-metric query parameters into an existing params dict.""" + @property + def _charted_slots(self) -> list[_MetricSlot]: + return [s for s in self._slots if s.first_conversion_select() is not None] + + def params( + self, + *, + environment_key: str, + feature_name: str, + window_start: datetime, + window_end: datetime, + ) -> dict[str, object]: + """Every parameter the results and conversions queries bind: the + exposure window, each metric's event, and the event lists each join + narrows to.""" + params = exposure_window_params( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + ) if not self._slots: - return + return params params["metric_events"] = [s.spec.event for s in self._slots] + params["conversion_events"] = [s.spec.event for s in self._charted_slots] for slot in self._slots: params[f"metric_{slot.index}_event"] = slot.spec.event + return params def decode_conversion_rows( - self, rows: Sequence[Sequence[Any]] - ) -> list[ConversionBucket]: - """Map conversions-query rows back to the metric behind each slot index.""" - return [ - ConversionBucket( - metric_id=self._slots[int(metric_index)].spec.metric_id, - variant=str(variant), - bucket=bucket, - converted_identities=int(converted_identities), + self, rows: Sequence[Sequence[Any]], column_names: Sequence[str] + ) -> dict[int, list[ConversionBucket]]: + """Group conversions-query rows by the metric behind each slot index. + Every charted metric gets a key, empty when nobody converted yet.""" + index = {name: position for position, name in enumerate(column_names)} + buckets: dict[int, list[ConversionBucket]] = { + slot.spec.metric_id: [] for slot in self._charted_slots + } + for row in rows: + metric_id = self._slots[int(row[index["metric_index"]])].spec.metric_id + buckets[metric_id].append( + ConversionBucket( + variant=str(row[index["variant"]]), + bucket=row[index["bucket"]], + converted_identities=int(row[index["converted_identities"]]), + ) ) - for variant, metric_index, bucket, converted_identities in rows - ] + return buckets def decode_rows( self, rows: list[Any], column_names: Sequence[str] diff --git a/api/experimentation/services.py b/api/experimentation/services.py index e339f47a92d8..058d7c6ea317 100644 --- a/api/experimentation/services.py +++ b/api/experimentation/services.py @@ -27,7 +27,6 @@ from experimentation.constants import ( CONTROL_VARIANT_KEY, EXPERIMENT_FLAG, - EXPOSURE_EVENT_NAME, EXPOSURE_HOURLY_BUCKET_MAX_WINDOW, RESULTS_MIN_CONVERSIONS_PER_VARIANT, RESULTS_MIN_IDENTITIES_PER_VARIANT, @@ -67,7 +66,11 @@ WarehouseDeliveryOutcome, WarehouseType, ) -from experimentation.results_query import _EXPOSURES_CTE, ResultsQueryBuilder +from experimentation.results_query import ( + _EXPOSURES_CTE, + ResultsQueryBuilder, + exposure_window_params, +) from experimentation.stats import ( Inference, VariantStats, @@ -373,22 +376,18 @@ def _exposures_timeseries( def _conversions_timeseries( - spec: MetricSpec, + metric_id: int, aggregates: ResultsAggregates, ) -> ConversionsTimeseries | None: - if ( - aggregates.granularity is None - or spec.aggregation != MetricAggregation.OCCURRENCE - ): + buckets = aggregates.conversion_buckets.get(metric_id) + if buckets is None: return None return ConversionsTimeseries( granularity=aggregates.granularity, points=[ ConversionsTimeseriesPoint(bucket=bucket, converted_identities=counts) for bucket, counts in _counts_by_bucket( - (b.bucket, b.variant, b.converted_identities) - for b in aggregates.conversion_buckets - if b.metric_id == spec.metric_id + (b.bucket, b.variant, b.converted_identities) for b in buckets ) ], ) @@ -397,8 +396,9 @@ def _conversions_timeseries( def _counts_by_bucket( rows: typing.Iterable[tuple[datetime, str, int]], ) -> list[tuple[str, dict[str, int]]]: - """Group (bucket, variant, count) rows into one per-variant dict per bucket, - keyed by the bucket's ISO timestamp, in bucket order.""" + """Group rows into one per-variant dict per bucket, in bucket order. The + bucket becomes an ISO string because the summary lands in a JSONField as + is, and datetimes wouldn't serialise.""" counts_by_bucket: dict[datetime, dict[str, int]] = {} for bucket, variant, count in rows: counts_by_bucket.setdefault(bucket, {})[variant] = count @@ -431,13 +431,12 @@ def get_exposure_buckets( EXPOSURE_BUCKETS_QUERY.format( bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] ), - { - "environment_key": environment_key, - "exposure_event": EXPOSURE_EVENT_NAME, - "feature_name": feature_name, - "window_start": window_start, - "window_end": window_end, - }, + exposure_window_params( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + ), ) return [ ExposureBucket( @@ -450,71 +449,67 @@ def get_exposure_buckets( ] -def get_metric_variant_stats( +def get_results_aggregates( *, environment_key: str, feature_name: str, window_start: datetime, window_end: datetime, specs: Sequence[MetricSpec], + granularity: ExposureGranularity, ) -> ResultsAggregates: - """Run the warehouse query, returning per-variant identity counts and, per - metric, per-variant sufficient statistics.""" - builder = ResultsQueryBuilder(specs) - params: dict[str, object] = { - "environment_key": environment_key, - "exposure_event": EXPOSURE_EVENT_NAME, - "feature_name": feature_name, - "window_start": window_start, - "window_end": window_end, - } - builder.add_metric_params(params) + """Run the warehouse reads behind one results refresh: per-variant identity + counts and sufficient statistics, exposure buckets, and per charted metric + the buckets of first post-exposure conversions. - rows, columns = _get_clickhouse_client( + Three separate reads, so events landing mid-run can leave the chart's last + point a few identities off the table until the next refresh.""" + builder = ResultsQueryBuilder(specs) + params = builder.params( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + ) + client = _get_clickhouse_client( send_receive_timeout=CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, - ).execute(builder.build_query(), params, with_column_types=True) + ) + + rows, columns = client.execute( + builder.build_query(), params, with_column_types=True + ) exposure_counts, metric_stats = builder.decode_rows( rows, [name for name, _type in columns] ) + conversion_buckets: dict[int, list[ConversionBucket]] = {} + conversions_query = builder.build_conversions_query( + bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] + ) + if conversions_query is not None: + rows, columns = client.execute( + conversions_query, params, with_column_types=True + ) + conversion_buckets = builder.decode_conversion_rows( + rows, [name for name, _type in columns] + ) + return ResultsAggregates( specs=list(specs), exposure_counts=exposure_counts, metric_stats=metric_stats, + granularity=granularity, + exposure_buckets=get_exposure_buckets( + environment_key=environment_key, + feature_name=feature_name, + window_start=window_start, + window_end=window_end, + granularity=granularity, + ), + conversion_buckets=conversion_buckets, ) -def get_conversion_buckets( - *, - environment_key: str, - feature_name: str, - window_start: datetime, - window_end: datetime, - specs: Sequence[MetricSpec], - granularity: ExposureGranularity, -) -> list[ConversionBucket]: - """Per occurrence metric, variant and time bucket: identities whose first - post-exposure conversion landed in that bucket.""" - builder = ResultsQueryBuilder(specs) - query = builder.build_conversions_query( - bucket_function=_EXPOSURE_BUCKET_FUNCTIONS[granularity] - ) - if query is None: - return [] - params: dict[str, object] = { - "environment_key": environment_key, - "exposure_event": EXPOSURE_EVENT_NAME, - "feature_name": feature_name, - "window_start": window_start, - "window_end": window_end, - } - builder.add_metric_params(params) - rows = _get_clickhouse_client( - send_receive_timeout=CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, - ).execute(query, params) - return builder.decode_conversion_rows(rows) - - def build_results_summary( aggregates: ResultsAggregates, *, @@ -538,16 +533,14 @@ def build_results_summary( inference=_metric_inference( spec, aggregates.metric_stats.get(spec.metric_id, {}) ), - timeseries=_conversions_timeseries(spec, aggregates), + conversions_timeseries=_conversions_timeseries( + spec.metric_id, aggregates + ), ) for spec in aggregates.specs ], - exposures_timeseries=( - _exposures_timeseries( - aggregates.exposure_buckets, granularity=aggregates.granularity - ) - if aggregates.granularity is not None - else None + exposures_timeseries=_exposures_timeseries( + aggregates.exposure_buckets, granularity=aggregates.granularity ), ) @@ -560,34 +553,13 @@ def compute_results_summary( ) -> ResultsSummary: """Gather an experiment's metric statistics and chart rows from the warehouse and reduce them to the stored results payload.""" - specs = _experiment_metric_specs(experiment) - environment_key = experiment.environment.api_key - feature_name = experiment.feature.name - granularity = _select_exposure_granularity(window_start, window_end) - aggregates = replace( - get_metric_variant_stats( - environment_key=environment_key, - feature_name=feature_name, - window_start=window_start, - window_end=window_end, - specs=specs, - ), - granularity=granularity, - exposure_buckets=get_exposure_buckets( - environment_key=environment_key, - feature_name=feature_name, - window_start=window_start, - window_end=window_end, - granularity=granularity, - ), - conversion_buckets=get_conversion_buckets( - environment_key=environment_key, - feature_name=feature_name, - window_start=window_start, - window_end=window_end, - specs=specs, - granularity=granularity, - ), + aggregates = get_results_aggregates( + environment_key=experiment.environment.api_key, + feature_name=experiment.feature.name, + window_start=window_start, + window_end=window_end, + specs=_experiment_metric_specs(experiment), + granularity=_select_exposure_granularity(window_start, window_end), ) return build_results_summary( aggregates, diff --git a/api/tests/unit/experimentation/test_models.py b/api/tests/unit/experimentation/test_models.py index db9115f9ba86..805574ba2c10 100644 --- a/api/tests/unit/experimentation/test_models.py +++ b/api/tests/unit/experimentation/test_models.py @@ -169,8 +169,10 @@ def _results_summary() -> ResultsSummary: "control": VariantStats(n=1000, sum=100.0, sum_squares=100.0) }, inference={}, + conversions_timeseries=None, ) ], + exposures_timeseries=ExposuresTimeseries(granularity="day", points=[]), ) @@ -198,10 +200,10 @@ def test_experiment_results__record_refresh__stores_payload_and_clears_error( "control": {"n": 1000, "sum": 100.0, "sum_squares": 100.0} }, "inference": {}, - "timeseries": None, + "conversions_timeseries": None, } ], - "exposures_timeseries": None, + "exposures_timeseries": {"granularity": "day", "points": []}, } assert results.as_of == as_of assert results.last_error_at is None diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 1daa7e8a004b..79d4e4247f6c 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -1,4 +1,4 @@ -from dataclasses import asdict +from dataclasses import asdict, replace from datetime import datetime, timezone from unittest.mock import MagicMock @@ -799,6 +799,9 @@ def _aggregates( specs=specs, exposure_counts=exposure_counts, metric_stats=metric_stats, + granularity="day", + exposure_buckets=[], + conversion_buckets={}, ) @@ -812,6 +815,16 @@ def _result_columns(metric_count: int) -> list[tuple[str, str]]: return columns +def _conversion_columns() -> list[tuple[str, str]]: + """Column metadata for the conversions query, in SELECT order.""" + return [ + ("variant", "String"), + ("metric_index", "UInt8"), + ("bucket", "DateTime('UTC')"), + ("converted_identities", "UInt64"), + ] + + def test_get_metric_variant_stats__metrics__queries_and_maps_rows( mocker: MockerFixture, ) -> None: @@ -832,7 +845,11 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( ), ] mock_client = mocker.Mock() - mock_client.execute.return_value = (rows, _result_columns(4)) + mock_client.execute.side_effect = [ + (rows, _result_columns(4)), + ([], _conversion_columns()), + [], + ] mock_get_client = mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, @@ -847,12 +864,13 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=window_start, window_end=window_end, specs=specs, + granularity="day", ) # Then per-variant counts and sufficient statistics are mapped per metric @@ -869,8 +887,9 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( assert aggregates.metric_stats[13]["variant_a"] == VariantStats( n=1000, sum=210.0, sum_squares=520.0 ) - # And the query joins post-exposure metric events and excludes quarantined - sql, params = mock_client.execute.call_args.args + # And the results query joins post-exposure metric events and excludes + # quarantined identities + sql, params = mock_client.execute.call_args_list[0].args assert "LEFT JOIN events AS m" in sql assert "m.timestamp >= e.first_exposure" in sql assert "m.timestamp >= %(window_start)s" in sql @@ -900,7 +919,9 @@ def test_get_metric_variant_stats__metrics__queries_and_maps_rows( assert params["metric_2_event"] == "page_view" assert params["metric_3_event"] == "session" assert params["window_end"] == window_end - mock_get_client.assert_called_once_with( + # And the conversions join is narrowed to the occurrence metric's event + assert params["conversion_events"] == ["purchase"] + mock_get_client.assert_called_with( send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, ) @@ -915,7 +936,11 @@ def test_get_metric_variant_stats__three_variants__maps_all_variants( ("variant_b", 950, 110.0, 110.0, 5100.0, 29000.0), ] mock_client = mocker.Mock() - mock_client.execute.return_value = (rows, _result_columns(2)) + mock_client.execute.side_effect = [ + (rows, _result_columns(2)), + ([], _conversion_columns()), + [], + ] mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, @@ -926,12 +951,13 @@ def test_get_metric_variant_stats__three_variants__maps_all_variants( ] # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), specs=specs, + granularity="day", ) # Then all three variants are decoded into counts and metric stats @@ -951,28 +977,32 @@ def test_get_metric_variant_stats__no_metrics__counts_variants_only( ) -> None: # Given an experiment with no attached metrics mock_client = mocker.Mock() - mock_client.execute.return_value = ( - [("control", 1000), ("variant_a", 900)], - _result_columns(0), - ) + mock_client.execute.side_effect = [ + ([("control", 1000), ("variant_a", 900)], _result_columns(0)), + [], + ] mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, ) # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), specs=[], + granularity="day", ) - # Then only the per-variant counts are returned, with no metric join + # Then only the per-variant counts are returned, with no metric join, and + # no conversions query is run assert aggregates.exposure_counts == {"control": 1000, "variant_a": 900} assert aggregates.metric_stats == {} - sql, params = mock_client.execute.call_args.args + assert aggregates.conversion_buckets == {} + assert mock_client.execute.call_count == 2 + sql, params = mock_client.execute.call_args_list[0].args assert "SELECT variant, count() AS n" in sql assert "LEFT JOIN" not in sql assert "metric_events" not in params @@ -993,7 +1023,11 @@ def test_get_metric_variant_stats__shuffled_columns__maps_by_name( ] rows = [(30000.0, "control", 100.0, 1000, 5000.0, 100.0)] mock_client = mocker.Mock() - mock_client.execute.return_value = (rows, columns) + mock_client.execute.side_effect = [ + (rows, columns), + ([], _conversion_columns()), + [], + ] mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, @@ -1004,12 +1038,13 @@ def test_get_metric_variant_stats__shuffled_columns__maps_by_name( ] # When - aggregates = services.get_metric_variant_stats( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), window_end=datetime(2026, 6, 10, tzinfo=timezone.utc), specs=specs, + granularity="day", ) # Then values are decoded by column name, not position @@ -1096,6 +1131,8 @@ def test_build_conversions_query__mixed_slots__occurrence_slots_only() -> None: assert "WHERE first_conversion IS NOT NULL" in sql assert "GROUP BY variant, metric_index, bucket" in sql assert "WHERE e.quarantined = 0" in sql + # And the join is narrowed to the charted metrics' events + assert "AND m.event IN %(conversion_events)s" in sql def test_build_conversions_query__no_occurrence_slots__returns_none() -> None: @@ -1111,8 +1148,9 @@ def test_build_conversions_query__no_occurrence_slots__returns_none() -> None: assert builder.build_conversions_query(bucket_function="toStartOfDay") is None -def test_decode_conversion_rows__rows__maps_slot_index_to_metric_id() -> None: - # Given occurrence slots at index 0 and 2, and one warehouse row for each +def test_decode_conversion_rows__rows__groups_by_metric_behind_slot_index() -> None: + # Given occurrence slots at index 0 and 2, rows for slot 0 only, and + # columns in a different order than the SELECT builder = ResultsQueryBuilder( [ _spec(metric_id=7, event="purchase", aggregation="occurrence"), @@ -1121,34 +1159,39 @@ def test_decode_conversion_rows__rows__maps_slot_index_to_metric_id() -> None: ] ) bucket = datetime(2026, 6, 1, tzinfo=timezone.utc) - rows = [("control", 0, bucket, 12), ("variant_a", 2, bucket, 3)] + columns = ["converted_identities", "variant", "bucket", "metric_index"] + rows = [(12, "control", bucket, 0), (3, "variant_a", bucket, 0)] # When - buckets = builder.decode_conversion_rows(rows) + buckets = builder.decode_conversion_rows(rows, columns) - # Then each row is attributed to the metric behind its slot index - assert buckets == [ - ConversionBucket( - metric_id=7, variant="control", bucket=bucket, converted_identities=12 - ), - ConversionBucket( - metric_id=11, variant="variant_a", bucket=bucket, converted_identities=3 - ), - ] + # Then rows are decoded by column name under the metric behind their slot, + # and the charted metric nobody converted on still gets an empty entry + assert buckets == { + 7: [ + ConversionBucket("control", bucket, converted_identities=12), + ConversionBucket("variant_a", bucket, converted_identities=3), + ], + 11: [], + } -def test_get_conversion_buckets__day_granularity__queries_and_maps_rows( +def test_get_results_aggregates__occurrence_metric__gathers_chart_rows( mocker: MockerFixture, ) -> None: - # Given the warehouse returns one conversion row per variant for the - # occurrence metric in slot 0 + # Given the warehouse answers the results query, then one conversion row + # per variant for the occurrence metric in slot 0, then one exposure row bucket = datetime(2026, 6, 1, tzinfo=timezone.utc) mock_client = mocker.Mock() - mock_client.execute.return_value = [ - ("control", 0, bucket, 12), - ("variant_a", 0, bucket, 15), + mock_client.execute.side_effect = [ + ([("control", 1000, 12.0, 12.0, 500.0, 900.0)], _result_columns(2)), + ( + [("control", 0, bucket, 12), ("variant_a", 0, bucket, 15)], + _conversion_columns(), + ), + [(0, "control", bucket, 1000)], ] - mock_get_client = mocker.patch( + mocker.patch( "experimentation.services._get_clickhouse_client", return_value=mock_client, ) @@ -1160,7 +1203,7 @@ def test_get_conversion_buckets__day_granularity__queries_and_maps_rows( window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) # When - result = services.get_conversion_buckets( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=window_start, @@ -1169,18 +1212,20 @@ def test_get_conversion_buckets__day_granularity__queries_and_maps_rows( granularity="day", ) - # Then the rows are mapped to dataclasses - assert result == [ - ConversionBucket( - metric_id=7, variant="control", bucket=bucket, converted_identities=12 - ), - ConversionBucket( - metric_id=7, variant="variant_a", bucket=bucket, converted_identities=15 - ), + # Then the chart rows are mapped, conversions under the occurrence metric only + assert aggregates.granularity == "day" + assert aggregates.conversion_buckets == { + 7: [ + ConversionBucket("control", bucket, converted_identities=12), + ConversionBucket("variant_a", bucket, converted_identities=15), + ] + } + assert aggregates.exposure_buckets == [ + ExposureBucket("control", bucket, first_exposed_identities=1000) ] - # And the query buckets first conversions by UTC day, over the same window - # and metric events as the results query - sql, params = mock_client.execute.call_args.args + # And the conversions query buckets first conversions by UTC day, joining + # only the occurrence metric's events over the results query's window + sql, params = mock_client.execute.call_args_list[1].args assert "toStartOfDay(first_conversion, 'UTC') AS bucket" in sql assert params == { "environment_key": "env-key-123", @@ -1189,22 +1234,28 @@ def test_get_conversion_buckets__day_granularity__queries_and_maps_rows( "window_start": window_start, "window_end": window_end, "metric_events": ["purchase", "revenue"], + "conversion_events": ["purchase"], "metric_0_event": "purchase", "metric_1_event": "revenue", } - mock_get_client.assert_called_once_with( - send_receive_timeout=services.CLICKHOUSE_BACKGROUND_QUERY_TIMEOUT_SECONDS, - ) -def test_get_conversion_buckets__no_occurrence_metrics__skips_query( +def test_get_results_aggregates__value_metrics_only__skips_conversions_query( mocker: MockerFixture, ) -> None: - # Given only value metrics are attached - mock_get_client = mocker.patch("experimentation.services._get_clickhouse_client") + # Given only a value metric is attached + mock_client = mocker.Mock() + mock_client.execute.side_effect = [ + ([("control", 1000, 500.0, 900.0)], _result_columns(1)), + [], + ] + mocker.patch( + "experimentation.services._get_clickhouse_client", + return_value=mock_client, + ) # When - result = services.get_conversion_buckets( + aggregates = services.get_results_aggregates( environment_key="env-key-123", feature_name="my-feature", window_start=datetime(2026, 6, 1, tzinfo=timezone.utc), @@ -1213,9 +1264,11 @@ def test_get_conversion_buckets__no_occurrence_metrics__skips_query( granularity="day", ) - # Then nothing is charted and the warehouse is not queried - assert result == [] - mock_get_client.assert_not_called() + # Then nothing is charted and only the results and exposures queries run + assert aggregates.conversion_buckets == {} + assert mock_client.execute.call_count == 2 + _, params = mock_client.execute.call_args_list[0].args + assert params["conversion_events"] == [] def test_build_results_summary__healthy_arms__infers_each_treatment() -> None: @@ -1418,9 +1471,9 @@ def test_build_results_summary__computed__serialises_to_wire_shape() -> None: "ci_high", "chance_to_win", } - # And a run without bucket rows still carries the chart keys, as null - assert payload["exposures_timeseries"] is None - assert payload["metrics"][0]["timeseries"] is None + # And the chart series sit alongside, empty for a run with no bucket rows + assert payload["exposures_timeseries"] == {"granularity": "day", "points": []} + assert payload["metrics"][0]["conversions_timeseries"] is None def test_build_results_summary__exposure_rows__attaches_exposures_timeseries() -> None: @@ -1438,6 +1491,7 @@ def test_build_results_summary__exposure_rows__attaches_exposures_timeseries() - ExposureBucket("variant_a", day_1, first_exposed_identities=1000), ExposureBucket("", day_1, first_exposed_identities=5, quarantined=True), ], + conversion_buckets={}, ) # When @@ -1472,19 +1526,22 @@ def test_build_results_summary__occurrence_metrics__attach_conversions() -> None exposure_counts={"control": 1000, "variant_a": 1000}, metric_stats={}, granularity="day", - conversion_buckets=[ - ConversionBucket(7, "variant_a", day_3, converted_identities=20), - ConversionBucket(7, "control", day_1, converted_identities=100), - ConversionBucket(7, "variant_a", day_1, converted_identities=100), - ConversionBucket(11, "control", day_1, converted_identities=7), - ], + exposure_buckets=[], + conversion_buckets={ + 7: [ + ConversionBucket("variant_a", day_3, converted_identities=20), + ConversionBucket("control", day_1, converted_identities=100), + ConversionBucket("variant_a", day_1, converted_identities=100), + ], + 11: [ConversionBucket("control", day_1, converted_identities=7)], + }, ) # When summary = services.build_results_summary(aggregates, expected_shares={}) # Then each metric gets only its own rows, per variant, in bucket order - assert summary.metrics[0].timeseries == ConversionsTimeseries( + assert summary.metrics[0].conversions_timeseries == ConversionsTimeseries( granularity="day", points=[ ConversionsTimeseriesPoint( @@ -1497,7 +1554,7 @@ def test_build_results_summary__occurrence_metrics__attach_conversions() -> None ), ], ) - assert summary.metrics[1].timeseries == ConversionsTimeseries( + assert summary.metrics[1].conversions_timeseries == ConversionsTimeseries( granularity="day", points=[ ConversionsTimeseriesPoint( @@ -1515,13 +1572,15 @@ def test_build_results_summary__value_metric__timeseries_none() -> None: exposure_counts={"control": 1000}, metric_stats={}, granularity="day", + exposure_buckets=[], + conversion_buckets={}, ) # When summary = services.build_results_summary(aggregates, expected_shares={}) # Then a value metric has no conversion rate to chart - assert summary.metrics[0].timeseries is None + assert summary.metrics[0].conversions_timeseries is None @pytest.mark.django_db @@ -1755,25 +1814,21 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( } }, ) - mock_stats = mocker.patch( - "experimentation.services.get_metric_variant_stats", - return_value=aggregates, - ) window_start = datetime(2026, 6, 1, tzinfo=timezone.utc) window_end = datetime(2026, 6, 10, tzinfo=timezone.utc) - mock_exposures = mocker.patch( - "experimentation.services.get_exposure_buckets", - return_value=[ - ExposureBucket("control", window_start, first_exposed_identities=1000) - ], - ) - mock_conversions = mocker.patch( - "experimentation.services.get_conversion_buckets", - return_value=[ - ConversionBucket( - metric.id, "control", window_start, converted_identities=100 - ) - ], + mock_gather = mocker.patch( + "experimentation.services.get_results_aggregates", + return_value=replace( + aggregates, + exposure_buckets=[ + ExposureBucket("control", window_start, first_exposed_identities=1000) + ], + conversion_buckets={ + metric.id: [ + ConversionBucket("control", window_start, converted_identities=100) + ] + }, + ), ) # When @@ -1783,36 +1838,21 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( window_end=window_end, ) - # Then the warehouse is queried with the experiment's metric specs - mock_stats.assert_called_once_with( + # Then the warehouse is read once for the experiment's metric specs, + # bucketed by day because the window is longer than 72 hours + mock_gather.assert_called_once_with( environment_key=environment.api_key, feature_name=feature.name, window_start=window_start, window_end=window_end, specs=expected_specs, + granularity="day", ) # And the summary carries the metric result with an SRM verdict from the - # configured 50/50 split + # configured 50/50 split, plus both chart series assert summary.srm_p_value == pytest.approx(1.0) assert summary.metrics[0].metric_id == metric.id assert summary.metrics[0].inference["variant_a"] is not None - # And the chart rows are gathered over the same window, bucketed by day - # because the window is longer than 72 hours - mock_exposures.assert_called_once_with( - environment_key=environment.api_key, - feature_name=feature.name, - window_start=window_start, - window_end=window_end, - granularity="day", - ) - mock_conversions.assert_called_once_with( - environment_key=environment.api_key, - feature_name=feature.name, - window_start=window_start, - window_end=window_end, - specs=expected_specs, - granularity="day", - ) assert summary.exposures_timeseries == ExposuresTimeseries( granularity="day", points=[ @@ -1821,7 +1861,7 @@ def test_compute_results_summary__experiment__queries_warehouse_and_builds( ) ], ) - assert summary.metrics[0].timeseries == ConversionsTimeseries( + assert summary.metrics[0].conversions_timeseries == ConversionsTimeseries( granularity="day", points=[ ConversionsTimeseriesPoint( diff --git a/api/tests/unit/experimentation/test_tasks.py b/api/tests/unit/experimentation/test_tasks.py index 8155cca3ac2e..b144b251eb45 100644 --- a/api/tests/unit/experimentation/test_tasks.py +++ b/api/tests/unit/experimentation/test_tasks.py @@ -453,8 +453,10 @@ def _results_summary() -> ResultsSummary: "control": VariantStats(n=1000, sum=100.0, sum_squares=100.0) }, inference={}, + conversions_timeseries=None, ) ], + exposures_timeseries=ExposuresTimeseries(granularity="day", points=[]), ) diff --git a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md index 2cd515ba005e..831f765dcf69 100644 --- a/docs/docs/deployment-self-hosting/observability/_events-catalogue.md +++ b/docs/docs/deployment-self-hosting/observability/_events-catalogue.md @@ -838,7 +838,7 @@ Attributes: ### `warehouse.connection.connected` Logged at `info` from: - - `api/experimentation/services.py:1287` + - `api/experimentation/services.py:1259` Attributes: - `environment.id` @@ -847,8 +847,8 @@ Attributes: ### `warehouse.connection.event_names_failed` Logged at `warning` from: - - `api/experimentation/services.py:270` - - `api/experimentation/services.py:1387` + - `api/experimentation/services.py:273` + - `api/experimentation/services.py:1359` Attributes: - `environment.id` @@ -858,7 +858,7 @@ Attributes: ### `warehouse.connection.event_stats_failed` Logged at `warning` from: - - `api/experimentation/services.py:1350` + - `api/experimentation/services.py:1322` Attributes: - `environment.id` @@ -867,7 +867,7 @@ Attributes: ### `warehouse.connection.test_event_sent` Logged at `info` from: - - `api/experimentation/services.py:1061` + - `api/experimentation/services.py:1033` Attributes: - `environment.id` @@ -876,7 +876,7 @@ Attributes: ### `warehouse.connection.verification_failed` Logged at `warning` from: - - `api/experimentation/services.py:1262` + - `api/experimentation/services.py:1234` Attributes: - `environment.id` @@ -886,7 +886,7 @@ Attributes: ### `warehouse.connection.verification_succeeded` Logged at `info` from: - - `api/experimentation/services.py:1272` + - `api/experimentation/services.py:1244` Attributes: - `environment.id` @@ -895,7 +895,7 @@ Attributes: ### `warehouse.delivery.all_objects_rejected` Logged at `error` from: - - `api/experimentation/services.py:1217` + - `api/experimentation/services.py:1189` Attributes: - `connection.id` @@ -906,7 +906,7 @@ Attributes: ### `warehouse.delivery.budget_exhausted` Logged at `info` from: - - `api/experimentation/services.py:1106` + - `api/experimentation/services.py:1078` Attributes: - `connection.id` @@ -917,7 +917,7 @@ Attributes: ### `warehouse.delivery.completed` Logged at `info` from: - - `api/experimentation/services.py:1227` + - `api/experimentation/services.py:1199` Attributes: - `connection.id` @@ -930,7 +930,7 @@ Attributes: ### `warehouse.delivery.failed` Logged at `error` from: - - `api/experimentation/services.py:1200` + - `api/experimentation/services.py:1172` Attributes: - `connection.id` @@ -941,7 +941,7 @@ Attributes: ### `warehouse.delivery.object_rejected` Logged at `error` from: - - `api/experimentation/services.py:1135` + - `api/experimentation/services.py:1107` Attributes: - `connection.id` @@ -953,7 +953,7 @@ Attributes: ### `warehouse.srm.overallocated` Logged at `error` from: - - `api/experimentation/services.py:657` + - `api/experimentation/services.py:629` Attributes: - `environment.id` @@ -963,7 +963,7 @@ Attributes: ### `warehouse.srm.unkeyed_variant` Logged at `error` from: - - `api/experimentation/services.py:643` + - `api/experimentation/services.py:615` Attributes: - `environment.id` From a08ccf5ab9d14e7a3b16ea5c44241357f54bc2df Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Thu, 3 Sep 2026 17:15:45 +0530 Subject: [PATCH 3/4] test(experimentation): pin empty conversion series for uncoverted occurrence metric --- .../unit/experimentation/test_services.py | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/api/tests/unit/experimentation/test_services.py b/api/tests/unit/experimentation/test_services.py index 79d4e4247f6c..4b4862249f2e 100644 --- a/api/tests/unit/experimentation/test_services.py +++ b/api/tests/unit/experimentation/test_services.py @@ -1583,6 +1583,28 @@ def test_build_results_summary__value_metric__timeseries_none() -> None: assert summary.metrics[0].conversions_timeseries is None +def test_build_results_summary__occurrence_metric_no_conversions__empty_series() -> ( + None +): + # Given an occurrence metric that was charted but nobody converted on yet + aggregates = ResultsAggregates( + specs=[_spec(metric_id=7, event="purchase")], + exposure_counts={"control": 1000}, + metric_stats={}, + granularity="day", + exposure_buckets=[], + conversion_buckets={7: []}, + ) + + # When + summary = services.build_results_summary(aggregates, expected_shares={}) + + # Then the chart is present and empty, not absent + assert summary.metrics[0].conversions_timeseries == ConversionsTimeseries( + granularity="day", points=[] + ) + + @pytest.mark.django_db def test_experiment_metric_specs__attached_metrics__maps_definition_and_direction( experiment: Experiment, From 04569b74a7c3e3a33515f98bf452e0f723ce94a7 Mon Sep 17 00:00:00 2001 From: Gagan Trivedi Date: Thu, 3 Sep 2026 17:26:46 +0530 Subject: [PATCH 4/4] docs(experimentation): state the running-total contract on exposures_timeseries --- api/experimentation/dataclasses.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/api/experimentation/dataclasses.py b/api/experimentation/dataclasses.py index 0a7bc3e96b87..c46f7cfc23ff 100644 --- a/api/experimentation/dataclasses.py +++ b/api/experimentation/dataclasses.py @@ -113,7 +113,10 @@ class ResultsSummary: srm_p_value: float | None metrics: list[MetricResult] # Denominator for the conversion charts, computed in the same run as the - # metrics so both sides of the rate share one as_of. + # metrics so both sides of the rate share one as_of. Exposures are bucketed + # by first exposure and conversions by first conversion, so only running + # totals divide: a bucket's own conversions over its own new identities + # compares different people and can exceed 100%. exposures_timeseries: ExposuresTimeseries