Lab 06 — Data Quality & Tests

The goal — a trust layer

So far mini-GridDP moves data: Postgres → bronze → silver → gold, on a schedule. But nothing yet checks that the data is correct. A duplicated rental, a customer that points at a tier that doesn't exist, a feed that silently stopped loading last Tuesday — all of these pass through happily and poison the dashboard. The fix is a layer of automated assertions that run with the pipeline and fail loudly when reality violates an expectation.

Tests sit after transformation and before serving — a gate the data must pass to reach the dashboard:

By the end of this lab the gate has three kinds of checks:

Schema tests on the gold models

dbt ships four generic tests you'll reach for constantly: not_null, unique, relationships (a foreign-key check), and accepted_values. You declare them in a YAML file beside your models — no SQL required. Lab 03 left a small _marts.yml with four tests; grow it into the full contract for the star schema. (A model may be described in only one YAML file, so extend this one rather than adding a schema.yml next to it.)

version: 2

models:
  - name: fct_rentals
    description: "One row per GPU rental, priced as-of the start time. The central fact of the star schema."
    columns:
      - name: rental_id
        description: "Surrogate key — unique per rental."
        tests:
          - not_null
          - unique
      - name: customer_id
        description: "FK to dim_customer."
        tests:
          - not_null
          - relationships:
              arguments:
                to: ref('dim_customer')
                field: customer_id
      - name: gpu_id
        description: "FK to dim_gpu."
        tests:
          - not_null
          - relationships:
              arguments:
                to: ref('dim_gpu')
                field: gpu_id
      - name: hours
        tests:
          - not_null
      - name: revenue
        tests:
          - not_null

  - name: dim_customer
    description: "Customer dimension (Type 1 — one row per customer, current values)."
    columns:
      - name: customer_id
        tests:
          - not_null
          - unique
      - name: tier
        description: "Pricing tier the customer is on."
        tests:
          - accepted_values:
              arguments:
                values: ['free', 'pro', 'enterprise']

Run just the tests (no rebuild) and read the summary:

uv run dbt test --select fct_rentals dim_customer

Freshness — is the data even current?

Schema tests check that the data is shaped right. They say nothing about whether it's recent. If the Postgres extract silently failed three days ago, every schema test still passes — the old rows are perfectly valid, just stale. Freshness is the check that catches a feed that quietly died.

dbt has freshness built in for sources. Every bronze row carries the loaded_at stamp that the Lab 02 extractor and the Lab 04 consumer write. In your sources file, point dbt at that column and set thresholds:

version: 2

sources:
  - name: bronze
    schema: bronze
    description: Raw tables landed by the Lab 02 ingest and the Lab 04 stream consumer.
    loaded_at_field: loaded_at        # the ingest timestamp we stamp on each row
    freshness:
      warn_after:  {count: 6,  period: hour}   # noisy but ok
      error_after: {count: 24, period: hour}   # a full day stale = broken
    tables:
      - name: customers
      - name: gpus
      - name: gpu_prices
      - name: rentals
      - name: telemetry
        freshness:                    # telemetry is a firehose — tighter bound
          warn_after:  {count: 30, period: minute}
          error_after: {count: 2,  period: hour}

Run the freshness check on its own:

uv run dbt source freshness

A custom assertion — enforce a business rule

The generic tests don't know your business. Revenue is never negative. A rental lasts more than zero hours. Every rental in bronze made it into the fact table. Those are singular tests — a single SQL file in tests/ that selects the rows that violate the rule. Remember the golden rule: a test passes when its query returns zero rows.

Write one that asserts rental hours and revenue are always positive:

-- Fail if any rental has non-positive hours or negative revenue.
-- A passing test returns ZERO rows.
select
    rental_id,
    hours,
    revenue
from {{ ref('fct_rentals') }}
where hours <= 0
   or revenue < 0

And a reconciliation test — every rental that landed in bronze should appear in the gold fact, so the counts must match:

-- Fail if the gold fact row count drifts from the bronze source.
-- Returns a single row only when the counts disagree.
with bronze as (
    select count(*) as n from {{ source('bronze', 'rentals') }}
),
gold as (
    select count(*) as n from {{ ref('fct_rentals') }}
)
select bronze.n as bronze_rows, gold.n as gold_rows
from bronze, gold
where bronze.n != gold.n

Run your custom tests:

