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:

Column	Type	Notes
timestamp	DateTime64(6)	Flush-window time; always bound it (daily partitions)
metric	LowCardinality(String)	See inventory below
type	LowCardinality(String)	Aggregation semantics: sum (default), max, avg
key	Map(String, String)	The actor: access as key['team_id'], key['distinct_id'], key['partition'], key['session_id']
value	Float64	The aggregated value for this flush window
count	UInt64	Observations in the window
pipeline	LowCardinality(String)	e.g. analytics
lane	LowCardinality(String)	main, overflow, historical, async, turbo
labels	Map(String, String)	Extra non-key labels

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:

Pattern	Reading
High events, normal ms/event	Hot key (volume) — overflow/rebalance is the lever
Low events, high ms/event	Expensive actor — fat person properties, merge-heavy, or contended writes; scaling out will not help

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).