Data Quality Checks That Run Every Load

A quality check is a query that returns the rows that should not exist. If it returns none, the data passed. Everything in this lesson is that one idea, applied at the point where bad data would otherwise become someone’s dashboard.

Setup

import duckdb
con = duckdb.connect("bookshop.duckdb")

con.execute("""
    create or replace table bronze_orders as
    select * from (values
        (1001,    1, date '2026-01-04', 'completed',  25.50),
        (1002,    2, date '2026-01-04', 'COMPLETED',  12.00),
        (1003,    1, date '2026-01-04', 'returned',   40.00),
        (1003,    1, date '2026-01-04', 'returned',   40.00),   -- duplicate
        (1004,    3, date '2026-01-04', 'completed',  -8.75),   -- negative
        (1005, NULL, date '2026-01-04', 'completed',  63.20),   -- null fk
        (1006,    9, date '2026-01-04', 'refunded',   19.99),   -- orphan
        (1007,    2, date '2026-01-04', 'unknown',     5.00)    -- bad enum
    ) t(order_id, customer_id, ordered_at, status, amount)
""")
con.execute("""
    create or replace table customers as
    select * from (values (1,'Ada','GB'),(2,'Grace','US'),(3,'Alan','GB'))
    t(customer_id, full_name, country_code)
""")

Six checks

CHECKS = {
    "unique_order_id": """
        select order_id, count(*) as n from bronze_orders
        group by 1 having count(*) > 1
    """,
    "not_null_customer_id": """
        select order_id from bronze_orders where customer_id is null
    """,
    "accepted_status": """
        select order_id, status from bronze_orders
        where lower(trim(status)) not in ('completed','returned','refunded','pending')
    """,
    "positive_amount": """
        select order_id, amount from bronze_orders where amount <= 0
    """,
    "customer_exists": """
        select o.order_id, o.customer_id from bronze_orders o
        left join customers c using (customer_id)
        where o.customer_id is not null and c.customer_id is null
    """,
    "amount_within_range": """
        select order_id, amount from bronze_orders where amount > 10000
    """,
}

def run_checks(table="bronze_orders"):
    results = []
    for name, sql in CHECKS.items():
        failures = con.execute(sql).fetchall()
        results.append((name, len(failures), failures[:2]))
    width = max(len(n) for n, _, _ in results)
    for name, n, sample in results:
        flag = "PASS" if n == 0 else "FAIL"
        print(f"{flag}  {name:<{width}}  {n:>3}  {sample if n else ''}")
    return sum(n for _, n, _ in results)

total = run_checks()
print(f"\n{total} failing rows")

FAIL  unique_order_id        1  [(1003, 2)]
FAIL  not_null_customer_id   1  [(1005,)]
FAIL  accepted_status        1  [(1007, 'unknown')]
FAIL  positive_amount        1  [(1004, Decimal('-8.75'))]
FAIL  customer_exists        1  [(1006, 9)]
PASS  amount_within_range    0  

5 failing rows

Each check names the rows, not just a count — which is the difference between a failure you can act on and one that starts an investigation.

Note 1002 with COMPLETED passed accepted_status, because the check lowercases. Deciding whether inconsistent casing is a quality failure or a cleaning step is a real choice: here it is cleaning, so the check tolerates it and silver normalises it.

Fail, warn, or quarantine

Not every failure deserves the same response:

SEVERITY = {
    "unique_order_id":      "error",    # breaks every downstream aggregate
    "not_null_customer_id": "error",
    "customer_exists":      "warn",     # known-imperfect source
    "accepted_status":      "warn",
    "positive_amount":      "warn",     # refunds arrive as negatives sometimes
    "amount_within_range":  "warn",
}

def gate():
    errors, warnings = [], []
    for name, sql in CHECKS.items():
        n = len(con.execute(sql).fetchall())
        if n:
            (errors if SEVERITY[name] == "error" else warnings).append((name, n))
    for name, n in warnings:
        print(f"WARN  {name}: {n} rows")
    if errors:
        raise SystemExit(f"BLOCKED: {errors}")
    print("gate passed")

gate()

WARN  customer_exists: 1 rows
WARN  accepted_status: 1 rows
WARN  positive_amount: 1 rows
BLOCKED: [('unique_order_id', 1), ('not_null_customer_id', 1)]

The pipeline stops, and yesterday’s correct marts stay in place rather than being overwritten from a broken load. That is the whole point of a gate: failing to publish is recoverable; publishing wrong numbers is not.

Quarantine instead of dropping

con.execute("""
    create or replace table quarantine as
    select order_id, customer_id, ordered_at, status, amount,
           list_value(
               case when customer_id is null then 'null_customer_id' end,
               case when amount <= 0 then 'non_positive_amount' end,
               case when lower(trim(status)) not in ('completed','returned','refunded','pending')
                    then 'bad_status' end,
               case when customer_id not in (select customer_id from customers)
                    then 'orphan_customer' end
           ).list_filter(x -> x is not null) as reasons,
           now() as quarantined_at
    from bronze_orders
    where customer_id is null
       or amount <= 0
       or lower(trim(status)) not in ('completed','returned','refunded','pending')
       or customer_id not in (select customer_id from customers)
""")

print(con.execute("select order_id, status, amount, reasons from quarantine order by order_id").df().to_string(index=False))

 order_id    status  amount                reasons
     1004 completed   -8.75  [non_positive_amount]
     1005 completed   63.20     [null_customer_id]
     1006  refunded   19.99      [orphan_customer]
     1007   unknown    5.00           [bad_status]

