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_check_definitions ||--o{ dd_preflight_log : check_id
    dd_check_definitions ||--o{ dd_run_attempts : check_id
    dd_preflight_log o|--o{ dd_run_attempts : preflight_id
    dd_run_attempts o|--o{ dd_run_attempts : rerun_of_run_id
    dd_run_attempts ||--o{ dd_run_results : run_id
    dd_check_definitions ||--o{ dd_run_slot_state : check_id
    dd_run_attempts ||--o| dd_run_slot_state : run_id
    dd_check_definitions ||--o{ dd_watermark_events : check_id
    dd_run_attempts ||--o| dd_watermark_events : evaluated_run_id
    dd_run_attempts o|--o{ dd_watermark_events : blocking_run_id
    dd_check_definitions ||--o| dd_watermark_state : check_id
    dd_run_attempts ||--o{ dd_watermark_state : last_evaluated_run_id
    dd_run_attempts o|--o{ dd_watermark_state : blocking_run_id

    dd_check_definitions["dd_check_definitions<br/>Versioned data-diff check definitions<br/>imported from YAML configuration"] {
        UUID check_id PK
        TEXT full_check_name "UQ with revision; UQ when is_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 is_current
        TIMESTAMPTZ created_at
        TIMESTAMPTZ superseded_at "nullable"
    }

    dd_preflight_log["dd_preflight_log<br/>Append-only preflight evidence and<br/>early validation errors for run attempts"] {
        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_run_attempts["dd_run_attempts<br/>Execution attempts updated from RUNNING<br/>to a terminal outcome"] {
        UUID run_id PK
        UUID check_id FK
        TIMESTAMPTZ scheduled_for "UQ with check_id and attempt"
        TIMESTAMPTZ window_start "before window_end"
        TIMESTAMPTZ window_end
        INTEGER attempt "UQ with check_id and scheduled_for"
        TEXT trigger_type "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_run_results["dd_run_results<br/>Append-only per-check-type outcomes<br/>for completed run attempts"] {
        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_run_slot_state["dd_run_slot_state<br/>Mutable selected terminal attempt for<br/>each check and 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_watermark_state["dd_watermark_state<br/>Mutable current verified interval and furthest<br/>observed window for each check definition"] {
        UUID check_id PK, FK
        TIMESTAMPTZ verified_start "at most verified_end"
        TIMESTAMPTZ verified_end "at most furthest_observed_end"
        TIMESTAMPTZ furthest_observed_end
        TEXT verified_status "CONTIGUOUS or BLOCKED"
        UUID blocking_run_id FK "nullable"
        UUID last_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_watermark_events["dd_watermark_events<br/>Append-only history of verified intervals<br/>and furthest observed window transitions"] {
        UUID watermark_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 verified_start "at most verified_end"
        TIMESTAMPTZ previous_verified_end "nullable"
        TIMESTAMPTZ verified_end "at most furthest_observed_end"
        TIMESTAMPTZ furthest_observed_end
        TEXT verified_status "CONTIGUOUS or BLOCKED"
        UUID blocking_run_id FK "nullable"
        TIMESTAMPTZ recorded_at
        TEXT reason
    }

    

Data-diff backend schema after migration 002

The schema has two related paths. The first records definitions and execution evidence. The second selects one terminal attempt per scheduled slot, folds those slot outcomes into the current watermark, and records every watermark transition:

dd_check_definitions
    ├── dd_preflight_log
    └── dd_run_attempts
            └── dd_run_results

dd_run_attempts
    └── dd_run_slot_state
            └── dd_watermark_state
                    └── dd_watermark_events

The second path describes processing rather than foreign-key ownership; the ERD above shows the exact database relationships.

PipelineWise keeps every attempt in dd_run_attempts and every watermark transition in dd_watermark_events. dd_run_slot_state materializes only the highest terminal attempt for each scheduled slot, while dd_watermark_state stores the current verified interval and the furthest observed window end. A new chronological slot updates that state directly. A replacement or out-of-order slot recalculates it from the slot-state rows without rescanning superseded attempts.

Reporting queries

Run these queries against the PipelineWise backend database.

Watermark per table:

SELECT d.full_check_name,
       d.revision,
       d.source_database,
       d.source_schema,
       d.source_table,
       d.target_database,
       d.target_schema,
       d.target_table,
       COALESCE(w.verified_status, 'NOT_RUN') AS verified_status,
       w.verified_start,
       w.verified_end,
       w.furthest_observed_end,
       CURRENT_TIMESTAMP - w.verified_end AS verification_lag,
       w.blocking_run_id,
       w.last_evaluated_run_id,
       w.updated_at AS watermark_updated_at,
       w.reason
  FROM public.dd_check_definitions d
  LEFT JOIN public.dd_watermark_state w
    ON w.check_id = d.check_id
 WHERE d.is_current
 ORDER BY d.target_id, d.tap_id,
          d.source_schema, d.source_table;

Successful and failed check results per table per UTC day. Historical definition revisions and every attempt, including retries and remediation runs, are included. The result counts are per check_type; terminal run-level errors without result rows are counted separately:

WITH outcomes AS (
    SELECT d.full_check_name,
           d.source_database,
           d.source_schema,
           d.source_table,
           d.target_database,
           d.target_schema,
           d.target_table,
           (COALESCE(a.finished_at, a.started_at)
               AT TIME ZONE 'UTC')::date AS check_date,
           a.status AS run_status,
           r.run_id AS result_run_id,
           r.status AS result_status
      FROM public.dd_run_attempts a
      JOIN public.dd_check_definitions d
        ON d.check_id = a.check_id
      LEFT JOIN public.dd_run_results r
        ON r.run_id = a.run_id
     WHERE a.status IN ('PASS', 'FAIL', 'ERROR')
)
SELECT full_check_name,
       source_database,
       source_schema,
       source_table,
       target_database,
       target_schema,
       target_table,
       check_date,
       COUNT(*) FILTER (
           WHERE result_status = 'PASS'
       ) AS successful_check_results,
       COUNT(*) FILTER (
           WHERE result_status IN ('FAIL', 'ERROR')
       ) AS failed_check_results,
       COUNT(*) FILTER (
           WHERE result_run_id IS NULL
             AND run_status = 'ERROR'
       ) AS run_level_errors
  FROM outcomes
 GROUP BY full_check_name,
          source_database, source_schema, source_table,
          target_database, target_schema, target_table,
          check_date
 ORDER BY check_date DESC, full_check_name;

Current unresolved failures. Only current check-definition revisions are shown; failures belonging to superseded revisions are intentionally excluded:

SELECT d.full_check_name,
       d.revision,
       d.source_schema,
       d.source_table,
       s.scheduled_for,
       s.window_start,
       s.window_end,
       s.run_id,
       a.attempt,
       a.trigger_type,
       COALESCE(r.check_type, '<run-level error>') AS failed_check,
       COALESCE(r.status, s.status) AS failure_status,
       COALESCE(r.error, a.error) AS error,
       COALESCE(s.run_id = w.blocking_run_id, FALSE) AS blocks_watermark,
       a.finished_at
  FROM public.dd_run_slot_state s
  JOIN public.dd_check_definitions d
    ON d.check_id = s.check_id
  JOIN public.dd_run_attempts a
    ON a.run_id = s.run_id
  LEFT JOIN public.dd_run_results r
    ON r.run_id = s.run_id
   AND r.status IN ('FAIL', 'ERROR')
  LEFT JOIN public.dd_watermark_state w
    ON w.check_id = s.check_id
 WHERE d.is_current
   AND s.status IN ('FAIL', 'ERROR')
 ORDER BY a.finished_at DESC,
          d.full_check_name,
          r.check_type;

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_run_attempts original
  JOIN public.dd_check_definitions checks
    ON checks.check_id = original.check_id
  LEFT JOIN public.dd_run_attempts remediation
    ON remediation.rerun_of_run_id = original.run_id
 WHERE original.rerun_of_run_id IS NULL
   AND original.status IN ('FAIL', 'ERROR');