diff --git a/.changeset/backfill-mv-replay-empty-target.md b/.changeset/backfill-mv-replay-empty-target.md new file mode 100644 index 0000000..c1c4574 --- /dev/null +++ b/.changeset/backfill-mv-replay-empty-target.md @@ -0,0 +1,5 @@ +--- +"@chkit/plugin-backfill": patch +--- + +Fix `mv_replay` backfill of a from-scratch empty aggregate target. Chunk planning now sizes chunks against the materialized view's source table (the one it reads `FROM`) instead of the target, so bootstrapping an empty rollup no longer fails with "No partitions found for <target>". The empty-check still guards the source, and multi-view fan-in from different sources keeps its existing behaviour. diff --git a/apps/docs/src/content/docs/plugins/backfill.md b/apps/docs/src/content/docs/plugins/backfill.md index e8ef24e..9102432 100644 --- a/apps/docs/src/content/docs/plugins/backfill.md +++ b/apps/docs/src/content/docs/plugins/backfill.md @@ -100,7 +100,7 @@ The plugin supports two strategies for backfilling data, chosen automatically ba **Table backfill** (`table` strategy): For direct table targets, inserts data by selecting from the same table within the time window. This is the most common case. -**Materialized view replay** (`mv_replay` strategy): When the target is the `to` table of one or more materialized views, the plugin replays each view's aggregation query over every chunk window, re-materializing the aggregation for that window and ensuring correctness for aggregate backfills. A target fed by several materialized views replays them all in a single `INSERT … SELECT … UNION ALL …` per chunk, so no view's rows are missed. Requires `requireIdempotencyToken: true` for safe resumable retries. +**Materialized view replay** (`mv_replay` strategy): When the target is the `to` table of one or more materialized views, the plugin replays each view's aggregation query over every chunk window, re-materializing the aggregation for that window and ensuring correctness for aggregate backfills. A target fed by several materialized views replays them all in a single `INSERT … SELECT … UNION ALL …` per chunk, so no view's rows are missed. Because the chunks are sized from the view's **source** table (the one it reads `FROM`) rather than the target, a from-scratch **empty** aggregate target is fine — it's the case being bootstrapped. Requires `requireIdempotencyToken: true` for safe resumable retries. ## Time column resolution diff --git a/packages/plugin-backfill/src/chunking/sql.test.ts b/packages/plugin-backfill/src/chunking/sql.test.ts index 564e0e9..6715b00 100644 --- a/packages/plugin-backfill/src/chunking/sql.test.ts +++ b/packages/plugin-backfill/src/chunking/sql.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from 'bun:test' import { + extractSourceTableRef, findTopLevelKeywords, injectSortKeyFilter, rewriteSelectColumns, @@ -142,6 +143,62 @@ describe('rewriteSelectColumns', () => { }) }) +describe('extractSourceTableRef', () => { + test('extracts a qualified database.table', () => { + expect(extractSourceTableRef('SELECT * FROM solana.raw_token_transfers')).toEqual({ + database: 'solana', + table: 'raw_token_transfers', + }) + }) + + test('stops at a trailing clause', () => { + expect( + extractSourceTableRef('SELECT a, count() AS c FROM app.events GROUP BY a ORDER BY c'), + ).toEqual({ database: 'app', table: 'events' }) + }) + + test('ignores a table alias', () => { + expect(extractSourceTableRef('SELECT * FROM app.events AS e WHERE e.x = 1')).toEqual({ + database: 'app', + table: 'events', + }) + expect(extractSourceTableRef('SELECT * FROM app.events e')).toEqual({ + database: 'app', + table: 'events', + }) + }) + + test('returns a bare table with no database', () => { + expect(extractSourceTableRef('SELECT count() FROM raw_events')).toEqual({ table: 'raw_events' }) + }) + + test('strips backtick-quoted identifiers', () => { + expect(extractSourceTableRef('SELECT * FROM `app`.`events`')).toEqual({ + database: 'app', + table: 'events', + }) + }) + + test('ignores a FROM keyword hiding inside a string literal', () => { + expect(extractSourceTableRef("SELECT 'FROM sneaky' AS note FROM app.events")).toEqual({ + database: 'app', + table: 'events', + }) + }) + + test('returns undefined for a subquery source', () => { + expect(extractSourceTableRef('SELECT * FROM (SELECT * FROM app.events) AS s')).toBeUndefined() + }) + + test('returns undefined for a table-function source', () => { + expect(extractSourceTableRef('SELECT * FROM numbers(10)')).toBeUndefined() + }) + + test('returns undefined when there is no FROM', () => { + expect(extractSourceTableRef('SELECT 1')).toBeUndefined() + }) +}) + describe('injectSortKeyFilter', () => { test('adds a WHERE clause when none exists', () => { const sql = injectSortKeyFilter('SELECT * FROM t', 'ts', 'numeric', '1', '5') diff --git a/packages/plugin-backfill/src/chunking/sql.ts b/packages/plugin-backfill/src/chunking/sql.ts index b25950a..6963bbb 100644 --- a/packages/plugin-backfill/src/chunking/sql.ts +++ b/packages/plugin-backfill/src/chunking/sql.ts @@ -145,6 +145,93 @@ export function rewriteSelectColumns(query: string, targetColumns: string[]): st return `${trimmed.slice(0, projectionStart)} ${prefix}${rewritten.join(', ')}\n${trimmed.slice(bounds.fromPos)}` } +/** + * Extract the primary source table an mv_replay backfill must chunk against — + * the table read by the first top-level `FROM` in a materialized view's + * `SELECT`. Chunk conditions (`_partition_id`, sort-key ranges) are injected + * into that SELECT and run against this source, so its physical metadata is + * what sizing must introspect, not the (legitimately empty) target. + * + * Handles `db.table`, bare `table`, and backtick-quoted identifiers, ignoring a + * trailing alias or clause. Returns `undefined` when the `FROM` target is a + * subquery, a table function, or otherwise not a plain table reference — the + * caller decides how to treat an unresolvable source. + */ +export function extractSourceTableRef( + query: string, +): { database?: string; table: string } | undefined { + const fromHit = findTopLevelKeywords(query, ['FROM']).find((hit) => hit.keyword === 'FROM') + if (!fromHit) return undefined + + const token = readFirstTableToken(query.slice(fromHit.position + 'FROM'.length)) + if (!token) return undefined + + return parseTableIdentifier(token) +} + +/** + * Read the identifier immediately following `FROM`. Stops at the first + * whitespace, comma, or closing paren (an alias or trailing clause). Returns + * `undefined` for a subquery (`FROM (…)`) or table function (`name(…)`), which + * are not plain table references. + */ +function readFirstTableToken(text: string): string | undefined { + let index = 0 + while (index < text.length && /\s/.test(text[index] ?? '')) index++ + if (index >= text.length || text[index] === '(') return undefined + + let token = '' + let inBacktick = false + for (; index < text.length; index++) { + const char = text[index] ?? '' + if (char === '`') { + inBacktick = !inBacktick + token += char + continue + } + if (inBacktick) { + token += char + continue + } + if (char === '(') return undefined // table function, e.g. numbers(10) + if (/\s/.test(char) || char === ',' || char === ')') break + token += char + } + + return token.length > 0 ? token : undefined +} + +/** Split a possibly-qualified, possibly-backticked identifier into db/table. */ +function parseTableIdentifier(ref: string): { database?: string; table: string } | undefined { + const segments: string[] = [] + let current = '' + let inBacktick = false + for (const char of ref) { + if (char === '`') { + inBacktick = !inBacktick + continue + } + if (char === '.' && !inBacktick) { + segments.push(current) + current = '' + continue + } + current += char + } + segments.push(current) + + if (segments.length === 1) { + const table = segments[0] ?? '' + return table ? { table } : undefined + } + if (segments.length === 2) { + const database = segments[0] ?? '' + const table = segments[1] ?? '' + return database && table ? { database, table } : undefined + } + return undefined // db.schema.table or malformed — unsupported +} + export function injectSortKeyFilter( query: string, sortKeyColumn: string, diff --git a/packages/plugin-backfill/src/detect.test.ts b/packages/plugin-backfill/src/detect.test.ts index 18f82b0..7bbba8d 100644 --- a/packages/plugin-backfill/src/detect.test.ts +++ b/packages/plugin-backfill/src/detect.test.ts @@ -3,7 +3,7 @@ import { describe, expect, test } from 'bun:test' import { table, materializedView } from '@chkit/core' import type { SchemaDefinition } from '@chkit/core' -import { detectCandidatesFromTable, extractSchemaTimeColumn, findMvsForTarget, findTableForTarget } from './detect.js' +import { detectCandidatesFromTable, extractSchemaTimeColumn, findMvsForTarget, findTableForTarget, resolveMvReplaySource } from './detect.js' describe('@chkit/plugin-backfill detect', () => { test('finds DateTime column in ORDER BY as top candidate', () => { @@ -294,4 +294,71 @@ describe('@chkit/plugin-backfill detect', () => { expect(found.map((mv) => mv.name)).toEqual(['hourly_mv', 'daily_mv']) }) + + test('resolveMvReplaySource resolves the qualified FROM table of a single MV', () => { + const mv = materializedView({ + database: 'app', + name: 'events_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT toStartOfHour(ts) AS ts, count() AS c FROM solana.raw_events GROUP BY ts', + }) + + expect(resolveMvReplaySource([mv])).toEqual({ database: 'solana', table: 'raw_events' }) + }) + + test('resolveMvReplaySource collapses multiple MVs sharing one source', () => { + const hourly = materializedView({ + database: 'app', + name: 'hourly_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT toStartOfHour(ts) AS ts, count() AS c FROM app.raw_events GROUP BY ts', + }) + const daily = materializedView({ + database: 'app', + name: 'daily_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT toStartOfDay(ts) AS ts, count() AS c FROM app.raw_events GROUP BY ts', + }) + + expect(resolveMvReplaySource([hourly, daily])).toEqual({ database: 'app', table: 'raw_events' }) + }) + + test('resolveMvReplaySource defaults an unqualified FROM to the MV database', () => { + const mv = materializedView({ + database: 'analytics', + name: 'events_mv', + to: { database: 'analytics', name: 'events_agg' }, + as: 'SELECT count() AS c FROM raw_events', + }) + + expect(resolveMvReplaySource([mv])).toEqual({ database: 'analytics', table: 'raw_events' }) + }) + + test('resolveMvReplaySource returns undefined when MVs fan in from different sources', () => { + const web = materializedView({ + database: 'app', + name: 'web_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT count() AS c FROM app.web_events', + }) + const api = materializedView({ + database: 'app', + name: 'api_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT count() AS c FROM app.api_events', + }) + + expect(resolveMvReplaySource([web, api])).toBeUndefined() + }) + + test('resolveMvReplaySource returns undefined when the FROM is a subquery', () => { + const mv = materializedView({ + database: 'app', + name: 'events_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT count() AS c FROM (SELECT * FROM app.raw_events) GROUP BY 1', + }) + + expect(resolveMvReplaySource([mv])).toBeUndefined() + }) }) diff --git a/packages/plugin-backfill/src/detect.ts b/packages/plugin-backfill/src/detect.ts index d15b7a6..a7a2b0d 100644 --- a/packages/plugin-backfill/src/detect.ts +++ b/packages/plugin-backfill/src/detect.ts @@ -1,5 +1,6 @@ import type { MaterializedViewDefinition, SchemaDefinition, TableDefinition } from '@chkit/core' +import { extractSourceTableRef } from './chunking/sql.js' import './table-config.js' import type { TimeColumnCandidate } from './types.js' @@ -40,6 +41,35 @@ export function findMvsForTarget( ) } +/** + * Resolve the single source table an mv_replay backfill should size its chunks + * against — the table the materialized views read `FROM`. The injected chunk + * conditions (`_partition_id`, sort-key ranges) run against that source, so the + * chunk planner must introspect it rather than the target, which is + * legitimately empty while a fresh aggregate is being bootstrapped. + * + * An unqualified `FROM` table defaults to the view's own database, matching + * ClickHouse name resolution. Returns `undefined` when a source can't be + * resolved to a single shared table — either a `FROM` we can't parse, or MVs + * fanning in from different sources (which one chunk plan can't drive) — so the + * caller falls back to introspecting the target, preserving multi-source replay. + */ +export function resolveMvReplaySource( + mvs: MaterializedViewDefinition[] +): { database: string; table: string } | undefined { + const sources = new Map() + + for (const mv of mvs) { + const ref = extractSourceTableRef(mv.as) + if (!ref) return undefined + const database = ref.database ?? mv.database + sources.set(`${database}.${ref.table}`, { database, table: ref.table }) + } + + const distinct = [...sources.values()] + return distinct.length === 1 ? distinct[0] : undefined +} + export function findTableForTarget( definitions: SchemaDefinition[], database: string, diff --git a/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts b/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts new file mode 100644 index 0000000..acf35a0 --- /dev/null +++ b/packages/plugin-backfill/src/mv-replay-plan.e2e.test.ts @@ -0,0 +1,218 @@ +import { afterAll, beforeAll, describe, expect, test } from 'bun:test' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { tmpdir } from 'node:os' + +import { resolveConfig } from '@chkit/core' +import type { ClickHouseExecutor } from '@chkit/clickhouse' +import { + createLiveExecutor, + createPrefix, + createStatelessLiveExecutor, + getRequiredEnv, + waitForTable, +} from '@chkit/clickhouse/e2e-testkit' + +import { executeBackfill } from './async-backfill.js' +import { buildChunkExecutionSql } from './chunking/sql.js' +import { generateIdempotencyToken } from './chunking/utils/ids.js' +import { PlanSchema } from './options.js' +import { buildBackfillPlan } from './planner.js' +import type { PlannerQuery } from './chunking/types.js' +import type { BackfillPlanState } from './types.js' + +// --------------------------------------------------------------------------- +// Regression e2e for chkit#187: an mv_replay backfill of a from-scratch EMPTY +// aggregate target must plan its chunks against the MV *source* (the table the +// view reads), not the target. Before the fix, planning introspected the empty +// target and failed with "No partitions found for ". +// +// This drives the full path against a live cluster: buildBackfillPlan (schema +// load → MV detection → source introspection → chunking) followed by +// executeBackfill running the generated INSERT…SELECTs, then verifies the +// populated target matches the forward MV output. +// --------------------------------------------------------------------------- + +const SOURCE_ROWS = 4000 +const BUCKETS = 4 + +// DDL / inserts / counts go through the session-bound executor (sequential). +let ddl: ClickHouseExecutor +// The execute loop submits + polls in parallel; a stateless executor avoids +// ObsessionDB session-locking errors under concurrency. +let runExecutor: ClickHouseExecutor +let plannerQuery: PlannerQuery +let db: string +let sourceTable: string +let targetTable: string +let sourceFqn: string +let targetFqn: string +let dir: string +let configPath: string + +async function aggregateByBucket(fqn: string, valueExpr: string): Promise> { + return ddl.query<{ bucket: string; total: string }>( + `SELECT toString(bucket) AS bucket, toString(${valueExpr}) AS total + FROM ${fqn} + GROUP BY bucket + ORDER BY bucket + SETTINGS select_sequential_consistency = 1`, + ) +} + +function schemaSource(): string { + // Plain-object definitions (no imports) so loadSchemaDefinitions can evaluate + // the file straight from a temp dir, matching the unit-test convention. + return `export const events_target = { + kind: 'table', + database: '${db}', + name: '${targetTable}', + columns: [ + { name: 'bucket', type: 'UInt8' }, + { name: 'total', type: 'UInt64' }, + ], + engine: 'SummingMergeTree', + primaryKey: ['bucket'], + orderBy: ['bucket'], +} +export const events_mv = { + kind: 'materialized_view', + database: '${db}', + name: '${sourceTable}_mv', + to: { database: '${db}', name: '${targetTable}' }, + as: 'SELECT bucket, sum(id) AS total FROM ${db}.${sourceTable} GROUP BY bucket', +} +` +} + +beforeAll(async () => { + const env = getRequiredEnv() + db = env.clickhouseDatabase + ddl = createLiveExecutor(env) + runExecutor = createStatelessLiveExecutor(env) + plannerQuery = async ( + sql: string, + settings?: Record, + ): Promise => runExecutor.query(sql, settings) + + const prefix = createPrefix('backfill_mvreplay') + sourceTable = `${prefix}source` + targetTable = `${prefix}agg` + sourceFqn = `${db}.${sourceTable}` + targetFqn = `${db}.${targetTable}` + + // Partitioned source with real data. + await ddl.command(` + CREATE TABLE IF NOT EXISTS ${sourceFqn} ( + id UInt64, + bucket UInt8, + payload String + ) ENGINE = MergeTree() + PARTITION BY bucket + ORDER BY id + `) + // Aggregate target that starts EMPTY — the scenario the bug blocked. + await ddl.command(` + CREATE TABLE IF NOT EXISTS ${targetFqn} ( + bucket UInt8, + total UInt64 + ) ENGINE = SummingMergeTree() + ORDER BY bucket + `) + await waitForTable(ddl, db, sourceTable) + await waitForTable(ddl, db, targetTable) + + const rows = Array.from({ length: SOURCE_ROWS }, (_, i) => ({ + id: i, + bucket: i % BUCKETS, + payload: 'x'.repeat(256), + })) + await ddl.insert({ table: sourceFqn, values: rows }) + + dir = await mkdtemp(join(tmpdir(), 'chkit-backfill-mvreplay-')) + configPath = join(dir, 'clickhouse.config.ts') + await writeFile(join(dir, 'schema.ts'), schemaSource()) +}, 120_000) + +afterAll(async () => { + if (sourceFqn) await ddl.command(`DROP TABLE IF EXISTS ${sourceFqn}`) + if (targetFqn) await ddl.command(`DROP TABLE IF EXISTS ${targetFqn}`) + if (dir) await rm(dir, { recursive: true, force: true }) + await runExecutor?.close() + await ddl?.close() +}) + +describe('e2e: mv_replay backfill of an empty aggregate target (chkit#187)', () => { + test('plans from the source, then executeBackfill populates the empty target to match the forward MV', async () => { + // Confirm the target really is empty before we plan against it. + expect(await aggregateByBucket(targetFqn, 'sum(total)')).toHaveLength(0) + + const config = resolveConfig({ schema: './schema.ts', metaDir: './chkit/meta' }) + + // Size chunks so each source partition is one chunk (partition-aligned, no + // intra-partition range splitting) — the same shape the copy e2e uses. This + // keeps the test on the part the fix touches (source introspection + the + // per-partition INSERT…SELECT) rather than the sort-key splitter. + const [bytesRow] = await ddl.query<{ total: string }>(` + SELECT toString(sum(data_uncompressed_bytes)) AS total + FROM system.parts + WHERE database = '${db}' AND table = '${sourceTable}' AND active = 1 + SETTINGS select_sequential_consistency = 1 + `) + const uncompressedBytes = Number(bytesRow?.total ?? 0) + expect(uncompressedBytes).toBeGreaterThan(0) + + const opts = PlanSchema.parse({ target: targetFqn, maxChunkBytes: uncompressedBytes }) + + // The bug: this threw "No partitions found for ". Now it plans off + // the source instead. + const output = await buildBackfillPlan({ + opts, + configPath, + config, + clickhouseQuery: plannerQuery, + querySettings: { enable_parallel_replicas: 0 }, + }) + + const plan: BackfillPlanState = output.plan + expect(plan.execution.mode).toBe('mv_replay') + // Chunk plan is sourced from the MV's FROM table, not the empty target. + expect(plan.chunkPlan.table.database).toBe(db) + expect(plan.chunkPlan.table.table).toBe(sourceTable) + // One chunk per source partition — a real multi-chunk plan over the source. + expect(plan.chunkPlan.chunks.length).toBe(BUCKETS) + + const result = await executeBackfill({ + executor: runExecutor, + planId: plan.planId, + chunks: plan.chunkPlan.chunks.map((chunk) => ({ id: chunk.id })), + buildQuery: ({ id }) => { + const planChunk = plan.chunkPlan.chunks.find((candidate) => candidate.id === id) + if (!planChunk) throw new Error(`Chunk ${id} not found in plan`) + return buildChunkExecutionSql({ + planId: plan.planId, + chunk: planChunk, + target: plan.target, + sourceTarget: plan.execution.sourceTarget, + table: plan.chunkPlan.table, + mvReplayQueries: plan.execution.mvReplayQueries, + targetColumns: plan.execution.targetColumns, + idempotencyToken: plan.execution.requireIdempotencyToken + ? generateIdempotencyToken(plan.planId, planChunk.id) + : '', + }) + }, + concurrency: 3, + pollIntervalMs: 1500, + }) + + expect(result.failed).toBe(0) + expect(result.completed).toBe(plan.chunkPlan.chunks.length) + + // Per-bucket values must match a forward run of the MV over the whole source. + const expected = await aggregateByBucket(sourceFqn, 'sum(id)') + const actual = await aggregateByBucket(targetFqn, 'sum(total)') + expect(expected).toHaveLength(BUCKETS) + expect(actual).toEqual(expected) + }, 180_000) +}) diff --git a/packages/plugin-backfill/src/planner.test.ts b/packages/plugin-backfill/src/planner.test.ts index 5c6abd7..a9717f5 100644 --- a/packages/plugin-backfill/src/planner.test.ts +++ b/packages/plugin-backfill/src/planner.test.ts @@ -45,6 +45,75 @@ function createMockQuery(opts: { } } +/** + * A mock that only knows about `sourceTable`: every introspection query scoped + * to any other table (e.g. the empty aggregate target) returns no rows. This + * lets a test assert that mv_replay planning introspects the MV *source* and + * not the target. + */ +function createSourceScopedMockQuery(opts: { + sourceTable: string + partitions?: Array<{ + partition_id: string + total_rows: string + total_bytes: string + total_uncompressed_bytes?: string + min_time: string + max_time: string + }> + sortingKey?: string + columnRows?: Array<{ name: string; type: string }> +}): (sql: string) => Promise { + const partitions = opts.partitions ?? [ + { + partition_id: '202606', + total_rows: '1000', + total_bytes: '500000', + total_uncompressed_bytes: '1000000', + min_time: '2026-06-01 00:00:00', + max_time: '2026-06-30 18:00:00', + }, + ] + const sortingKey = opts.sortingKey ?? 'id' + const columnRows = opts.columnRows ?? [{ name: 'id', type: 'UInt64' }] + const table = opts.sourceTable + + return async (sql: string) => { + if (sql.includes('SELECT 1 FROM')) return [{ ok: 1 }] as T[] + if (sql.includes('FROM system.parts')) { + return (sql.includes(`table = '${table}'`) ? partitions : []) as T[] + } + if (sql.includes('FROM system.tables')) { + return (sql.includes(`name = '${table}'`) ? [{ sorting_key: sortingKey }] : []) as T[] + } + if (sql.includes('FROM system.columns')) { + return (sql.includes(`table = '${table}'`) ? columnRows : []) as T[] + } + return [] as T[] + } +} + +const MV_REPLAY_SCHEMA = `export const events_target = { + kind: 'table', + database: 'app', + name: 'events_agg', + columns: [ + { name: 'id', type: 'UInt64' }, + { name: 'count', type: 'UInt64' }, + ], + engine: 'SummingMergeTree', + primaryKey: ['id'], + orderBy: ['id'], +} +export const events_mv = { + kind: 'materialized_view', + database: 'app', + name: 'events_mv', + to: { database: 'app', name: 'events_agg' }, + as: 'SELECT id, count() AS count FROM app.raw_events GROUP BY id', +} +` + describe('@chkit/plugin-backfill planning', () => { test('each plan gets a unique random id and canonical chunk plan', async () => { const dir = await mkdtemp(join(tmpdir(), 'chkit-backfill-plugin-')) @@ -370,6 +439,82 @@ export const api_mv = { } }) + test('mv_replay chunks the MV source even when the target aggregate is empty', async () => { + const dir = await mkdtemp(join(tmpdir(), 'chkit-backfill-plugin-')) + const configPath = join(dir, 'clickhouse.config.ts') + const schemaPath = join(dir, 'schema.ts') + + try { + await writeFile(schemaPath, MV_REPLAY_SCHEMA) + + const config = resolveConfig({ schema: './schema.ts', metaDir: './chkit/meta' }) + const opts = PlanSchema.parse({ target: 'app.events_agg' }) + + // The target (events_agg) is empty; only the MV source (raw_events) has + // partitions. Before the fix this threw "No partitions found for + // app.events_agg" because planning introspected the target. + const output = await buildBackfillPlan({ + opts, + configPath, + config, + clickhouseQuery: createSourceScopedMockQuery({ sourceTable: 'raw_events' }), + }) + + expect(output.plan.execution.mode).toBe('mv_replay') + // Chunk plan is sourced from the MV's FROM table, not the target. + expect(output.plan.chunkPlan.table.database).toBe('app') + expect(output.plan.chunkPlan.table.table).toBe('raw_events') + expect(output.plan.chunkPlan.chunks.length).toBeGreaterThan(0) + + const chunk = output.plan.chunkPlan.chunks[0] + const sql = chunk + ? buildChunkExecutionSql({ + planId: output.plan.planId, + chunk, + target: output.plan.target, + sourceTarget: output.plan.execution.sourceTarget, + table: output.plan.chunkPlan.table, + mvReplayQueries: output.plan.execution.mvReplayQueries, + targetColumns: output.plan.execution.targetColumns, + idempotencyToken: generateIdempotencyToken(output.plan.planId, chunk.id), + }) + : '' + + expect(sql).toContain('INSERT INTO app.events_agg') + expect(sql).toContain('FROM app.raw_events') + // Chunk conditions reference the source's real partition, not the target. + expect(sql).toContain("_partition_id = '202606'") + } finally { + await rm(dir, { recursive: true, force: true }) + } + }) + + test('mv_replay still fails fast when the MV source itself is empty', async () => { + const dir = await mkdtemp(join(tmpdir(), 'chkit-backfill-plugin-')) + const configPath = join(dir, 'clickhouse.config.ts') + const schemaPath = join(dir, 'schema.ts') + + try { + await writeFile(schemaPath, MV_REPLAY_SCHEMA) + + const config = resolveConfig({ schema: './schema.ts', metaDir: './chkit/meta' }) + const opts = PlanSchema.parse({ target: 'app.events_agg' }) + + // No partitions for the source either — the empty-check must still guard, + // now pointed at the source rather than the target. + await expect( + buildBackfillPlan({ + opts, + configPath, + config, + clickhouseQuery: createSourceScopedMockQuery({ sourceTable: 'raw_events', partitions: [] }), + }) + ).rejects.toThrow('No partitions found for app.raw_events') + } finally { + await rm(dir, { recursive: true, force: true }) + } + }) + test('rejects persisted legacy plans with an actionable error', async () => { const dir = await mkdtemp(join(tmpdir(), 'chkit-backfill-plugin-')) const configPath = join(dir, 'clickhouse.config.ts') diff --git a/packages/plugin-backfill/src/planner.ts b/packages/plugin-backfill/src/planner.ts index 34b5aca..e5d40c2 100644 --- a/packages/plugin-backfill/src/planner.ts +++ b/packages/plugin-backfill/src/planner.ts @@ -1,11 +1,11 @@ import { dirname } from 'node:path' -import type { ResolvedChxConfig } from '@chkit/core' +import type { MaterializedViewDefinition, ResolvedChxConfig } from '@chkit/core' import { loadSchemaDefinitions } from '@chkit/core/schema-loader' import { encodeChunkPlanForPersistence } from './chunking/boundary-codec.js' import { generateChunkPlan } from './chunking/planner.js' -import { findMvsForTarget } from './detect.js' +import { findMvsForTarget, resolveMvReplaySource } from './detect.js' import { BackfillConfigError } from './errors.js' import type { PlanOptions } from './options.js' import { @@ -17,6 +17,46 @@ import { } from './state.js' import type { BuildBackfillPlanOutput } from './types.js' +interface BackfillStrategy { + mvs: MaterializedViewDefinition[] + mvReplayQueries?: string[] + targetColumns?: string[] +} + +/** + * Inspect the schema to decide how the target gets populated. When one or more + * materialized views feed it, this is an mv_replay backfill and their queries + * drive the insert; otherwise it's a plain copy. A schema that can't be loaded + * falls back to copy — the same lenient behaviour as before. + */ +async function detectBackfillStrategy(input: { + schema: ResolvedChxConfig['schema'] + configDir: string + database: string + table: string +}): Promise { + try { + const definitions = await loadSchemaDefinitions(input.schema, { cwd: input.configDir }) + const mvs = findMvsForTarget(definitions, input.database, input.table) + if (mvs.length === 0) return { mvs: [] } + + const tableDef = definitions.find( + (definition) => + definition.kind === 'table' && + definition.database === input.database && + definition.name === input.table + ) + return { + mvs, + mvReplayQueries: mvs.map((mv) => mv.as), + targetColumns: tableDef?.kind === 'table' ? tableDef.columns.map((column) => column.name) : undefined, + } + } catch { + // Schema load failed, fall back to direct copy. + return { mvs: [] } + } +} + export async function buildBackfillPlan(input: { opts: PlanOptions configPath: string @@ -31,9 +71,23 @@ export async function buildBackfillPlan(input: { throw new BackfillConfigError('Invalid target format. Expected .') } - const chunkPlan = await generateChunkPlan({ + // Detect the execution strategy before chunk planning: an mv_replay backfill + // sizes its chunks against the MV *source* (the table its SELECT reads), + // because the injected chunk conditions run against that source — not the + // target, which is legitimately empty when bootstrapping an aggregate. Only + // the copy path introspects the target itself. + const strategy = await detectBackfillStrategy({ + schema: input.config.schema, + configDir: dirname(input.configPath), database, table, + }) + const replaySource = strategy.mvReplayQueries ? resolveMvReplaySource(strategy.mvs) : undefined + const chunkSource = replaySource ?? { database, table } + + const chunkPlan = await generateChunkPlan({ + database: chunkSource.database, + table: chunkSource.table, from: opts.from, to: opts.to, targetChunkBytes: opts.maxChunkBytes, @@ -44,7 +98,7 @@ export async function buildBackfillPlan(input: { const firstPartition = chunkPlan.partitions[0] if (!firstPartition) { throw new BackfillConfigError( - `No partitions found for ${opts.target}${opts.from || opts.to ? ' within the specified time range' : ''}. The table may be empty.` + `No partitions found for ${chunkSource.database}.${chunkSource.table}${opts.from || opts.to ? ' within the specified time range' : ''}. The table may be empty.` ) } @@ -61,26 +115,7 @@ export async function buildBackfillPlan(input: { const stateDir = computeBackfillStateDir(input.config, input.configPath, opts.stateDir) const paths = backfillPaths(stateDir, chunkPlan.planId) - let mvReplayQueries: string[] | undefined - let targetColumns: string[] | undefined - - try { - const definitions = await loadSchemaDefinitions(input.config.schema, { - cwd: dirname(input.configPath), - }) - const mvs = findMvsForTarget(definitions, database, table) - if (mvs.length > 0) { - mvReplayQueries = mvs.map((mv) => mv.as) - const tableDef = definitions.find( - (definition) => definition.kind === 'table' && definition.database === database && definition.name === table - ) - if (tableDef?.kind === 'table') { - targetColumns = tableDef.columns.map((column) => column.name) - } - } - } catch { - // Schema load failed, fall back to direct copy. - } + const { mvReplayQueries, targetColumns } = strategy const plan = { planId: chunkPlan.planId,