Two structural decisions that determine whether ingestion can be tested at all, and whether bad data can reach a dashboard. Neither is about ingestion technology, and both are lost by default.
Stand 29.07.2026
#The transform is a function; the expectations are a dict
What
A decorated function cannot be unit-tested: importing the module registers a dataset, and calling the function needs a pipeline context. The fix is not a test framework, it is a two-file split — the logic in a module that imports nothing from the pipeline API, and a three-line wrapper that does.
Why
Without the split, the only way to test a transform is to run a pipeline: three to eight minutes per attempt, on shared infrastructure, with a cluster start in the middle. What happens next is universal — people stop testing and start looking at the output table.
Without the shared dict, each rule exists twice, once in the pipeline and once in the test, and they drift inside a sprint. The test then proves that the test's copy of the rule is satisfied, which is a fact about the test.
How
pythonsrc/dwh/pipelines/transforms/silver_sales_order.py — pure, importable, no pipeline API
from pyspark.sql import DataFrame, functions as F
EXPECTATIONS = {
"line_key_present": "order_id IS NOT NULL AND line_number IS NOT NULL",
"known_currency": "currency IN ('EUR', 'CHF')",
"rate_resolved": "fx_rate IS NOT NULL",
"amount_positive": "amount_eur > 0 OR change_type = 'D'",
"order_not_ancient": "order_ts >= TIMESTAMP'2020-01-01 00:00:00'",
}
def to_silver_sales_order(lines: DataFrame, rates: DataFrame) -> DataFrame:
"""SAP item fields in, canonical names out, valued in EUR at the order date.
lines: bronze.sap_vbap change records.
rates: silver.fx_rate — (currency, rate_date, fx_rate), one row per day.
Business rule 2: amount_eur = ROUND(quantity * unit_price * fx_rate, 2).
"""
return (
lines.selectExpr(
"VBELN AS order_id",
"POSNR AS line_number",
"MATNR AS product_id",
"KWMENG AS quantity",
"NETWR / NULLIF(KWMENG, 0) AS unit_price",
"WAERK AS currency",
"CAST(ERDAT AS TIMESTAMP) AS order_ts",
"change_type",
"change_seq",
)
.withColumn("rate_date", F.to_date("order_ts"))
.join(rates, on=["currency", "rate_date"], how="left")
.withColumn(
"amount_eur",
F.round(F.col("quantity") * F.col("unit_price") * F.col("fx_rate"), 2),
)
.drop("rate_date")
)
pythonsrc/dwh/pipelines/silver_sales_order.py — the wrapper; same module as the AUTO CDC flows above
from pyspark import pipelines as dp
from dwh.pipelines.transforms.silver_sales_order import EXPECTATIONS, to_silver_sales_order
@dp.temporary_view(name="vbap_typed")
@dp.expect_all_or_drop(EXPECTATIONS)
def vbap_typed():
return to_silver_sales_order(
spark.readStream.table("bronze.sap_vbap"),
spark.read.table("silver.fx_rate"),
)
This is the vbap_typed view the AUTO CDC flow above consumes, and the placement is a choice rather than a constraint. You can put expectations on an AUTO CDC target: dp.create_streaming_table() — the same call that declares sales_order_line above — takes expect_all, expect_all_or_drop and expect_all_or_fail as parameters, with the same behaviour and the same syntax as the decorators.
Put them on the source anyway, and for a specific reason: the target sees applied rows, the source sees change records. A row dropped at the source never becomes a version of that key at all. A row dropped at the target has already been sequenced, so what you removed is one change out of an ordered series — the delete that closed an order, the correction that fixed an amount — and the version that survives is a state the source system was never in. Dropping at the source loses a row; dropping at the target rewrites history. The one real prohibition is elsewhere: expectations are not supported with AUTO CDC FROM SNAPSHOT, which is the KNA1 flow, and is written up under Snapshot comparison.
rate_resolved earns its place: the rate join is a LEFT join, so a missing ECB rate produces fx_rate = NULL, then amount_eur = NULL, then a revenue total that is quietly short by one currency for one day. Nothing fails. amount_positive allows deletes through because a delete record carries no amount; rule 5 — positive amounts on non-cancelled lines — is completed in gold, where status exists.
The test imports the same dict. A rule is written once and asserted in two places, which is the only version of the rule is enforced that survives a refactor.
Three actions, three consequences
Decorator
Violating row
Update
Use for
expect_all
kept
succeeds
anything you want measured before you dare enforce it
expect_all_or_drop
dropped, counted in the event log
succeeds
rows unusable downstream and safe to lose
expect_all_or_fail
written nowhere
fails immediately
invariants where publishing nothing beats publishing something wrong
Expectations require the ADVANCED pipeline edition. CORE is streaming ingest only, PRO adds CDC, ADVANCED adds expectations — and ADVANCED is the default, so this only bites the team that changed the edition to save money.
Before any of that can be read, one setting. By default only the pipeline's run-as user can query the event log, and the event_log() table-valued function can be called only by the pipeline owner — a view built over it cannot be shared, so granting on the view does not help either. Publish the log to Unity Catalog in the pipeline definition and it becomes an ordinary table you can GRANT on, join to, and alert from.
yamlresources/pipelines/dwh_ingest.yml — the event log, published and grantable
sqlWhat actually happened — the event log, not the run status. Catalog is a parameter, never a literal.
GRANT SELECT ON TABLE
IDENTIFIER(:catalog || '.' || :ops_schema || '.dwh_ingest_event_log')
TO `data-platform-rota`;
SELECT
e.timestamp,
x.name AS expectation,
x.dataset,
x.passed_records,
x.failed_records
FROM IDENTIFIER(:catalog || '.' || :ops_schema || '.dwh_ingest_event_log') AS e
LATERAL VIEW explode(
from_json(
e.details:flow_progress.data_quality.expectations,
'array<struct<name: string, dataset: string,
passed_records: bigint, failed_records: bigint>>'
)
) AS x
WHERE e.event_type = 'flow_progress'
AND e.timestamp > current_timestamp() - INTERVAL 1 DAY
AND x.dataset = 'sales_order_line' -- the published log covers the whole pipeline
ORDER BY x.failed_records DESC;
The published log is per pipeline, not per table, which is why the query filters on dataset. The unpublished form — event_log(TABLE(...)) — is still the right tool at a laptop while you are developing; it is only unfit as the source of an alert somebody else has to answer.
Doing it
Put every transform in transforms/ with no pipeline importMocking does not make a decorated module testable.
Keep EXPECTATIONS beside the function producing the rows it constrains, and import it in both places
Start every new rule at warn; promote to drop or fail after watching failed-records for a week
Read the event log per run rather than the run statusA drop expectation is a success at job level.
Publish the event log to Unity Catalog in the pipeline definition (event_log: catalog / schema / name)Grant SELECT to the on-call group, before anyone builds a dashboard on it.
Embedding it
Have the team convert one pipeline file to the two-file split and time both feedback loops themselvesThe number does the persuading.
In review, ask where the rule is writtenIf the answer is two places, the review just found a bug that has not happened yet.
Run the over-strict exercise: set a rule that drops 5 % of valid rowsSee how long anyone takes to notice from job status alone.
Defaults
transforms/ imports nothing from the pipeline API. Ever
EXPECTATIONS is a plain dict, imported by the pipeline and by the test
New rules land as warn; promotion is a decision with evidence behind it
Alerts read the event log — published to Unity Catalog and granted to the rotaThe unpublished log is readable only by the pipeline's owner. The run status is not a quality signal.
Expectations sit on the flow's source view, not on the AUTO CDC targetA choice about change semantics, not a platform limit.
Gotchas
expect_all_or_drop on a rule that is slightly too strictRows disappear, the pipeline is green, the number is quietly low — the most expensive expectation failure, because the system reports success while losing data.
Testing the pipeline module instead of the transformImporting it registers datasets and the test needs a pipeline context; the usual response is to mock the decorator, which tests the mock.
Rules copied into the test fileThey drift, always in the same direction: the test relaxes, because relaxing is what makes it pass.
Reading only the run statusA drop expectation that removed 400,000 rows and a clean run are indistinguishable from the jobs UI.
Building the expectation alert on event_log() and testing it as yourselfThe function can be called only by the pipeline owner and the view over it cannot be shared, so the dashboard works for the person who built it and is empty for the rota it is supposed to wake — an alert that never fires reads exactly like a system with no failures.
Attaching an expectation to the AUTO CDC target because it felt tidierIt is allowed, so nothing warns you; what it drops is a sequenced change rather than an input row, and the surviving version of that order is a state SAP never held.
The tempting design is one pipeline: bronze, silver, a few validation tables, gold. It reads as one unit of work and it gates nothing. Inside a single pipeline, whether gold refreshes depends on the dependency graph, not the quality outcome — and a validation table that nothing reads is a table that nothing waits for.
Forcing the edge is worse. Make gold read the validation table so it 'depends' on it, and lineage now claims a fact table is derived from a row count, while every full refresh recomputes the check as part of publishing the thing it was meant to guard.
Why
The gate exists to answer one question at 03:00: is it better to publish yesterday's gold, or today's suspect gold? In a single pipeline that question is never asked — the answer was baked in when the DAG was drawn, and it is always publish.
Splitting also separates two failure classes needing different people. An ingest failure is a source problem; a publish failure is a modelling problem. One pipeline gives both the same alert, and whoever is on call must diagnose which kind it is before acting.
How
textTwo pipelines, one gate between them
dwh_ingest bronze.* -> silver.* expectations live here
|
gate one SQL task; raise_error on breach, counts in the message
|
dwh_publish silver.* -> gold.* does not start if the gate failed
sqltests/sql/gate_silver_to_gold.sql — rule 7 with a tolerance, because the unknown member is legal
-- owner: data-platform rota
-- tolerance: 0.5 %, agreed with sales ops 06.2026
WITH recent AS (
SELECT o.order_id,
c.customer_id IS NULL AS unresolved
FROM IDENTIFIER(:catalog || '.silver.sales_order') o
LEFT JOIN IDENTIFIER(:catalog || '.silver.customer') c
ON c.customer_id = o.customer_id
AND c._tech_is_current
WHERE o.order_date >= current_date() - INTERVAL 1 DAY
)
SELECT CASE
WHEN COUNT(*) = 0 THEN 'ok - no orders in window'
WHEN SUM(IF(unresolved, 1, 0)) / COUNT(*) <= 0.005 THEN 'ok'
ELSE raise_error(CONCAT(
'gate silver->gold: ',
CAST(SUM(IF(unresolved, 1, 0)) AS STRING), ' of ',
CAST(COUNT(*) AS STRING),
' orders resolve to no current customer'))
END AS gate
FROM recent;
yamlresources/jobs/dwh_daily.yml — the gate is a task, never a table
The gate SQL above reads c._tech_is_current, which makes it path B in the snapshot topic's sense. On path A the same predicate is c.__END_AT IS NULL. Gates are the place the two interval contracts collide first, because a gate that fails to compile gets disabled rather than fixed.
Doing it
Two pipelines, one job: dwh_ingest ends at silver, dwh_publish starts at silver
Write gates as SQL files in the repo, one assertion per fileLet a breach call raise_error so the task fails with both counts in the message.
Parameterise the catalogA gate file with dwh_prod baked in passes the test deployment by reading production.
Give every gate a tolerance and a named owner in a comment at the topA gate with neither gets disabled the third time it fires.
Decide per gate whether failing means stop or publish-and-alertPut the decision in the runbook.
Embedding it
Ask the team what happens to gold when a silver expectation drops 10 % of rowsIf the answer is 'the pipeline is green', they have found the gap themselves.
Have them write the first gate including the toleranceTake that tolerance to the business for a number rather than inventing one.
Break the gate deliberately during the game dayCheck whether anyone can tell, from the alert alone, which pipeline to open.
Defaults
Two pipelines with a job between them; the gate is a task, never a table
A breach raises; the message carries both counts, so the alert already says how bad it is
Tolerances are business numbers, agreed and written down
One alert per chain, routed to a rota — see the monitoring route
Gotchas
A validation table inside the publishing pipelineIt computes, it goes green, and gold publishes regardless, because nothing downstream reads it. The team believes there is a gate for months; there has never been one.
expect_all_or_fail used as the gateOne malformed row from one source stops the entire update including the eleven tables that were fine — and at 03:00 nobody consciously chose to trade a wrong number for no number.
Forcing a dependency edge so gold 'waits' for validationLineage then shows fct_order_line derived from a quality metric, and the next person to read the DAG is misled about where the numbers come from.
A gate with no toleranceIt fires on three rows, someone re-runs the job with the check commented out, and the commented line is still there in December.
A second bundle file defining a job that is also called dwh_dailyThe two definitions do not merge and they do not clash loudly — one wins, and the tasks the other one contributed are simply not scheduled any more. The BI route's version of dwh_daily is the same job, not a second one.