Backfill¶

Resumable, chunked, idempotent historical loads for any source/sink route.

This guide covers the self-service workflow (dpone backfill), the manifest contract, resume semantics, verification, and interval-aware Airflow integration. For the strategy semantics see Load strategies → backfill.

When to use backfill¶

Initial historical load of a new route (initial_backfill).
Range replays after upstream corrections (range_replay).
Re-population after schema evolution (schema_backfill).
Interval-driven idempotent scheduled loads (Airflow catchup/backfill).

Quick start¶

Declare the campaign in the manifest:

name: orders_backfill

source:
  type: mssql
  connection_id: mssql_oltp
  table:
    schema: dbo
    name: orders

sink:
  type: clickhouse
  connection_id: clickhouse_dwh
  table:
    schema: analytics
    name: orders
  strategy:
    mode: backfill
    backfill:
      inner_mode: partition_replace
      chunk:
        column: business_date
        from: "2025-01-01"
        to: "2025-12-31"
        step: 1d
      parallel_workers: 1

Review the deterministic plan (no data movement):

dpone backfill plan orders_backfill.yml

Execute; the run is dry-run by default, --execute loads pending chunks:

dpone backfill run orders_backfill.yml --execute

If the campaign is interrupted, resume all non-committed chunks:

dpone backfill resume orders_backfill.yml

Inspect progress at any time:

dpone backfill status orders_backfill.yml --format json

Operate an interrupted or long-running campaign:

dpone backfill retry-failed orders_backfill.yml --execute
dpone backfill cancel orders_backfill.yml --reason "source maintenance" --requested-by "data-ops"
dpone backfill doctor orders_backfill.yml
dpone backfill plan orders_backfill.yml --advisor

dpone run orders_backfill.yml is equivalent to backfill run --execute: the runtime detects mode: backfill with a chunk block and orchestrates the campaign automatically, which is what Airflow pods execute.

Chunk windows¶

Kind	Boundary semantics	Predicate
`date`	`to` inclusive; half-open windows	`column >= 'start' AND column < 'end'`
`timestamp`	`to` exclusive; half-open windows	`column >= 'start' AND column < 'end'`
`integer`	inclusive	`column >= start AND column <= end`

Steps: Nh (timestamp only), Nd, Nw, Nmo, Ny, or an integer for kind: integer. Kind is inferred from from and can be overridden with chunk.kind.

Half-open windows guarantee adjacent chunks never overlap, so a re-run of one chunk replaces exactly its own slice — the core idempotency invariant.

Resume semantics and the ledger¶

Every campaign has a deterministic run_key derived from the dataset and the full chunk configuration. Progress is persisted to a human-readable JSON ledger:

.dpone/backfill/<run_key>.json

Committed chunks are skipped on re-run; execution continues from the first non-committed chunk.
Changing any boundary/step starts a fresh campaign (new run_key).
Override the directory with backfill.state_dir or DPONE_BACKFILL_STATE_DIR; pin an explicit key with backfill.backfill_id or --backfill-id.
Every chunk records its idempotency_key, attempts, row counts and the run_id/load_id of its ETL run for lineage.
A pinned backfill_id cannot silently resume a different campaign shape: plan/config hash drift fails before data movement.
Campaign and chunk leases prevent two workers from starting the same work concurrently. Expired running chunks are marked lease_expired and become retryable.
A campaign lock is acquired before source IO. If the same campaign is already active, the second run fails fast instead of racing the same chunks.
resume runs every non-committed chunk (pending, failed, lease_expired). retry-failed sets retry_policy: failed_only and runs only chunks whose last status is failed; pending future chunks are left untouched.
Unknown retry_policy values fail before source IO. This is deliberate: a typo must not silently widen a targeted retry into a broader reload.
cancel is cooperative: it blocks new chunks while already-running chunks either finish or expire by their lease TTL.

Operator runbook¶

Situation	Command	What moves data
First execution	`dpone backfill run manifest.yml --execute`	all planned chunks
Normal continuation after an interruption	`dpone backfill resume manifest.yml`	every non-committed chunk
Re-test only failed chunks after fixing a transient issue	`dpone backfill retry-failed manifest.yml --execute`	failed chunks only
Stop scheduling new chunks	`dpone backfill cancel manifest.yml --reason ... --requested-by ...`	no new chunks
Explain current state	`dpone backfill doctor manifest.yml`	no data movement

