From b3a92d7a4c23e167d695a36e7b1c5517321a6496 Mon Sep 17 00:00:00 2001 From: Ritik Jain Date: Mon, 24 Aug 2026 18:15:27 +0530 Subject: [PATCH 1/3] perf(cohorts): re-key the cohort summary MVs for the criteria that read them profile_event_summary_mv is ordered (project_id, profile_id, name, event_date), and the property variant similarly with property_key after name. Cohort criteria filter on name, a date range and optionally a property, then GROUP BY profile_id. They never filter profile_id, so with profile_id second the usable key prefix ends at project_id and every criterion reads the project's entire slice of the MV however narrow the criterion is. Cohorts run one such query per criterion, on a schedule. Measured on a ~100M-row summary MV: a criterion that can only prune on project_id reads 103,985,861 rows (10.9s); the same criterion with name and event_date in the key prefix reads 115,141 rows (0.11s). Reproduced from scratch on a seeded 373K-row MV: 373,500 rows read vs 28,348. A sort key cannot be altered in place, so migration 20 creates replacement MVs keyed for the consumer: event_profile_summary_mv (project_id, name, event_date, profile_id) event_property_profile_summary_mv (project_id, name, property_key, property_value, event_date, profile_id) The SELECT bodies, the identity filter and the aggregate columns are copied verbatim, so the new tables hold exactly the same rows as the old ones and only the physical order differs. Verified locally: both old and new MVs receive identical row counts from the same inserts. populate: false, with history backfilled by a companion script that is deliberately not a numbered migration so it cannot run inside the migration container. These are AggregatingMergeTree tables, so re-running a populated range double counts; the script is month-partition-aligned with an explicit --replace retry path and requires --until to avoid double counting the window the live trigger already covers. The old MVs are left in place and keep receiving inserts, so this is reversible by pointing cohort.service back at them. Dropping them is a one-line follow-up migration once the new ones are verified. Co-Authored-By: Claude Fable 5 --- .../20-cohort-summary-mv-sort-key.ts | 151 ++++++++++ .../backfill-cohort-summary-mvs.ts | 271 ++++++++++++++++++ packages/db/src/clickhouse/client.ts | 5 + packages/db/src/services/cohort.service.ts | 8 +- packages/db/src/services/delete.service.ts | 2 + 5 files changed, 433 insertions(+), 4 deletions(-) create mode 100644 packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts create mode 100644 packages/db/code-migrations/backfill-cohort-summary-mvs.ts diff --git a/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts b/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts new file mode 100644 index 000000000..dc2821f99 --- /dev/null +++ b/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts @@ -0,0 +1,151 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { TABLE_NAMES } from '../src/clickhouse/client'; +import { + createMaterializedView, + getExistingTables, + runClickhouseMigrationCommands, +} from '../src/clickhouse/migration'; +import { getIsCluster } from './helpers'; + +/** + * Re-key the cohort summary MVs for the queries that actually read them. + * + * profile_event_summary_mv is ordered (project_id, profile_id, name, + * event_date), and the property variant similarly with property_key after + * name. Cohort criteria filter on name, a date range and optionally a + * property, then GROUP BY profile_id โ€” they never filter profile_id. With + * profile_id second, the usable key prefix ends at project_id, so every + * criterion reads the project's entire slice of the MV however narrow the + * criterion is. Cohorts run one such query per criterion, on a schedule. + * + * Measured on a ~100M-row summary MV: a criterion that can only prune on + * project_id reads 103,985,861 rows (10.9s); the same criterion with name + * and event_date in the key prefix reads 115,141 rows (0.11s). + * + * A sort key cannot be altered in place, so this creates replacement MVs + * keyed for the consumer: + * + * event_profile_summary_mv + * (project_id, name, event_date, profile_id) + * event_property_profile_summary_mv + * (project_id, name, property_key, property_value, event_date, profile_id) + * + * The SELECT bodies, the identity filter and the aggregate columns are + * unchanged, so the new tables hold exactly the same rows as the old ones. + * Only the physical order differs. + * + * populate: false โ€” these index events inserted after CREATE. History must + * be backfilled with the companion script, which is deliberately not a + * numbered migration so it cannot run inside the migration container: + * + * packages/db/code-migrations/backfill-cohort-summary-mvs.ts + * + * These are AggregatingMergeTree tables, so re-running an already-populated + * range double counts; the script is month-partition-aligned with an + * explicit --replace path for retries. Until it completes, cohorts with + * relative timeframes compute from partial history, so run it promptly. + * + * The old MVs are left in place and keep receiving inserts. Once the new + * ones are verified, dropping them is a one-line follow-up migration. + */ +export async function up() { + const replicatedVersion = '1'; + const existingTables = await getExistingTables(); + const isClustered = getIsCluster(); + const sqls: string[] = []; + + if ( + !existingTables.includes( + `${TABLE_NAMES.event_profile_summary_mv}_distributed`, + ) && + !existingTables.includes(TABLE_NAMES.event_profile_summary_mv) + ) { + sqls.push( + ...createMaterializedView({ + name: TABLE_NAMES.event_profile_summary_mv, + tableName: 'events', + engine: 'AggregatingMergeTree()', + orderBy: ['project_id', 'name', 'event_date', 'profile_id'], + partitionBy: 'toYYYYMM(event_date)', + query: `SELECT + project_id, + profile_id, + name, + toStartOfDay(created_at) AS event_date, + countState() AS event_count, + minState(created_at) AS first_event_time, + maxState(created_at) AS last_event_time, + sumState(duration) AS total_duration + FROM {events} + WHERE profile_id != device_id + GROUP BY project_id, profile_id, name, event_date`, + distributionHash: 'cityHash64(project_id, profile_id)', + replicatedVersion, + isClustered, + populate: false, + }), + ); + } + + if ( + !existingTables.includes( + `${TABLE_NAMES.event_property_profile_summary_mv}_distributed`, + ) && + !existingTables.includes(TABLE_NAMES.event_property_profile_summary_mv) + ) { + sqls.push( + ...createMaterializedView({ + name: TABLE_NAMES.event_property_profile_summary_mv, + tableName: 'events', + engine: 'AggregatingMergeTree()', + orderBy: [ + 'project_id', + 'name', + 'property_key', + 'property_value', + 'event_date', + 'profile_id', + ], + partitionBy: 'toYYYYMM(event_date)', + query: `SELECT + project_id, + profile_id, + name, + property_key, + property_value, + toStartOfDay(created_at) AS event_date, + countState() AS event_count, + minState(created_at) AS first_event_time, + maxState(created_at) AS last_event_time + FROM {events} + ARRAY JOIN mapKeys(properties) AS property_key, mapValues(properties) AS property_value + WHERE profile_id != device_id + AND property_key != '' + AND property_value != '' + GROUP BY project_id, profile_id, name, property_key, property_value, event_date`, + distributionHash: 'cityHash64(project_id, profile_id)', + replicatedVersion, + isClustered, + populate: false, + }), + ); + } + + fs.writeFileSync( + path.join(import.meta.filename.replace('.ts', '.sql')), + sqls + .map((sql) => sql.trim().replace(/;$/, '').replace(/\n{2,}/g, '\n').concat(';')) + .join('\n\n---\n\n'), + ); + + if (process.argv.includes('--dry')) { + console.log('๐Ÿ” DRY RUN โ€” CREATE statements:'); + for (const sql of sqls) { + console.log(`\n${sql}\n`); + } + return; + } + + await runClickhouseMigrationCommands(sqls); +} diff --git a/packages/db/code-migrations/backfill-cohort-summary-mvs.ts b/packages/db/code-migrations/backfill-cohort-summary-mvs.ts new file mode 100644 index 000000000..9d81288d8 --- /dev/null +++ b/packages/db/code-migrations/backfill-cohort-summary-mvs.ts @@ -0,0 +1,271 @@ +import { TABLE_NAMES } from '../src/clickhouse/client'; +import { + chMigrationClient, + runClickhouseMigrationCommands, +} from '../src/clickhouse/migration'; +import { getIsCluster } from './helpers'; + +/** + * Backfill the re-keyed cohort summary MVs from migration 20 with history. + * + * - event_profile_summary_mv + * - event_property_profile_summary_mv + * + * Both are created with populate: false, so they only index events inserted + * after CREATE. This feeds them everything before that point. + * + * NOT a numbered migration on purpose: migrate.ts only auto-runs files whose + * name starts with a number, so this can never execute inside the migration + * container, where an eviction mid-run would force a full re-run. Run it + * supervised: + * + * CLICKHOUSE_URL=... jiti packages/db/code-migrations/backfill-cohort-summary-mvs.ts --until='YYYY-MM-DD hh:mm:ss' + * + * RESTART SAFETY + * AggregatingMergeTree is not idempotent under re-insert: countState rows + * merge additively, so re-running a range double counts. The safe unit of + * retry is the MONTH, because batches are aligned to the tables' toYYYYMM + * partitions. If a month fails or is interrupted, re-run just that month + * with --replace, which drops the month's partition on the target first. + * + * --until is REQUIRED (except with --dry). Pass the CREATE time of the MVs + * in UTC: events after that are already indexed by the live trigger, so + * backfilling past it double counts the overlap. Find it with: + * SELECT metadata_modification_time FROM system.tables + * WHERE name = 'event_profile_summary_mv' + * + * Flags: + * --dry Print the per-month plan and the first batch; run nothing. + * --from=YYYYMM First month (default: month of min(created_at) in events). + * --to=YYYYMM Last month (default: current month). + * --until=DATETIME Upper bound on created_at (see above). + * --batch-days=N Days per INSERT within a month (default 2). + * --parallel=N Concurrent batches (default 2). + * --replace DROP PARTITION on the target before each month (retry mode). + * --only=summary|property Backfill just one of the two tables. + */ + +const DEFAULT_BATCH_DAYS = 2; +const DEFAULT_PARALLEL = 2; + +// Spill the per-batch GROUP BY rather than OOM on the ARRAY JOIN fan-out; +// max_insert_threads parallelises the part-building stage, which is +// otherwise single-threaded and leaves cores idle during a backfill. +const INSERT_SETTINGS = + 'SETTINGS max_bytes_before_external_group_by = 4294967296, max_insert_threads = 8'; + +type Batch = { label: string; sql: string }; + +function getArg(name: string): string | undefined { + const prefix = `--${name}=`; + return process.argv.find((a) => a.startsWith(prefix))?.slice(prefix.length); +} + +function resolveTarget(base: string, isClustered: boolean): string { + return isClustered ? `${base}_replicated` : base; +} + +function monthStart(yyyymm: number): string { + const y = Math.floor(yyyymm / 100); + const m = yyyymm % 100; + return `${y}-${String(m).padStart(2, '0')}-01 00:00:00`; +} + +function nextMonth(yyyymm: number): number { + const y = Math.floor(yyyymm / 100); + const m = yyyymm % 100; + return m === 12 ? (y + 1) * 100 + 1 : yyyymm + 1; +} + +// Column projections and the identity filter are byte-identical to the MV +// definitions in migration 20; only the time bounds differ. +function summarySelect(start: string, end: string): string { + return `SELECT + project_id, + profile_id, + name, + toStartOfDay(created_at) AS event_date, + countState() AS event_count, + minState(created_at) AS first_event_time, + maxState(created_at) AS last_event_time, + sumState(duration) AS total_duration +FROM events +WHERE created_at >= toDateTime('${start}') + AND created_at < toDateTime('${end}') + AND profile_id != device_id +GROUP BY project_id, profile_id, name, event_date`; +} + +function propertySelect(start: string, end: string): string { + return `SELECT + project_id, + profile_id, + name, + property_key, + property_value, + toStartOfDay(created_at) AS event_date, + countState() AS event_count, + minState(created_at) AS first_event_time, + maxState(created_at) AS last_event_time +FROM events +ARRAY JOIN mapKeys(properties) AS property_key, mapValues(properties) AS property_value +WHERE created_at >= toDateTime('${start}') + AND created_at < toDateTime('${end}') + AND profile_id != device_id + AND property_key != '' + AND property_value != '' +GROUP BY project_id, profile_id, name, property_key, property_value, event_date`; +} + +function monthBatches( + month: number, + until: string, + batchDays: number, + target: string, + select: (start: string, end: string) => string, +): Batch[] { + const batches: Batch[] = []; + const start = new Date(`${monthStart(month).replace(' ', 'T')}Z`); + const monthEnd = new Date(`${monthStart(nextMonth(month)).replace(' ', 'T')}Z`); + const untilDate = new Date(`${until.replace(' ', 'T')}Z`); + const end = monthEnd < untilDate ? monthEnd : untilDate; + + let cursor = new Date(start); + while (cursor < end) { + const next = new Date(cursor); + next.setUTCDate(next.getUTCDate() + batchDays); + const batchEnd = next > end ? end : next; + const s = cursor.toISOString().slice(0, 19).replace('T', ' '); + const e = batchEnd.toISOString().slice(0, 19).replace('T', ' '); + batches.push({ + label: `${s} -> ${e}`, + sql: `INSERT INTO ${target}\n${select(s, e)}\n${INSERT_SETTINGS}`, + }); + cursor = batchEnd; + } + return batches; +} + +// N workers drain the batch list in order. Batches are independent time +// ranges, so ordering does not matter; parts merge asynchronously. +async function runPool( + batches: Batch[], + parallel: number, + onDone: (batch: Batch, seconds: number) => void, +): Promise { + let next = 0; + await Promise.all( + Array.from({ length: Math.max(1, Math.min(parallel, batches.length)) }, async () => { + while (next < batches.length) { + const batch = batches[next++]!; + const t0 = Date.now(); + await chMigrationClient.command({ query: batch.sql }); + onDone(batch, Math.round((Date.now() - t0) / 1000)); + } + }), + ); +} + +async function getMinMonth(): Promise { + const res = await chMigrationClient.query({ + query: 'SELECT toYYYYMM(min(created_at)) AS m FROM events', + format: 'JSONEachRow', + }); + const rows = await res.json<{ m: string }>(); + return Number(rows[0]?.m ?? 0); +} + +export async function up() { + const isClustered = getIsCluster(); + const isDry = process.argv.includes('--dry'); + const replace = process.argv.includes('--replace'); + const only = getArg('only'); + const batchDays = Number.parseInt(getArg('batch-days') ?? String(DEFAULT_BATCH_DAYS), 10); + const parallel = Number.parseInt(getArg('parallel') ?? String(DEFAULT_PARALLEL), 10); + + const now = new Date(); + const currentMonth = now.getUTCFullYear() * 100 + (now.getUTCMonth() + 1); + const fromMonth = Number.parseInt(getArg('from') ?? String(await getMinMonth()), 10); + const toMonth = Number.parseInt(getArg('to') ?? String(currentMonth), 10); + const until = getArg('until'); + + if (!until && !isDry) { + console.error( + 'โŒ --until= is required. Without it the window already indexed by the live MV trigger is double counted.', + ); + process.exit(1); + } + const untilStr = until ?? now.toISOString().slice(0, 19).replace('T', ' '); + + const targets: Array<{ label: string; table: string; select: (s: string, e: string) => string }> = []; + if (only !== 'property') { + targets.push({ + label: TABLE_NAMES.event_profile_summary_mv, + table: resolveTarget(TABLE_NAMES.event_profile_summary_mv, isClustered), + select: summarySelect, + }); + } + if (only !== 'summary') { + targets.push({ + label: TABLE_NAMES.event_property_profile_summary_mv, + table: resolveTarget(TABLE_NAMES.event_property_profile_summary_mv, isClustered), + select: propertySelect, + }); + } + + const months: number[] = []; + for (let m = fromMonth; m <= toMonth; m = nextMonth(m)) { + months.push(m); + } + + console.log(''); + console.log('๐Ÿ“ฆ Cohort summary MV backfill'); + console.log(` Months: ${fromMonth} -> ${toMonth} (${months.length})`); + console.log(` Until: ${untilStr}`); + console.log(` Batch days: ${batchDays} Parallel: ${parallel} Replace: ${replace}`); + console.log(` Targets: ${targets.map((t) => t.label).join(', ')}`); + console.log(` Mode: ${isDry ? 'DRY RUN' : 'EXECUTE'}`); + + if (isDry) { + for (const target of targets) { + const sample = monthBatches(months[0]!, untilStr, batchDays, target.table, target.select); + console.log(`\n-- ${target.label}: ${sample.length} batches for ${months[0]} --`); + console.log(sample[0]?.sql); + } + return; + } + + const startedAt = Date.now(); + for (const target of targets) { + console.log(`\n๐Ÿš€ ${target.label}`); + for (const month of months) { + const batches = monthBatches(month, untilStr, batchDays, target.table, target.select); + if (batches.length === 0) { + continue; + } + if (replace) { + await runClickhouseMigrationCommands([ + `ALTER TABLE ${target.table} DROP PARTITION '${month}'`, + ]); + } + const t0 = Date.now(); + await runPool(batches, parallel, (batch, seconds) => { + console.log(` ยท ${batch.label} (${seconds}s)`); + }); + console.log( + ` โœ… ${month} in ${Math.round((Date.now() - t0) / 1000)}s (${batches.length} batches, elapsed=${Math.round((Date.now() - startedAt) / 1000)}s)`, + ); + } + } + console.log('\nโœ… Backfill complete.'); +} + +// Allow direct execution. +if (import.meta.url === `file://${process.argv[1]}`) { + up() + .then(() => process.exit(0)) + .catch((err) => { + console.error(err); + process.exit(1); + }); +} diff --git a/packages/db/src/clickhouse/client.ts b/packages/db/src/clickhouse/client.ts index fc4941af8..d78a5220c 100644 --- a/packages/db/src/clickhouse/client.ts +++ b/packages/db/src/clickhouse/client.ts @@ -67,6 +67,11 @@ export const TABLE_NAMES = { cohort_members: 'cohort_members', cohort_metadata: 'cohort_metadata', profile_event_summary_mv: 'profile_event_summary_mv', + // Same content as the two MVs above, keyed for the cohort criteria that + // read them (event + window first, profile last) rather than by profile. + // See migration 20. + event_profile_summary_mv: 'event_profile_summary_mv', + event_property_profile_summary_mv: 'event_property_profile_summary_mv', profile_event_property_summary_mv: 'profile_event_property_summary_mv', }; diff --git a/packages/db/src/services/cohort.service.ts b/packages/db/src/services/cohort.service.ts index 8a20fe06d..dacf352ad 100644 --- a/packages/db/src/services/cohort.service.ts +++ b/packages/db/src/services/cohort.service.ts @@ -192,7 +192,7 @@ export function buildEventCriteriaQuery( const frequencyOp = getFrequencyOperator(frequency); return ` SELECT profile_id - FROM ${TABLE_NAMES.profile_event_property_summary_mv} + FROM ${TABLE_NAMES.event_property_profile_summary_mv} WHERE project_id = ${sqlstring.escape(projectId)} AND name = ${sqlstring.escape(name)} AND ${timeConstraint.replace('created_at', 'event_date')} @@ -204,7 +204,7 @@ export function buildEventCriteriaQuery( return ` SELECT DISTINCT profile_id - FROM ${TABLE_NAMES.profile_event_property_summary_mv} + FROM ${TABLE_NAMES.event_property_profile_summary_mv} WHERE project_id = ${sqlstring.escape(projectId)} AND name = ${sqlstring.escape(name)} AND ${timeConstraint.replace('created_at', 'event_date')} @@ -216,7 +216,7 @@ export function buildEventCriteriaQuery( const frequencyOp = getFrequencyOperator(frequency); return ` SELECT profile_id - FROM ${TABLE_NAMES.profile_event_summary_mv} + FROM ${TABLE_NAMES.event_profile_summary_mv} WHERE project_id = ${sqlstring.escape(projectId)} AND name = ${sqlstring.escape(name)} AND ${timeConstraint.replace('created_at', 'event_date')} @@ -227,7 +227,7 @@ export function buildEventCriteriaQuery( return ` SELECT DISTINCT profile_id - FROM ${TABLE_NAMES.profile_event_summary_mv} + FROM ${TABLE_NAMES.event_profile_summary_mv} WHERE project_id = ${sqlstring.escape(projectId)} AND name = ${sqlstring.escape(name)} AND ${timeConstraint.replace('created_at', 'event_date')} diff --git a/packages/db/src/services/delete.service.ts b/packages/db/src/services/delete.service.ts index b8466f333..7dace2640 100644 --- a/packages/db/src/services/delete.service.ts +++ b/packages/db/src/services/delete.service.ts @@ -51,6 +51,8 @@ export async function deleteFromClickhouse(projectIds: string[]) { TABLE_NAMES.cohort_metadata, TABLE_NAMES.profile_event_summary_mv, TABLE_NAMES.profile_event_property_summary_mv, + TABLE_NAMES.event_profile_summary_mv, + TABLE_NAMES.event_property_profile_summary_mv, ]; for (const table of tables) { From 0113bcfc9c3591c8970ce589835a957bdfeeed15 Mon Sep 17 00:00:00 2001 From: Ritik Jain Date: Thu, 27 Aug 2026 17:18:23 +0530 Subject: [PATCH 2/3] feat(cohorts): run the summary MV backfill as an idempotent migration The backfill was a hand-run script: it appended to the MVs, so it needed an exact --until to avoid double counting rows the live trigger had already written, and it refused to run without one. It also mixed topologies, resolving its target to the node-local _replicated while reading from the Distributed events. Make it migration 21 and rebuild rather than append. Each month's partition is dropped and rebuilt from events, which is idempotent by construction: an evicted migration pod is resumed by running again, and a month that was double counted heals on the next pass. That in turn removes the need to know when the MVs were created, so no operator input is required and a small deployment gets working cohorts with no flags. The bound is now read per month, straight after that month's drop. A run can take hours and the current month is dropped at the end of it, so a single bound taken at the start would delete every row the trigger wrote during the run and then decline to rebuild them. Boundary strings keep milliseconds for the same reason: created_at is DateTime64(3), and truncating the final batch to whole seconds drops the events in the boundary second after their trigger rows are gone. Reads and writes both go through the Distributed tables, so one run from one node covers every shard. DROP PARTITION is the exception, since the Distributed engine rejects partitioning, so it goes ON CLUSTER against the local table. Rebuilding a large events table unattended is still not wanted, so it steps aside above COHORT_BACKFILL_MAX_EVENTS rows and prints the manual command. The same file remains directly executable for supervised runs. Verified on standalone 26.1.3 and a keeper-backed 2-shard cluster 25.3: fresh install no-ops, history plus live-trigger overlap matches ground truth computed from events for both tables, one run writes both shards, re-runs from either node are stable, and a deliberately double-counted month returns to exact. Co-Authored-By: Claude Opus 5 --- .../20-cohort-summary-mv-sort-key.ts | 12 +- .../21-backfill-cohort-summary-mvs.ts | 402 ++++++++++++++++++ .../backfill-cohort-summary-mvs.ts | 271 ------------ 3 files changed, 404 insertions(+), 281 deletions(-) create mode 100644 packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts delete mode 100644 packages/db/code-migrations/backfill-cohort-summary-mvs.ts diff --git a/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts b/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts index dc2821f99..786c6ffb4 100644 --- a/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts +++ b/packages/db/code-migrations/20-cohort-summary-mv-sort-key.ts @@ -35,16 +35,8 @@ import { getIsCluster } from './helpers'; * unchanged, so the new tables hold exactly the same rows as the old ones. * Only the physical order differs. * - * populate: false โ€” these index events inserted after CREATE. History must - * be backfilled with the companion script, which is deliberately not a - * numbered migration so it cannot run inside the migration container: - * - * packages/db/code-migrations/backfill-cohort-summary-mvs.ts - * - * These are AggregatingMergeTree tables, so re-running an already-populated - * range double counts; the script is month-partition-aligned with an - * explicit --replace path for retries. Until it completes, cohorts with - * relative timeframes compute from partial history, so run it promptly. + * populate: false โ€” these index events inserted after CREATE. History is + * filled by migration 21, which rebuilds them month by month from events. * * The old MVs are left in place and keep receiving inserts. Once the new * ones are verified, dropping them is a one-line follow-up migration. diff --git a/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts b/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts new file mode 100644 index 000000000..87be81c38 --- /dev/null +++ b/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts @@ -0,0 +1,402 @@ +import { TABLE_NAMES } from '../src/clickhouse/client'; +import { + chMigrationClient, + runClickhouseMigrationCommands, +} from '../src/clickhouse/migration'; +import { getIsCluster } from './helpers'; + +/** + * Fill the re-keyed cohort summary MVs from migration 20 with history. + * + * - event_profile_summary_mv + * - event_property_profile_summary_mv + * + * Both are created with populate: false, so the live trigger only indexes + * events inserted after CREATE. Everything older has to be aggregated from + * the events table, which is what this does. + * + * REBUILD, NOT APPEND + * Each month's partition is dropped and rebuilt from events rather than + * appended to. These are AggregatingMergeTree tables, so an append that + * overlaps rows the trigger already wrote would double count them; a + * rebuild cannot, because the trigger's copies for that month are dropped + * first. That makes the work idempotent: an interrupted run is resumed by + * running it again, and a month can be redone at any time. + * + * It also removes the need to know when the MVs were created. Each month is + * bounded by a now64(3) read immediately after that month's DROP: rows older + * than the bound had their trigger-written copies dropped and are rebuilt + * here, rows newer are left to the trigger. Each event is counted once. + * + * The bound has to be per month, not per run. A run can take hours, and the + * current month is dropped at the end of it, so a bound captured at the + * start would delete every row the trigger wrote while the run was going and + * then decline to rebuild them. Taking it after each drop narrows the + * exposure to the round trip between the drop and the read, in which an + * arriving event can be counted twice. Pass --until to cut somewhere fixed + * instead, which is what a supervised re-run of a closed month wants. + * + * The tradeoff a rebuild makes is visibility: while a month is being + * rebuilt its partition is incomplete, so a cohort computed in that window + * under-counts that month. It converges as soon as the month finishes. + * + * AUTOMATIC BY DEFAULT + * This runs as a normal migration so a new install ends up with working + * cohorts without anyone reading this file. Because the rebuild is + * idempotent, an evicted migration pod is safe: the migration is not + * recorded, and the next boot redoes it. + * + * Rebuilding the full history of a large events table is not something to + * start unattended, though, so it steps aside above + * COHORT_BACKFILL_MAX_EVENTS rows (default 100,000,000, roughly the point + * where this stops being minutes) and prints the command to run by hand. + * Raise the variable, or pass --force, to run it anyway. + * + * MANUAL USE + * The same file is the supervised tool. Run it directly to control when + * the work happens, or to redo part of it: + * + * CLICKHOUSE_URL=... jiti packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts + * + * Flags: + * --dry Print the plan and the first batch; run nothing. + * --force Ignore COHORT_BACKFILL_MAX_EVENTS. + * --from=YYYYMM First month (default: month of min(created_at)). + * --to=YYYYMM Last month (default: current month). + * --until=DATETIME Fixed upper bound on created_at for every month + * (default: now64(3) taken after each month's drop). + * --batch-days=N Days per INSERT within a month (default 2). + * --parallel=N Concurrent batches (default 2). + * --only=summary|property Rebuild just one of the two tables. + * + * CLUSTERS + * Reads and writes go through the Distributed tables, so one run from one + * node covers every shard. The one statement that cannot be distributed is + * DROP PARTITION, which the Distributed engine rejects outright, so that + * goes ON CLUSTER against the local table instead. Still a single run. + */ + +const DEFAULT_BATCH_DAYS = 2; +const DEFAULT_PARALLEL = 2; +const DEFAULT_MAX_EVENTS = 100_000_000; + +// Spill the per-batch GROUP BY rather than OOM on the ARRAY JOIN fan-out; +// max_insert_threads parallelises the part-building stage, which is +// otherwise single-threaded and leaves cores idle during a rebuild. +// distributed_foreground_insert so a finished batch means the rows have +// actually landed on their shards, not that they were queued for delivery. +const INSERT_SETTINGS = `SETTINGS + max_bytes_before_external_group_by = 4294967296, + max_insert_threads = 8, + distributed_foreground_insert = 1`; + +type Batch = { label: string; sql: string }; + +function getArg(name: string): string | undefined { + const prefix = `--${name}=`; + return process.argv.find((a) => a.startsWith(prefix))?.slice(prefix.length); +} + +function getPositiveInt(value: string | undefined, fallback: number): number { + if (value === undefined) { + return fallback; + } + if (!/^\d+$/.test(value)) { + throw new Error(`expected a positive integer, got "${value}"`); + } + const parsed = Number.parseInt(value, 10); + if (!Number.isSafeInteger(parsed) || parsed < 1) { + throw new Error(`expected a positive integer, got "${value}"`); + } + return parsed; +} + +// DROP PARTITION is the one statement that cannot go through the Distributed +// table: the engine rejects partitioning outright. ON CLUSTER against the +// local table reaches every shard in a single statement instead. +function dropPartition( + table: string, + month: number, + isClustered: boolean, +): string { + return isClustered + ? `ALTER TABLE ${table}_replicated ON CLUSTER '{cluster}' DROP PARTITION '${month}'` + : `ALTER TABLE ${table} DROP PARTITION '${month}'`; +} + +function monthStart(yyyymm: number): string { + const y = Math.floor(yyyymm / 100); + const m = yyyymm % 100; + return `${y}-${String(m).padStart(2, '0')}-01 00:00:00`; +} + +function nextMonth(yyyymm: number): number { + const y = Math.floor(yyyymm / 100); + const m = yyyymm % 100; + return m === 12 ? (y + 1) * 100 + 1 : yyyymm + 1; +} + +// Column projections and the identity filter are byte-identical to the MV +// definitions in migration 20; only the time bounds differ. +function summarySelect(start: string, end: string): string { + return `SELECT + project_id, + profile_id, + name, + toStartOfDay(created_at) AS event_date, + countState() AS event_count, + minState(created_at) AS first_event_time, + maxState(created_at) AS last_event_time, + sumState(duration) AS total_duration +FROM ${TABLE_NAMES.events} +WHERE created_at >= toDateTime64('${start}', 3) + AND created_at < toDateTime64('${end}', 3) + AND profile_id != device_id +GROUP BY project_id, profile_id, name, event_date`; +} + +function propertySelect(start: string, end: string): string { + return `SELECT + project_id, + profile_id, + name, + property_key, + property_value, + toStartOfDay(created_at) AS event_date, + countState() AS event_count, + minState(created_at) AS first_event_time, + maxState(created_at) AS last_event_time +FROM ${TABLE_NAMES.events} +ARRAY JOIN mapKeys(properties) AS property_key, mapValues(properties) AS property_value +WHERE created_at >= toDateTime64('${start}', 3) + AND created_at < toDateTime64('${end}', 3) + AND profile_id != device_id + AND property_key != '' + AND property_value != '' +GROUP BY project_id, profile_id, name, property_key, property_value, event_date`; +} + +function monthBatches( + month: number, + until: string, + batchDays: number, + target: string, + select: (start: string, end: string) => string, +): Batch[] { + const batches: Batch[] = []; + const start = new Date(`${monthStart(month).replace(' ', 'T')}Z`); + const monthEnd = new Date( + `${monthStart(nextMonth(month)).replace(' ', 'T')}Z`, + ); + const untilDate = new Date(`${until.replace(' ', 'T')}Z`); + const end = monthEnd < untilDate ? monthEnd : untilDate; + + let cursor = new Date(start); + while (cursor < end) { + const next = new Date(cursor); + next.setUTCDate(next.getUTCDate() + batchDays); + const batchEnd = next > end ? end : next; + // Keep milliseconds: created_at is DateTime64(3) and the final batch end + // is the month's bound, so truncating to whole seconds would drop every + // event in the boundary second after its trigger row had been dropped. + const s = cursor.toISOString().slice(0, 23).replace('T', ' '); + const e = batchEnd.toISOString().slice(0, 23).replace('T', ' '); + batches.push({ + label: `${s} -> ${e}`, + sql: `INSERT INTO ${target}\n${select(s, e)}\n${INSERT_SETTINGS}`, + }); + cursor = batchEnd; + } + return batches; +} + +// N workers drain the batch list in order. Batches are independent time +// ranges within an already-dropped month, so ordering does not matter. +async function runPool( + batches: Batch[], + parallel: number, + onDone: (batch: Batch, seconds: number) => void, +): Promise { + let next = 0; + await Promise.all( + Array.from( + { length: Math.max(1, Math.min(parallel, batches.length)) }, + async () => { + while (next < batches.length) { + const batch = batches[next++]!; + const startedAt = Date.now(); + await chMigrationClient.command({ query: batch.sql }); + onDone(batch, Math.round((Date.now() - startedAt) / 1000)); + } + }, + ), + ); +} + +async function scalar(query: string): Promise { + const res = await chMigrationClient.query({ query, format: 'JSONEachRow' }); + const rows = await res.json>(); + return rows[0] ? Object.values(rows[0])[0] : undefined; +} + +export async function up() { + const isClustered = getIsCluster(); + const isDry = process.argv.includes('--dry'); + const force = process.argv.includes('--force'); + const only = getArg('only'); + const batchDays = getPositiveInt(getArg('batch-days'), DEFAULT_BATCH_DAYS); + const parallel = getPositiveInt(getArg('parallel'), DEFAULT_PARALLEL); + const maxEvents = getPositiveInt( + process.env.COHORT_BACKFILL_MAX_EVENTS, + DEFAULT_MAX_EVENTS, + ); + + const totalEvents = Number( + (await scalar(`SELECT count() AS c FROM ${TABLE_NAMES.events}`)) ?? 0, + ); + if (totalEvents === 0) { + console.log('๐Ÿ“ฆ Cohort summary MVs: no events to aggregate, nothing to do'); + return; + } + if (totalEvents > maxEvents && !force && !isDry) { + console.log(''); + console.log( + `โญ๏ธ Cohort summary MVs: skipping the automatic rebuild (${totalEvents.toLocaleString()} events > COHORT_BACKFILL_MAX_EVENTS=${maxEvents.toLocaleString()}).`, + ); + console.log( + ' Cohorts compute from partial history until this is run. Run it when it suits you:', + ); + console.log( + ' CLICKHOUSE_URL=... jiti packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts', + ); + console.log( + ' It rebuilds month by month and is safe to interrupt and re-run.', + ); + console.log(''); + return; + } + + // A fixed --until applies to every month. Without one, each month takes its + // own bound straight after its drop (see the header) so a long run cannot + // lose the rows the trigger wrote while it was running. Read from + // ClickHouse rather than the local clock to stay in created_at's clock + // domain. + const fixedUntil = getArg('until'); + const planningUntil = + fixedUntil ?? (await scalar('SELECT toString(now64(3)) AS t')); + if (!planningUntil) { + throw new Error('could not resolve the rebuild upper bound'); + } + + const currentMonth = Number( + await scalar( + `SELECT toString(toYYYYMM(toDateTime64('${planningUntil}', 3))) AS m`, + ), + ); + const fromMonth = getPositiveInt( + getArg('from') ?? + (await scalar( + `SELECT toString(toYYYYMM(min(created_at))) AS m FROM ${TABLE_NAMES.events}`, + )), + currentMonth, + ); + const toMonth = getPositiveInt(getArg('to'), currentMonth); + + const targets: Array<{ + label: string; + table: string; + select: (start: string, end: string) => string; + }> = []; + if (only !== 'property') { + targets.push({ + label: TABLE_NAMES.event_profile_summary_mv, + table: TABLE_NAMES.event_profile_summary_mv, + select: summarySelect, + }); + } + if (only !== 'summary') { + targets.push({ + label: TABLE_NAMES.event_property_profile_summary_mv, + table: TABLE_NAMES.event_property_profile_summary_mv, + select: propertySelect, + }); + } + + const months: number[] = []; + for (let m = fromMonth; m <= toMonth; m = nextMonth(m)) { + months.push(m); + } + + console.log(''); + console.log('๐Ÿ“ฆ Cohort summary MV rebuild'); + console.log(` Events: ${totalEvents.toLocaleString()}`); + console.log(` Months: ${fromMonth} -> ${toMonth} (${months.length})`); + console.log(` Until: ${fixedUntil ?? 'per month, after each drop'}`); + console.log(` Batch days: ${batchDays} Parallel: ${parallel}`); + console.log(` Targets: ${targets.map((t) => t.label).join(', ')}`); + console.log(` Clustered: ${isClustered}`); + console.log(` Mode: ${isDry ? 'DRY RUN' : 'EXECUTE'}`); + + if (isDry) { + for (const target of targets) { + const sample = monthBatches( + months[0]!, + planningUntil, + batchDays, + target.table, + target.select, + ); + console.log( + `\n-- ${target.label}: ${sample.length} batches for ${months[0]} --`, + ); + console.log(dropPartition(target.table, months[0]!, isClustered)); + console.log(sample[0]?.sql); + } + return; + } + + const startedAt = Date.now(); + for (const target of targets) { + console.log(`\n๐Ÿš€ ${target.label}`); + for (const month of months) { + // Drop first, then take the bound, so anything the trigger writes from + // here on is kept by the trigger and excluded from the rebuild. + await runClickhouseMigrationCommands([ + dropPartition(target.table, month, isClustered), + ]); + const until = + fixedUntil ?? (await scalar('SELECT toString(now64(3)) AS t')); + if (!until) { + throw new Error(`${month}: could not resolve the rebuild upper bound`); + } + const batches = monthBatches( + month, + until, + batchDays, + target.table, + target.select, + ); + if (batches.length === 0) { + continue; + } + const monthStartedAt = Date.now(); + await runPool(batches, parallel, (batch, seconds) => { + console.log(` ยท ${batch.label} (${seconds}s)`); + }); + console.log( + ` โœ… ${month} in ${Math.round((Date.now() - monthStartedAt) / 1000)}s (${batches.length} batches, elapsed=${Math.round((Date.now() - startedAt) / 1000)}s)`, + ); + } + } + console.log('\nโœ… Rebuild complete.'); +} + +// Allow direct execution for the supervised path. +if (import.meta.url === `file://${process.argv[1]}`) { + up() + .then(() => process.exit(0)) + .catch((err) => { + console.error(err); + process.exit(1); + }); +} diff --git a/packages/db/code-migrations/backfill-cohort-summary-mvs.ts b/packages/db/code-migrations/backfill-cohort-summary-mvs.ts deleted file mode 100644 index 9d81288d8..000000000 --- a/packages/db/code-migrations/backfill-cohort-summary-mvs.ts +++ /dev/null @@ -1,271 +0,0 @@ -import { TABLE_NAMES } from '../src/clickhouse/client'; -import { - chMigrationClient, - runClickhouseMigrationCommands, -} from '../src/clickhouse/migration'; -import { getIsCluster } from './helpers'; - -/** - * Backfill the re-keyed cohort summary MVs from migration 20 with history. - * - * - event_profile_summary_mv - * - event_property_profile_summary_mv - * - * Both are created with populate: false, so they only index events inserted - * after CREATE. This feeds them everything before that point. - * - * NOT a numbered migration on purpose: migrate.ts only auto-runs files whose - * name starts with a number, so this can never execute inside the migration - * container, where an eviction mid-run would force a full re-run. Run it - * supervised: - * - * CLICKHOUSE_URL=... jiti packages/db/code-migrations/backfill-cohort-summary-mvs.ts --until='YYYY-MM-DD hh:mm:ss' - * - * RESTART SAFETY - * AggregatingMergeTree is not idempotent under re-insert: countState rows - * merge additively, so re-running a range double counts. The safe unit of - * retry is the MONTH, because batches are aligned to the tables' toYYYYMM - * partitions. If a month fails or is interrupted, re-run just that month - * with --replace, which drops the month's partition on the target first. - * - * --until is REQUIRED (except with --dry). Pass the CREATE time of the MVs - * in UTC: events after that are already indexed by the live trigger, so - * backfilling past it double counts the overlap. Find it with: - * SELECT metadata_modification_time FROM system.tables - * WHERE name = 'event_profile_summary_mv' - * - * Flags: - * --dry Print the per-month plan and the first batch; run nothing. - * --from=YYYYMM First month (default: month of min(created_at) in events). - * --to=YYYYMM Last month (default: current month). - * --until=DATETIME Upper bound on created_at (see above). - * --batch-days=N Days per INSERT within a month (default 2). - * --parallel=N Concurrent batches (default 2). - * --replace DROP PARTITION on the target before each month (retry mode). - * --only=summary|property Backfill just one of the two tables. - */ - -const DEFAULT_BATCH_DAYS = 2; -const DEFAULT_PARALLEL = 2; - -// Spill the per-batch GROUP BY rather than OOM on the ARRAY JOIN fan-out; -// max_insert_threads parallelises the part-building stage, which is -// otherwise single-threaded and leaves cores idle during a backfill. -const INSERT_SETTINGS = - 'SETTINGS max_bytes_before_external_group_by = 4294967296, max_insert_threads = 8'; - -type Batch = { label: string; sql: string }; - -function getArg(name: string): string | undefined { - const prefix = `--${name}=`; - return process.argv.find((a) => a.startsWith(prefix))?.slice(prefix.length); -} - -function resolveTarget(base: string, isClustered: boolean): string { - return isClustered ? `${base}_replicated` : base; -} - -function monthStart(yyyymm: number): string { - const y = Math.floor(yyyymm / 100); - const m = yyyymm % 100; - return `${y}-${String(m).padStart(2, '0')}-01 00:00:00`; -} - -function nextMonth(yyyymm: number): number { - const y = Math.floor(yyyymm / 100); - const m = yyyymm % 100; - return m === 12 ? (y + 1) * 100 + 1 : yyyymm + 1; -} - -// Column projections and the identity filter are byte-identical to the MV -// definitions in migration 20; only the time bounds differ. -function summarySelect(start: string, end: string): string { - return `SELECT - project_id, - profile_id, - name, - toStartOfDay(created_at) AS event_date, - countState() AS event_count, - minState(created_at) AS first_event_time, - maxState(created_at) AS last_event_time, - sumState(duration) AS total_duration -FROM events -WHERE created_at >= toDateTime('${start}') - AND created_at < toDateTime('${end}') - AND profile_id != device_id -GROUP BY project_id, profile_id, name, event_date`; -} - -function propertySelect(start: string, end: string): string { - return `SELECT - project_id, - profile_id, - name, - property_key, - property_value, - toStartOfDay(created_at) AS event_date, - countState() AS event_count, - minState(created_at) AS first_event_time, - maxState(created_at) AS last_event_time -FROM events -ARRAY JOIN mapKeys(properties) AS property_key, mapValues(properties) AS property_value -WHERE created_at >= toDateTime('${start}') - AND created_at < toDateTime('${end}') - AND profile_id != device_id - AND property_key != '' - AND property_value != '' -GROUP BY project_id, profile_id, name, property_key, property_value, event_date`; -} - -function monthBatches( - month: number, - until: string, - batchDays: number, - target: string, - select: (start: string, end: string) => string, -): Batch[] { - const batches: Batch[] = []; - const start = new Date(`${monthStart(month).replace(' ', 'T')}Z`); - const monthEnd = new Date(`${monthStart(nextMonth(month)).replace(' ', 'T')}Z`); - const untilDate = new Date(`${until.replace(' ', 'T')}Z`); - const end = monthEnd < untilDate ? monthEnd : untilDate; - - let cursor = new Date(start); - while (cursor < end) { - const next = new Date(cursor); - next.setUTCDate(next.getUTCDate() + batchDays); - const batchEnd = next > end ? end : next; - const s = cursor.toISOString().slice(0, 19).replace('T', ' '); - const e = batchEnd.toISOString().slice(0, 19).replace('T', ' '); - batches.push({ - label: `${s} -> ${e}`, - sql: `INSERT INTO ${target}\n${select(s, e)}\n${INSERT_SETTINGS}`, - }); - cursor = batchEnd; - } - return batches; -} - -// N workers drain the batch list in order. Batches are independent time -// ranges, so ordering does not matter; parts merge asynchronously. -async function runPool( - batches: Batch[], - parallel: number, - onDone: (batch: Batch, seconds: number) => void, -): Promise { - let next = 0; - await Promise.all( - Array.from({ length: Math.max(1, Math.min(parallel, batches.length)) }, async () => { - while (next < batches.length) { - const batch = batches[next++]!; - const t0 = Date.now(); - await chMigrationClient.command({ query: batch.sql }); - onDone(batch, Math.round((Date.now() - t0) / 1000)); - } - }), - ); -} - -async function getMinMonth(): Promise { - const res = await chMigrationClient.query({ - query: 'SELECT toYYYYMM(min(created_at)) AS m FROM events', - format: 'JSONEachRow', - }); - const rows = await res.json<{ m: string }>(); - return Number(rows[0]?.m ?? 0); -} - -export async function up() { - const isClustered = getIsCluster(); - const isDry = process.argv.includes('--dry'); - const replace = process.argv.includes('--replace'); - const only = getArg('only'); - const batchDays = Number.parseInt(getArg('batch-days') ?? String(DEFAULT_BATCH_DAYS), 10); - const parallel = Number.parseInt(getArg('parallel') ?? String(DEFAULT_PARALLEL), 10); - - const now = new Date(); - const currentMonth = now.getUTCFullYear() * 100 + (now.getUTCMonth() + 1); - const fromMonth = Number.parseInt(getArg('from') ?? String(await getMinMonth()), 10); - const toMonth = Number.parseInt(getArg('to') ?? String(currentMonth), 10); - const until = getArg('until'); - - if (!until && !isDry) { - console.error( - 'โŒ --until= is required. Without it the window already indexed by the live MV trigger is double counted.', - ); - process.exit(1); - } - const untilStr = until ?? now.toISOString().slice(0, 19).replace('T', ' '); - - const targets: Array<{ label: string; table: string; select: (s: string, e: string) => string }> = []; - if (only !== 'property') { - targets.push({ - label: TABLE_NAMES.event_profile_summary_mv, - table: resolveTarget(TABLE_NAMES.event_profile_summary_mv, isClustered), - select: summarySelect, - }); - } - if (only !== 'summary') { - targets.push({ - label: TABLE_NAMES.event_property_profile_summary_mv, - table: resolveTarget(TABLE_NAMES.event_property_profile_summary_mv, isClustered), - select: propertySelect, - }); - } - - const months: number[] = []; - for (let m = fromMonth; m <= toMonth; m = nextMonth(m)) { - months.push(m); - } - - console.log(''); - console.log('๐Ÿ“ฆ Cohort summary MV backfill'); - console.log(` Months: ${fromMonth} -> ${toMonth} (${months.length})`); - console.log(` Until: ${untilStr}`); - console.log(` Batch days: ${batchDays} Parallel: ${parallel} Replace: ${replace}`); - console.log(` Targets: ${targets.map((t) => t.label).join(', ')}`); - console.log(` Mode: ${isDry ? 'DRY RUN' : 'EXECUTE'}`); - - if (isDry) { - for (const target of targets) { - const sample = monthBatches(months[0]!, untilStr, batchDays, target.table, target.select); - console.log(`\n-- ${target.label}: ${sample.length} batches for ${months[0]} --`); - console.log(sample[0]?.sql); - } - return; - } - - const startedAt = Date.now(); - for (const target of targets) { - console.log(`\n๐Ÿš€ ${target.label}`); - for (const month of months) { - const batches = monthBatches(month, untilStr, batchDays, target.table, target.select); - if (batches.length === 0) { - continue; - } - if (replace) { - await runClickhouseMigrationCommands([ - `ALTER TABLE ${target.table} DROP PARTITION '${month}'`, - ]); - } - const t0 = Date.now(); - await runPool(batches, parallel, (batch, seconds) => { - console.log(` ยท ${batch.label} (${seconds}s)`); - }); - console.log( - ` โœ… ${month} in ${Math.round((Date.now() - t0) / 1000)}s (${batches.length} batches, elapsed=${Math.round((Date.now() - startedAt) / 1000)}s)`, - ); - } - } - console.log('\nโœ… Backfill complete.'); -} - -// Allow direct execution. -if (import.meta.url === `file://${process.argv[1]}`) { - up() - .then(() => process.exit(0)) - .catch((err) => { - console.error(err); - process.exit(1); - }); -} From 70427bf8058da2421edd29390b06eed29bbf4989 Mon Sep 17 00:00:00 2001 From: Ritik Jain Date: Thu, 27 Aug 2026 17:59:24 +0530 Subject: [PATCH 3/3] feat(cohorts): drop the size guard, rebuild newest month first The COHORT_BACKFILL_MAX_EVENTS threshold and its --force escape hatch are gone. The constant was an invented number standing in for "how long will this take", which row count is a poor proxy for, and it made the automatic path conditional on a guess. Months now rebuild newest first. Cohort criteria mostly use relative timeframes, so the recent months are the ones that make cohorts correct again, and an interrupted run leaves the useful end done. Also drop the caveat about cohorts under-counting mid-rebuild needing attention: the cohortRefresh cron already recomputes every non-static cohort every 30 minutes, so it resolves on its own. Re-verified on standalone 26.1.3 and a 2-shard cluster 25.3 from empty databases: both tables match ground truth computed from events, re-runs from either node are stable, and a deliberately double-counted month returns to exact. Co-Authored-By: Claude Opus 5 --- .../21-backfill-cohort-summary-mvs.ts | 41 ++++--------------- 1 file changed, 8 insertions(+), 33 deletions(-) diff --git a/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts b/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts index 87be81c38..0206fe8c7 100644 --- a/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts +++ b/packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts @@ -38,7 +38,10 @@ import { getIsCluster } from './helpers'; * * The tradeoff a rebuild makes is visibility: while a month is being * rebuilt its partition is incomplete, so a cohort computed in that window - * under-counts that month. It converges as soon as the month finishes. + * under-counts that month. Nothing needs doing about it, the cohortRefresh + * cron recomputes every non-static cohort every 30 minutes. Months are + * rebuilt newest first so the ones cohorts actually read, which mostly use + * relative timeframes, are correct soonest. * * AUTOMATIC BY DEFAULT * This runs as a normal migration so a new install ends up with working @@ -46,12 +49,6 @@ import { getIsCluster } from './helpers'; * idempotent, an evicted migration pod is safe: the migration is not * recorded, and the next boot redoes it. * - * Rebuilding the full history of a large events table is not something to - * start unattended, though, so it steps aside above - * COHORT_BACKFILL_MAX_EVENTS rows (default 100,000,000, roughly the point - * where this stops being minutes) and prints the command to run by hand. - * Raise the variable, or pass --force, to run it anyway. - * * MANUAL USE * The same file is the supervised tool. Run it directly to control when * the work happens, or to redo part of it: @@ -60,7 +57,6 @@ import { getIsCluster } from './helpers'; * * Flags: * --dry Print the plan and the first batch; run nothing. - * --force Ignore COHORT_BACKFILL_MAX_EVENTS. * --from=YYYYMM First month (default: month of min(created_at)). * --to=YYYYMM Last month (default: current month). * --until=DATETIME Fixed upper bound on created_at for every month @@ -78,7 +74,6 @@ import { getIsCluster } from './helpers'; const DEFAULT_BATCH_DAYS = 2; const DEFAULT_PARALLEL = 2; -const DEFAULT_MAX_EVENTS = 100_000_000; // Spill the per-batch GROUP BY rather than OOM on the ARRAY JOIN fan-out; // max_insert_threads parallelises the part-building stage, which is @@ -242,15 +237,9 @@ async function scalar(query: string): Promise { export async function up() { const isClustered = getIsCluster(); const isDry = process.argv.includes('--dry'); - const force = process.argv.includes('--force'); const only = getArg('only'); const batchDays = getPositiveInt(getArg('batch-days'), DEFAULT_BATCH_DAYS); const parallel = getPositiveInt(getArg('parallel'), DEFAULT_PARALLEL); - const maxEvents = getPositiveInt( - process.env.COHORT_BACKFILL_MAX_EVENTS, - DEFAULT_MAX_EVENTS, - ); - const totalEvents = Number( (await scalar(`SELECT count() AS c FROM ${TABLE_NAMES.events}`)) ?? 0, ); @@ -258,24 +247,6 @@ export async function up() { console.log('๐Ÿ“ฆ Cohort summary MVs: no events to aggregate, nothing to do'); return; } - if (totalEvents > maxEvents && !force && !isDry) { - console.log(''); - console.log( - `โญ๏ธ Cohort summary MVs: skipping the automatic rebuild (${totalEvents.toLocaleString()} events > COHORT_BACKFILL_MAX_EVENTS=${maxEvents.toLocaleString()}).`, - ); - console.log( - ' Cohorts compute from partial history until this is run. Run it when it suits you:', - ); - console.log( - ' CLICKHOUSE_URL=... jiti packages/db/code-migrations/21-backfill-cohort-summary-mvs.ts', - ); - console.log( - ' It rebuilds month by month and is safe to interrupt and re-run.', - ); - console.log(''); - return; - } - // A fixed --until applies to every month. Without one, each month takes its // own bound straight after its drop (see the header) so a long run cannot // lose the rows the trigger wrote while it was running. Read from @@ -322,10 +293,14 @@ export async function up() { }); } + // Newest first: cohort criteria mostly use relative timeframes, so the + // recent months are the ones that make cohorts correct again. It also means + // an interrupted run leaves the useful end done. const months: number[] = []; for (let m = fromMonth; m <= toMonth; m = nextMonth(m)) { months.push(m); } + months.reverse(); console.log(''); console.log('๐Ÿ“ฆ Cohort summary MV rebuild');