uv run dbt test --select test_type:singular

Break it, then fix it

A test you've never seen fail is a test you don't trust. Let's prove the trust layer works by injecting a bad row — a rental with negative hours — straight into bronze, rebuilding, and running the suite.

# open a DuckDB shell on the warehouse (you are in dbt/, so one level up)
duckdb ../warehouse.duckdb

-- inside duckdb: a rental that lasted -5 hours. Nonsense, but realistic
-- (a clock skew or a bad upstream record could produce exactly this).
-- rental_id 0 can never collide with a SERIAL id and leaves the Lab 02 watermark alone.
INSERT INTO bronze.rentals (rental_id, customer_id, gpu_id, hours, started_at)
VALUES (0, 1, 7, -5, now());
.quit

Now rebuild gold so the bad row flows through. dbt build runs the fact's tests right after the model, so the alarm sounds here already; dbt test then runs the whole suite:

uv run dbt build --select fct_rentals
uv run dbt test

3 of 11 FAIL 1 assert_rentals_positive .................. [FAIL 1 in 0.05s]
[ERROR]: in test assert_rentals_positive (tests/assert_rentals_positive.sql)
  Got 1 result, configured to fail if != 0
  compiled code at target/compiled/mini_griddp/tests/assert_rentals_positive.sql
Done. PASS=10 WARN=0 ERROR=1 SKIP=0 NO-OP=0 REUSED=0 TOTAL=11

Now fix it — remove the bad row, rebuild, and re-test:

duckdb ../warehouse.duckdb "DELETE FROM bronze.rentals WHERE rental_id = 0;"
uv run dbt build --select fct_rentals
uv run dbt test

Wire tests into the Dagster pipeline

Tests that only run when you remember to type dbt test aren't a trust layer — they're a good intention. The point is that quality runs automatically with the pipeline, so bad data is caught the moment it's produced, on every scheduled tick. In Dagster, the clean way to express "this data must pass these checks" is an asset check.

The schema and singular tests are already wired in. Look back at the Lab 05 dbt_build asset: it runs dbt build, and build is run + test. A failing test makes dbt exit non-zero, _run raises, the asset goes red and fct_telemetry_hourly never starts. You proved that by hand a moment ago; the scheduler gets the same gate for free. Freshness is the one check that does not run inside dbt build, so add it as a Dagster asset check on the bronze asset it describes:

from dagster import asset_check, AssetCheckResult   # add to the existing import


# A freshness gate, expressed as a Dagster asset check on bronze_business.
# It fails the check if `dbt source freshness` errors on bronze.rentals,
# and blocking=True stops the run there instead of building gold on stale data.
@asset_check(asset=bronze_business, blocking=True)
def rentals_are_fresh(context) -> AssetCheckResult:
    result = subprocess.run(
        ["dbt", "source", "freshness", "--profiles-dir", ".",
         "--select", "source:bronze.rentals"],
        cwd=DBT_DIR, capture_output=True, text=True,
    )
    context.log.info(result.stdout.strip())
    return AssetCheckResult(
        passed=result.returncode == 0,          # dbt exits 1 on ERROR STALE
        metadata={"detail": MetadataValue.text("dbt source freshness on bronze.rentals")},
    )

Reload the Dagster code location and materialize the pipeline (from the repo root):

cd ..
uv run dagster dev          # then "Materialize all" in the UI, or headless:
uv run dagster job execute -j full_pipeline -m dagster_app.definitions

Commit Lab 06

You've added a whole layer to the platform. Commit it so the trust layer is part of the story your git history tells.

git add dbt/models/marts/_marts.yml \
        dbt/models/staging/_sources.yml \
        dbt/tests/ \
        dagster_app/assets.py
git status                              # confirm only the trust-layer files are staged
git commit -m "Lab 06: data quality — schema tests, freshness, custom assertions, Dagster checks"

✓ Check yourself

Answer from memory before opening each one.

1. A join in fct_rentals fans out and every rental appears twice, doubling revenue on the dashboard. Every column is still populated and every foreign key still resolves. Which test goes red, and what does its query return?

   Show answer

   unique on rental_id. The test compiles to a SELECT for key values that occur more than once, and any rows returned are the failures. not_null and relationships stay green, because each duplicated row is perfectly well-formed. See Schema tests and the zero-rows rule.

