Lab: Build a quality gate
Write the check runner, wire it after the pipeline as a blocking gate, then plant a corrupted batch and watch the gate catch all five problems.
This lab makes Harbor Lane's warehouse defended. You'll write checks.py — a runner that executes the test suite against harbor.duckdb and exits nonzero on any failure — then prove it works the only way that counts: by feeding it corrupted data and watching it refuse.
'''Harbor Lane quality gate. Exit 0 = safe to expose; exit 1 = block the load.'''
import sys
import duckdb
con = duckdb.connect('harbor.duckdb', read_only=True)
failures = []
def check(name, sql):
'''sql returns one number: the violation count (0 = pass).'''
bad = con.execute(sql).fetchone()[0]
print(f" {'PASS' if bad == 0 else 'FAIL'} {name}" + ('' if bad == 0 else f' [{bad}]'))
if bad != 0:
failures.append(name)
# --- row-level tests ---------------------------------------------------------
check('order_id unique',
'SELECT count(*) - count(DISTINCT order_id) FROM fct_order_lines')
check('sku resolves to dim_product', '''
SELECT count(*) FROM fct_order_lines f
LEFT JOIN dim_product p USING (sku) WHERE p.sku IS NULL''')
check('qty positive and present',
'SELECT count(*) FROM fct_order_lines WHERE qty IS NULL OR qty <= 0')
check('unit_price in sane range',
'SELECT count(*) FROM fct_order_lines WHERE unit_price <= 0 OR unit_price > 500')
check('status within contract', '''
SELECT count(*) FROM fct_order_lines
WHERE status NOT IN ('completed', 'cancelled', 'refunded')''')
# --- table-level monitors ------------------------------------------------------
check('freshness: newest partition within 1 day of latest expected', '''
SELECT CASE WHEN max(order_date) >= (SELECT max(order_date) FROM fct_order_lines) THEN 0 ELSE 1 END
FROM fct_order_lines''') # placeholder - see step 2, you will make this real
check('volume: latest day within 50-150% of trailing-7 average', '''
WITH daily AS (
SELECT order_date, count(*) AS n FROM fct_order_lines GROUP BY 1),
latest AS (SELECT n FROM daily ORDER BY order_date DESC LIMIT 1),
base AS (
SELECT avg(n) AS a FROM (
SELECT n FROM daily ORDER BY order_date DESC LIMIT 7 OFFSET 1))
SELECT CASE WHEN (SELECT n FROM latest)
BETWEEN 0.5 * (SELECT a FROM base)
AND 1.5 * (SELECT a FROM base)
THEN 0 ELSE 1 END''')
print(f'\n{len(failures)} failing check(s)')
sys.exit(1 if failures else 0)Note the freshness check as written is a placeholder (it compares the table to itself — always passes). A check that cannot fail is worse than no check, because it manufactures confidence. Here is the finished version — replace the placeholder check('freshness: ...') call in checks.py with this one:
# The fact table is FRESH when its newest partition matches the newest
# orders_*.csv on disk. glob() lists the files; regexp_extract pulls the
# date out of each filename. Returns 0 (fresh) or 1 (stale).
check('freshness: fact keeps up with newest export file', '''
SELECT CASE WHEN
(SELECT max(order_date) FROM fct_order_lines)
>= (SELECT max(regexp_extract(file, '\\d{4}-\\d{2}-\\d{2}')::DATE)
FROM glob('data/raw/orders_*.csv'))
THEN 0 ELSE 1 END''')Delete the newest orders_*.csv and re-run to prove the check now can fail: the fact table's newest partition is suddenly ahead of the newest file on disk, so freshness flips to FAIL. (Writing this query yourself first, before pasting, is a worthwhile optional stretch — it's the one bit of genuinely fiddly DuckDB SQL in the lab.)
Now attack your own gate
'''Simulate the upstream incident: a corrupted daily export.'''
import csv
with open('data/raw/orders_2026-05-31.csv', 'w', newline='') as f:
w = csv.writer(f)
w.writerow(['order_id','channel','customer_ref','sku','qty','unit_price','ordered_at','status'])
rows = [
['O9001', 'web', 'W0001', 'SKU-001', 2, 9.50, '2026-05-31T09:00:00', 'completed'],
['O9001', 'web', 'W0001', 'SKU-001', 2, 9.50, '2026-05-31T09:00:00', 'completed'], # dup id
['O9002', 'web', 'W0002', 'SKU-999', 1, 12.00, '2026-05-31T10:00:00', 'completed'], # ghost sku
['O9003', 'pos', 'P0003', 'SKU-004', '', 8.00, '2026-05-31T11:00:00', 'completed'], # null qty
['O9004', 'web', 'W0004', 'SKU-005', 1, 7.25, '2026-05-31T12:00:00', 'complete'], # bad enum
]
w.writerows(rows) # also only 5 rows: ~5% of normal volume
print('corrupted export planted: orders_2026-05-31.csv')python3 checks.py # all PASS on clean data
python3 plant_bad_batch.py
python3 pipeline.py # loads the poison (pipelines don't judge)
python3 checks.py # FAIL x5: uniqueness, RI, qty, enum, volume
echo $? # 1 -> a scheduler would now block + page- 1Run the healthy pass, plant the bad batch, and confirm the gate catches five distinct failures (four row-level + volume). Match each FAIL line to the exact planted row that caused it.
- 2Fix the freshness check: paste the finished version above over the placeholder in
checks.py, then delete the newestorders_*.csvand confirm the check now can fail. (Optional stretch: write theglob()+regexp_extractquery yourself before looking at the provided one.) - 3Clean up: delete
orders_2026-05-31.csv, re-runpipeline.py— and notice the poison rows are still in the fact table (the partition overwrite only rewrites dates present in the batch). Write the one-line repair (DELETE FROM fct_order_lines WHERE order_date = '2026-05-31'), then write down the lesson: gates should run before exposure, because un-loading is manual. - 4Stretch: create a
check_runslog table (run timestamp, check name, violation count) written bychecks.py, and aprofile_dailytable storing per-day row counts and null rates. You've just built the storage layer of a data-observability product.
The problem set gives you six incident narratives (late file, double-send, enum drift, slow null creep, timezone shift, partial backfill) and asks which check family catches each and how long the blind spot lasts with daily batch. Then it has you calibrate z-score thresholds against a noisy series — the exact tuning exercise Vigil's monitors automate with learned baselines.