feat(weave): add intent cluster assignment tables

[gtarpenning]

Summary

• Clustering is a periodic batch that cannot know its answer when an occurrence is written, so inline cluster_* columns on the occurrence row leave no writer able to produce a complete row. Assignment gets its own tables instead. Stacked on feat(weave): add intent records storage schema #7598, which no longer creates those columns, so 041 is purely additive.
• Assignments key on signature_id, so an occurrence arriving after a run resolves to its cluster with no backfill.
• A run's children are immutable. intent_clusters, intent_cluster_assignments, and intent_cluster_daily are plain MergeTree with no version column. A retry is a new run id rather than a rewrite, so nothing needs replacing, no read needs FINAL, and a partially written run is abandoned and expired rather than repaired. Only intent_cluster_runs keeps a lifecycle, and it holds one row per attempt.
• Immutability has a price the schema now pays explicitly: replacement was the only thing making a retried INSERT idempotent. Every child insert must carry an insert_deduplication_token, and the three child tables set non_replicated_deduplication_window = 1000 so that token is honored on the non-replicated path. Verified: with a shared token a replayed fold stays at one row, without one it lands twice and doubles occurrences.
• intent_cluster_runs keys on (project_id, cluster_run_id) with lens out of the key, so two lenses cannot share a run id here while their children merge into one indistinguishable pile. Readers of promoted_at must collapse versions with argMax before ordering, since a max() over raw RMT rows reports a demoted run as live.
• Immutability is what allows the reverse-lookup projection on assignments. ClickHouse rejects ADD PROJECTION on ReplacingMergeTree outright (deduplicate_merge_projection_mode = throw), so "which signatures are in this cluster" could not be indexed at all while the table stayed RMT.
• A run covers exactly one lens under an opaque id, so lens leaves every child key. Previously intent cluster 0 and failure cluster 0 collided under one run id. This is the same "fold the dimension in so it stays out of ORDER BY" pattern signature_id already uses.
• signature_count per day was wrong on any multi-day read, counting a signature once per day it appeared while users beside it merged correctly. It is now a mergeable state alongside users and conversations. Those states move from uniq to uniqHLL12.
• Every run-scoped table shares one TTL, so a cluster can no longer outlive its own membership, and runs carry an explicit promoted_at pointer so an unpromoted experimental run cannot go live by being newest.
• Splitting signature payload and vectors into a separate table is deliberately not included: measured dedup on real data is 1.24:1, where it returns only 1.70x storage and adds a join. Worth revisiting at ~3:1.

Query patterns

Every read resolves the promoted run first, then hits one of three shapes. No read uses FINAL, and only intent_cluster_runs needs argMax.

-- The run every other query is scoped to. An experimental run that completed
-- more recently is not promoted, so it cannot be picked up here.
SELECT cluster_run_id, pipeline_version, window_start, window_end
FROM (
    SELECT cluster_run_id,
           argMax(status,           record_version) AS status,
           argMax(promoted_at,      record_version) AS promoted_at,
           argMax(pipeline_version, record_version) AS pipeline_version,
           argMax(window_start,     record_version) AS window_start,
           argMax(window_end,       record_version) AS window_end
    FROM intent_cluster_runs
    WHERE project_id = {project_id:String} AND lens = {lens:String}
    GROUP BY cluster_run_id
)
WHERE status = 'complete' AND promoted_at > toDateTime64(0, 3, 'UTC')
ORDER BY promoted_at DESC
LIMIT 1

-- "What are my clusters." Never touches an occurrence row. Every distinct-entity
-- figure merges rather than sums, so a signature or user spanning several days
-- counts once across the range.
SELECT d.cluster_id AS cluster_id, c.label AS label,
       sum(d.occurrences) AS occurrences,
       uniqHLL12Merge(d.signatures) AS signatures
FROM intent_cluster_daily AS d
LEFT JOIN (
    SELECT cluster_id, label
    FROM intent_clusters
    WHERE project_id = {project_id:String} AND cluster_run_id = {cluster_run_id:String}
) AS c USING (cluster_id)
WHERE d.project_id = {project_id:String}
  AND d.cluster_run_id = {cluster_run_id:String}
  AND d.day BETWEEN {day_start:Date} AND {day_end:Date}
  AND d.cluster_id != -1                     -- -1 is the HDBSCAN noise bucket
GROUP BY cluster_id, label
ORDER BY occurrences DESC
LIMIT 25