retry-failed uses operation-level exit semantics: it exits with code 0 when the selected failed chunks finish successfully, even if future chunks are still pending. The full campaign status remains visible through status, doctor, the ledger and Airflow XCom.

doctor is the self-service decision helper:

Campaign state	Doctor next action
`running` chunks exist	inspect `status` and wait for lease TTL before retrying
`failed` chunks exist	run `retry-failed`, then inspect `status`
only `pending` chunks remain	run `resume`
all chunks are committed	no data movement action

State backends¶

Default CLI/dev state remains local_file (.dpone/backfill). Production Airflow deployments should mirror campaign state into the audit schema:

sink:
  strategy:
    mode: backfill
    backfill:
      state:
        backend: audit_schema
        schema: DWH_Tech

The audit-schema backend writes snapshots to:

DWH_Tech.__dpone__backfill_campaigns;
DWH_Tech.__dpone__backfill_chunks.

The backend keeps the local JSON ledger as an operational cache and mirrors every campaign/chunk transition to SQL. The SQL campaign snapshot is also a durable read source: status, doctor and the advisor can recover the latest campaign state from the audit schema when the local Airflow/Kubernetes pod cache is already gone, as long as the caller provides the sink connector through the runtime/Python API.

No DAG-side dependency injection is required for the SQL backend. During dpone run, the runtime builds the matching ClickHouse/Postgres/MSSQL state store from the sink connector and backfill.state manifest block. During read-only diagnostics (plan --advisor, status, doctor, cancel), dpone can resolve the same sink connector from the manifest via the existing runtime SinkFactory; embedded Python/API callers may still inject the connector directly. If the backend is audit_schema but the sink connector is unavailable, the run fails before source IO instead of falling back to local-only state silently. The run result includes backfill.state_backend so operators and acceptance reports can distinguish local-only state from SQL-audited state. It also includes backfill.state_capabilities:

Field	Meaning
`backend`	`local_file` or `audit_schema`
`dialect`	file/clickhouse/postgres/mssql implementation
`durable_read` / `durable_write`	whether state survives the current process
`lock_scope`	current campaign/chunk lock scope
`distributed_lock`	whether the backend is certified for cross-pod mutual exclusion

Current backend capability matrix:

Backend	Dialect	Distributed campaign lock	Lock scope
`local_file`	file	no	`process`
`audit_schema`	ClickHouse	no	`local_cache`
`audit_schema`	Postgres	yes	`postgres_advisory_campaign`
`audit_schema`	MSSQL	yes	`mssql_application_campaign`

Postgres uses a session-level advisory campaign lock before mutating the local ledger. MSSQL uses a session-level application lock through sp_getapplock. If the external campaign lock is already held, the campaign does not start and the local cache is left unchanged. ClickHouse audit state currently provides durable read/write evidence only; this is safe for normal single-run Airflow execution and diagnostics, but it is not advertised as a distributed lock.

If your deployment requires a certified cross-pod campaign lock before source IO, set:

state:
  backend: audit_schema
  schema: DWH_Tech
  require_distributed_lock: true

For uncertified dialects this fails closed with a clear blocker instead of pretending the lock is stronger than it is. For Postgres/MSSQL audit-schema state it enforces the external-lock-backed route.

For CLI-only local usage, local_file remains the zero-dependency default. For production Airflow/API usage, prefer audit_schema: it makes campaign state visible in DWH_Tech together with __dpone__loads/__dpone__load_steps and lets external diagnostics read the same source of truth as the runtime.

Verification¶

Two layers, both automatic:

Row-count parity — each committed chunk compares extracted vs loaded rows; mismatches surface as a verification.status: warning section in the run result and the ledger.
Reconciliation bridge — after the last chunk commits, the orchestrator writes <run_key>.execution.json next to the ledger. Feed it to the existing evidence tooling for deep source/sink reconciliation:

dpone ops route-refresh-capture-snapshots \
  --execution .dpone/backfill/<run_key>.execution.json \
  --output-dir .dpone/backfill/verify ...
dpone ops route-refresh-verify ...

dpone backfill plan --advisor adds a conservative performance recommendation without mutating the manifest. Profiles:

Profile	Use when	Default behavior
`balanced`	normal production run	small bounded parallelism when many chunks exist
`worker_safety`	weak worker / tight disk or memory	one chunk at a time
`source_safety`	OLTP source protection	one chunk at a time
`speed`	certified source and sink capacity	bounded parallelism with source-pressure warning

