Querying Tophog

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:

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

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

   hogli metabase:databases --region eu

Pick "PostHog ClickHouse PROD <REGION> Data Tier" (the data tier, not

the query tier — tophog lives with the events data).

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