Class ResultsQueryBuilder

api/experimentation/results_query.py:116–174

Assembles and decodes the experimentation results ClickHouse query.

class ResultsQueryBuilder:
    """Assembles and decodes the experimentation results ClickHouse query."""

    def __init__(self, specs: Sequence[MetricSpec]) -> None:
        self._slots = [_MetricSlot(spec, i) for i, spec in enumerate(specs)]

    def build_query(self) -> str:
        if not self._slots:
            return _EXPOSURES_COUNT_ONLY_QUERY

        unit_selects = ",\n        ".join(s.unit_select() for s in self._slots)
        outer_selects = ",\n    ".join(s.outer_select() for s in self._slots)

        return (
            _EXPOSURES_CTE
            + f""",
unit_values AS (
    SELECT
        e.variant AS variant,
        {unit_selects}
    FROM exposures AS e
{_METRIC_JOIN}
    WHERE e.quarantined = 0
    GROUP BY e.identifier, e.variant
)
SELECT variant, count() AS n,
    {outer_selects}
FROM unit_values
GROUP BY variant"""
        )

    def add_metric_params(self, params: dict[str, object]) -> None:
        """Add per-metric query parameters into an existing params dict."""
        if not self._slots:
            return
        params["metric_events"] = [s.spec.event for s in self._slots]
        for slot in self._slots:
            params[f"metric_{slot.index}_event"] = slot.spec.event

    def decode_rows(
        self, rows: list[Any], column_names: Sequence[str]
    ) -> tuple[dict[str, int], dict[int, dict[str, VariantStats]]]:
        """Decode raw ClickHouse rows into exposure counts and per-metric stats.

        Columns are located by name, so a missing one raises KeyError rather than
        silently reading a neighbour's value.
        """
        index = {name: position for position, name in enumerate(column_names)}
        exposure_counts: dict[str, int] = {}
        metric_stats: dict[int, dict[str, VariantStats]] = {
            slot.spec.metric_id: {} for slot in self._slots
        }
        for row in rows:
            variant = str(row[index["variant"]])
            n = int(row[index["n"]])
            exposure_counts[variant] = n
            for slot in self._slots:
                metric_stats[slot.spec.metric_id][variant] = slot.decode(n, row, index)