Course 6 · Capstone

Lab 06 — Data Quality & Tests

A pipeline that runs is not the same as a pipeline you can trust. In this lab you add the trust layer to mini-GridDP: dbt schema tests on the gold star schema, a freshness check that catches stale data, and a custom assertion that enforces a business rule. Then you'll deliberately break a row, watch a test fail with a clear message, fix it, and watch it go green — and finally wire dbt test into the Dagster pipeline so quality runs every time the data does.

Setup — do this first

You need two things from earlier labs running: the dbt project from Lab 03 (the dbt/ folder with dim_customer, dim_gpu, fct_rentals, and fct_telemetry_hourly built into DuckDB), and the Dagster pipeline from Lab 05. From your repo root, confirm the warehouse is populated:

terminal

cd mini-griddp/dbt
dbt build          # ingest + model so gold tables exist and are fresh
dbt ls             # you should see dim_customer, fct_rentals, fct_telemetry_hourly

If dbt build succeeds and the gold models exist, you're ready. Everything in this lab is added inside the dbt/ project and the dagster_app/ folder.

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:

INGEST MODEL (dbt) QUALITY GATE SERVE ┌────────┐ ┌──────────────┐ ┌────────────────────────┐ ┌──────────┐ │ bronze │───▶│ silver →gold │───▶│ dbt test │───▶│ semantic │ │ raw │ │ star schema │ │ ├ schema tests │ │ layer + │ └────────┘ └──────────────┘ │ │ not_null, unique │ │ dashboard│ │ │ relationships │ └──────────┘ │ │ accepted_values │ │ ├ freshness check │ │ └ custom assertions │ ▲ └────────────┬─────────────┘ │ │ ✗ FAIL → stop ──────┘ ▼ blocks bad data from serving

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

Check	Catches	How
Schema tests	Nulls in keys, duplicate keys, broken joins, invalid categories	dbt generic tests in `schema.yml`
Freshness	A feed that silently stopped — yesterday's data masquerading as today's	dbt `source freshness`
Custom assertions	Business rules: negative revenue, zero-hour rentals, row counts that don't reconcile	dbt singular tests (SQL)

A test is just a query that should return zero rows

Every dbt test compiles to a SELECT. If that query returns any rows, those rows are the failures and the test fails. Hold onto this idea — it makes both the built-in tests and the custom ones you'll write feel like the same simple thing.

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 schema.yml beside your models — no SQL required. Create or open the schema file for the gold marts and describe your star schema's contracts.

dbt/models/marts/schema.yml

version: 2

models:
  - name: fct_rentals
    description: "One row per GPU rental. 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:
              to: ref('dim_customer')
              field: customer_id
      - name: gpu_id
        description: "FK to dim_gpu."
        tests:
          - not_null
          - relationships:
              to: ref('dim_gpu')
              field: gpu_id
      - name: rental_hours
        tests:
          - not_null

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

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

terminal

dbt test --select fct_rentals dim_customer

✓ You should see

A line per test and a green summary like PASS=7 WARN=0 ERROR=0 SKIP=0 TOTAL=7. dbt names each test automatically — e.g. unique_fct_rentals_rental_id and relationships_fct_rentals_customer_id__customer_id__ref_dim_customer_. Every one passing means: no duplicate or null rental IDs, every rental points at a real customer and GPU, and every customer sits on a known tier. That is a real contract, enforced.

Why these four are the workhorses

unique + not_null on every primary key catches the single most common warehouse bug — a fan-out join that silently doubles your rows and inflates revenue. relationships catches the second — an orphaned fact whose dimension was never loaded. Put them on every key in every gold table and you've eliminated a whole class of "the numbers look wrong" incidents.

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. In your sources file, point dbt at the column that carries the load/event time and set thresholds:

dbt/models/staging/sources.yml

version: 2

sources:
  - name: bronze
    schema: bronze
    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: raw_rentals
      - name: raw_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:

terminal

dbt source freshness

✓ You should see

One line per source table with a verdict — PASS, WARN, or ERROR — and the actual age, e.g. PASS freshness of bronze.raw_rentals (max loaded_at 0:04:11 old). If you just ran dbt build, everything is freshly loaded and passes. Stop your ingest, wait past a threshold, and the verdict flips to ERROR — exactly the alarm you want when a feed dies.

Freshness vs. a row test

Freshness reads the max of a timestamp column and compares it to "now" — it doesn't scan every row, so it's cheap to run often. Reach for it whenever "did this feed update?" matters more than "is each row correct?". For mini-GridDP, run it at the top of every pipeline tick so a stale feed stops the run before it builds a misleading dashboard.

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:

dbt/tests/assert_rentals_positive.sql

-- Fail if any rental has non-positive hours or negative revenue.
-- A passing test returns ZERO rows.
select
    rental_id,
    rental_hours,
    revenue
from {{ ref('fct_rentals') }}
where rental_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:

dbt/tests/assert_rental_count_reconciles.sql

-- 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', 'raw_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:

terminal

dbt test --select test_type:singular

✓ You should see

Both pass: PASS assert_rentals_positive and PASS assert_rental_count_reconciles. Each one ran its SELECT, got zero rows back, and reported green. You've just encoded two business invariants as code that runs forever — no more "I assume revenue is positive," only "the test proves it."

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.

terminal — inject a bad row

