Data-diff backend database

Data-diff uses a PostgreSQL control-plane database to store check definitions, execution history, results, and coverage. It is independent of source and target connections and must not also serve as a replication target.

See Data-diff checks to define, schedule, and remediate data-diff checks.

Configuration

Add backend_db to config.yml:

backend_db:
  host: "backend.example.com"
  port: 5432
  user: "pipelinewise"
  password: "<vault encrypted>"
  dbname: "pipelinewise"
  sslmode: "verify-full"
  ddl_user: "pipelinewise_ddl"
  ddl_password: "<vault encrypted>"

When PostgreSQL is also the replication target, give the backend its own service or database.

backend_db enables data-diff. Without it, import_config warns and ignores every data_diff block. Replication never reads the backend, so an outage pauses reconciliation only. However, import_config fails when it cannot persist check definitions.

ddl_user runs Alembic migrations and owns the schema. The application user can therefore hold DML grants only. The migration grants user what it needs, so that role requires nothing beyond CONNECT. Set ddl_user to the application credentials when separate roles are not required.

Schema

        %%{init: {"er": {"diagramPadding": 8, "entityPadding": 8}}}%%
erDiagram
    direction LR
    dd_checks ||--o{ dd_preflights : check_id
    dd_checks ||--o{ dd_runs : dd_check_id
    dd_preflights o|--o{ dd_runs : preflight_id
    dd_runs o|--o{ dd_runs : rerun_of_run_id
    dd_runs ||--o{ dd_results : run_id
    dd_checks ||--o{ dd_effective_attempts : check_id
    dd_runs ||--o| dd_effective_attempts : run_id
    dd_checks ||--o{ dd_coverage_events : check_id
    dd_runs ||--o| dd_coverage_events : evaluated_run_id
    dd_runs o|--o{ dd_coverage_events : blocking_run_id
    dd_checks ||--o| dd_coverage_state : check_id
    dd_runs ||--o{ dd_coverage_state : evaluated_run_id
    dd_runs o|--o{ dd_coverage_state : blocking_run_id

    dd_checks["dd_checks<br/>Versioned check definitions<br/>imported from YAML config"] {
        UUID check_id PK
        TEXT full_check_name "UQ with revision; UQ when current"
        INTEGER revision "UQ with full_check_name"
        CHAR(64) config_hash
        JSONB canonical_config
        TEXT target_id
        TEXT tap_id
        TEXT source_type
        TEXT target_type
        TEXT source_database
        TEXT target_database
        TEXT source_schema
        TEXT source_table
        TEXT target_schema
        TEXT target_table
        TEXT source_key_column
        TEXT target_key_column
        TEXT source_timestamp_column
        TEXT target_timestamp_column
        JSONB checks
        TEXT frequency
        BIGINT window_start_seconds "greater than 0"
        BIGINT window_end_seconds "at least 0"
        BIGINT statement_timeout_seconds "greater than 0"
        BOOLEAN current
        TIMESTAMPTZ created_at
        TIMESTAMPTZ superseded_at "nullable"
    }

    dd_preflights["dd_preflights<br/>Source size and index validation<br/>recorded before each check execution"] {
        UUID preflight_id PK
        UUID check_id FK
        TEXT status "PASS, BLOCKED, or ERROR"
        TIMESTAMPTZ checked_at
        CHAR(64) query_fingerprint
        JSONB index_metadata
        JSONB findings
        TEXT error "nullable"
        BIGINT table_rows "nullable"
        BIGINT row_limit "nullable"
        BOOLEAN has_leading_index "nullable"
    }

    dd_runs["dd_runs<br/>Shared execution log<br/>one row per scheduled job attempt"] {
        UUID run_id PK
        UUID dd_check_id FK
        TIMESTAMPTZ scheduled_for "UQ with dd_check_id and attempt"
        TIMESTAMPTZ window_start "before window_end"
        TIMESTAMPTZ window_end
        INTEGER attempt "UQ with dd_check_id and scheduled_for"
        TEXT trigger "SCHEDULED, MANUAL, RETRY, or REMEDIATION"
        TEXT status "RUNNING, PASS, FAIL, or ERROR"
        UUID rerun_of_run_id FK "nullable self-reference"
        TEXT remediation_reference "nullable"
        UUID preflight_id FK "nullable"
        TIMESTAMPTZ started_at
        TIMESTAMPTZ finished_at "nullable"
        TEXT error "nullable"
    }

    dd_results["dd_results<br/>Per-check-type outcome detail<br/>for each data-diff run"] {
        UUID run_id PK, FK
        TEXT check_type PK
        TEXT status "PASS, FAIL, or ERROR"
        JSONB source_value "nullable"
        JSONB target_value "nullable"
        DOUBLE_PRECISION source_query_seconds "nullable"
        DOUBLE_PRECISION target_query_seconds "nullable"
        TEXT error "nullable"
    }

    dd_effective_attempts["dd_effective_attempts<br/>Latest terminal attempt for each<br/>data-diff scheduled slot"] {
        UUID check_id PK, FK
        TIMESTAMPTZ scheduled_for PK
        UUID run_id FK, UK
        INTEGER attempt "greater than 0"
        TIMESTAMPTZ window_start "before window_end"
        TIMESTAMPTZ window_end
        TEXT status "PASS, FAIL, or ERROR"
    }

    dd_coverage_state["dd_coverage_state<br/>Materialized current verified-through<br/>watermark for each data-diff check"] {
        UUID check_id PK, FK
        TIMESTAMPTZ coverage_start "at most verified_through"
        TIMESTAMPTZ verified_through "at most max_observed_end"
        TIMESTAMPTZ max_observed_end
        TEXT coverage_status "CONTIGUOUS or BLOCKED"
        UUID blocking_run_id FK "nullable"
        UUID evaluated_run_id FK
        TEXT event_type "INITIALIZE, ADVANCE, INVALIDATE, BLOCK, or CONFIRM"
        BIGINT state_version "greater than 0"
        TIMESTAMPTZ updated_at
        TEXT reason
    }

    dd_coverage_events["dd_coverage_events<br/>Append-only log tracking the data-diff<br/>verified-through watermark"] {
        UUID coverage_event_id PK
        BIGSERIAL event_sequence UK
        UUID check_id FK
        UUID evaluated_run_id FK, UK
        TEXT event_type "INITIALIZE, ADVANCE, INVALIDATE, BLOCK, or CONFIRM"
        TIMESTAMPTZ coverage_start "at most verified_through"
        TIMESTAMPTZ previous_verified_through "nullable"
        TIMESTAMPTZ verified_through "at most max_observed_end"
        TIMESTAMPTZ max_observed_end
        TEXT coverage_status "CONTIGUOUS or BLOCKED"
        UUID blocking_run_id FK "nullable"
        TIMESTAMPTZ recorded_at
        TEXT reason
    }

    

Data-diff backend schema after migration 001

PipelineWise keeps every attempt in dd_runs and every coverage transition in dd_coverage_events. dd_effective_attempts materializes only the highest terminal attempt for each scheduled slot, while dd_coverage_state stores the current watermark. A new chronological slot updates that state directly. A replacement or out-of-order slot recalculates it from the effective rows without rescanning superseded attempts.

Reporting queries

Run these queries against the PipelineWise backend database.

Current coverage watermarks:

SELECT state.check_id, checks.full_check_name,
       checks.revision, checks.source_schema, checks.source_table,
       checks.source_timestamp_column, state.coverage_start,
       state.verified_through, state.max_observed_end,
       state.coverage_status, state.blocking_run_id,
       state.evaluated_run_id, state.event_type,
       state.updated_at AS verified_at, state.reason
  FROM public.dd_coverage_state state
  JOIN public.dd_checks checks
    ON checks.check_id = state.check_id;

Failed runs and their remediation attempts:

SELECT checks.full_check_name, checks.revision,
       original.run_id AS failed_run_id,
       original.status AS failed_status,
       original.window_start, original.window_end,
       original.finished_at AS failed_at,
       remediation.run_id AS remediation_run_id,
       remediation.attempt AS remediation_attempt,
       remediation.status AS remediation_status,
       remediation.remediation_reference,
       remediation.finished_at AS remediation_finished_at,
       COALESCE(remediation.status = 'PASS', FALSE) AS recovered
  FROM public.dd_runs original
  JOIN public.dd_checks checks
    ON checks.check_id = original.dd_check_id
  LEFT JOIN public.dd_runs remediation
    ON remediation.rerun_of_run_id = original.run_id
 WHERE original.rerun_of_run_id IS NULL
   AND original.status IN ('FAIL', 'ERROR');