@chkit/plugin-backfill

Beta plugin for checkpointed SQL backfill operations in chkit. Its API, configuration, and CLI flags may change between releases.

Part of the chkit monorepo. This plugin extends the chkit CLI with data backfill commands.

Install

bun add -d @chkit/plugin-backfill

Usage

Register the plugin in your config:

// clickhouse.config.ts
import { defineConfig } from '@chkit/core'
import { backfill } from '@chkit/plugin-backfill'

export default defineConfig({
  schema: './src/db/schema/**/*.ts',
  outDir: './chkit',
  plugins: [
    backfill(),
  ],
  clickhouse: {
    url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
  },
})

Documentation

See the backfill plugin documentation (beta).

SDK

The package root is limited to the plugin registration API. Everything used by the CLI itself — the chunk planner, SQL builders, async executor, logging — is also exported from the @chkit/plugin-backfill/sdk subpath so you can build your own backfill scripts without going through the CLI.

import {
  generateChunkPlan,
  buildChunkExecutionSql,
  executeBackfill,
  getBackfillLogger,
  type ChunkPlan,
  type PlannerQuery,
} from '@chkit/plugin-backfill/sdk'

The pipeline has three stages, and you can use any subset:

1. Plan — generateChunkPlan(...) introspects a table and returns a ChunkPlan describing how to partition the work into roughly equal-sized chunks.
2. Build SQL — buildChunkExecutionSql(...) turns a single Chunk into an INSERT … SELECT statement.
3. Execute — executeBackfill(...) submits chunks against a real ClickHouseExecutor with deterministic query IDs, polling, and resume support.

Plan a backfill

generateChunkPlan is decoupled from any ClickHouse client. You pass in a query function with the PlannerQuery shape and the planner uses it for every introspection / probe / split query. This makes the planner trivial to instrument or run against alternative clients.

import { createClient } from '@clickhouse/client'
import { generateChunkPlan, type PlannerQuery } from '@chkit/plugin-backfill/sdk'

const client = createClient({ url: process.env.CLICKHOUSE_URL })

const query: PlannerQuery = async (sql, settings) => {
  const result = await client.query({
    query: sql,
    format: 'JSONEachRow',
    clickhouse_settings: settings as Record<string, string | number | boolean>,
  })
  return result.json()
}

const plan = await generateChunkPlan({
  database: 'analytics',
  table: 'events',
  from: '2025-01-01T00:00:00Z',
  to: '2025-02-01T00:00:00Z',
  targetChunkBytes: 1_000_000_000, // ~1 GiB per chunk
  query,
  // 'count' is exact but slower; 'explain-estimate' is faster but approximate
  rowProbeStrategy: 'count',
})

console.log(`${plan.chunks.length} chunks, ${plan.totalRows.toLocaleString()} rows`)

Execute chunks against a target

buildChunkExecutionSql produces the per-chunk INSERT … SELECT and executeBackfill runs them with concurrency, polling, and progress callbacks. Persist the progress argument anywhere you like to support resume.

import { createClickHouseExecutor } from '@chkit/clickhouse'
import {
  buildChunkExecutionSql,
  executeBackfill,
  type BackfillProgress,
} from '@chkit/plugin-backfill/sdk'

const executor = createClickHouseExecutor({
  url: process.env.CLICKHOUSE_URL!,
  username: 'default',
  password: process.env.CLICKHOUSE_PASSWORD!,
  database: 'analytics',
})

const chunksById = new Map(plan.chunks.map((chunk) => [chunk.id, chunk]))
let saved: BackfillProgress | undefined // load from disk for resume

const result = await executeBackfill({
  executor,
  planId: plan.planId,
  chunks: plan.chunks,
  buildQuery: ({ id }) =>
    buildChunkExecutionSql({
      planId: plan.planId,
      chunk: chunksById.get(id)!,
      target: 'analytics.events_backfill',
      table: plan.table,
    }),
  concurrency: 4,
  pollIntervalMs: 5_000,
  resumeFrom: saved,
  onProgress: async (progress) => {
    saved = progress
    // persist to disk / state store
  },
})

console.log(`done=${result.completed} failed=${result.failed}`)

Plan persistence

Plans contain string boundaries that may include non-UTF-8 bytes (the planner uses latin1-encoded byte ranges for string sort keys), so JSON-serializing a ChunkPlan directly will lose information. Use the codec helpers when you need to round-trip a plan through storage:

import {
  encodeChunkPlanForPersistence,
  decodeChunkPlanFromPersistence,
} from '@chkit/plugin-backfill/sdk'

const json = JSON.stringify(encodeChunkPlanForPersistence(plan))
// later …
const plan2 = decodeChunkPlanFromPersistence(JSON.parse(json))

Logging

The planner emits structured logs via @logtape/logtape under the ['chkit', 'backfill'] category. Configure a sink at process start to see them — slow-query warnings (>5 s) are emitted at warning level, planning progress at info, and per-strategy decisions at debug.

import { configureSync, getConsoleSink, getTextFormatter } from '@chkit/plugin-backfill/sdk'

configureSync({
  sinks: { console: getConsoleSink({ formatter: getTextFormatter({ timestamp: 'time' }) }) },
  loggers: [{ category: 'chkit', sinks: ['console'], lowestLevel: 'info' }],
  reset: true,
})

To capture every SQL statement the planner runs (with timing, server-side stats, and per-strategy classification), wrap your query function instead of relying solely on logging — the wrapper sees the raw SQL and settings on every call and can record query IDs, response headers, and durations alongside the structured logs.

License

MIT