# open a DuckDB shell on the warehouse
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).
INSERT INTO bronze.raw_rentals (rental_id, customer_id, gpu_id, rental_hours, revenue, loaded_at)
VALUES ('BAD-001', 'C-100', 'G-7', -5, 240.00, now());
.quit

Now rebuild gold so the bad row flows through, and run the full test suite:

terminal

dbt build --select fct_rentals
dbt test

✓ You should see — the FAIL

A red failure with a message that points right at the problem:

output

Failure 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 TOTAL=11

One ERROR, ten PASS. dbt even tells you how to inspect the offending rows — run the compiled SQL and you'll see rental_id = 'BAD-001' with rental_hours = -5. The gate caught it before it could reach the dashboard.

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

terminal — fix it

duckdb warehouse.duckdb "DELETE FROM bronze.raw_rentals WHERE rental_id = 'BAD-001';"
dbt build --select fct_rentals
dbt test

✓ You should see — the PASS

Back to green: Done. PASS=11 WARN=0 ERROR=0 SKIP=0 TOTAL=11. That fail → fix → pass cycle is the whole point of a trust layer. The test didn't just sit there being green; it actively caught a real defect, named it, and went green again only once the data was correct.

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.

In Lab 05 you exposed your dbt models as Dagster assets via @dbt_assets. dbt's tests come along for free: the dagster-dbt integration emits each dbt test as an asset check on the model it covers, as long as you run them. Make your dbt asset run build (which is run + test together) instead of bare run:

dagster_app/assets.py

from dagster import AssetCheckResult, asset_check
from dagster_dbt import DbtCliResource, dbt_assets

@dbt_assets(manifest=dbt_manifest_path)
def griddp_dbt_assets(context, dbt: DbtCliResource):
    # `build` = run + test. Each dbt test surfaces as an asset check
    # on its model, so a failing test marks the asset check as failed.
    yield from dbt.cli(["build"], context=context).stream()


# A standalone freshness gate, expressed as a Dagster asset check.
# It fails the check (and the run) if `dbt source freshness` errors.
@asset_check(asset="raw_rentals")
def rentals_are_fresh(dbt: DbtCliResource):
    result = dbt.cli(
        ["source", "freshness"], raise_on_error=False
    ).wait()
    return AssetCheckResult(
        passed=result.is_successful(),
        metadata={"detail": "dbt source freshness on bronze.raw_rentals"},
    )

Reload the Dagster code location and materialize the pipeline:

terminal

cd ../dagster_app
dagster dev          # then materialize the dbt assets from the UI, or:
dagster asset materialize --select griddp_dbt_assets

✓ You should see

In the Dagster UI, each gold asset now shows a Checks tab listing its dbt tests — unique_fct_rentals_rental_id, assert_rentals_positive, and the rest — each with a green check. Re-inject BAD-001, materialize again, and the run goes red: the asset check fails and the failure is attached to fct_rentals, so you can see exactly which asset broke and why. Quality now runs every time the pipeline does — no remembering required.

Block bad data from serving

Because the check is attached to the asset, you can make it a hard gate: downstream assets (the semantic layer in Lab 07) won't materialize if the upstream check failed. That's the difference between a test that reports a problem and a gate that stops bad data — set the check's severity to ERROR so a failure halts the run.

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.

terminal

cd ..
git add dbt/models/marts/schema.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"

✓ You should see

A commit summarizing the new schema.yml contracts, the freshness config, two singular tests, and the wired-in Dagster asset checks. Run git log --oneline and you'll see Lab 06 sitting right after Lab 05 — the pipeline now ships with its own quality gate.

✓ Check yourself

Can you explain why "a dbt test is a query that should return zero rows" — and how both generic and singular tests fit that rule?
Which test catches a fan-out join that doubles your rows? Which catches a feed that silently stopped loading?
What's the difference between a freshness check and a row-level test, and why is freshness cheap to run often?
Why is running dbt build (not dbt run) from the Dagster asset what turns your tests into a real gate?

Exercise — Assert telemetry utilization is a valid percentage (15 minutes)

The fct_telemetry_hourly rollup has a util_pct column — GPU utilization as a percentage. A percentage must sit between 0 and 100; anything outside that range is a bug (a bad reading, a divide error, a unit mix-up). Add a test that asserts util_pct is always in [0, 100], run it, and confirm it passes.

Do it the singular-test way (one SQL file). If you want a bonus, also express the same rule the generic way using the accepted_range test from the dbt_utils package.

solution — dbt/tests/assert_util_pct_in_range.sql

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

bonus — generic version in marts/schema.yml

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

run it

dbt test --select fct_telemetry_hourly
# -> PASS assert_util_pct_in_range

To really convince yourself, inject a row with util_pct = 142 into the rollup's source, rebuild, and watch this test fail with Got 1 result — then remove it and watch it pass. Same fail → fix → pass loop, now guarding your telemetry.

With a trusted star schema, you can finally define metrics once and serve them everywhere — turning raw columns into the business language of revenue, utilization, and churn. → Lab 07 — The Semantic Layer & Metrics

The goal — a trust layer

Schema tests on the gold models

Freshness — is the data even current?

A custom assertion — enforce a business rule

Break it, then fix it

Wire tests into the Dagster pipeline

Commit Lab 06

✓ Check yourself

Next