-- Reach, for the rendered page only. Merging uniq states costs the same whatever
-- the row count, so it is scoped to the ids the page above returned.
SELECT cluster_id,
       uniqHLL12Merge(users) AS users,
       uniqHLL12Merge(conversations) AS conversations
FROM intent_cluster_daily
WHERE project_id = {project_id:String}
  AND cluster_run_id = {cluster_run_id:String}
  AND day BETWEEN {day_start:Date} AND {day_end:Date}
  AND cluster_id IN {cluster_ids:Array(Int32)}
GROUP BY cluster_id

-- "Is this signature in a cluster." Full sorting-key prefix, one granule, no
-- aggregation: a run writes each signature exactly once.
SELECT cluster_id, cluster_confidence, umap_x, umap_y
FROM intent_cluster_assignments
WHERE project_id = {project_id:String}
  AND cluster_run_id = {cluster_run_id:String}
  AND signature_id = unhex({signature_hex:String})

-- "Which signatures are in this cluster." Served by proj_by_cluster as a
-- contiguous range read rather than a scan of the run's every assignment.
SELECT hex(signature_id) AS signature_hex, cluster_confidence, umap_x, umap_y
FROM intent_cluster_assignments
WHERE project_id = {project_id:String}
  AND cluster_run_id = {cluster_run_id:String}
  AND cluster_id = {cluster_id:Int32}
ORDER BY cluster_confidence DESC
LIMIT 50

-- Drill down to the occurrences behind a cluster. Plain IN, not GLOBAL IN: every
-- intent table shards on project_id, so the subquery stays shard-local. Keep the
-- source-time bound, and do NOT dedupe with GROUP BY id: that costs more than
-- the scan it sits on. Dedupe the returned page instead.
SELECT trace_id, span_id, signature, source_started_at
FROM intent_records
WHERE project_id = {project_id:String}
  AND pipeline_version = {pipeline_version:UInt32}
  AND source_started_at >= {start_ts:DateTime64(6)}
  AND source_started_at <  {end_ts:DateTime64(6)}
  AND signature_id IN (
      SELECT signature_id
      FROM intent_cluster_assignments
      WHERE project_id = {project_id:String}
        AND cluster_run_id = {cluster_run_id:String}
        AND cluster_id = {cluster_id:Int32}
  )
ORDER BY source_started_at DESC
LIMIT 100

-- "What cluster is this trace in." Resolve the signature first, then point-look
-- up the full key. Never JOIN with assignments as the right table for a single
-- occurrence: that builds a hash table over the run's every assignment.
SELECT cluster_id, cluster_confidence
FROM intent_cluster_assignments
WHERE project_id = {project_id:String}
  AND cluster_run_id = {cluster_run_id:String}
  AND signature_id IN (
      SELECT signature_id FROM intent_records
      WHERE project_id = {project_id:String} AND trace_id = {trace_id:String}
  )

Writing the rollup

The fold is the run's last step and stays sourced from occurrences. It is chunked
over disjoint cluster_id ranges, which is what keeps its memory bounded:

INSERT INTO intent_cluster_daily (project_id, cluster_run_id, day, cluster_id,
    occurrences, signatures, users, conversations)
SELECT o.project_id, {run:String}, toDate(o.source_started_at), a.cluster_id,
       count(), uniqHLL12State(o.signature_id),
       uniqHLL12State(o.user_id), uniqHLL12State(o.conversation_id)
FROM <occurrences deduplicated per id, scoped to the run's lens,
      pipeline_version and source-time window> AS o
INNER JOIN (
    SELECT signature_id, cluster_id           -- no argMax: assignments are immutable
    FROM intent_cluster_assignments
    WHERE project_id = {project_id:String}
      AND cluster_run_id = {run:String}
      AND cluster_id BETWEEN {lo:Int32} AND {hi:Int32}
) AS a USING (signature_id)
GROUP BY o.project_id, toDate(o.source_started_at), a.cluster_id

Chunk on cluster_id specifically. Chunking on a signature prefix benchmarks
better and is silently wrong: each chunk emits partial aggregates for the same
(day, cluster_id) key, and with nothing to sum them the rollup ends up holding
one arbitrary chunk. Measured, that read back 12.5M of 100M occurrences.

