Skip to content

Backfill integration suite

End-to-end certification of the chunked backfill subsystem (user guide) against real services: every case runs the full runtime path — bounded extraction, staging, inner-strategy finalization per chunk — and asserts row/checksum parity plus ledger invariants.

Where it lives

Location Contents
tests/integration/backfill/backfill_toolkit.py Shared seed spec, chunk-window builders, LoadConfig builder, parity/ledger assertions
tests/integration/backfill/endpoints.py Engine endpoints (Postgres/ClickHouse/MSSQL/Kafka): DDL, deterministic seed, checksum dialects, source/sink factories
tests/integration/backfill/conftest.py MSSQL and Kafka session fixtures with skip-first UX
tests/integration/backfill/test_backfill_*_integration.py Route x inner-strategy matrix per route
tests/integration/backfill/test_backfill_scenarios_integration.py Resume, idempotent re-run, parallel workers, verification bridge, invalid combinations
tests/integration/airflow/test_airflow_backfill_interval_integration.py Interval-driven dpone run via the real CLI with Airflow-rendered DPONE_* env
tests/test_backfill_orchestrator.py Runtime campaign lock, retry-failed selection, fail-closed retry policy, verification evidence
tests/test_backfill_campaign_state.py Campaign locks, chunk leases, cancel, retryable chunks, config drift
tests/test_backfill_diagnostic_connector.py CLI/API diagnostics resolve sink connectors through the standard runtime factory
tests/test_backfill_audit_schema_state.py SQL audit-schema mirrors and read-side campaign recovery
tests/test_backfill_audit_schema_service.py status/doctor service reads audit-schema state through injected sink connector
tests/test_airflow_interval_env_rendering.py Real-Airflow Jinja rendering contract (runs where Airflow is importable)

Marker: integration_backfill (declared in pyproject.toml).

How to run locally

  1. Start the disposable stack (idempotent):
docker compose -f docker/docker-compose.integration.yml up -d --wait \
  postgres clickhouse mssql kafka schema-registry
  1. Run the suite (defaults match the compose port forwards):
DPONE_RUN_INTEGRATION=1 uv run pytest -m integration_backfill \
  tests/integration/backfill tests/integration/airflow -rs -q

Overrides use the same DPONE_IT_* variables as the rest of the integration layer (DPONE_IT_PG_*, DPONE_IT_CH_*, DPONE_IT_MSSQL_*, DPONE_IT_KAFKA_BOOTSTRAP).

  1. Optional: validate the scheduler-side rendering contract under a real Airflow (mirrors .github/workflows/airflow-pack-compat.yml):
uv venv .venv-airflow --python 3.12
uv pip install --python .venv-airflow/bin/python \
  "apache-airflow==3.0.2" "apache-airflow-providers-cncf-kubernetes==10.4.2" \
  ./packages/dpone-airflow-pack pytest
.venv-airflow/bin/python -m pytest tests/test_airflow_interval_env_rendering.py -q

Skip semantics

  • Without DPONE_RUN_INTEGRATION=1 every module skips.
  • MSSQL cases skip with a clear reason when ODBC Driver 18 for SQL Server (plus bcp from mssql-tools18) is not installed or the service is unreachable; they always run in CI where the driver is provisioned.
  • Kafka cases skip when confluent-kafka is missing or the broker is unreachable.

Case matrix

Deterministic dataset per route: 120 rows, id 1..120, business_date over 6 days, bucket 1..4, amount = id * 7. Parity contract: (count, sum(id), sum(amount)) source window vs target.

Route replace (date chunks) incremental_merge (date chunks) partition_replace (bucket chunks) full_refresh (single chunk)
postgres -> clickhouse yes yes yes (native REPLACE PARTITION) yes
mssql -> clickhouse yes yes yes covered by pg->ch (sink-side semantics)
postgres -> mssql yes yes yes (delete+insert fallback) covered by pg->ch
clickhouse -> mssql yes yes yes (delete+insert fallback) covered by pg->ch
postgres -> postgres yes (pre-created target) yes yes covered by pg->ch
postgres -> kafka n/a (append-only log) yes (keyed upsert replay, consumed back) rejected by config validation n/a