2. The Postgres extract died three days ago. Since then every schema test and both singular tests have passed on every run. Why — and which check would have caught it on day one?

   Show answer

   Those tests judge the shape of the rows that are there, and three-day-old rows are still unique, non-null and positive. Staleness is about rows that never arrived: dbt source freshness compares max(loaded_at) with now and errors once the gap passes the threshold. See Freshness.

3. Marketing launches a startup tier and new customers begin arriving with it. What happens at the next dbt test, and what is the right fix?

   Show answer

   accepted_values on dim_customer.tier fails and returns the value it does not recognise. The data is not wrong — the contract is out of date. Confirm the tier is real, then add it to the values list in _marts.yml; filtering the rows out or disabling the test throws away the check that will catch the next value that is a typo. See Schema tests on the gold models.

4. An at-least-once ingest lands one rental twice in bronze, and staging deduplicates it as designed. What does assert_rental_count_reconciles report, and is it right to?

   Show answer

   It fails: bronze has one more row than gold, so the query returns a row. But gold is correct — the test compares count(*) on both sides, so it reads legitimate deduplication (or any row staging deliberately filters out) as loss. Reconcile on what ought to match, such as count(distinct rental_id) under the same filters; a test that fails on correct data teaches people to ignore it. See A custom assertion.

5. The Lab 05 dbt_build asset calls dbt build rather than dbt run. What would the pipeline do with a negative-hours rental under run, and what changes under build?

   Show answer

   run only builds models — no test executes, so the bad row flows into fct_rentals and on toward the dashboard while the run stays green. build is run + test: the tests execute right after their models, dbt exits non-zero on the failure, _run raises, the asset turns red and nothing downstream starts. Tests that run only when somebody remembers to type the command are not a gate. See Wire tests into the Dagster pipeline.

6. The pipeline loads rentals once a day at 06:00 and checks source freshness at the top of each run, with error_after set to 24 hours. What happens on a morning when the run starts ten minutes late, and how should the threshold be chosen?

   Show answer

   At the top of the run the newest loaded_at is still yesterday's load — about 24 h 10 min old — so freshness errors before today's ingest has had a chance to run. A threshold equal to the load cadence has no slack. Choose it from the expected gap between loads plus headroom for ordinary delay: too tight and it cries wolf until people ignore it, too loose and stale data is served for hours unnoticed. Telemetry gets the tighter bound because it is expected continuously. See Freshness.

-- Fail if any hourly utilization falls outside 0..100.
-- Returns the offending rows; zero rows = pass.
select
    gpu_id,
    hour,
    avg_util_pct
from {{ ref('fct_telemetry_hourly') }}
where avg_util_pct < 0
   or avg_util_pct > 100

  - name: fct_telemetry_hourly
    columns:
      - name: avg_util_pct
        tests:
          - dbt_utils.accepted_range:
              arguments:
                min_value: 0
                max_value: 100
                inclusive: true

uv run dbt test --select fct_telemetry_hourly
# -> PASS assert_util_pct_in_range

Takeaway

• A pipeline that runs is not one you can trust; add schema tests for shape, freshness checks for recency and singular tests for business rules.
• unique plus not_null on every primary key catches the commonest warehouse bug, a fan-out join that doubles rows and inflates revenue, and relationships catches orphaned facts.
• Schema tests pass on stale data, so freshness compares the newest loaded_at with now to catch a feed that silently stopped.
• A singular test is a SQL file that selects the bad rows, such as negative revenue, and a reconciliation test checks that every bronze rental reached the gold fact.
• A test you have never seen fail is not trusted, so inject a negative-hours row, watch the suite go red, then fix it and watch it go green.
• A dbt test failure blocks downstream because the whole dbt build fails, while a Dagster asset check only reports unless it is marked blocking=True.

With a trusted star schema you can define metrics once and serve them everywhere, which is the next lab. → Lab 07 — The Semantic Layer & Metrics

Further reading

• dbt: Data tests — dbt Labs docs, accessed 2026. Generic and singular tests and how failing rows are reported.
• dbt: Freshness — dbt Labs docs, accessed 2026. Warn and error thresholds on a source's load time.
• Dagster: Testing assets with asset checks — Dagster Labs docs, accessed 2026. Asset checks and the blocking option.
• dbt: build command — dbt Labs docs, accessed 2026. Why a failed test stops downstream models.