feat(experiments): extend metric-events precomputation to count/sum mean metrics

products/experiments/backend/hogql_queries/LAZY_COMPUTATION.md

@@ -265,3 +265,5 @@ GROUP BY variant
 5. **Fallback**: If precomputation fails or isn't ready, the query falls back to scanning the events table directly.
 
 6. **Future: GROUP BY (entity_id, variant)**: Currently the query groups by entity_id only and resolves the variant during precomputation (e.g., argMin for first-seen). This bakes the variant handling strategy into the stored data, so switching between "first seen" and "exclude multiple" requires recomputation. A better approach is to GROUP BY (entity_id, variant) so we store one row per user-variant pair, then resolve multiple variants at query time.
+
+7. **Metric events**: The same lazy computation system also precomputes the metric_events half into `experiment_metric_events_preaggregated`, for ordered funnel metrics (one row per matching event with step indicators in `steps`) and count/sum mean metrics (one row per matching event with its value in `numeric_value`). Eligibility is gated by `_metric_events_precompute_applicable()` in `experiment_query_runner.py`; breakdowns, CUPED, and data warehouse sources fall back to a direct scan of the metric_events CTE.

products/experiments/backend/hogql_queries/experiment_mean_query_builder.py

@@ -6,6 +6,7 @@
 from posthog.hogql.parser import parse_expr, parse_select
 
 from products.experiments.backend.hogql_queries.base_query_utils import (
+    event_or_action_to_filter,
     is_session_property_metric,
     validate_session_property,
 )
@@ -126,20 +127,50 @@ def get_mean_query_common_ctes(self) -> str:
             entity_metric_selects = """
                     {value_agg} AS value"""
 
+        if self._b.metric_events_preaggregation_job_ids:
+            # Read from the precomputed table instead of scanning events. Only reachable
+            # for eligible metrics (no breakdowns/CUPED/DW/session properties), so the
+            # branches above never coexist with this one.
+            # Filter by experiment date range: jobs can cover broader time ranges than
+            # the experiment for cache reusability, so we must filter on read. The upper
+            # bound includes the conversion window since metric events can occur after
+            # experiment end, mirroring build_metric_predicate() on the direct path.
+            # GROUP BY collapses replayed rows by event identity: ReplacingMergeTree only
+            # dedups at merge time and this read doesn't use FINAL, so a re-applied build
+            # INSERT would otherwise double-count sums. Same defense as the exposures read;
+            # the funnel read skips it because funnel evaluation tolerates duplicate events.
+            entity_id_cast = "toUUID(t.entity_id)" if self._b.entity_key == "person_id" else "t.entity_id"
+            metric_events_cte = f"""
+            metric_events AS (
+                SELECT
+                    {entity_id_cast} AS entity_id,
+                    t.timestamp AS timestamp,
+                    any(t.numeric_value) AS value
+                FROM experiment_metric_events_preaggregated AS t
+                WHERE t.job_id IN {{metric_events_job_ids}}
+                    AND t.team_id = {{metric_events_team_id}}
+                    AND t.timestamp >= {{metric_events_date_from}}
+                    AND t.timestamp < {{metric_events_date_to}} + toIntervalSecond({{metric_events_conversion_window_seconds}})
+                GROUP BY t.entity_id, t.timestamp, t.event_uuid
+            )"""
+        else:
+            metric_events_cte = """
+            metric_events AS (
+                SELECT
+                    {entity_key} AS entity_id,
+                    {metric_timestamp_field} AS timestamp,
+                    {value_expr} AS value
+                    -- breakdown columns added programmatically below
+                FROM {metric_table}
+                WHERE {metric_predicate}
+            )"""
+
         return f"""
             exposures AS (
                 {{exposure_select_query}}
             ),
 
-            metric_events AS (
-                SELECT
-                    {{entity_key}} AS entity_id,
-                    {{metric_timestamp_field}} AS timestamp,
-                    {{value_expr}} AS value
-                    -- breakdown columns added programmatically below
-                FROM {{metric_table}}
-                WHERE {{metric_predicate}}
-            ),
+            {metric_events_cte},
 
             entity_metrics AS (
                 SELECT
@@ -220,6 +251,15 @@ def get_mean_query_common_placeholders(self) -> dict:
                 value_expr=self._b._build_windowed_metric_value_expr(cuped_pre_window_predicate)
             )
 
+        if self._b.metric_events_preaggregation_job_ids:
+            placeholders["metric_events_job_ids"] = ast.Constant(value=self._b.metric_events_preaggregation_job_ids)
+            placeholders["metric_events_team_id"] = ast.Constant(value=self._b.team.id)
+            placeholders["metric_events_date_from"] = self._b.date_range_query.date_from_as_hogql()
+            placeholders["metric_events_date_to"] = self._b.date_range_query.date_to_as_hogql()
+            placeholders["metric_events_conversion_window_seconds"] = ast.Constant(
+                value=self._b._get_conversion_window_seconds()
+            )
+
         # Add join condition for data warehouse
         if source_info.kind == "datawarehouse":
             placeholders["join_condition"] = parse_expr(
@@ -245,6 +285,55 @@ def get_session_property_placeholders(self) -> dict:
             "session_conversion_window_predicate": self._b._build_session_conversion_window_predicate(),
         }
 
+    def get_mean_metric_events_query_for_precomputation(self) -> tuple[str, dict[str, ast.Expr]]:
+        """
+        Returns the SELECT query that the lazy computation system wraps in an
+        INSERT INTO experiment_metric_events_preaggregated. This is the write
+        path — it scans the events table and stores one row per matching metric
+        event with its per-event value in numeric_value. The value is already
+        coalesced to a non-null float by _build_value_expr(), so storing it in
+        the non-nullable numeric_value column is lossless.
+
+        The query uses {time_window_min} and {time_window_max} placeholders filled
+        by the lazy computation system for each daily bucket. The experiment date
+        bounds must stay named placeholders (the caller declares experiment_date_to
+        a sentinel) rather than reusing _build_metric_predicate(), which bakes the
+        resolved dates into the AST — a running experiment's moving window end
+        would then change the job hash and defeat cache reuse.
+
+        Returns:
+            Tuple of (query_string, placeholders_dict)
+        """
+        assert isinstance(self._b.metric, ExperimentMeanMetric)
+        source = self._b.metric.source
+        assert isinstance(source, (ActionsNode, EventsNode))
+
+        query_string = """
+            SELECT
+                {entity_key} AS entity_id,
+                timestamp AS timestamp,
+                uuid AS event_uuid,
+                `$session_id` AS session_id,
+                {value_expr} AS numeric_value
+            FROM events
+            WHERE timestamp >= {time_window_min}
+                AND timestamp < {time_window_max}
+                AND timestamp >= {experiment_date_from}
+                AND timestamp < {experiment_date_to} + toIntervalSecond({conversion_window_seconds})
+                AND {metric_event_filter}
+        """
+
+        placeholders: dict[str, ast.Expr] = {
+            "entity_key": parse_expr(self._b.entity_key),
+            "value_expr": self._b._build_value_expr(),
+            "experiment_date_from": self._b.date_range_query.date_from_as_hogql(),
+            "experiment_date_to": self._b.date_range_query.date_to_as_hogql(),
+            "conversion_window_seconds": ast.Constant(value=self._b._get_conversion_window_seconds()),
+            "metric_event_filter": event_or_action_to_filter(self._b.team, source),
+        }
+
+        return query_string, placeholders
+
     def build_mean_query(self) -> ast.SelectQuery:
         """
         Builds query for mean metrics (count, sum, avg, etc.)