con.execute("""
    create or replace table silver_orders as
    select order_id, customer_id, ordered_at, lower(trim(status)) as status, amount
    from (select *, row_number() over (partition by order_id order by order_id) rn
          from bronze_orders)
    where rn = 1
      and order_id not in (select order_id from quarantine)
""")
print(con.execute("select count(*) from silver_orders").fetchone())

(3,)

Eight rows in, three clean rows out, four quarantined with reasons, one deduplicated. Nothing was lost, and the four bad rows are now a queryable backlog rather than a gap.

print(con.execute("""
    select unnest(reasons) as reason, count(*) as n
    from quarantine group by 1 order by n desc
""").df().to_string(index=False))

              reason  n
 non_positive_amount  1
    null_customer_id  1
     orphan_customer  1
          bad_status  1

That grouping is what you take to the upstream team — a named problem with a count, not “your data is bad”.

Freshness and volume

Column checks cannot see a load that simply did not happen.

con.execute("""
    create or replace table load_history as
    select * from (values
        (date '2026-01-01', 4820), (date '2026-01-02', 4712), (date '2026-01-03', 4902),
        (date '2026-01-04', 4688), (date '2026-01-05', 4771), (date '2026-01-06', 4810),
        (date '2026-01-07', 1204)
    ) t(load_date, rows_loaded)
""")

print(con.execute("""
    select
        load_date, rows_loaded,
        round(avg(rows_loaded) over (order by load_date rows between 6 preceding and 1 preceding), 0) as trailing_avg,
        round(100.0 * rows_loaded / avg(rows_loaded) over (order by load_date rows between 6 preceding and 1 preceding), 1) as pct_of_avg
    from load_history order by load_date desc limit 3
""").df().to_string(index=False))

  load_date  rows_loaded  trailing_avg  pct_of_avg
 2026-01-07         1204        4783.8        25.2
 2026-01-06         4810        4778.6       100.7
 2026-01-05         4771        4780.5        99.8

25% of the trailing average — a partial load. Every check in the previous section passes on those 1,204 rows, because they are individually valid. Volume is the check that catches a truncated file, a source outage, or a filter someone tightened.

def volume_gate(threshold=0.5):
    row = con.execute("""
        select rows_loaded,
               avg(rows_loaded) over (order by load_date rows between 6 preceding and 1 preceding) as avg7
        from load_history order by load_date desc limit 1
    """).fetchone()
    if row[1] and row[0] < row[1] * threshold:
        raise SystemExit(f"volume gate: {row[0]} rows is {row[0]/row[1]:.0%} of the 7-day average")

volume_gate()

volume gate: 1204 rows is 25% of the 7-day average

Freshness is the same shape:

print(con.execute("""
    select max(load_date) as latest,
           datediff('day', max(load_date), current_date) as days_stale
    from load_history
""").df().to_string(index=False))

     latest  days_stale
 2026-01-07         245

Reconciliation

The check that catches what per-column tests cannot — does the mart still agree with its source?

print(con.execute("""
    with source as (select sum(amount) as total from silver_orders where status = 'completed'),
         mart   as (select sum(revenue) as total from gold_daily_revenue)
    select source.total as source_total, mart.total as mart_total,
           round(source.total - mart.total, 2) as difference
    from source, mart
""").df().to_string(index=False))

 source_total  mart_total  difference
        37.50       25.50       12.00

£12 missing, because an inner join in the mart dropped an order whose customer is absent. Every column check passed. Reconciliation tests between a layer and the one below it are the highest-value checks in a pipeline, and the ones most often missing.

Recording quality over time

con.execute("""
    create table if not exists quality_runs (
        run_date date, check_name varchar, failing_rows bigint, severity varchar, checked_at timestamp
    )
""")

def record(run_date):
    for name, sql in CHECKS.items():
        n = len(con.execute(sql).fetchall())
        con.execute("insert into quality_runs values (?, ?, ?, ?, now())",
                    [run_date, name, n, SEVERITY[name]])

record("2026-01-04")

print(con.execute("""
    select check_name, failing_rows,
           avg(failing_rows) over (partition by check_name order by run_date
                                   rows between 7 preceding and 1 preceding) as trailing
    from quality_runs where run_date = '2026-01-04'
    order by failing_rows desc limit 3
""").df().to_string(index=False))

        check_name  failing_rows  trailing
   unique_order_id             1       0.0
 accepted_status               1       0.1
 positive_amount               1       0.9

unique_order_id went from a trailing zero to one — a new problem. positive_amount at 1 against a trailing 0.9 is business as usual. Alerting on the change rather than the absolute count is what stops a quality dashboard becoming noise everyone mutes.

Tools

The patterns above are what dbt tests, Great Expectations, Soda and Databricks pipeline expectations all implement. Reach for one when you want a shared vocabulary and reporting rather than a bespoke script — but the decisions are the same either way: which checks, what severity, and whether a failure blocks publication.

Practice

CHECKS["no_future_dates"] = "select order_id, ordered_at from bronze_orders where ordered_at > current_date"

FAIL  no_future_dates  1  [(1009, datetime.date(2027, 3, 1))]

              reason  n
 non_positive_amount  1
    null_customer_id  1
     orphan_customer  1

volume gate: 1204 rows is 25% of the 7-day average

 source_total  mart_total  difference
        37.50       25.50       12.00

Next: incremental loads and change data capture.