Two ideas carry the route: a test is a query that should return nothing, and a directory of such queries becomes a suite once twelve lines of Python have looked at it.
An assertion is a SELECT that returns the rows violating a rule. Empty result set, pass. There is no library and no DSL — the query is simultaneously the check and the error message, because the rows it returns on failure are the evidence needed to fix it.
Why
These catch the class of bug a PySpark unit test structurally cannot reach: the transform was correct on the twelve rows in the factory and wrong on the warehouse, because two sources both supplied customer 4711, because silver.fx_rate had no row for a bank holiday, because a job was re-run and a MERGE matched twice.
They are also the only tests that run against production data. A unit test proves the code you shipped; an assertion proves the table you have — the second is what a stakeholder means when they ask whether the numbers are right.
How
sqltests/sql/dim_customer_unique_key.sql
-- SCD2 dimension: the business key plus the interval start must be unique.
-- Zero rows = pass.
SELECT
customer_id,
_tech_valid_from,
COUNT(*) AS versions
FROM gold.dim_customer
GROUP BY customer_id, _tech_valid_from
HAVING COUNT(*) > 1
Compare the shape most teams write first. It passes and fails just as accurately and tells you nothing:
sqlThe same check, useless on failure
SELECT CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END AS result
FROM (
SELECT customer_id FROM gold.dim_customer
GROUP BY customer_id, _tech_valid_from HAVING COUNT(*) > 1
)
One file, one rule, one statement, named <table>_<property>.sql. The file name becomes the test name, so a red build says dim_customer_scd2_no_overlap rather than test_sql_3. Nothing inside is catalog-qualified — the harness sets the catalog, which is what lets one file run against a seeded CI schema and against production.
Doing it
Write the assertion in the same pull request as the tableOne added later is calibrated against data that already exists, which is the opposite of a test.
Select the identifying columns plus the values that explain the violationKey alone is reproducible; key plus expected and actual is diagnosable.
Run the suite as a job task after the nightly load, not only in CIProduction data breaks assertions that seeded fixtures never will.
Embedding it
Hand the team a failing assertion with no explanationIf the returned rows do not explain the failure, the assertion is under-selected — a discovery lands better than a style rule.
Ask in review which of the seven business rules a new gold table can now violateThat answer is the assertion backlog, and they generated it.
Re-run last month's suite against a time-travel snapshotAn assertion going red on signed-off data teaches coverage faster than any talk.
Defaults
One rule per file; the file name is the test name
Return the offending rows, never a verdict
Never catalog-qualify inside the .sql file — the harness owns the catalog
An assertion that has never been red is unprovenBreak it deliberately once.
Gotchas
An assertion written after the table exists is fitted to the data, not to the ruleIt stays green through the bug it was written for, because that bug was already in the table when the author calibrated it.
NOT IN with a nullable column is the quietest failure herecountry NOT IN ('DE','AT','CH','NL') is NULL when country is NULL, so the row is not returned and the check passes on exactly the rows that are most broken. Every domain assertion needs an explicit IS NULL OR arm.
Assertions that only run in CICI runs against fixtures, and fixtures never contain the duplicate SAP sent at 03:00 on a Sunday.
Selecting only the key columnThe failure is reproducible but not diagnosable, so the on-call engineer re-runs it by hand with more columns — every time, at 02:00, forever.
The whole runner. It globs the directory, parametrises pytest with the file stems as test IDs, and asserts the result is empty.
Why
A harness this small is a governance decision disguised as code. Adding a check costs one file and no Python, so analysts add checks. The moment a test requires learning a framework, the only people writing tests are the people who already write tests — and the business rules stay untested because whoever knows them cannot express them.
The parametrised shape also gives the failure list for free: a broken load produces four red test IDs rather than one red suite, and the names describe the breakage before anyone opens a log.
How
pythontests/sql/test_assertions.py — the entire harness
import pathlib
import pytest
SQL_DIR = pathlib.Path(__file__).parent
CASES = sorted(SQL_DIR.glob("*.sql"))
assert CASES, f"no .sql assertions found under {SQL_DIR}"
@pytest.mark.parametrize("case", CASES, ids=[c.stem for c in CASES])
def test_assertion(spark, catalog, case):
spark.sql(f"USE CATALOG {catalog}")
bad = spark.sql(case.read_text()).take(20)
assert not bad, "{}: {} violating row(s) shown, capped at 20\n{}".format(
case.stem, len(bad), "\n".join(str(r.asDict()) for r in bad)
)
Three details are load-bearing. ids=[c.stem ...] makes the report read test_assertion[fct_order_line_orphans], so the failure names the property. take(20) bounds the blast radius. And assert CASES exists because pytest skips an empty parametrize list — without it, a packaging change that leaves the files out produces a green run with zero tests.
The spark and catalog fixtures come from the same conftest.py as the PySpark suite — catalog is the session-scoped fixture that returns os.environ["DWH_CATALOG"], resolved lazily so the local PySpark suite, which never requests it, needs no environment at all. That single variable is where the whole test estate learns which catalog it is pointed at. One difference matters here: this suite has no local mode. No local session contains gold.dim_customer, so it always runs remotely — against dwh_test after CI has loaded, or against dwh_prod on a schedule.
yamlresources/assertions.job.yml — the scheduled production run
resources:
jobs:
dwh_assertions:
name: "dwh · SQL assertions · ${bundle.target}"
tasks:
- task_key: assertions
# NO environment_key here. Serverless notebook tasks are reported to
# reject it — register key serverless-environment-key, status hot and
# explicitly UNVERIFIED. Reproduce on the client workspace before you
# either rely on it or design around it.
notebook_task:
# thin wrapper: runs pytest over tests/sql and writes the JUnit XML
# to the ci Volume. All the logic is in test_assertions.py above.
notebook_path: ../tests/sql/run_assertions.py
base_parameters:
catalog: ${var.catalog}
schedule:
quartz_cron_expression: "0 30 5 * * ?"
timezone_id: "Europe/Zurich"
email_notifications:
on_failure: ["${var.oncall_rota}"]
Doing it
Reuse the conftest.py from the PySpark suite, catalog fixture includedTwo session builders drift, and the drift shows up as a test that passes in one suite and fails in the other; two ways of naming the catalog show up as a suite that asserts against dwh_test while the job that scheduled it loaded dwh_prod.
Write the JUnit XML to the ci Volume in dwh_test so the report survives the cluster
Point the scheduled job at the on-call rota, never at an individual and never at your own address
Embedding it
Ask an analyst, not an engineer, to add the ninth assertionIf they need help, fix the harness rather than train the analyst.
Delete one .sql file in front of the team and let CI stay greenMissing tests are invisible in the abstract and obvious once watched.
When a production assertion goes red, sit next to whoever is on call and let them use the rowsDo not fix it.
Defaults
Twelve lines. If the harness grows, the growth belongs in the .sql files
File stem as test ID, so failures are self-describing
Always remote — there is no local mode for warehouse assertions
Bounded collection: take(n), never collect()
Gotchas
collect() instead of take()A missing join predicate turns an orphan check into a cross join, the driver pulls millions of rows, and the suite dies of an OOM twenty minutes in — which reads as infrastructure flakiness and gets retried rather than fixed.
An empty glob is a silent greenpytest skips a zero-length parametrize list, so a packaging change produces a passing run with no tests and nobody reads the collected-count line.
Gating CI by grepping the pytest summary for 'failed'A missing table or a parse error raises inside the test body, so pytest reports it as FAILED and the grep does catch it — but pytest reports ERROR, not FAILED, when a FIXTURE fails. Expired credentials or an unreachable workspace kill the session-scoped spark fixture at setup, every test is errored rather than failed, the word 'failed' never appears in the summary, and the gate reports green on a run that asserted nothing. Gate on the exit code, which is non-zero for both.
Running assertions before the load in the same jobThey pass against yesterday's data and the job is green while today's load is still failing. Depend on the load task explicitly.
Assuming the serverless task takes an environment_keyIt is reported to be rejected and the report is not doc-confirmed, so a copied job definition can fail at deploy time with a message about a field that is documented elsewhere. Deploy the job to a scratch target once before you promise a schedule.