products/experiments/backend/hogql_queries/experiment_query_builder.py

@@ -655,19 +655,30 @@ def get_exposure_query_for_precomputation(self) -> tuple[str, dict[str, ast.Expr
         """
         return self._exposure_query_builder().precomputation_query()
 
-    def get_funnel_metric_events_query_for_precomputation(self) -> tuple[str, dict[str, ast.Expr]]:
+    def get_metric_events_query_for_precomputation(self) -> tuple[str, dict[str, ast.Expr]]:
         """
         Returns the SELECT query that the lazy computation system wraps in an
-        INSERT INTO experiment_metric_events_preaggregated. This is the write
-        path — it scans the events table and stores one row per matching event
-        with step indicators packed into an Array(UInt8).
+        INSERT INTO experiment_metric_events_preaggregated, dispatched by metric
+        type. This is the write path — it scans the events table and stores one
+        row per matching event: funnel metrics pack step indicators into an
+        Array(UInt8), mean metrics store the per-event value in numeric_value.
 
         The query uses {time_window_min} and {time_window_max} placeholders filled
         by the lazy computation system for each daily bucket.
 
         Returns:
             Tuple of (query_string, placeholders_dict)
         """
+        match self.metric:
+            case ExperimentFunnelMetric():
+                return self._funnel_query_builder().get_funnel_metric_events_query_for_precomputation()
+            case ExperimentMeanMetric():
+                return self._mean_query_builder().get_mean_metric_events_query_for_precomputation()
+            case _:
+                raise NotImplementedError(f"Metric-events precomputation is not supported for {type(self.metric)}")
+
+    def get_funnel_metric_events_query_for_precomputation(self) -> tuple[str, dict[str, ast.Expr]]:
+        """Funnel-specific write query; prefer get_metric_events_query_for_precomputation()."""
         return self._funnel_query_builder().get_funnel_metric_events_query_for_precomputation()
 
     def _build_variant_expr_for_mean(self) -> ast.Expr:

products/experiments/backend/hogql_queries/experiment_query_runner.py

@@ -11,12 +11,15 @@
 from rest_framework.exceptions import ValidationError
 
 from posthog.schema import (
+    ActionsNode,
     CachedExperimentQueryResponse,
+    EventsNode,
     ExperimentActorsQuery,
     ExperimentBreakdownResult,
     ExperimentDataWarehouseNode,
     ExperimentFunnelMetric,
     ExperimentMeanMetric,
+    ExperimentMetricMathType,
     ExperimentQuery,
     ExperimentQueryResponse,
     ExperimentRatioMetric,
@@ -48,7 +51,11 @@
 )
 from products.cohorts.backend.models.cohort import Cohort
 from products.experiments.backend.hogql_queries import MULTIPLE_VARIANT_KEY, get_baseline_variant_key
-from products.experiments.backend.hogql_queries.base_query_utils import experiment_window, experiment_window_end
+from products.experiments.backend.hogql_queries.base_query_utils import (
+    experiment_window,
+    experiment_window_end,
+    is_session_property_metric,
+)
 from products.experiments.backend.hogql_queries.cuped_config import get_cuped_config
 from products.experiments.backend.hogql_queries.error_handling import experiment_error_handler
 from products.experiments.backend.hogql_queries.experiment_query_builder import (
@@ -349,20 +356,20 @@ def _ensure_exposures_precomputed(self, builder: ExperimentQueryBuilder) -> Lazy
 
     def _ensure_metric_events_precomputed(self, builder: ExperimentQueryBuilder) -> LazyComputationResult:
         """
-        Ensures lazy-computed funnel metric event data exists for this experiment.
+        Ensures lazy-computed metric event data exists for this experiment.
 
-        Stores one row per matching event with step indicators in the
-        experiment_metric_events_preaggregated table.
+        Stores one row per matching event in the experiment_metric_events_preaggregated
+        table: funnel metrics store step indicators, mean metrics the per-event value.
         """
-        query_string, placeholders = builder.get_funnel_metric_events_query_for_precomputation()
+        query_string, placeholders = builder.get_metric_events_query_for_precomputation()
 
         if not self.experiment.start_date:
             raise ValidationError("Experiment must have a start date for lazy computation")
 
         date_from = self.experiment.start_date
         date_to = experiment_window_end(self.experiment, self.as_of)
 
-        # Extend time range by conversion window — funnel step events can occur after experiment end
+        # Extend time range by conversion window — metric events can occur after experiment end
         conversion_window_seconds = builder._get_conversion_window_seconds()
         if conversion_window_seconds > 0:
             date_to = date_to + timedelta(seconds=conversion_window_seconds)
@@ -423,14 +430,26 @@ def _precompute_skip_reason(self) -> Optional[str]:
         return None  # precompute was attempted; a direct path means the build failed / wasn't ready
 
     def _metric_events_precompute_applicable(self) -> bool:
-        """Metric-events precompute only supports ordered funnels without breakdowns, CUPED, or data warehouse."""
-        return (
-            isinstance(self.metric, ExperimentFunnelMetric)
-            and (self.metric.funnel_order_type or "ordered") == "ordered"
-            and not self._get_breakdowns_for_builder()
-            and not self.cuped_config.enabled
-            and not self.is_data_warehouse_query
-        )
+        """
+        Metric-events precompute supports ordered funnels and count/sum-style mean
+        metrics, in both cases without breakdowns, CUPED, or data warehouse sources.
+        """
+        if self._get_breakdowns_for_builder() or self.cuped_config.enabled or self.is_data_warehouse_query:
+            return False
+        if isinstance(self.metric, ExperimentFunnelMetric):
+            return (self.metric.funnel_order_type or "ordered") == "ordered"
+        if isinstance(self.metric, ExperimentMeanMetric):
+            source = self.metric.source
+            if not isinstance(source, (EventsNode, ActionsNode)):
+                return False
+            # Session-property means aggregate via a per-session dedup CTE that the
+            # precomputed table can't feed; ID-valued math (unique session/DAU/group)
+            # and HogQL expressions don't fit the Float64 numeric_value column.
+            if is_session_property_metric(source):
+                return False
+            math_type = getattr(source, "math", None) or ExperimentMetricMathType.TOTAL
+            return math_type in (ExperimentMetricMathType.TOTAL, ExperimentMetricMathType.SUM)
+        return False
 
     def _get_experiment_query(self) -> ast.SelectQuery:
         """
@@ -495,10 +514,11 @@ def _get_experiment_query(self) -> ast.SelectQuery:
                     },
                 )
 
-            # Precompute metric events for ordered funnel metrics. CUPED extends the
-            # funnel scan back by `lookback_days` to source the pre-exposure covariate;
-            # the precomputed metric_events table only covers the experiment window, so
-            # skip precomputation here and let the builder issue a fresh scan.
+            # Precompute metric events for eligible metrics (ordered funnels, count/sum
+            # means). CUPED extends the metric scan back by `lookback_days` to source the
+            # pre-exposure covariate; the precomputed metric_events table only covers the
+            # experiment window, so skip precomputation here and let the builder issue a
+            # fresh scan.
             if self._metric_events_precompute_applicable():
                 try:
                     with tags_context(