From d09eabb3bfe27b35e8ddf97a561fb434136c28da Mon Sep 17 00:00:00 2001 From: Rodrigo Iloro Date: Mon, 8 Jun 2026 04:23:06 -0300 Subject: [PATCH 1/2] feat(experiments): add metric breakdown injector for mean metrics. --- .../experiment_mean_query_builder.py | 13 +- .../hogql_queries/experiment_query_runner.py | 11 +- .../metric_breakdown_injector.py | 175 ++++- .../experiment_query_runner/test_breakdown.py | 105 +-- .../test_metric_breakdown.py | 717 ++++++++++++++++++ .../test/test_metric_breakdown_injector.py | 158 ++++ 6 files changed, 1065 insertions(+), 114 deletions(-) create mode 100644 products/experiments/backend/hogql_queries/test/experiment_query_runner/test_metric_breakdown.py diff --git a/products/experiments/backend/hogql_queries/experiment_mean_query_builder.py b/products/experiments/backend/hogql_queries/experiment_mean_query_builder.py index fc3fb5fa0a5d..35b9ae21f0da 100644 --- a/products/experiments/backend/hogql_queries/experiment_mean_query_builder.py +++ b/products/experiments/backend/hogql_queries/experiment_mean_query_builder.py @@ -10,7 +10,6 @@ is_session_property_metric, validate_session_property, ) -from products.experiments.backend.hogql_queries.breakdown_injector import BreakdownInjector from products.experiments.backend.hogql_queries.metric_source import MetricSourceInfo if TYPE_CHECKING: @@ -393,9 +392,9 @@ def build_mean_query(self) -> ast.SelectQuery: assert isinstance(query, ast.SelectQuery) - # Inject breakdown columns into the query AST. The metric-event injector is funnel-only, - # so mean metrics always use the property-from-exposure BreakdownInjector. - if isinstance(self._b.breakdown_injector, BreakdownInjector): + # Inject breakdown columns into the query AST. Both injectors expose + # inject_mean_breakdown_columns (old: exposure source; new: metric-event source + limit). + if self._b.breakdown_injector: self._b.breakdown_injector.inject_mean_breakdown_columns(query, final_cte_name="entity_metrics") return query @@ -492,9 +491,9 @@ def build_mean_query_with_winsorization(self) -> ast.SelectQuery: assert isinstance(query, ast.SelectQuery) - # Inject breakdown columns into the query AST. The metric-event injector is funnel-only, - # so mean metrics always use the property-from-exposure BreakdownInjector. - if isinstance(self._b.breakdown_injector, BreakdownInjector): + # Inject breakdown columns into the query AST. Both injectors expose + # inject_mean_breakdown_columns (old: exposure source; new: metric-event source + limit). + if self._b.breakdown_injector: self._b.breakdown_injector.inject_mean_breakdown_columns(query, final_cte_name="winsorized_entity_metrics") return query diff --git a/products/experiments/backend/hogql_queries/experiment_query_runner.py b/products/experiments/backend/hogql_queries/experiment_query_runner.py index 0dfb04e9b826..f8bb56aeb973 100644 --- a/products/experiments/backend/hogql_queries/experiment_query_runner.py +++ b/products/experiments/backend/hogql_queries/experiment_query_runner.py @@ -544,12 +544,15 @@ def _get_experiment_query(self) -> ast.SelectQuery: breakdowns = self._get_breakdowns_for_builder() breakdown_injector: MetricBreakdownInjector | None = None - # Data warehouse funnels route through the legacy UNION ALL builder, which replaces the - # metric_events CTE and drops the injected breakdown columns, so gate them out until the - # injector supports that path. + # Metric types migrated to the metric-event breakdown injector. Others fall back to the + # old BreakdownInjector until they migrate (then BreakdownInjector is deleted). + migrated_metric = isinstance(self.metric, ExperimentFunnelMetric | ExperimentMeanMetric) + # Data warehouse sources aren't supported by the injector yet: funnels route through the + # legacy builder that drops the injected columns, and DW mean metrics select from the + # warehouse table where event/person/session breakdown chains don't resolve. if ( breakdowns - and isinstance(self.metric, ExperimentFunnelMetric) + and migrated_metric and not self.is_data_warehouse_query and self._metric_event_breakdowns_enabled() ): diff --git a/products/experiments/backend/hogql_queries/metric_breakdown_injector.py b/products/experiments/backend/hogql_queries/metric_breakdown_injector.py index 2c50ae6ccf4b..91c93d5d2808 100644 --- a/products/experiments/backend/hogql_queries/metric_breakdown_injector.py +++ b/products/experiments/backend/hogql_queries/metric_breakdown_injector.py @@ -12,12 +12,15 @@ migrate to metric-event breakdowns. """ -from typing import cast +from typing import Union, cast from posthog.schema import ( Breakdown, BreakdownAttributionType, ExperimentFunnelMetric, + ExperimentMeanMetric, + ExperimentRatioMetric, + ExperimentRetentionMetric, MultipleBreakdownType, StepOrderValue, ) @@ -29,9 +32,11 @@ from posthog.hogql_queries.insights.trends.utils import get_properties_chain from posthog.hogql_queries.insights.utils.breakdowns import BREAKDOWN_NULL_STRING_LABEL, BREAKDOWN_OTHER_STRING_LABEL +MetricType = Union[ExperimentFunnelMetric, ExperimentMeanMetric, ExperimentRatioMetric, ExperimentRetentionMetric] + class MetricBreakdownInjector: - def __init__(self, breakdowns: list[Breakdown], metric: ExperimentFunnelMetric): + def __init__(self, breakdowns: list[Breakdown], metric: MetricType): self.breakdowns = breakdowns self.metric = metric @@ -94,6 +99,7 @@ def _attribution_agg_and_steps(self) -> tuple[str, list[int]]: - step (unordered, "Any step"): argMinIf across all metric step columns, since an unordered funnel has no fixed step position and any matching step attributes. """ + assert isinstance(self.metric, ExperimentFunnelMetric) attribution = self.metric.breakdownAttributionType or BreakdownAttributionType.FIRST_TOUCH num_metric_steps = len(self.metric.series) is_unordered = self.metric.funnel_order_type == StepOrderValue.UNORDERED @@ -180,20 +186,27 @@ def _top_breakdowns_subquery(self, aliases: list[str]) -> ast.SelectQuery: ) return subquery - def _inject_final_breakdown_columns(self, query: ast.SelectQuery, aliases: list[str]) -> None: + def _inject_final_breakdown_columns( + self, query: ast.SelectQuery, aliases: list[str], final_cte_name: str = "entity_metrics" + ) -> None: """Surface breakdown columns in the outer SELECT (after variant) and GROUP BY. Applies the top-N + "Other" limit: breakdown tuples beyond the limit (ranked by user count) are relabeled to BREAKDOWN_OTHER_STRING_LABEL before the final GROUP BY, capping output cardinality. The relabel collapses the whole tuple together so all breakdown columns of an "Other" row carry the Other label. + + ``final_cte_name`` is the CTE the outer SELECT reads from (``entity_metrics`` for funnels + and non-winsorized mean, ``winsorized_entity_metrics`` for winsorized mean). Ranking is + always computed from ``entity_metrics`` since winsorization doesn't change the set of + (entity, variant, breakdown) tuples. """ - # Membership test against the top-N set. Single breakdown compares the scalar value - # directly; multiple breakdowns compare the value tuple element-wise. + # Membership test against the top-N set, always ranked from entity_metrics (pre-winsorization). + # Single breakdown compares the scalar value directly; multiple breakdowns compare the tuple. if len(aliases) == 1: - left: ast.Expr = ast.Field(chain=["entity_metrics", aliases[0]]) + left: ast.Expr = ast.Field(chain=[final_cte_name, aliases[0]]) else: - left = ast.Tuple(exprs=[ast.Field(chain=["entity_metrics", alias]) for alias in aliases]) + left = ast.Tuple(exprs=[ast.Field(chain=[final_cte_name, alias]) for alias in aliases]) in_top = ast.CompareOperation( op=ast.CompareOperationOp.In, left=left, @@ -205,7 +218,7 @@ def _inject_final_breakdown_columns(self, query: ast.SelectQuery, aliases: list[ "if({in_top}, {value}, {other})", placeholders={ "in_top": in_top, - "value": ast.Field(chain=["entity_metrics", alias]), + "value": ast.Field(chain=[final_cte_name, alias]), "other": ast.Constant(value=BREAKDOWN_OTHER_STRING_LABEL), }, ) @@ -267,3 +280,149 @@ def inject_funnel_breakdown_columns_optimized(self, query: ast.SelectQuery) -> N ) self._inject_final_breakdown_columns(query, aliases) + + def inject_mean_breakdown_columns(self, query: ast.SelectQuery, final_cte_name: str = "entity_metrics") -> None: + """Inject breakdown columns into a mean query. + + Reads the breakdown property off the *metric event* (in ``metric_events``) and + attributes each user to a single bucket via **first-touch** — ``argMin`` over the + metric event timestamp. Mean has no attribution modes (unlike funnels), so the + breakdown is always taken from the user's first qualifying metric event. The + per-user conversion window is already enforced by the ``entity_metrics`` join, so + no extra condition is needed on the ``argMin``. + + ``final_cte_name`` is ``entity_metrics`` for the plain query or + ``winsorized_entity_metrics`` for the winsorized query; winsorization needs the + breakdown carried through its percentiles/join CTEs. + + Session-property metrics have an extra ``metric_events_by_session`` layer; their + ``metric_events`` CTE joins exposures and can't see raw event properties, so the + breakdown is read (and deduped) in the by-session layer instead. See + ``_read_breakdown_into_metric_events``. + """ + if not self._has_breakdown(): + return + + aliases = self._get_breakdown_aliases() + + # The session-property shape reads the breakdown a layer earlier; the attribution + # timestamp is the session's first event time rather than the metric event timestamp. + is_session_property = bool(query.ctes and "metric_events_by_session" in query.ctes) + timestamp_field = "first_event_timestamp" if is_session_property else "timestamp" + + self._read_breakdown_into_metric_events(query, aliases, is_session_property) + + # Attribute first-touch per user: argMin over the (session-)first event timestamp. + if query.ctes and "entity_metrics" in query.ctes: + entity_metrics_cte = query.ctes["entity_metrics"] + if isinstance(entity_metrics_cte, ast.CTE) and isinstance(entity_metrics_cte.expr, ast.SelectQuery): + for alias in aliases: + entity_metrics_cte.expr.select.append( + ast.Alias( + alias=alias, + expr=ast.Call( + name="argMin", + args=[ + ast.Field(chain=["metric_events", alias]), + ast.Field(chain=["metric_events", timestamp_field]), + ], + ), + ) + ) + + # Carry the attributed breakdown through the winsorization CTEs. + self._inject_winsorization_breakdown_columns(query, aliases, final_cte_name) + + self._inject_final_breakdown_columns(query, aliases, final_cte_name=final_cte_name) + + def _read_breakdown_into_metric_events( + self, query: ast.SelectQuery, aliases: list[str], is_session_property: bool + ) -> None: + """Surface the breakdown value (and an attribution timestamp) in the ``metric_events`` CTE. + + Standard path: read the breakdown directly off the metric event in ``metric_events``. + + Session-property path: ``metric_events`` joins exposures and can't see raw event + properties, so read the breakdown off raw events in ``metric_events_by_session`` + (deduped per session via ``any()``, mirroring ``session_value``), then carry both the + breakdown and ``first_event_timestamp`` up into ``metric_events``. + """ + if not query.ctes: + return + + breakdown_exprs = self.build_breakdown_exprs(table_alias="") + + if not is_session_property: + metric_events_cte = query.ctes.get("metric_events") + if isinstance(metric_events_cte, ast.CTE) and isinstance(metric_events_cte.expr, ast.SelectQuery): + for alias, expr in breakdown_exprs: + metric_events_cte.expr.select.append(ast.Alias(alias=alias, expr=expr)) + return + + # Read + dedupe the breakdown per session in the by-session layer. + by_session_cte = query.ctes.get("metric_events_by_session") + if isinstance(by_session_cte, ast.CTE) and isinstance(by_session_cte.expr, ast.SelectQuery): + for alias, expr in breakdown_exprs: + by_session_cte.expr.select.append(ast.Alias(alias=alias, expr=ast.Call(name="any", args=[expr]))) + + # Carry the breakdown and first_event_timestamp up into metric_events for attribution. + metric_events_cte = query.ctes.get("metric_events") + if isinstance(metric_events_cte, ast.CTE) and isinstance(metric_events_cte.expr, ast.SelectQuery): + for alias in aliases: + metric_events_cte.expr.select.append( + ast.Alias(alias=alias, expr=ast.Field(chain=["metric_events_by_session", alias])) + ) + metric_events_cte.expr.select.append( + ast.Alias( + alias="first_event_timestamp", + expr=ast.Field(chain=["metric_events_by_session", "first_event_timestamp"]), + ) + ) + + def _inject_winsorization_breakdown_columns( + self, query: ast.SelectQuery, aliases: list[str], final_cte_name: str + ) -> None: + """Carry breakdown columns through the percentiles and winsorized_entity_metrics CTEs. + + Winsorization computes per-breakdown-group percentile thresholds (preserved here, not + changed), so the breakdown must be grouped in ``percentiles`` and joined into + ``winsorized_entity_metrics`` on the breakdown columns. + """ + if not (query.ctes and "percentiles" in query.ctes): + return + + percentiles_cte = query.ctes["percentiles"] + if isinstance(percentiles_cte, ast.CTE) and isinstance(percentiles_cte.expr, ast.SelectQuery): + for alias in aliases: + percentiles_cte.expr.select.append( + ast.Alias(alias=alias, expr=ast.Field(chain=["entity_metrics", alias])) + ) + if percentiles_cte.expr.group_by is None: + percentiles_cte.expr.group_by = [] + for alias in aliases: + percentiles_cte.expr.group_by.append(ast.Field(chain=["entity_metrics", alias])) + + if final_cte_name == "winsorized_entity_metrics" and "winsorized_entity_metrics" in query.ctes: + winsorized_cte = query.ctes["winsorized_entity_metrics"] + if isinstance(winsorized_cte, ast.CTE) and isinstance(winsorized_cte.expr, ast.SelectQuery): + for alias in aliases: + winsorized_cte.expr.select.append( + ast.Alias(alias=alias, expr=ast.Field(chain=["entity_metrics", alias])) + ) + # Convert the CROSS JOIN with percentiles into a per-breakdown-group INNER JOIN. + if winsorized_cte.expr.select_from: + join_expr = winsorized_cte.expr.select_from.next_join + if join_expr and isinstance(join_expr, ast.JoinExpr): + join_expr.join_type = "JOIN" + join_conditions: list[ast.Expr] = [ + ast.CompareOperation( + op=ast.CompareOperationOp.Eq, + left=ast.Field(chain=["percentiles", alias]), + right=ast.Field(chain=["entity_metrics", alias]), + ) + for alias in aliases + ] + condition_expr = join_conditions[0] + for condition in join_conditions[1:]: + condition_expr = ast.And(exprs=[condition_expr, condition]) + join_expr.constraint = ast.JoinConstraint(expr=condition_expr, constraint_type="ON") diff --git a/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_breakdown.py b/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_breakdown.py index 3583ca83e2e8..3cc8c1b2b8e8 100644 --- a/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_breakdown.py +++ b/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_breakdown.py @@ -9,7 +9,6 @@ from posthog.schema import ( Breakdown, - BreakdownAttributionType, BreakdownFilter, EventsNode, ExperimentFunnelMetric, @@ -453,13 +452,12 @@ def test_funnel_breakdown_source_respects_flag(self, _name, flag_on, expected_bu @parameterized.expand( [ - ("first_touch", BreakdownAttributionType.FIRST_TOUCH, None, {"StepZeroBrowser"}), - ("last_touch", BreakdownAttributionType.LAST_TOUCH, None, {"StepOneBrowser"}), - ("step_1", BreakdownAttributionType.STEP, 1, {"StepOneBrowser"}), + ("flag_off", False, {"ExposureBrowser"}), + ("flag_on", True, {"MetricBrowser"}), ] ) @freeze_time("2020-01-01T12:00:00Z") - def test_funnel_attribution_modes_bucket_correctly(self, _name, attribution, attribution_value, expected_buckets): + def test_mean_breakdown_source_respects_flag(self, _name, flag_on, expected_buckets): from unittest.mock import patch feature_flag = self.create_feature_flag() @@ -467,11 +465,9 @@ def test_funnel_attribution_modes_bucket_correctly(self, _name, attribution, att experiment.stats_config = {"method": "frequentist"} experiment.save() - metric = ExperimentFunnelMetric( - series=[EventsNode(event="step_zero"), EventsNode(event="step_one")], + metric = ExperimentMeanMetric( + source=EventsNode(event="purchase", math=ExperimentMetricMathType.SUM, math_property="amount"), breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), - breakdownAttributionType=attribution, - breakdownAttributionValue=attribution_value, ) experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) experiment.metrics = [metric.model_dump(mode="json")] @@ -479,7 +475,7 @@ def test_funnel_attribution_modes_bucket_correctly(self, _name, attribution, att feature_flag_property = f"$feature/{feature_flag.key}" - # Step 0 and step 1 carry different breakdown values, so each attribution mode resolves differently. + # Exposure event carries "ExposureBrowser"; the metric event carries "MetricBrowser". for variant in ["control", "test"]: distinct_id = f"user_{variant}" _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) @@ -492,28 +488,22 @@ def test_funnel_attribution_modes_bucket_correctly(self, _name, attribution, att feature_flag_property: variant, "$feature_flag_response": variant, "$feature_flag": feature_flag.key, + "$browser": "ExposureBrowser", }, ) _create_event( team=self.team, - event="step_zero", + event="purchase", distinct_id=distinct_id, timestamp="2020-01-02T12:01:00Z", - properties={feature_flag_property: variant, "$browser": "StepZeroBrowser"}, - ) - _create_event( - team=self.team, - event="step_one", - distinct_id=distinct_id, - timestamp="2020-01-02T12:02:00Z", - properties={feature_flag_property: variant, "$browser": "StepOneBrowser"}, + properties={feature_flag_property: variant, "$browser": "MetricBrowser", "amount": 10}, ) flush_persons_and_events() with patch( "products.experiments.backend.hogql_queries.experiment_query_runner.posthoganalytics.feature_enabled", - return_value=True, + return_value=flag_on, ): query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) result = cast(ExperimentQueryResponse, query_runner.calculate()) @@ -522,81 +512,6 @@ def test_funnel_attribution_modes_bucket_correctly(self, _name, attribution, att buckets = {value for br in result.breakdown_results for value in br.breakdown_value} self.assertEqual(buckets, expected_buckets) - @freeze_time("2020-01-01T12:00:00Z") - def test_funnel_breakdown_limit_collapses_to_other(self): - from unittest.mock import patch - - from posthog.hogql_queries.insights.utils.breakdowns import BREAKDOWN_OTHER_STRING_LABEL - - feature_flag = self.create_feature_flag() - experiment = self.create_experiment(feature_flag=feature_flag) - experiment.stats_config = {"method": "frequentist"} - experiment.save() - - metric = ExperimentFunnelMetric( - series=[EventsNode(event="purchase")], - breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")], breakdown_limit=2), - ) - experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) - experiment.metrics = [metric.model_dump(mode="json")] - experiment.save() - - feature_flag_property = f"$feature/{feature_flag.key}" - - # Four browser values with descending user counts: Chrome(4) > Safari(3) > Firefox(2) > Edge(1). - # With breakdown_limit=2, only Chrome + Safari survive; Firefox + Edge collapse to "Other". - browser_counts = {"Chrome": 4, "Safari": 3, "Firefox": 2, "Edge": 1} - user_index = 0 - for browser, count in browser_counts.items(): - for _ in range(count): - variant = "control" if user_index % 2 == 0 else "test" - distinct_id = f"user_{user_index}" - user_index += 1 - _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) - _create_event( - team=self.team, - event="$feature_flag_called", - distinct_id=distinct_id, - timestamp="2020-01-02T12:00:00Z", - properties={ - feature_flag_property: variant, - "$feature_flag_response": variant, - "$feature_flag": feature_flag.key, - }, - ) - _create_event( - team=self.team, - event="purchase", - distinct_id=distinct_id, - timestamp="2020-01-02T12:01:00Z", - properties={feature_flag_property: variant, "$browser": browser}, - ) - - flush_persons_and_events() - - with patch( - "products.experiments.backend.hogql_queries.experiment_query_runner.posthoganalytics.feature_enabled", - return_value=True, - ): - query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) - result = cast(ExperimentQueryResponse, query_runner.calculate()) - - assert result.breakdown_results is not None - buckets = {value for br in result.breakdown_results for value in br.breakdown_value} - # Top 2 by frequency survive; the rest collapse into a single "Other" bucket. - self.assertEqual(buckets, {"Chrome", "Safari", BREAKDOWN_OTHER_STRING_LABEL}) - - # The "Other" bucket is ordered last in the breakdown list. - ordered_values = [value for br in result.breakdown_results for value in br.breakdown_value] - self.assertEqual(ordered_values[-1], BREAKDOWN_OTHER_STRING_LABEL) - - # The "Other" bucket preserves totals: per-breakdown baseline samples sum to the overall. - assert result.baseline is not None - per_breakdown_baseline = sum( - br.baseline.number_of_samples for br in result.breakdown_results if br.baseline is not None - ) - self.assertEqual(per_breakdown_baseline, result.baseline.number_of_samples) - @freeze_time("2020-01-01T12:00:00Z") @snapshot_clickhouse_queries def test_ratio_metric_with_breakdown(self): diff --git a/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_metric_breakdown.py b/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_metric_breakdown.py new file mode 100644 index 000000000000..312156104cff --- /dev/null +++ b/products/experiments/backend/hogql_queries/test/experiment_query_runner/test_metric_breakdown.py @@ -0,0 +1,717 @@ +from typing import cast + +from freezegun import freeze_time +from posthog.test.base import _create_event, _create_person, flush_persons_and_events +from unittest.mock import patch + +from django.test import override_settings + +from parameterized import parameterized + +from posthog.schema import ( + Breakdown, + BreakdownAttributionType, + BreakdownFilter, + EventsNode, + ExperimentFunnelMetric, + ExperimentMeanMetric, + ExperimentMetricMathType, + ExperimentQuery, + ExperimentQueryResponse, +) + +from posthog.hogql_queries.insights.utils.breakdowns import BREAKDOWN_NULL_STRING_LABEL, BREAKDOWN_OTHER_STRING_LABEL + +from products.experiments.backend.hogql_queries.experiment_query_runner import ExperimentQueryRunner +from products.experiments.backend.hogql_queries.test.experiment_query_runner.base import ExperimentQueryRunnerBaseTest + + +@override_settings(IN_UNIT_TESTING=True) +class TestMetricBreakdown(ExperimentQueryRunnerBaseTest): + def setUp(self): + super().setUp() + # The metric-event breakdown injector is gated by a feature flag; force it on for this + # entire suite so individual tests exercise the new injector without per-test plumbing. + flag_patcher = patch( + "products.experiments.backend.hogql_queries.experiment_query_runner.posthoganalytics.feature_enabled", + return_value=True, + ) + flag_patcher.start() + self.addCleanup(flag_patcher.stop) + + @parameterized.expand( + [ + ("first_touch", BreakdownAttributionType.FIRST_TOUCH, None, {"StepZeroBrowser"}), + ("last_touch", BreakdownAttributionType.LAST_TOUCH, None, {"StepOneBrowser"}), + ("step_1", BreakdownAttributionType.STEP, 1, {"StepOneBrowser"}), + ] + ) + @freeze_time("2020-01-01T12:00:00Z") + def test_funnel_attribution_modes_bucket_correctly(self, _name, attribution, attribution_value, expected_buckets): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentFunnelMetric( + series=[EventsNode(event="step_zero"), EventsNode(event="step_one")], + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), + breakdownAttributionType=attribution, + breakdownAttributionValue=attribution_value, + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Step 0 and step 1 carry different breakdown values, so each attribution mode resolves differently. + for variant in ["control", "test"]: + distinct_id = f"user_{variant}" + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="step_zero", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={feature_flag_property: variant, "$browser": "StepZeroBrowser"}, + ) + _create_event( + team=self.team, + event="step_one", + distinct_id=distinct_id, + timestamp="2020-01-02T12:02:00Z", + properties={feature_flag_property: variant, "$browser": "StepOneBrowser"}, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + buckets = {value for br in result.breakdown_results for value in br.breakdown_value} + self.assertEqual(buckets, expected_buckets) + + @freeze_time("2020-01-01T12:00:00Z") + def test_funnel_breakdown_limit_collapses_to_other(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentFunnelMetric( + series=[EventsNode(event="purchase")], + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")], breakdown_limit=2), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Four browser values with descending user counts: Chrome(4) > Safari(3) > Firefox(2) > Edge(1). + # With breakdown_limit=2, only Chrome + Safari survive; Firefox + Edge collapse to "Other". + browser_counts = {"Chrome": 4, "Safari": 3, "Firefox": 2, "Edge": 1} + user_index = 0 + for browser, count in browser_counts.items(): + for _ in range(count): + variant = "control" if user_index % 2 == 0 else "test" + distinct_id = f"user_{user_index}" + user_index += 1 + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={feature_flag_property: variant, "$browser": browser}, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + buckets = {value for br in result.breakdown_results for value in br.breakdown_value} + # Top 2 by frequency survive; the rest collapse into a single "Other" bucket. + self.assertEqual(buckets, {"Chrome", "Safari", BREAKDOWN_OTHER_STRING_LABEL}) + + # The "Other" bucket is ordered last in the breakdown list. + ordered_values = [value for br in result.breakdown_results for value in br.breakdown_value] + self.assertEqual(ordered_values[-1], BREAKDOWN_OTHER_STRING_LABEL) + + # The "Other" bucket preserves totals: per-breakdown baseline samples sum to the overall. + assert result.baseline is not None + per_breakdown_baseline = sum( + br.baseline.number_of_samples for br in result.breakdown_results if br.baseline is not None + ) + self.assertEqual(per_breakdown_baseline, result.baseline.number_of_samples) + + @freeze_time("2020-01-01T12:00:00Z") + def test_session_property_mean_breakdown_from_metric_event(self): + import uuid + + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + # Session-property metric (aggregates a session-level property) with an event-property breakdown. + metric = ExperimentMeanMetric( + source=EventsNode( + event="$pageview", + math=ExperimentMetricMathType.AVG, + math_property="$session_duration", + math_property_type="session_properties", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Exposure carries "ExposureBrowser"; the metric event carries "MetricBrowser". + for variant in ["control", "test"]: + distinct_id = f"user_{variant}" + session_id = str(uuid.uuid4()) + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + "$browser": "ExposureBrowser", + "$session_id": session_id, + }, + ) + _create_event( + team=self.team, + event="$pageview", + distinct_id=distinct_id, + timestamp="2020-01-02T12:05:00Z", + properties={feature_flag_property: variant, "$browser": "MetricBrowser", "$session_id": session_id}, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + buckets = {value for br in result.breakdown_results for value in br.breakdown_value} + # The breakdown is attributed from the metric event, not the exposure. + self.assertEqual(buckets, {"MetricBrowser"}) + + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_limit_collapses_to_other(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.SUM, + math_property="amount", + ), + breakdownFilter=BreakdownFilter( + breakdowns=[Breakdown(property="$browser")], + breakdown_limit=2, + ), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Four browser values with descending user counts: Chrome(4) > Safari(3) > Firefox(2) > Edge(1). + # With breakdown_limit=2, only Chrome + Safari survive; Firefox + Edge collapse to "Other". + browser_counts = {"Chrome": 4, "Safari": 3, "Firefox": 2, "Edge": 1} + user_index = 0 + for browser, count in browser_counts.items(): + for _ in range(count): + variant = "control" if user_index % 2 == 0 else "test" + distinct_id = f"user_mean_limit_{user_index}" + user_index += 1 + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={ + feature_flag_property: variant, + "$browser": browser, + "amount": 10, + }, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + buckets = {value for br in result.breakdown_results for value in br.breakdown_value} + # Top 2 by frequency survive; the rest collapse into a single "Other" bucket. + self.assertEqual(buckets, {"Chrome", "Safari", BREAKDOWN_OTHER_STRING_LABEL}) + + # The "Other" bucket is ordered last. + ordered_values = [value for br in result.breakdown_results for value in br.breakdown_value] + self.assertEqual(ordered_values[-1], BREAKDOWN_OTHER_STRING_LABEL) + + # Per-breakdown baseline samples sum to the overall baseline. + assert result.baseline is not None + per_breakdown_samples = sum( + br.baseline.number_of_samples for br in result.breakdown_results if br.baseline is not None + ) + self.assertEqual(per_breakdown_samples, result.baseline.number_of_samples) + + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_first_touch_with_changing_metric_event_values(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.SUM, + math_property="amount", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Each variant has one user who makes TWO purchases: one with "FirstBrowser" (earlier) + # and one with "SecondBrowser" (later). First-touch attribution must pick "FirstBrowser". + for variant in ["control", "test"]: + distinct_id = f"user_changing_{variant}" + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + # First metric event — should be picked by argMin + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={feature_flag_property: variant, "$browser": "FirstBrowser", "amount": 10}, + ) + # Second metric event — later timestamp, different breakdown value + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:02:00Z", + properties={feature_flag_property: variant, "$browser": "SecondBrowser", "amount": 20}, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + buckets = {value for br in result.breakdown_results for value in br.breakdown_value} + # Both users' first metric event had "FirstBrowser", so only that bucket should appear. + self.assertEqual(buckets, {"FirstBrowser"}) + self.assertNotIn("SecondBrowser", buckets) + + # The single bucket must have one user per variant. + first_browser_breakdown = next( + (br for br in result.breakdown_results if br.breakdown_value == ["FirstBrowser"]), None + ) + self.assertIsNotNone(first_browser_breakdown) + assert first_browser_breakdown is not None + self.assertIsNotNone(first_browser_breakdown.baseline) + self.assertEqual(first_browser_breakdown.baseline.number_of_samples, 1) + + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_with_winsorization_flag_on(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.SUM, + math_property="amount", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), + lower_bound_percentile=0.05, + upper_bound_percentile=0.95, + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Two browsers on the metric event side (exposure carries no browser property). + for i in range(4): + browser = "Chrome" if i < 2 else "Safari" + variant = "control" if i % 2 == 0 else "test" + distinct_id = f"user_winsor_{i}" + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={ + feature_flag_property: variant, + "$browser": browser, + "amount": 10 + i * 5, + }, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + buckets = {value for br in result.breakdown_results for value in br.breakdown_value} + # Buckets come from the metric event, not the exposure event. + self.assertEqual(buckets, {"Chrome", "Safari"}) + for breakdown_result in result.breakdown_results: + self.assertIsNotNone(breakdown_result.baseline) + + @parameterized.expand( + [ + ("two_breakdowns", ["$browser", "$os"], 2), + ("three_breakdowns", ["$browser", "$os", "$device_type"], 3), + ] + ) + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_with_multiple_breakdowns_flag_on(self, _name, breakdown_properties, expected_tuple_len): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.SUM, + math_property="amount", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property=p) for p in breakdown_properties]), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Seed two users per variant with deterministic property combinations. + combos = [ + {"$browser": "Chrome", "$os": "Windows", "$device_type": "Desktop"}, + {"$browser": "Safari", "$os": "Mac", "$device_type": "Mobile"}, + ] + for variant in ["control", "test"]: + for idx, combo in enumerate(combos): + distinct_id = f"user_multi_{variant}_{idx}" + metric_props = {k: v for k, v in combo.items() if k in breakdown_properties} + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={ + feature_flag_property: variant, + "amount": 10, + **metric_props, + }, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + # Every breakdown_value tuple must have the expected number of elements. + for breakdown_result in result.breakdown_results: + self.assertEqual(len(breakdown_result.breakdown_value), expected_tuple_len) + + # The two seeded combinations should appear. + breakdown_values = [tuple(br.breakdown_value) for br in result.breakdown_results] + expected_chrome = tuple(combos[0][p] for p in breakdown_properties) + expected_safari = tuple(combos[1][p] for p in breakdown_properties) + self.assertIn(expected_chrome, breakdown_values) + self.assertIn(expected_safari, breakdown_values) + + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_null_values_flag_on(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.TOTAL, + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Two users per variant: one with $browser on the metric event, one without. + for variant in ["control", "test"]: + for idx, has_browser in enumerate([True, False]): + distinct_id = f"user_null_{variant}_{idx}" + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + purchase_props: dict = {feature_flag_property: variant} + if has_browser: + purchase_props["$browser"] = "Chrome" + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties=purchase_props, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + breakdown_values = [br.breakdown_value for br in result.breakdown_results] + # The real bucket and the null/missing bucket must both appear. + self.assertIn(["Chrome"], breakdown_values) + self.assertIn([BREAKDOWN_NULL_STRING_LABEL], breakdown_values) + + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_person_property_flag_on(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.SUM, + math_property="amount", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="country", type="person")]), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Two users per variant; person property carries country (no event-level country). + country_by_variant = {"control": "US", "test": "UK"} + for variant, country in country_by_variant.items(): + for i in range(2): + distinct_id = f"user_person_{variant}_{i}" + _create_person(distinct_ids=[distinct_id], team_id=self.team.pk, properties={"country": country}) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id=distinct_id, + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: variant, + "$feature_flag_response": variant, + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="purchase", + distinct_id=distinct_id, + timestamp="2020-01-02T12:01:00Z", + properties={feature_flag_property: variant, "amount": 10}, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + assert result.breakdown_results is not None + breakdown_values = [br.breakdown_value for br in result.breakdown_results] + self.assertIn(["US"], breakdown_values) + self.assertIn(["UK"], breakdown_values) + + us_breakdown = next((br for br in result.breakdown_results if br.breakdown_value == ["US"]), None) + self.assertIsNotNone(us_breakdown) + assert us_breakdown is not None + self.assertEqual(us_breakdown.baseline.number_of_samples, 2) + + uk_breakdown = next((br for br in result.breakdown_results if br.breakdown_value == ["UK"]), None) + self.assertIsNotNone(uk_breakdown) + assert uk_breakdown is not None + self.assertEqual(len(uk_breakdown.variants), 1) + self.assertEqual(uk_breakdown.variants[0].number_of_samples, 2) + + @freeze_time("2020-01-01T12:00:00Z") + def test_mean_breakdown_unexposed_user_flag_on(self): + feature_flag = self.create_feature_flag() + experiment = self.create_experiment(feature_flag=feature_flag) + experiment.stats_config = {"method": "frequentist"} + experiment.save() + + metric = ExperimentMeanMetric( + source=EventsNode( + event="purchase", + math=ExperimentMetricMathType.SUM, + math_property="amount", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")]), + ) + experiment_query = ExperimentQuery(experiment_id=experiment.id, kind="ExperimentQuery", metric=metric) + experiment.metrics = [metric.model_dump(mode="json")] + experiment.save() + + feature_flag_property = f"$feature/{feature_flag.key}" + + # Exposed and converted — should appear in "Chrome" bucket. + _create_person(distinct_ids=["user_converting_control"], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id="user_converting_control", + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: "control", + "$feature_flag_response": "control", + "$feature_flag": feature_flag.key, + }, + ) + _create_event( + team=self.team, + event="purchase", + distinct_id="user_converting_control", + timestamp="2020-01-02T12:01:00Z", + properties={feature_flag_property: "control", "$browser": "Chrome", "amount": 10}, + ) + + # Exposed but did NOT purchase — contributes to exposure count but no metric event. + _create_person(distinct_ids=["user_exposed_only_control"], team_id=self.team.pk) + _create_event( + team=self.team, + event="$feature_flag_called", + distinct_id="user_exposed_only_control", + timestamp="2020-01-02T12:00:00Z", + properties={ + feature_flag_property: "control", + "$feature_flag_response": "control", + "$feature_flag": feature_flag.key, + }, + ) + + flush_persons_and_events() + + query_runner = ExperimentQueryRunner(query=experiment_query, team=self.team) + result = cast(ExperimentQueryResponse, query_runner.calculate()) + + # Query must run without error and the converting user's bucket must be present. + assert result.breakdown_results is not None + breakdown_values = [br.breakdown_value for br in result.breakdown_results] + self.assertIn(["Chrome"], breakdown_values) diff --git a/products/experiments/backend/hogql_queries/test/test_metric_breakdown_injector.py b/products/experiments/backend/hogql_queries/test/test_metric_breakdown_injector.py index c4fff4223674..20dd0f286cea 100644 --- a/products/experiments/backend/hogql_queries/test/test_metric_breakdown_injector.py +++ b/products/experiments/backend/hogql_queries/test/test_metric_breakdown_injector.py @@ -4,8 +4,12 @@ Breakdown, BreakdownAttributionType, BreakdownFilter, + BreakdownType, EventsNode, + ExperimentDataWarehouseNode, ExperimentFunnelMetric, + ExperimentMeanMetric, + ExperimentMetricMathType, StepOrderValue, ) @@ -36,6 +40,67 @@ def _unwrap_attribution(expr: ast.Expr) -> ast.Call: return attributed +def _mean_metric(breakdown_limit=None): + return ExperimentMeanMetric( + source=EventsNode(event="purchase"), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="$browser")], breakdown_limit=breakdown_limit), + ) + + +def _dw_mean_metric(): + return ExperimentMeanMetric( + source=ExperimentDataWarehouseNode( + table_name="usage", + events_join_key="properties.$user_id", + data_warehouse_join_key="userid", + timestamp_field="ds", + math=ExperimentMetricMathType.SUM, + math_property="usage", + ), + breakdownFilter=BreakdownFilter(breakdowns=[Breakdown(property="plan", type=BreakdownType.DATA_WAREHOUSE)]), + ) + + +def _mean_query() -> ast.SelectQuery: + # Minimal stand-in mirroring the standard mean shape: metric_events + entity_metrics. + query = parse_select("SELECT variant FROM entity_metrics") + assert isinstance(query, ast.SelectQuery) + me = parse_select("SELECT entity_id, timestamp, value FROM events") + em = parse_select( + "SELECT exposures.entity_id, exposures.variant, sum(value) AS value FROM exposures GROUP BY exposures.entity_id, exposures.variant" + ) + assert isinstance(me, ast.SelectQuery) and isinstance(em, ast.SelectQuery) + query.ctes = { + "metric_events": ast.CTE(name="metric_events", expr=me, cte_type="subquery"), + "entity_metrics": ast.CTE(name="entity_metrics", expr=em, cte_type="subquery"), + } + return query + + +def _session_mean_query() -> ast.SelectQuery: + # Stand-in for the session-property mean shape: + # metric_events_by_session -> metric_events -> entity_metrics. + query = parse_select("SELECT variant FROM entity_metrics") + assert isinstance(query, ast.SelectQuery) + mebs = parse_select( + "SELECT entity_id, `$session_id` AS session_id, any(session.x) AS session_value, " + "min(timestamp) AS first_event_timestamp FROM events GROUP BY entity_id, `$session_id`" + ) + me = parse_select( + "SELECT exposures.entity_id AS entity_id, exposures.variant AS variant, " + "metric_events_by_session.session_value AS value FROM exposures " + "INNER JOIN metric_events_by_session ON exposures.entity_id = metric_events_by_session.entity_id" + ) + em = parse_select("SELECT entity_id, variant, sum(value) AS value FROM metric_events GROUP BY entity_id, variant") + assert isinstance(mebs, ast.SelectQuery) and isinstance(me, ast.SelectQuery) and isinstance(em, ast.SelectQuery) + query.ctes = { + "metric_events_by_session": ast.CTE(name="metric_events_by_session", expr=mebs, cte_type="subquery"), + "metric_events": ast.CTE(name="metric_events", expr=me, cte_type="subquery"), + "entity_metrics": ast.CTE(name="entity_metrics", expr=em, cte_type="subquery"), + } + return query + + def _optimized_query() -> ast.SelectQuery: query = parse_select("SELECT variant FROM base_events AS entity_metrics") assert isinstance(query, ast.SelectQuery) @@ -178,3 +243,96 @@ def test_step_attribution_out_of_range_raises(self): raise AssertionError("expected ValueError") except ValueError: pass + + +class TestMetricBreakdownInjectorMean: + def test_breakdown_read_from_metric_event(self): + metric = _mean_metric() + injector = MetricBreakdownInjector(metric.breakdownFilter.breakdowns, metric) + query = _mean_query() + + injector.inject_mean_breakdown_columns(query) + + assert query.ctes is not None + me_cte = query.ctes["metric_events"] + assert isinstance(me_cte, ast.CTE) and isinstance(me_cte.expr, ast.SelectQuery) + me_aliases = [c.alias for c in me_cte.expr.select if isinstance(c, ast.Alias)] + assert "breakdown_value_1" in me_aliases + + def test_entity_metrics_uses_first_touch_argmin_from_metric_event(self): + metric = _mean_metric() + injector = MetricBreakdownInjector(metric.breakdownFilter.breakdowns, metric) + query = _mean_query() + + injector.inject_mean_breakdown_columns(query) + + expr = _entity_metrics_aliases(query)["breakdown_value_1"] + # argMin(metric_events.breakdown_value_1, metric_events.timestamp) — first metric event per user. + assert isinstance(expr, ast.Call) + assert expr.name == "argMin" + assert expr.args[0] == ast.Field(chain=["metric_events", "breakdown_value_1"]) + assert expr.args[1] == ast.Field(chain=["metric_events", "timestamp"]) + + def test_final_select_applies_top_n_other_limit(self): + metric = _mean_metric(breakdown_limit=3) + injector = MetricBreakdownInjector(metric.breakdownFilter.breakdowns, metric) + query = _mean_query() + + injector.inject_mean_breakdown_columns(query) + + final_alias = next(c for c in query.select if isinstance(c, ast.Alias) and c.alias == "breakdown_value_1") + assert isinstance(final_alias.expr, ast.Call) + assert final_alias.expr.name == "if" + other = final_alias.expr.args[2] + assert isinstance(other, ast.Constant) + assert other.value == "$$_posthog_breakdown_other_$$" + + def test_session_property_reads_breakdown_in_by_session_cte(self): + metric = _mean_metric() + injector = MetricBreakdownInjector(metric.breakdownFilter.breakdowns, metric) + query = _session_mean_query() + + injector.inject_mean_breakdown_columns(query) + + assert query.ctes is not None + # Breakdown is read off raw events in the by-session layer (deduped via any()), since the + # metric_events CTE there has no access to raw event properties. + mebs_cte = query.ctes["metric_events_by_session"] + assert isinstance(mebs_cte, ast.CTE) and isinstance(mebs_cte.expr, ast.SelectQuery) + mebs_aliases = [c.alias for c in mebs_cte.expr.select if isinstance(c, ast.Alias)] + assert "breakdown_value_1" in mebs_aliases + + def test_session_property_first_touch_uses_first_event_timestamp(self): + metric = _mean_metric() + injector = MetricBreakdownInjector(metric.breakdownFilter.breakdowns, metric) + query = _session_mean_query() + + injector.inject_mean_breakdown_columns(query) + + expr = _entity_metrics_aliases(query)["breakdown_value_1"] + # First-touch over the session's first_event_timestamp (carried up from the by-session layer). + assert isinstance(expr, ast.Call) + assert expr.name == "argMin" + assert expr.args[0] == ast.Field(chain=["metric_events", "breakdown_value_1"]) + assert expr.args[1] == ast.Field(chain=["metric_events", "first_event_timestamp"]) + + def test_data_warehouse_breakdown_reads_warehouse_column(self): + metric = _dw_mean_metric() + injector = MetricBreakdownInjector(metric.breakdownFilter.breakdowns, metric) + query = _mean_query() + + injector.inject_mean_breakdown_columns(query) + + assert query.ctes is not None + me_cte = query.ctes["metric_events"] + assert isinstance(me_cte, ast.CTE) and isinstance(me_cte.expr, ast.SelectQuery) + bd = next(c for c in me_cte.expr.select if isinstance(c, ast.Alias) and c.alias == "breakdown_value_1") + # A DW breakdown is a direct warehouse column (the metric_events CTE reads FROM the warehouse + # table), so it must read bare `plan`, NOT be wrapped in an event `properties` chain. + coalesce_call = bd.expr + assert isinstance(coalesce_call, ast.Call) and coalesce_call.name == "coalesce" + to_string = coalesce_call.args[0] + assert isinstance(to_string, ast.Call) and to_string.name == "toString" + field = to_string.args[0] + assert isinstance(field, ast.Field) + assert field.chain == ["plan"] From 57d2d2bc6880e030b1fbda6236022aa1682605d1 Mon Sep 17 00:00:00 2001 From: Rodrigo Iloro Date: Fri, 14 Aug 2026 17:19:34 -0300 Subject: [PATCH 2/2] docs(experiments): document CUPED pre-window attribution limitation for mean metrics --- .../metric_breakdown_injector.py | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/products/experiments/backend/hogql_queries/metric_breakdown_injector.py b/products/experiments/backend/hogql_queries/metric_breakdown_injector.py index 91c93d5d2808..338470cb9c58 100644 --- a/products/experiments/backend/hogql_queries/metric_breakdown_injector.py +++ b/products/experiments/backend/hogql_queries/metric_breakdown_injector.py @@ -1,13 +1,13 @@ """ -Metric-event breakdown injection for experiment funnel queries. +Metric-event breakdown injection for experiment queries. Unlike ``BreakdownInjector`` (which attributes the breakdown value from the exposure event), this injector reads the breakdown property off the *metric* -event and attributes it across funnel steps using the metric's configured -``breakdownAttributionType`` — aligning experiment funnel breakdowns with how -insights funnels behave. +event. Funnels attribute across steps using the metric's configured +``breakdownAttributionType``; mean metrics attribute first-touch. This aligns +experiment breakdowns with how insights behave. -Scope: funnel metrics only. This is a standalone class (it does not subclass +Scope: funnel and mean metrics. This is a standalone class (it does not subclass ``BreakdownInjector``); the old injector is removed once all metric types migrate to metric-event breakdowns. """ @@ -287,9 +287,12 @@ def inject_mean_breakdown_columns(self, query: ast.SelectQuery, final_cte_name: Reads the breakdown property off the *metric event* (in ``metric_events``) and attributes each user to a single bucket via **first-touch** — ``argMin`` over the metric event timestamp. Mean has no attribution modes (unlike funnels), so the - breakdown is always taken from the user's first qualifying metric event. The - per-user conversion window is already enforced by the ``entity_metrics`` join, so - no extra condition is needed on the ``argMin``. + breakdown is always taken from the user's first qualifying metric event. + + Known limitation: with CUPED the ``entity_metrics`` join widens to (conversion window + OR pre-exposure lookback), so ``argMin`` can attribute the breakdown from an event that + happened before exposure. Attribution here does not add its own exposure-time guard; a + follow-up should exclude pre-window events from the attribution when CUPED is enabled. ``final_cte_name`` is ``entity_metrics`` for the plain query or ``winsorized_entity_metrics`` for the winsorized query; winsorization needs the