When a ledger already exists, the advisor becomes evidence-driven. By default it reads persisted campaign state: committed/failed chunk counts, row counts, attempts, lease_expired errors and per-chunk start/finish timestamps. You can also attach external runtime evidence produced by acceptance, load-step audit or integration matrix jobs:

dpone backfill plan orders_backfill.yml \
  --advisor \
  --advisor-evidence test_artifacts/acceptance/load_steps.json \
  --advisor-evidence test_artifacts/live_certification_vendor/backfill_matrix/certification_report.json \
  --format json

Accepted JSON shapes are intentionally small and connector-neutral:

{ "load_steps": [...] } or a raw list of load-step records;
{ "chunks": [...] } for an exported chunk ledger;
{ "matrix_report": {...} }, { "certification_report": {...} }, or a matrix report containing total_cases.

From that evidence it can recommend:

increasing chunk.step when many small successful chunks are fast and stable;
increasing chunk.step when many empty chunks prove the window is too narrow for the data distribution;
increasing parallel_workers only up to the bounded cap;
avoiding more parallelism when sink_finalize is the bottleneck compared with source_read;
reducing to one worker when failed chunks exist;
extending lease_ttl_minutes when workers expire leases before finishing.
blocking throughput tuning when a vendor/live matrix report is red. In that case the recommendation becomes resolve_vendor_certification_failures, risk=high, and recommended_max_parallel_chunks=1 until the matrix blockers are fixed.

Load-step rates are aggregated conservatively: if the same stage appears several times, the advisor uses the lowest positive rows/sec for that stage. That prevents one fast chunk from hiding a slow source read, finalize, cleanup or network-bound stage.

The output is advisory by design. It includes recommended_step, recommended_max_parallel_chunks, recommended_lease_ttl_minutes, risk, warnings, recommended_overrides and the evidence summary used for the recommendation, evidence_sources (chunk_ledger, load_steps, matrix_report), but it never edits the manifest or silently widens a running campaign. The evidence summary separates failed_chunks from failed_matrix_reports / failed_matrix_cases; a green matrix report is not counted as an empty chunk and cannot accidentally trigger a wider chunk window. recommended_overrides is intentionally machine-readable so CI, MR comments and runbooks can show a concrete manifest patch without applying it.

Watermark safety¶

Chunked backfills never mutate incremental watermarks: extraction uses the chunk predicate, not saved incremental state. When an incremental route needs an explicit rewind before replaying history, use the gated operation:

dpone state rewind --backend postgres --state-type xmin analytics.orders \
  --to "2025-01-01" --reason "range replay" \
  --yes --approved-by "data-arch" \
  --evidence-output .dpone/state/rewind-orders.json

The rewind stays in preview mode until both --yes and --approved-by are provided; the JSON payload doubles as approval evidence.

Airflow integration¶

Interval contract¶

The GitOps Airflow pack templates the DAG-run interval into every dpone pod via KubernetesPodOperator.env_vars (rendered per task instance):

Variable	Template
`DPONE_DAG_ID`	`{{ dag.dag_id }}`
`DPONE_DAG_RUN_ID`	`{{ run_id }}`
`DPONE_TRY_NUMBER`	`{{ ti.try_number }}`
`DPONE_LOGICAL_DATE`	`{{ logical_date }}`
`DPONE_INTERVAL_START`	`{{ data_interval_start }}`
`DPONE_INTERVAL_END`	`{{ data_interval_end }}`

dpone run consumes the same variables (or explicit --interval-start / --interval-end / --execution-date flags): they feed run-state identity and resolve {{ token }} placeholders inside manifests.

Idempotent interval runs (canonical pattern)¶

Template the chunk window from the interval so Airflow catchup=True, task clears and airflow dags backfill each replace exactly their own slice:

sink:
  strategy:
    mode: backfill
    backfill:
      inner_mode: partition_replace
      chunk:
        column: business_date
        from: "{{ data_interval_start }}"
        to: "{{ data_interval_end }}"
        step: 1d

Supported tokens: data_interval_start, data_interval_end, logical_date, ds (logical date as YYYY-MM-DD), dag_run_id.

Example DAGs¶

See examples/dags/ in the repository root:

