fix(backfill): chunk mv_replay from the MV source so empty targets bootstrap (#187)

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

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
 

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

packages/plugin-backfill/src/chunking/sql.ts

@@ -145,6 +145,93 @@
   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,

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()
+  })
 })

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<string, { database: string; table: string }>()
+
+  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,