CI/CD for Data: Gating Promotion with Real Tests

Promote a staging table to production only when it passes real checks, then watch a broken build get blocked automatically instead of reaching anyone querying prod.

Two environments, one gate function

import duckdb
con = duckdb.connect()
con.execute("create schema staging")
con.execute("create schema prod")

# prod already has yesterday's good data
con.execute("create table prod.orders as select * from (values (1, 25.5), (2, 12.0)) t(order_id, amount)")

def run_gate_checks(schema_table):
    checks = con.execute(f"""
        select count(*) as rows,
               count(*) filter (order_id is null) as null_ids,
               count(*) - count(distinct order_id) as duplicate_ids
        from {schema_table}
    """).fetchone()
    rows, nulls, dupes = checks
    failures = []
    if rows == 0:
        failures.append("zero rows - refusing to promote an empty table")
    if nulls:
        failures.append(f"{nulls} null order_id(s)")
    if dupes:
        failures.append(f"{dupes} duplicate order_id(s)")
    return failures

def deploy(candidate_schema, prod_schema, table):
    failures = run_gate_checks(f"{candidate_schema}.{table}")
    if failures:
        print(f"BLOCKED: {', '.join(failures)}")
        print(f"  {prod_schema}.{table} left untouched")
        return False
    con.execute(f"create or replace table {prod_schema}.{table} as select * from {candidate_schema}.{table}")
    print(f"DEPLOYED: {candidate_schema}.{table} promoted to {prod_schema}.{table}")
    return True

A clean build promotes normally

con.execute("create or replace table staging.orders as select * from "
            "(values (1, 25.5), (2, 12.0), (3, 8.75)) t(order_id, amount)")
deploy("staging", "prod", "orders")
print(f"prod.orders now has {con.execute('select count(*) from prod.orders').fetchone()[0]} rows")

$ python cicd.py
== good candidate build: three clean rows ==
DEPLOYED: staging.orders promoted to prod.orders
prod.orders now has 3 rows

A build with a bug gets blocked before it reaches prod

con.execute("create or replace table staging.orders as select * from "
            "(values (1, 25.5), (2, 12.0), (2, 999.0)) t(order_id, amount)")
deploy("staging", "prod", "orders")
print(f"prod.orders still has {con.execute('select count(*) from prod.orders').fetchone()[0]} rows "
      f"(the good build from before, untouched)")

== broken candidate build: a bug introduces a duplicate order_id ==
BLOCKED: 1 duplicate order_id(s)
  prod.orders left untouched
prod.orders still has 3 rows (the good build from before, untouched)

Order 2 appears twice in the second build — some upstream join bug, easy to imagine shipping unnoticed. run_gate_checks catches it before deploy ever runs the promotion statement, so prod.orders stays exactly as it was: 3 rows, no duplicates, from the last build that actually passed. Nobody querying prod ever sees the broken version. This is the same shape as a CI pipeline blocking a merge on a failing test suite — the “test suite” here just happens to run against real data instead of only against code.

Practice

BLOCKED: 1 negative amount(s)
  prod.orders left untouched

def deploy(candidate_schema, prod_schema, table):
    failures = run_gate_checks(f"{candidate_schema}.{table}")
    if failures:
        print(f"BLOCKED: {failures}")
        return False
    con.execute(f"create or replace table {prod_schema}.{table}_previous as select * from {prod_schema}.{table}")
    con.execute(f"create or replace table {prod_schema}.{table} as select * from {candidate_schema}.{table}")
    return True

Next: disaster recovery — backup, restore, and the RPO/RTO numbers that decide how much data a platform can afford to lose.