Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -546,7 +546,9 @@ def _get_experiment_query(self) -> ast.SelectQuery:
breakdown_injector: MetricBreakdownInjector | None = None
# 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)
migrated_metric = isinstance(
self.metric, ExperimentFunnelMetric | ExperimentMeanMetric | ExperimentRetentionMetric
)
# 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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
data_warehouse_node_to_filter,
event_or_action_to_filter,
)
from products.experiments.backend.hogql_queries.breakdown_injector import BreakdownInjector

if TYPE_CHECKING:
from products.experiments.backend.hogql_queries.experiment_query_builder import ExperimentQueryBuilder
Expand Down Expand Up @@ -359,9 +358,9 @@ def build_retention_query(self) -> ast.SelectQuery:
else:
start_events_cte.expr.having = ast.And(exprs=[start_events_cte.expr.having, retention_maturity])

# Inject breakdown columns if breakdown filter is present. The metric-event injector is
# funnel-only, so retention metrics always use the property-from-exposure BreakdownInjector.
if isinstance(self._b.breakdown_injector, BreakdownInjector):
# Inject breakdown columns if breakdown filter is present. Both injectors expose
# inject_retention_breakdown_columns (old: exposure source; new: start-event source + limit).
if self._b.breakdown_injector:
self._b.breakdown_injector.inject_retention_breakdown_columns(query)

return query
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
ExperimentRatioMetric,
ExperimentRetentionMetric,
MultipleBreakdownType,
StartHandling,
StepOrderValue,
)

Expand Down Expand Up @@ -429,3 +430,56 @@ def _inject_winsorization_breakdown_columns(
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")

def inject_retention_breakdown_columns(self, query: ast.SelectQuery) -> None:
"""Inject breakdown columns into a retention query.

Reads the breakdown off the **start event** (in ``start_events``) and attributes each
user from the start event that anchors their retention window: ``argMin`` (earliest) for
FIRST_SEEN start handling, ``argMax`` (latest) for LAST_SEEN, so the bucket comes from the
same event the window anchors on. Retention is a ratio ``retained / cohort``; the cohort
(denominator) is defined by the start event and every retained user has one, so the start
event is the only attribution source that keeps the per-bucket denominator well-defined.
The completion event is NOT used for the breakdown — non-retained users have no completion
event, which would break the per-bucket denominator.

``start_events`` groups by entity, so the breakdown is deduped per user inside it; the
attributed value is then carried into ``entity_metrics``.
"""
if not self._has_breakdown():
return

aliases = self._get_breakdown_aliases()
breakdown_exprs = self.build_breakdown_exprs(table_alias="")

# Attribute from the start event that anchors the window: latest for LAST_SEEN, earliest
# otherwise, so the breakdown bucket matches the anchoring start event.
assert isinstance(self.metric, ExperimentRetentionMetric)
attribution_agg = "argMax" if self.metric.start_handling == StartHandling.LAST_SEEN else "argMin"

# Read + attribute the breakdown off the start event, deduped per entity.
if query.ctes and "start_events" in query.ctes:
start_events_cte = query.ctes["start_events"]
if isinstance(start_events_cte, ast.CTE) and isinstance(start_events_cte.expr, ast.SelectQuery):
for alias, expr in breakdown_exprs:
start_events_cte.expr.select.append(
ast.Alias(
alias=alias,
expr=ast.Call(name=attribution_agg, args=[expr, ast.Field(chain=["timestamp"])]),
)
)

# Carry the attributed breakdown into entity_metrics and group by it.
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.Field(chain=["start_events", alias]))
)
if entity_metrics_cte.expr.group_by is None:
entity_metrics_cte.expr.group_by = []
for alias in aliases:
entity_metrics_cte.expr.group_by.append(ast.Field(chain=["start_events", alias]))

self._inject_final_breakdown_columns(query, aliases)
Loading
Loading