dpone_interval_catchup_dag.py — interval-aware daily DAG with catchup=True (idempotent per-interval runs).
dpone_backfill_campaign_dag.py — manually triggered, parameterized campaign DAG (params: from/to/step) driving dpone backfill run --execute.

XCom evidence¶

The gitops.airflow_xcom_summary payload now carries the run interval and a bounded backfill progress section (campaign counters, selected chunk count, operation_status and retry_policy only — per-chunk details stay in the runtime evidence and the ledger), so downstream tasks and monitoring see exactly which window was loaded and whether the run was a normal resume or a targeted failed-chunk retry. Schema: airflow-xcom-summary.schema.json.

Bounded Airflow backfill mapping¶

The default internal mode keeps one Airflow task and lets dpone execute the whole campaign. Use mapped modes only when operators need bounded per-range Grid visibility:

Mode	Airflow task instances	dpone execution
`internal`	one	existing ledger and optional local workers
`visible`	one per chunk, at most 200	one selected chunk per mapped task
`summary`	at most `max_items`	one ordered contiguous chunk range per mapped task

gitops:
  airflow:
    mapping:
      mode: summary
      max_items: 40
      max_active: 8
      pool: dpone_backfill

Mapped modes require a chunked backfill, parallel_workers: 1, a non-empty Airflow pool and PostgreSQL or MSSQL audit_schema state with require_distributed_lock: true. max_items is 1..200; max_active is 1..64 and cannot exceed max_items. Static check and pack build reject task explosion, unsupported state and multiplied parallelism before publication.

The pack contains a deterministic mapping plan. Each mapped pod receives only its range indexes and plan digests. Before connector I/O, runtime rebuilds the chunk plan, verifies the hash and contiguous range, initializes the shared ledger under a short lock, then acquires a distinct lease for every selected chunk. Terminal transitions use owner compare-and-set, so a stale task attempt cannot overwrite a newer owner. A summary item executes its chunks sequentially.

Airflow task state is presentation and retry intent; the dpone ledger remains authoritative. Clearing a successful mapped task safely skips chunks already committed in the ledger. XCom contains a bounded backfill.mapping summary but never the chunk ledger, predicates or row data. Full progress remains in state and evidence.

Data-aware scheduling (Assets)¶

Declare produced datasets in the workload catalog to schedule downstream DAGs on data readiness:

airflow:
  execution:
    outlets:
      - "clickhouse://analytics/orders"

The pack builder copies the block into airflow-pack.json; the scheduler-side provider converts URIs into Asset (Airflow 3) or Dataset (Airflow 2.4+) outlets on the runtime task.

CLI reference¶

Command	Purpose
`dpone backfill plan MANIFEST`	Deterministic chunk plan, no data movement
`dpone backfill run MANIFEST [--execute]`	Dry-run by default; `--execute` loads pending chunks
`dpone backfill resume MANIFEST`	Continue an interrupted campaign
`dpone backfill retry-failed MANIFEST`	Retry failed/non-committed chunks only
`dpone backfill status MANIFEST`	Ledger progress
`dpone backfill cancel MANIFEST`	Cooperative cancellation for existing campaign
`dpone backfill doctor MANIFEST`	Campaign health and next action hints
`dpone state rewind ... --to W`	Gated watermark rewind with evidence

Window overrides work on every subcommand: --column, --from, --to, --step, --kind, --inner-mode, --parallel-workers, --max-chunks, --state-dir, --backfill-id. All subcommands support --format text|json|md.

Troubleshooting¶

Symptom	Cause	Fix
`backfill would produce N chunks, above max_chunks`	step too small for the window	increase `step` or `max_chunks`
`inner_mode=full_refresh ... only valid for a single-chunk plan`	truncate+load cannot run per chunk	use `partition_replace`/`replace`, or a single chunk
`Kafka backfill supports only keyed upsert replay`	topics are append-only	omit `inner_mode` or set `incremental_merge`
campaign restarted from chunk 1	chunk config changed → new `run_key`	pin `--backfill-id` to reuse the previous ledger intentionally
`verification.status: warning`	extracted vs loaded row counts differ	inspect the mismatched chunks, then run the reconciliation bridge
`backfill campaign config hash changed`	pinned `backfill_id` points to a different chunk plan	use a new `backfill_id` or restore the original window
`lease_expired`	a worker died while a chunk was running	inspect logs, then run `dpone backfill retry-failed`
`cancel_requested`	operator stopped the campaign	resume with a new campaign id when ready