Querying tophog

tophog is the ingestion pipeline's heavy-hitter tracker: workers accumulate per-key aggregates (counts, timers) in memory and periodically flush them to the tophog ClickHouse table via Kafka (clickhouse_tophog topic). It answers "which actor is responsible" questions that fleet-level Prometheus metrics cannot — per-metric label cardinality is unbounded (distinct_id, session_id), so this data lives only in ClickHouse. Retention is 30 days.

The staff-only Django admin has a dashboard over it, but for agent-driven triage query it directly through the internal Metabase.

Access — internal Metabase, never Grafana

The production ClickHouse clusters hold customer data, so there is deliberately no ClickHouse datasource for agents in Grafana. The sanctioned path is the internal Metabase using the engineer's own SSO session — per-person identity, attributable in Metabase's query history, no standing credential. General mechanics live in the query-clickhouse-via-metabase skill; the short version:

1. The user must run login themselves (the agent shell cannot access the Keychain): hogli metabase:login --region eu (or us). macOS will prompt about "Chrome Safe Storage" — that's browser_cookie3 decrypting the browser's cookie store to capture the SSO session; one-time Allow is the right choice.

2. Discover the database id — it is not stable across Metabase rebuilds:

   hogli metabase:databases --region eu

   Pick "PostHog ClickHouse PROD Data Tier" (the data tier, not the query tier — tophog lives with the events data).

3. Run queries; the cookie is read internally and never enters the transcript:

   hogli metabase:query --region eu --database-id <id> <<'SQL'
   SELECT ...
   SQL

Schema

Table tophog (Distributed over sharded_tophog), ordered by (pipeline, lane, metric, timestamp, key), partitioned by day:

One row is one worker's flush window for one (metric, key) — always aggregate on read. Read-side semantics per type (matches the admin dashboard):

CASE type
    WHEN 'max' THEN max(value)
    WHEN 'avg' THEN sum(value * count) / sum(count)
    ELSE sum(value)
END

Metric inventory — discover live, don't trust lists

Metrics are defined inline in the ingestion pipelines (grep topHog( / timer( / sum( in nodejs/src/ingestion/pipelines/analytics/), so the set evolves. Always start with discovery:

SELECT metric, type, count() AS rows, sum(count) AS observations
FROM tophog
WHERE timestamp > now() - INTERVAL 1 HOUR
GROUP BY metric, type
ORDER BY metric

As of 2026-07-06 (master), the analytics metrics include process_persons_time (timer; key: team_id, distinct_id, partition), emitted_events[_per_distinct_id|_per_partition], transformations_run[_per_partition], events_dropped_by_transformation[_per_partition], merge_events_per_distinct_id (merge-intent events: $create_alias / $merge_dangerously with alias, $identify with $anon_distinct_id), group_identify_events_per_distinct_id, resolved_teams, and session-replay *_by_session_id timers.

Dimensions are deploy-gated and rows are immutable: the partition key on process_persons_time and the merge/group-identify metrics merged 2026-07-06 and only exist in data written after that deploy reaches the environment. Check before relying on them:

SELECT countIf(key['partition'] != '') AS with_partition, count() AS total
FROM tophog
WHERE timestamp > now() - INTERVAL 1 HOUR AND metric = 'process_persons_time'

The lens: cost vs volume

This is the load-bearing idea. Volume ranking finds busy actors; cost ranking finds slow ones — and a single lagging partition is usually a cost problem (a real incident: the top-cost actor was invisible in every volume view). Rank by summed timer value, and compute the per-event ratio to classify what you found:

Canned queries

Top actors by person-processing cost (the incident query):

SELECT
    key['team_id'] AS team_id,
    key['distinct_id'] AS distinct_id,
    round(sum(value)) AS total_ms,
    sum(count) AS events,
    round(sum(value) / sum(count), 1) AS ms_per_event,
    arraySort(groupUniqArray(lane)) AS lanes
FROM tophog
WHERE timestamp > now() - INTERVAL 1 HOUR
  AND metric = 'process_persons_time'
GROUP BY team_id, distinct_id
ORDER BY total_ms DESC
LIMIT 10

Scoped to one lagging partition (data written after the partition dimension deployed):

-- add to WHERE:
  AND key['partition'] = '434'

Merge storms (merges are the classic person-processing cost driver):

SELECT key['team_id'] AS team_id, key['distinct_id'] AS distinct_id,
       key['partition'] AS partition, sum(value) AS merge_events
FROM tophog
WHERE timestamp > now() - INTERVAL 1 HOUR
  AND metric = 'merge_events_per_distinct_id'
GROUP BY team_id, distinct_id, partition
ORDER BY merge_events DESC
LIMIT 10

Generic top-10 per metric with correct type semantics (the admin dashboard's query shape) — filter by pipeline / lane as needed:

SELECT metric, type, key, total, obs
FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY metric, type ORDER BY total DESC) AS rn
    FROM (
        SELECT metric, type, key,
               CASE type
                   WHEN 'max' THEN max(value)
                   WHEN 'avg' THEN sum(value * count) / sum(count)
                   ELSE sum(value)
               END AS total,
               sum(count) AS obs
        FROM tophog
        WHERE timestamp > now() - INTERVAL 1 HOUR
          AND pipeline = 'analytics' AND lane = 'main'
        GROUP BY metric, type, key
    )
)
WHERE rn <= 10
ORDER BY metric, type, rn

Cautions

• distinct_id and session_id values are customer PII (often emails). Internal triage use only — never paste them into public PRs, issues, or commit messages.
• Always bound timestamp — the table is partitioned by day and holds 30 days.
• Queries run under the engineer's Metabase identity and appear in their query history.

Related

• monitoring-ingestion-pipeline — the Grafana-side diagnosis, including the single-partition-lag playbook that hands off to this skill for actor identification.
• The pganalyze MCP — the next hop when person processing is implicated (query-level view of the persons Postgres).