The clickhouse -> mssql route additionally certifies the ClickHouse -> SQL Server DDL type mapping (MSSQLTypeMapper): width-exact integer mappings (Int32 -> int, UInt64 -> decimal(20,0), ...), Nullable(...) / LowCardinality(...) unwrapping, and DateTime64 -> datetime2, so repeated chunks never trigger fake breaking schema-evolution changes (unit contract: tests/test_mssql_type_mapper_contracts.py).

Scenario coverage on postgres -> clickhouse:

  • empty/no-op chunks: a wider campaign window can include chunks before/after the available source data; those chunks still acquire a lease, execute, commit with rows_extracted=0 / rows_loaded=0, and do not block later non-empty chunks;
  • resume: chunk 2 fails once; the re-run skips the committed chunk (attempts stay at 1), retries the failed one, and reaches full parity;
  • campaign lock: a second active runner fails before source IO instead of racing the same ledger;
  • retry-failed: only chunks with status failed are retried; pending future chunks stay pending;
  • unknown retry_policy: fails before source IO, so typos cannot silently widen a targeted retry;
  • stale running chunk recovery: expired leases become retryable instead of blocking the campaign forever;
  • cancellation: cancel_requested blocks new chunks and records the operator reason;
  • idempotent re-run: a completed campaign re-run moves zero data and creates zero duplicates;
  • parallel_workers=2 over independent partition chunks with one connection per worker (pod-per-chunk topology);
  • verification bridge: <run_key>.execution.json consumable by dpone ops route-refresh-capture-snapshots / route-refresh-verify;
  • invalid combinations fail fast: multi-chunk full_refresh, unknown inner_mode, incremental_merge without unique_key, Kafka with a partitioned inner mode.

Airflow coverage: two data intervals rendered exactly as Airflow renders the pack env_vars; each dpone run (real CLI subprocess, env-based credentials) loads only its own window, a task-clear re-run of the same interval resumes the same campaign without duplicates, and the XCom summary carries the interval and bounded backfill sections.

CI

.github/workflows/backfill-integration.yml runs the suite with disposable compose services on path-filtered PRs, nightly, and on dispatch, and uploads junit + coverage evidence. MSSQL ODBC/bcp tooling is installed on the runner, so the MSSQL routes are always exercised in CI even when skipped locally.

Vendor-live backfill matrix

Managed/vendor systems are not exercised by the PR backfill workflow. They run through the manual vendor_live profile in .github/workflows/live-certification.yml:

profile=vendor_live
run_vendor_live=true
vendor_backfill_source_filter=*
vendor_backfill_sink_filter=*
vendor_backfill_case_id_filter=*

The workflow does not start Docker services in this profile. It expects the CI secret manager to provide the provider credentials. Before running managed tests, it executes a redacted readiness gate:

uv run dpone ops managed-credentials-readiness \
  --profile vendor_live \
  --required-env APPSFLYER_API_TOKEN \
  --required-env GOOGLE_ADS_DEVELOPER_TOKEN \
  --output-dir test_artifacts/live_certification_vendor/managed-credentials \
  --format json

The readiness artifact contains only variable names and present / missing status, never values. If this gate fails, the managed matrix does not start. When it passes, the workflow runs the source/sink matrix with:

DPONE_RUN_INTEGRATION=1 \
DPONE_RUN_INTEGRATION_MATRIX=1 \
DPONE_MATRIX_RUN_MODE=vendor_live \
DPONE_MATRIX_STRATEGY=backfill \
DPONE_MATRIX_ARTIFACT_DIR=test_artifacts/live_certification_vendor/backfill_matrix \
uv run pytest -m integration_matrix tests/integration/matrix -q

Use the source/sink/case filters to constrain blast radius before running a managed backfill certification. Keep run_vendor_live=false for routine PRs; this layer is release-candidate or connector-change evidence only.

The generated matrix/load-step artifacts can be fed back into the advisor:

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

This lets the advisor recommend chunk.step, parallel_workers and lease TTL from real campaign timings instead of only static chunk-count heuristics.

If the vendor-live matrix is red, the advisor treats it as certification evidence, not as a failed data chunk. The safe next action is to fix the matrix blockers first; the advisor pins the recommendation to one worker and returns resolve_vendor_certification_failures. Only a green matrix is allowed to participate in performance tuning. Stage throughput from repeated load-step records is also conservative: the slowest positive rows/sec per stage is used for bottleneck detection.