Unit tests prove the rules are right. They cannot prove the deployed pipeline is running them. That takes a real update against known data, and one query afterwards.
One job, three tasks strictly ordered — write known data, run the pipeline over it, assert what happened — plus a teardown carrying run_if: ALL_DONE so the run schema is dropped whether the assertion passed or not. The middle task is a pipeline_task with full_refresh: true, and the reason for that flag is in the gotchas: it makes the update's semantics declared rather than an accident of how the seed happens to write.
Why
Unit tests cannot see deployment. They cannot catch a pipeline deployed on the wrong product edition, a libraries glob that stopped matching after a file moved, a configuration key renamed on one side only, or an expectation that is present in the code and absent from the running graph because the module was never loaded. Those are all real, all cheap to cause, and all invisible until an update runs.
Product editions are CORE, PRO and ADVANCED, and expectations require ADVANCED. Selecting a cheaper edition on a pipeline that declares expectations fails with an explanatory error rather than silently skipping the checks — the good outcome, but only if someone reads it. Under deadline pressure the temptation is to delete the expectations rather than change one line of YAML.
Where that error appears is the part that decides this route's architecture.bundle deploy creates or updates the pipeline definition; it does not compile the graph. The edition mismatch is caught during graph analysis, which happens when the first update runs — so the deploy returns green and the failure arrives minutes later, in a run. That is the argument for the pipeline task in CI rather than against it: if the platform checked at deploy time, a deploy step would be enough and the seed-run-assert job would be redundant; because it checks at update time, an update in the build is the only thing that finds it before production does.
One caveat on the flag itself: edition is a classic-compute setting and has no documented meaning for a serverless pipeline, which this handbook recommends elsewhere. On serverless the line is inert rather than protective, and the CI update is then the only evidence that expectations were evaluated at all.
How
yamlresources/ci.yml — the pipeline gets its own schema, the job gets three tasks and a teardown
resources:
pipelines:
silver:
name: dwh-silver-${bundle.target}
catalog: ${var.catalog}
schema: ${var.schema} # PUBLISHED into. CI: run_<job_run_id>. prod: silver
edition: ADVANCED # expectations require it — CLASSIC compute only;
# edition has no documented meaning for serverless
continuous: false # triggered: the update has to END
configuration:
# READ from. A separate variable, not ${var.schema}: in prod the
# pipeline publishes to silver and reads bronze. They coincide only in
# CI, where the seed writes into the same run schema.
dwh.source_schema: ${var.catalog}.${var.source_schema}
libraries:
- glob:
include: ../src/dwh/pipelines/**
jobs:
ci_assert:
name: dwh-ci-assert-${bundle.target}
max_concurrent_runs: 1
tasks:
- task_key: seed
notebook_task:
notebook_path: ../src/dwh/ci/seed.py
source: WORKSPACE
base_parameters:
catalog: ${var.catalog}
schema: ${var.source_schema} # the seed writes the SOURCE tables
- task_key: pipeline
depends_on:
- task_key: seed
pipeline_task:
pipeline_id: ${resources.pipelines.silver.id}
full_refresh: true
- task_key: assert
depends_on:
- task_key: pipeline
notebook_task:
notebook_path: ../src/dwh/ci/assert_expectations.py
source: WORKSPACE
base_parameters:
pipeline_id: ${resources.pipelines.silver.id}
- task_key: teardown
depends_on:
- task_key: assert
run_if: ALL_DONE # the point: also runs when assert failed
notebook_task:
notebook_path: ../src/dwh/ci/teardown.py
source: WORKSPACE
base_parameters:
catalog: ${var.catalog}
schema: ${var.schema}
The seed is seven rows: six that are legitimate and exactly one that violates exactly one rule. That discipline is what lets the assertion be an equality rather than a greater-than — and it is harder than it looks. The bad row is a USD line, so fx_rate has to carry a USD rate for that day; leave it out and the row violates known_currency and fx_rate_resolved together, and one seeded defect has become two expected counts.
pythonsrc/dwh/ci/seed.py — six good lines and one USD
catalog = dbutils.widgets.get("catalog")
schema = dbutils.widgets.get("schema")
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")
lines = [
# order_id, line_number, currency, status, quantity, unit_price, order_ts
("4711", 1, "EUR", "SHIPPED", 2.0, 125.00, "2026-07-01 09:00:00"),
("4711", 2, "EUR", "OPEN", 1.0, 49.50, "2026-07-01 09:00:00"),
("4712", 1, "CHF", "SHIPPED", 10.0, 12.00, "2026-07-01 11:15:00"),
("4713", 1, "EUR", "CANCELLED", 3.0, 80.00, "2026-07-02 08:00:00"),
("4714", 1, "EUR", "SHIPPED", 1.0, 999.00, "2026-07-02 16:40:00"),
("4715", 1, "EUR", "SHIPPED", 4.0, 25.00, "2020-01-01 00:00:00"), # boundary: must PASS
("4716", 1, "USD", "SHIPPED", 1.0, 10.00, "2026-07-03 10:00:00"), # the one bad row
]
# No country column: an order line has none. Seeding one so that a country
# rule can pass is how the harness ends up proving the wrong thing.
#
# Declare the temporal and decimal columns as STRING/DOUBLE and CAST.
# createDataFrame refuses a str for a TimestampType or DateType field and a
# float for a DecimalType field — it raises TypeError before the seed writes
# anything, and the message reads like a Spark bug rather than a schema typo.
cols = "order_id string, line_number int, currency string, \
status string, quantity double, unit_price double, order_ts string"
spark.createDataFrame(lines, cols) \
.selectExpr("order_id", "line_number", "currency", "status",
"CAST(quantity AS DECIMAL(12,3)) AS quantity",
"CAST(unit_price AS DECIMAL(18,2)) AS unit_price",
"CAST(order_ts AS TIMESTAMP) AS order_ts") \
.write.mode("overwrite").saveAsTable(f"{catalog}.{schema}.sap_vbap")
# EUR is absent on purpose — the transform defaults the base currency by name,
# so an EUR row here would mask a regression in that branch.
# USD IS present, and that is the subtle part: without a rate the bad row would
# violate fx_rate_resolved as well as known_currency, and EXPECTED_FAILURES
# could no longer be an equality per rule.
rates = [("CHF", "2026-07-01", 0.940000),
("USD", "2026-07-03", 0.860000)]
spark.createDataFrame(rates, "currency string, rate_date string, fx_rate double") \
.selectExpr("currency",
"CAST(rate_date AS DATE) AS rate_date",
"CAST(fx_rate AS DECIMAL(18,6)) AS fx_rate") \
.write.mode("overwrite").saveAsTable(f"{catalog}.{schema}.fx_rate")
Doing it
Deploy the CI bundle with run-specific schema variables, so each run owns its pipeline, seed and outputdatabricks bundle deploy -t test --var schema=run_$GITHUB_RUN_ID --var source_schema=run_$GITHUB_RUN_ID. The two are the same string only in CI; keep them two variables so prod cannot inherit the collapse.
Seed exactly one violation of exactly one ruleTwo violations and every assertion becomes 'at least', which passes for the wrong reason.
Include the boundary rowsThe 2020-01-01 order, the CANCELLED line at zero value, one row per currency the business trades in.
Add a teardown task with run_if: ALL_DONE so it fires on failure too, dropping the run schemaA leaked schema per CI run is visible within a week.
Embedding it
Have the team add the seventh row — the bad one — themselves, then watch the assert task go redOwning the failure is what makes the harness theirs.
Ask them to break the harness deliberately: point the pipeline at production bronzeThey see that everything still passes. That is the lesson about parameterised sources, delivered by the harness rather than by you.
Give the teardown step to whoever complains first about test-schema clutterIt is fifteen minutes of work and it converts an irritation into ownership.
Let the CI job stay red for one afternoon rather than fixing it yourselfWho fixes it tells you who has adopted it.
Defaults
Triggered pipeline, full_refresh: true, three tasks, strict depends_on
One schema per CI run; teardown runs even when the tests failed
edition: ADVANCED declared explicitly rather than relied on as the defaultA copied pipeline then cannot arrive on PRO unnoticed — and it is understood as a classic-compute setting, inert on serverless.
The seed is code in the repository, not a fixture table someone loaded once by hand
Gotchas
Omitting full_refresh and trusting whichever mechanism you were told aboutIn THIS harness the omission is survivable and for a reason worth knowing: the seed writes with .mode("overwrite") on every run, so the source gets a new Delta version containing every row, and a materialized view over an overwritten source cannot refresh incrementally — it recomputes, and the numbers come out right by accident. Change the seed to an append, or point the job at a streaming table that remembers its checkpoint, and the same run against the same data processes zero rows: every rule reports zero passed and zero failed, and the assertion passes vacuously. State full_refresh: true so the semantics are declared rather than inherited from how the seed happens to write, and guard it anyway: assert passed + failed > 0 per rule.
Two CI runs sharing one pipelineThe second update overwrites the first's tables, the assert task reads the other branch's numbers, and the developer debugs their own correct code for an afternoon.
A seed row that violates two rulesThe assertion for the second rule now expects 1 for reasons nobody wrote down, and when someone edits the seed a year later two tests fail with no explanation of what the row was for. The USD line is one edit away from this: drop the USD rate and it fails known_currency and fx_rate_resolved at once.
A seed that invents a column so a rule can passAdd country to the seeded sap_vbap and a country expectation goes green in CI — while the real bronze table has no such column and the first production update fails at analysis. The seed is a fixture for the model, not a place to negotiate with it: every column it writes must exist in the model, or the harness proves the wrong thing confidently.
Teardown as a plain last taskWithout run_if: ALL_DONE it only runs on success, so the schemas that leak are precisely the ones from failing runs, which are the ones you wanted to inspect and now cannot find among the hundred others.
Pointing dwh.source_schema at the schema the pipeline publishes intoIn CI the seed writes into the run schema and the pipeline reads it back, so it works and the build is green. In production the pipeline publishes to silver and would read silver.sap_vbap, which does not exist — a TABLE_OR_VIEW_NOT_FOUND on the first prod deploy, weeks after CI proved nothing about it. Read schema and publish schema are two variables.
Expectation results are events, not table rows. Every update writes flow_progress events whose details column carries a JSON blob; the data quality metrics live under data_quality.expectations as an array of {name, dataset, passed_records, failed_records}. Unpacking it needs an explicit from_json schema — there is nothing to infer from.
Why
Without this query, 'the pipeline is green' means the update did not crash. It says nothing about whether the rules ran, how many rows each removed, or whether a rule exists at all — the three facts that separate a pipeline that works from a pipeline you can hand over.
It also gives the drop a direction. failed_records trending upward on known_currency means the source started sending a currency nobody added to the domain: a business event — a new market, a new sales channel — arriving through a data quality metric, days before anyone files a ticket about missing revenue.
How
sqlsrc/dwh/ci/metrics_sql.py — METRICS_SQL: one update's expectation metrics, per rule
WITH latest AS (
-- "The latest update" is safe here on ONE invariant, and it is not visible
-- in this SQL: the CI bundle deploys a pipeline PER RUN and the job sets
-- max_concurrent_runs: 1, so this pipeline has exactly one update, ever.
-- Consolidate CI onto a shared pipeline and this reads another run's numbers.
SELECT origin.update_id AS update_id
FROM event_log('<pipeline_id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
unpacked AS (
SELECT explode(
from_json(
details:flow_progress.data_quality.expectations,
'array<struct<name: string, dataset: string,
passed_records: bigint, failed_records: bigint>>'
)
) AS e
FROM event_log('<pipeline_id>')
WHERE event_type = 'flow_progress'
AND origin.update_id = (SELECT update_id FROM latest)
)
SELECT e.dataset,
e.name AS rule,
SUM(e.passed_records) AS passed,
SUM(e.failed_records) AS failed
FROM unpacked
GROUP BY e.dataset, e.name
ORDER BY e.dataset, e.name
The assertion is then arithmetic. known_currency must have dropped exactly the one seeded row; every other rule must have dropped nothing; and every rule declared in EXPECTATIONS must appear, which is what catches a rule commented out during debugging and never restored.
pythonsrc/dwh/ci/assert_expectations.py
from dwh.ci.metrics_sql import METRICS_SQL
from dwh.pipelines.transforms.silver_sales_order import EXPECTATIONS
pipeline_id = dbutils.widgets.get("pipeline_id")
DATASET = "sales_order_line"
EXPECTED_FAILURES = {"known_currency": 1} # the single seeded USD row
rows = spark.sql(METRICS_SQL.replace("<pipeline_id>", pipeline_id)).collect()
# Key by (dataset, name), then narrow. Expectation names are unique per dataset,
# not per pipeline: a dict keyed by name alone loses a rule silently the day a
# second dataset declares its own known_currency.
by_key = {(r["dataset"], r["rule"]): r for r in rows}
by_rule = {rule: row for (dataset, rule), row in by_key.items() if dataset == DATASET}
assert by_rule, f"no expectation metrics for {DATASET} — did the pipeline actually run?"
assert set(by_rule) == set(EXPECTATIONS), (
f"declared {sorted(EXPECTATIONS)} but the update evaluated {sorted(by_rule)}"
)
for rule, row in sorted(by_rule.items()):
evaluated = row["passed"] + row["failed"]
assert evaluated > 0, f"{rule} saw zero rows — the update processed nothing"
expected = EXPECTED_FAILURES.get(rule, 0)
assert row["failed"] == expected, (
f"{rule}: expected {expected} failed record(s), got {row['failed']} "
f"of {evaluated} evaluated"
)
Doing it
Assert equality on failed_records, never a thresholdA threshold passes when the number is wrong in the direction you did not think about.
Assert that the set of rules in the log equals the set of keys in EXPECTATIONSThis is the only check that catches a deleted rule.
Key the metrics by (dataset, name), not by name aloneExpectation names are unique per dataset, and the second table with a known_currency rule silently overwrites the first in a dict keyed by name.
Write down next to the query what makes 'the latest update' correct, or pin the update idHere it is the per-run pipeline plus max_concurrent_runs: 1. An invariant nobody can see in the code is one refactor from gone.
In production, chart failed_records per rule per day and alert on the change, not on the levelThe level is a policy decision; the change is an event.
Publish the production event log to Unity Catalog and grant SELECT to the on-call groupThe default log is queryable by the pipeline owner only, which during an incident is the person on holiday.
Embedding it
Have the team write the from_json schema themselves against a real updateTen minutes with the raw details column teaches more about the pipeline runtime than any diagram.
Ask them to predict the six numbers this query will return before running itWrong predictions are exactly where their mental model of the pipeline is wrong.
Hand the production event-log dashboard to the team in week one and stop citing it yourselfIf they are reading it without you by the mid-point, the observability handover is done.
When an expectation fires in production, let the team explain the number to the businessThey run the query themselves. That is the moment data quality stops being an engineering concern.
Defaults
Exact-equality assertions, one seeded violation, one expected count
Set equality between declared rules and evaluated rules, every run
A passed + failed > 0 guard on every rule, so a no-op update cannot pass
The update the query reads is either pinned by id or made unambiguous by per-run isolationWhich one it is, is written down.
Production event log published to Unity Catalog, granted to a group, never dropped
Metrics keyed by dataset and name together, from the first pipeline onward
Gotchas
Taking 'the latest update' on a pipeline more than one run can touchTwo updates in flight and the assertion reads the other one's metrics — a flaky test whose flakiness is caused by the test, not the code, which is the hardest kind to diagnose. The query above is exempt only because the harness gives each run its own pipeline and caps the job at one concurrent run; the day someone consolidates CI onto a shared pipeline, both halves of that invariant disappear and nothing in the SQL objects.
Inferring the JSON schema instead of declaring itfrom_json with a wrong or partial schema returns NULLs rather than an error, SUM over NULLs returns NULL, and NULL == 0 is NULL — so the assertion neither passes nor fails in the way you expect and the build goes green on nothing.
Dropping the event log to reclaim storageIt is small and it is the only history you have; the day you need it is the day the auditor asks how long the number has been wrong.
Asserting failed_records >= 1It passes when the rule dropped four hundred rows instead of one, which is precisely the over-strict failure the whole route exists to catch.