Data quality, tests & contracts

What can be wrong with a table

Data problems cluster into a handful of dimensions, and each has a cheap check. Freshness: the newest row is older than expected. Volume: far fewer or more rows than usual for the period. Schema: a column missing, renamed or of a new type. Uniqueness: duplicate keys. Completeness: NULLs where values are required. Distribution: values outside a plausible range, a category that never appeared before, a mean that jumped. Referential integrity: keys pointing at rows that do not exist. Most incidents are one of these; a table with a check for each rarely surprises anyone.

Where checks live

Checks belong at three points. At ingestion, on raw or bronze data: schema and basic validity, to catch upstream changes before they spread; bad rows are quarantined with a reason, not dropped. After transformation, on every model: the dbt tests from the previous module — uniqueness, nullability, accepted values, relationships, and business rules as singular tests. Before publication, on the marts that dashboards and downstream systems consume: freshness, volume and distribution checks that gate the publish step. The severity differs: an ingestion schema change and a mart uniqueness failure should block; a distribution drift might warn and open a ticket.

import polars as pl

MAX_REJECT_RATE = 0.01


def check_and_split(df: pl.DataFrame) -> tuple[pl.DataFrame, pl.DataFrame]:
    rules = {
        "order_id_null":   pl.col("order_id").is_null(),
        "amount_negative": pl.col("amount") < 0,
        "amount_huge":     pl.col("amount") > 50_000,
        "status_unknown":  ~pl.col("status").is_in(["placed", "paid", "shipped", "refunded"]),
        "future_date":     pl.col("ordered_at") > pl.lit(pl.datetime(2100, 1, 1)),
    }
    tagged = df.with_columns(
        pl.concat_list([pl.when(expr).then(pl.lit(name)).otherwise(None) for name, expr in rules.items()])
          .list.drop_nulls().alias("failed_rules")
    )
    good = tagged.filter(pl.col("failed_rules").list.len() == 0).drop("failed_rules")
    bad = tagged.filter(pl.col("failed_rules").list.len() > 0)
    rate = bad.height / max(df.height, 1)
    if rate > MAX_REJECT_RATE:
        raise RuntimeError(f"reject rate {rate:.2%} exceeds {MAX_REJECT_RATE:.0%}; not loading")
    return good, bad     # caller writes `bad` to orders_quarantine with run id and timestamp

Quarantine matters because dropping rows hides the problem and failing the whole load on one bad row makes the pipeline brittle. A reject-rate threshold separates "a few malformed rows, load and report" from "something upstream broke, stop". The quarantine table gets its own freshness and volume checks; a quarantine that grows quietly is a problem in disguise.

Writing checks that stay useful

A check that fires every Monday because Monday volume is always lower gets muted, and then it never catches anything. Specific, stable checks compare like with like: volume against the same weekday over recent weeks, not against yesterday; freshness against the source's actual delivery schedule; distributions with tolerances derived from history rather than guessed. Anomaly-based checks (learned baselines) help for high-cardinality metrics, but a handful of explicit assertions written by someone who understands the data catch more, with fewer false alarms.

-- tests/assert_fct_orders_volume_plausible.sql
with daily as (
    select order_date, count(*) as n
    from {{ ref('fct_orders') }}
    where order_date >= current_date - 35
    group by 1
),
baseline as (
    select
        order_date,
        n,
        avg(n) over (partition by extract(dow from order_date)
                     order by order_date rows between 4 preceding and 1 preceding) as same_weekday_avg
    from daily
)
select *
from baseline
where order_date = current_date - 1
  and same_weekday_avg is not null
  and (n < 0.7 * same_weekday_avg or n > 1.5 * same_weekday_avg)

• Every check has an owner and a runbook line: what it means, first thing to look at.
• Severity is explicit: error blocks the build and pages; warn records and notifies asynchronously.
• Checks are versioned with the models they protect and reviewed in the same pull request.
• Thresholds are numbers in config with a comment explaining where they came from.
• A check that never fails in six months and could not plausibly fail is a candidate for removal; one that fails weekly is either catching real issues or needs tuning — decide which.

Data contracts

Most quality incidents originate upstream: an application team renames a field, changes a type, or stops populating a column, and the first sign is a broken report days later. A data contract makes the interface explicit: the producer publishes a schema (names, types, nullability, semantics, allowed values), a delivery expectation (frequency, freshness), and a versioning policy; the consumer builds against it; and both sides enforce it in CI — the producer's tests fail if their output no longer matches the contract, and the consumer's ingestion checks fail if the received data does not. Contracts turn a surprise into a pull request conversation.

dataset: shop.orders
owner: shop-backend@acme.example
version: 2
schema:
  - name: order_id
    type: integer
    nullable: false
    unique: true
    description: Natural key from the shop database.
  - name: customer_id
    type: integer
    nullable: false
  - name: ordered_at
    type: timestamp
    nullable: false
    description: Order creation time, UTC.
  - name: status
    type: string
    nullable: false
    enum: [placed, paid, shipped, refunded, cancelled]
  - name: amount_cents
    type: integer
    nullable: false
    description: Order total in minor units; never negative.
delivery:
  frequency: every 15 minutes
  freshness_sla: 1 hour
  partition: ordered_at (date)
changes:
  additive: allowed without version bump (new nullable columns)
  breaking: requires version bump and 30 days notice (rename, type change, removal, new enum value)

Enforce the contract mechanically. The producer generates the schema from their code and diffs it against the contract in CI. The consumer's ingestion step validates incoming data against the same file (the pydantic models from the Python module are one way; a schema registry is another for streaming). When a breaking change is needed, the producer publishes version 2 alongside version 1 for the notice period, exactly like an API.

When a check fails

A failing check is an incident with a small blast radius, and the response follows the SRE track's shape. Block the affected data from reaching consumers: dbt build skips dependents; a publish gate holds the mart; a dashboard shows its last good version with a banner. Alert with context: table, check, observed versus expected value, the run, a link to the data. Root-cause from the top of the lineage down: was the source late, did a load duplicate, did a schema change. Fix and rerun with the idempotent backfill. Record it, and add the check that would have caught it earlier if one did not exist. A short log of quality incidents, with time to detect and time to fix, is the quality program's own dashboard.