Each chunk insert carries insert_deduplication_token keyed on the run id,
table, and chunk sequence. Block-checksum dedup alone does not cover a replay:
now64() defaults make two logically identical inserts different blocks. The
full writer contract, including who reclaims an abandoned attempt and why
expire_at ordering is an invariant, is stated at the top of the migration.

Measured profile

Local ClickHouse 26.6, one project holding 100M occurrences over 80.6M signatures
(the measured 1.24:1 dedup) across 90 days, 441 clusters, so ~161k signatures per
cluster. Four other projects hold 1M more occurrences so project_id pruning is
real. read_rows and peak memory from system.query_log, granules from
EXPLAIN indexes = 1, warm execution. read_rows sums every table a query touches.

operation	before this pass	after
what are my clusters	13,671 rows / 11 ms	unchanged, and one less argMax
is this signature clustered	8,192 rows / 3 ms	unchanged, 1 granule of 9,845
which signatures in cluster	80.7M rows / 295 ms	188K rows / 7 ms
occurrences in cluster	114M rows / 4,601 ms / 7.62 GiB	145 ms / 38 MiB
reach, 25 rendered clusters	551 MiB / 337 ms	53.5 MiB / 76 ms
fold, whole project	15.39 GiB / 61 s	3.73 GiB / 44.7 s
rollup storage	18.4 KB per row	4.4 KB per row

Costs, stated plainly: the projection adds 1.25 GiB to a 3.06 GiB assignments
table and 26 s to build at 80.6M rows, and uniqHLL12 trades 2.1% error on a
63k-user count for the tenfold memory cut.

Index confirmation

EXPLAIN indexes = 1 pushes the full key prefix down on every read above.

• Rollup reads use the whole intent_cluster_daily key. toYYYYMM(day) prunes parts first, then project_id, cluster_run_id, day, cluster_id prunes granules. Reads stay in the tens of thousands of rows and do not move with occurrence count, which is the reason the table exists.
• Signature lookups use the whole intent_cluster_assignments key: 1 granule of 9,845. Dropping cluster_run_id still prunes to one granule per run, because project_id pins the prefix and signature_id stays sorted inside each run.
• Membership reads resolve to the projection, ReadFromMergeTree (proj_by_cluster), 22 granules of 9,845. The same table still answers signature lookups from its base order, so both directions are served without a second table.
• Occurrence drill-down is a bounded scan, not a point lookup. A cluster's ~161k signatures are a large scattered fraction of the table, so no index can make it cheap: idx_signature_id keeps 123 of 123 granules at that fan-out, while pruning 246 to 5 for a single signature, which is what it is for. What made this read expensive was the GROUP BY id dedup, not the scan: dropping it is 7.62 GiB to 51.9 MiB and 4,601 ms to 551 ms.

What this does not fix

At 1.24:1 dedup, signatures track occurrences at ~0.8x, so assignments are ~80M rows per attempt and the noise bucket alone holds 9.68M signatures. Every signature-keyed structure grows with ingest. More pressing at this scale, 441 clusters over 80.6M signatures means HDBSCAN over 80.6M vectors, which is the real ceiling. If clustering ends up running on a sample, most occurrences have no assignment row and "what cluster is this trace in" returns nothing for them. That is a product decision this schema does not settle.

Testing

• uv run --extra trace_server --group test pytest tests/trace_server_migrator/test_migrator_functional.py tests/trace_server/test_clickhouse_trace_server_migrator.py on local ClickHouse 26.6: 137 passed, 1 pre-existing failure (test_distributed_legacy_replicated_management_db, an exact engine_full assertion differing only by ORDER BY (db_name) parentheses on a newer local server than CI's pinned 26.4).
• The functional test pins all four engines and sorting keys, the shared TTL, the projection's existence and order, that two attempts of a run coexist without replacing each other, that an unpromoted newer run does not go live until promoted, the per-day rollup, that a multi-day read counts a spanning signature and user once, and post-run resolution of a late occurrence. It issues no FINAL.
• It also replays the rollup fold under a shared insert_deduplication_token and asserts the counts do not move, which is the only guard left once immutability removed replacement.
• Note for future edits to this file: prose comments cannot contain a semicolon. test_split_migration_sql_equivalent_on_all_shipped_migrations pins the comment-aware splitter against the old naive ; split, and a semicolon mid-comment makes the two disagree.
• Focused Ruff check/format on the touched test file: passed.

Breaking changes

None. This adds four new tables plus their migrator sharding entries. 041 is unreleased, so these changes edit it in place rather than adding a follow-up migration.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!