dbt Interview Questions and Answers Part 4 - Testing and Data Quality

dbt Interview Q&A · Part 4 of 10

In this post we are going to cover testing and data quality in dbt: data tests, severity thresholds, singular and generic tests, unit tests, model contracts and source freshness. These are 7 real dbt interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario where your manager asks you to build a testing strategy for a project that has no tests today.

Testing is how a dbt project stops bad data before it reaches a dashboard. dbt gives you several tools for it, and each one catches a different kind of problem: data tests check the rows, unit tests check the logic, contracts check the shape, and freshness checks that data is still arriving. Interviewers ask about it because every production incident ends with the same question: which test would have caught this?

Part 4 of the Real Time dbt Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's create the project and some sample data
3. Question 1: Duplicate customer IDs reached the finance dashboard. Which dbt tests would have caught it, and how do you add them?
4. Question 2: A test fails on three known bad rows from an old system. How do you use severity, warn_if and error_if instead of disabling the test?
5. Question 3: The business rule is that refunds can never exceed the order amount. How do you write a singular test or a custom generic test?
6. Question 4: A teammate changed a CASE statement and broke the revenue logic, but all data tests still passed. How do unit tests help?
7. Question 5: A downstream team needs the columns and types of your model to never change by surprise. How do model contracts work?
8. Question 6: Raw data stopped arriving and nobody noticed for two days. How do you set up source freshness?
9. Scenario 7: Your manager asks you to build a testing strategy for a dbt project that has no tests today. Where do you start?
10. The full script

The short answer

Say this first

Put unique and not_null on every key and run dbt build, so a failing test skips everything downstream. Tune known issues with warn_if / error_if, write business rules as singular or generic tests, check logic with unit tests, protect shared models with contracts, and watch feeds with source freshness.

Let's create the project and some sample data

Our project is a small shop. A seed called raw_customers stands in for the raw customer table, with two problems already in it: customer 102 is in there twice, and three customers from the legacy system have no email. A staging view cleans it up and a mart table, dim_customers, feeds the dashboard. Here are the starting files. The full script at the end also writes a profiles.yml that points at a local DuckDB file.

dbt_project.yml

name: shop
version: '1.0'
profile: shop

models:
  shop:
    staging:
      +materialized: view
    marts:
      +materialized: table

seeds/raw_customers.csv

customer_id,email,source_system,updated_at
101,ali@shop.com,crm,2026-09-01 10:00:00
102,sara@shop.com,crm,2026-09-02 11:00:00
102,sara.lee@shop.com,crm,2026-10-01 09:30:00
103,,legacy,2019-03-04 00:00:00
104,,legacy,2019-05-11 00:00:00
105,,legacy,2019-08-20 00:00:00
106,omar@shop.com,crm,2026-10-02 14:00:00

models/staging/stg_customers.sql

select
    customer_id,
    email,
    source_system,
    cast(updated_at as timestamp) as updated_at
from {{ ref('raw_customers') }}

models/marts/dim_customers.sql

select customer_id, email, source_system, updated_at
from {{ ref('stg_customers') }}

Question 1: Duplicate customer IDs reached the finance dashboard. Which dbt tests would have caught it, and how do you add them?

The primary key tests: unique and not_null on customer_id. They are a few lines of YAML, under data_tests, in the model's properties file.

Let's add them to stg_customers and run dbt build on dim_customers and everything upstream of it.

models/staging/stg_customers.yml

models:
  - name: stg_customers
    columns:
      - name: customer_id
        data_tests:
          - unique
          - not_null

Terminal (expected error)

$ dbt build -s +dim_customers
 Succeeded seed  main.raw_customers (table)
 Succeeded model main.stg_customers (view)
    Passed test  not_null_stg_customers_customer_id
    Failed test  unique_stg_customers_customer_id (models/staging/stg_customers.yml:6:13)
   Skipped model main.dim_customers (table)

====================== Test Failures =======================
Test failed (1 failed row(s)): unique_stg_customers_customer_id

Finished 'build' with 1 error for target 'dev'
Processed: 2 models | 2 tests | 1 seed
Summary: 5 total | 3 success | 1 error | 1 skipped

The unique test fails with 1 failed row(s): customer 102 is in the source twice. And because build runs the tests right after their model, dim_customers is skipped. The summary says 1 error | 1 skipped, so the duplicates never reach the dashboard.

Then we fix the cause. Staging now keeps the latest row per customer, with qualify and row_number(). Let's build again.

models/staging/stg_customers.sql

select
    customer_id,
    email,
    source_system,
    cast(updated_at as timestamp) as updated_at
from {{ ref('raw_customers') }}
-- one row per customer: keep the latest version
qualify row_number() over (
    partition by customer_id order by updated_at desc
) = 1

Terminal

$ dbt build -s +dim_customers
 Succeeded seed  main.raw_customers (table)
 Succeeded model main.stg_customers (view)
    Passed test  not_null_stg_customers_customer_id
    Passed test  unique_stg_customers_customer_id
 Succeeded model main.dim_customers (table)

Finished 'build' successfully for target 'dev'
Processed: 2 models | 2 tests | 1 seed
Summary: 5 total | 5 success

Both tests pass and dim_customers builds: 5 total | 5 success.

On a real project I put unique and not_null on the key of every model, and relationships tests on the foreign keys. For a composite key, the dbt_utils package has unique_combination_of_columns.

Real project tip

Run dbt build, not dbt run and then dbt test. With build, a failing test stops bad data from flowing downstream.

Question 2: A test fails on three known bad rows from an old system. How do you use severity, warn_if and error_if instead of disabling the test?

Don't disable it. A disabled test also stops catching new bad rows. Set thresholds instead, so the known rows warn and anything new fails.

First, let's add a not_null test on email and build.

models/staging/stg_customers.yml

models:
  - name: stg_customers
    columns:
      - name: customer_id
        data_tests:
          - unique
          - not_null
      - name: email
        data_tests:
          - not_null

Terminal (expected error)

$ dbt build -s +dim_customers
 Succeeded seed  main.raw_customers (table)
 Succeeded model main.stg_customers (view)
    Passed test  not_null_stg_customers_customer_id
    Failed test  not_null_stg_customers_email (models/staging/stg_customers.yml:10:13)
    Passed test  unique_stg_customers_customer_id
   Skipped model main.dim_customers (table)

====================== Test Failures =======================
Test failed (3 failed row(s)): not_null_stg_customers_email

Finished 'build' with 1 error for target 'dev'
Processed: 2 models | 3 tests | 1 seed
Summary: 6 total | 4 success | 1 error | 1 skipped

The three legacy customers have no email, so the test fails with 3 failed row(s) and dim_customers is skipped. Now we set thresholds in the test's config. With severity: error, dbt checks error_if first: more than three failing rows is an error. Then warn_if: more than zero is a warning.

models/staging/stg_customers.yml

models:
  - name: stg_customers
    columns:
      - name: customer_id
        data_tests:
          - unique
          - not_null
      - name: email
        data_tests:
          - not_null:
              config:
                severity: error
                warn_if: '>0'
                error_if: '>3'

Terminal

$ dbt build -s +dim_customers
 Succeeded seed  main.raw_customers (table)
 Succeeded model main.stg_customers (view)
    Passed test  not_null_stg_customers_customer_id
    Warned test  not_null_stg_customers_email (models/staging/stg_customers.yml:10:13)
    Passed test  unique_stg_customers_customer_id
 Succeeded model main.dim_customers (table)

Finished 'build' with 1 warning for target 'dev'
Processed: 2 models | 3 tests | 1 seed
Summary: 6 total | 5 success | 1 warn

The test now shows Warned on the three known rows, and dim_customers builds. The build finishes with 1 warning. Then a fourth bad row arrives: customer 107, from the CRM, without an email.

seeds/raw_customers.csv

customer_id,email,source_system,updated_at
101,ali@shop.com,crm,2026-09-01 10:00:00
102,sara@shop.com,crm,2026-09-02 11:00:00
102,sara.lee@shop.com,crm,2026-10-01 09:30:00
103,,legacy,2019-03-04 00:00:00
104,,legacy,2019-05-11 00:00:00
105,,legacy,2019-08-20 00:00:00
106,omar@shop.com,crm,2026-10-02 14:00:00
107,,crm,2026-10-03 08:00:00

Terminal (expected error)

$ dbt build -s +dim_customers
 Succeeded seed  main.raw_customers (table)
 Succeeded model main.stg_customers (view)
    Passed test  not_null_stg_customers_customer_id
    Failed test  not_null_stg_customers_email (models/staging/stg_customers.yml:10:13)
    Passed test  unique_stg_customers_customer_id
   Skipped model main.dim_customers (table)

====================== Test Failures =======================
Test failed (4 failed row(s)): not_null_stg_customers_email

Finished 'build' with 1 error for target 'dev'
Processed: 2 models | 3 tests | 1 seed
Summary: 6 total | 4 success | 1 error | 1 skipped

The test fails again, with 4 failed row(s), and dim_customers is skipped. That is exactly what we want.

There are two more options worth knowing:

• A where config that excludes the legacy rows is often better: any new null fails, even if an old row was fixed in the meantime.
• severity: warn on its own never fails, because dbt then skips error_if.

Question 3: The business rule is that refunds can never exceed the order amount. How do you write a singular test or a custom generic test?

A singular test is a SQL file in the tests folder that returns the bad rows. A custom generic test is a reusable test block that you attach to columns in YAML. Zero rows returned means the test passes, for both.

Let's start with the generic test. The not_greater_than test block takes the model, the column and our own argument, other_column. We attach it under arguments: refund_amount must not be greater than order_amount. We add orders and refunds, build fct_refunds, and look at the refunds.

seeds/raw_orders.csv

order_id,customer_id,amount,status
1,101,100.00,completed
2,102,50.00,shipped
3,106,75.00,cancelled

seeds/raw_refunds.csv

refund_id,order_id,refund_amount
1,1,30.00
2,1,80.00
3,2,60.00
4,3,20.00

models/staging/stg_orders.sql

select
    order_id,
    customer_id,
    cast(amount as decimal(10, 2)) as amount,
    status
from {{ ref('raw_orders') }}

models/marts/fct_refunds.sql

select
    r.refund_id,
    r.order_id,
    cast(r.refund_amount as decimal(10, 2)) as refund_amount,
    o.amount as order_amount
from {{ ref('raw_refunds') }} as r
join {{ ref('stg_orders') }} as o on r.order_id = o.order_id

tests/generic/not_greater_than.sql

{% test not_greater_than(model, column_name, other_column) %}

select *
from {{ model }}
where {{ column_name }} > {{ other_column }}

{% endtest %}

models/marts/fct_refunds.yml

models:
  - name: fct_refunds
    columns:
      - name: refund_amount
        data_tests:
          - not_greater_than:
              arguments:
                other_column: order_amount

Terminal (expected error)

$ dbt build -s +fct_refunds
 Succeeded seed  main.raw_orders (table)
 Succeeded seed  main.raw_refunds (table)
 Succeeded model main.stg_orders (view)
 Succeeded model main.fct_refunds (table)
    Failed test  not_greater_than_fct_refunds_refund_amount__order_amount
      (models/marts/fct_refunds.yml:6:13)

====================== Test Failures =======================
Test failed (1 failed row(s)): not_greater_than_fct_refunds_refund_amount__order_amount

Finished 'build' with 1 error for target 'dev'
Processed: 2 models | 1 test | 2 seeds
Summary: 5 total | 4 success | 1 error

SQL - check the table (DuckDB)

select refund_id, order_id, refund_amount, order_amount from fct_refunds order by refund_id

Result4 rows

refund_id	order_id	refund_amount	order_amount
1	1	30.00	100.00
2	1	80.00	100.00
3	2	60.00	50.00
4	3	20.00	75.00

The generic test finds one row: refund 3, 60.00 on an order of 50.00. Now the singular test. It totals the refunds per order and returns the orders where they add up to more than the order amount. We run it with dbt test.

tests/assert_refunds_not_more_than_order.sql

-- all refunds of an order must not add up to more than the order
select
    order_id,
    sum(refund_amount) as total_refunded,
    max(order_amount) as order_amount
from {{ ref('fct_refunds') }}
group by order_id
having sum(refund_amount) > max(order_amount)

Terminal (expected error)

$ dbt test -s assert_refunds_not_more_than_order
    Failed test  assert_refunds_not_more_than_order (tests/assert_refunds_not_more_than_order.sql)

====================== Test Failures =======================
Test failed (2 failed row(s)): assert_refunds_not_more_than_order

Finished 'test' with 1 error for target 'dev'
Processed: 1 test
Summary: 1 total | 1 error

It finds 2 failed row(s). Order 1 had two refunds, 30.00 and 80.00. Each one passes on its own, but together they add up to 110, more than the 100.00 order.

That is how I choose:

• A one-off rule, or one that needs joins or totals: a singular test
• A rule you will reuse on many columns: a generic test

Both kinds take the same configs, like severity and store_failures, and both run with dbt build.

Question 4: A teammate changed a CASE statement and broke the revenue logic, but all data tests still passed. How do unit tests help?

Data tests check the data after the build. A unit test checks the logic itself, with fixed inputs and expected outputs. A not_null test on revenue still passes when the CASE is wrong, because zero is not null.

The unit test lives in YAML, next to the model. Given three staging rows, completed, shipped and cancelled, it expects revenue of 100, 50 and 0. You only list the columns that matter. Let's build fct_orders.

models/marts/fct_orders.sql

select
    order_id,
    customer_id,
    status,
    case
        when status in ('completed', 'shipped') then amount
        else 0
    end as revenue
from {{ ref('stg_orders') }}

models/marts/fct_orders.yml

models:
  - name: fct_orders
    columns:
      - name: revenue
        data_tests:
          - not_null

unit_tests:
  - name: revenue_counts_completed_and_shipped
    model: fct_orders
    given:
      - input: ref('stg_orders')
        rows:
          - {order_id: 1, amount: 100, status: completed}
          - {order_id: 2, amount: 50, status: shipped}
          - {order_id: 3, amount: 75, status: cancelled}
    expect:
      rows:
        - {order_id: 1, revenue: 100}
        - {order_id: 2, revenue: 50}
        - {order_id: 3, revenue: 0}

Terminal

$ dbt build -s +fct_orders
 Succeeded seed  main.raw_orders (table)
 Succeeded model main.stg_orders (view)
    Passed unit_test main_dbt_test__audit.revenue_counts_completed_and_shipped
 Succeeded model main.fct_orders (table)
    Passed test  not_null_fct_orders_revenue

Finished 'build' successfully for target 'dev'
Processed: 2 models | 1 test | 1 seed | 1 unit test
Summary: 5 total | 5 success

The unit test and the data test both pass. Now the teammate changes the CASE to count only completed orders, and we build again.

models/marts/fct_orders.sql

select
    order_id,
    customer_id,
    status,
    case
        when status = 'completed' then amount
        else 0
    end as revenue
from {{ ref('stg_orders') }}

Terminal (expected error)

$ dbt build -s +fct_orders
 Succeeded seed  main.raw_orders (table)
 Succeeded model main.stg_orders (view)
    Failed unit_test main_dbt_test__audit.revenue_counts_completed_and_shipped
      (models/marts/fct_orders.yml:9:11)
   Skipped model main.fct_orders (table)
   Skipped test  not_null_fct_orders_revenue (models/marts/fct_orders.yml:6:13)

====================== Test Failures =======================
Test failed: revenue_counts_completed_and_shipped
+----------+-------------+
| order_id | revenue     |
+----------+-------------+
| 1        | 100.0       |
| 2        | 50.0 -> 0.0 |
| 3        | 0.0         |
+----------+-------------+
3 row(s) differ.

Finished 'build' with 1 error for target 'dev'
Processed: 2 models | 1 test | 1 seed | 1 unit test
Summary: 5 total | 2 success | 1 error | 2 skipped

The unit test fails, and the diff points at order 2: 50.0 -> 0.0. The unit test runs before the model is built, so fct_orders is skipped, and so is its not_null test. The summary says 1 error | 2 skipped.

One detail on ordering. The v2 docs say a build runs all unit tests first, using column type awareness. Here, on the dbt OSS build, as in dbt v1, each unit test runs just before its model, and its parents must exist.

Production tip

dbt recommends running unit tests in development and CI, not in production. In prod, exclude them with --exclude-resource-type unit_test.

Question 5: A downstream team needs the columns and types of your model to never change by surprise. How do model contracts work?

A contract makes the shape of the model a promise. You set contract: enforced: true and declare every column with its data type. Constraints, like not_null, go into the table's DDL.

Let's add a contract to dim_customers and build it.

models/marts/dim_customers.yml

models:
  - name: dim_customers
    config:
      contract:
        enforced: true
    columns:
      - name: customer_id
        data_type: integer
        constraints:
          - type: not_null
      - name: email
        data_type: varchar
      - name: source_system
        data_type: varchar
      - name: updated_at
        data_type: timestamp

Terminal

$ dbt build -s dim_customers
 Succeeded model main.dim_customers (table)

Finished 'build' successfully for target 'dev'
Processed: 1 model
Summary: 1 total | 1 success

It passes. dbt checks the columns of the query against the contract before it creates the table. Now a teammate casts customer_id to text.

models/marts/dim_customers.sql

select
    cast(customer_id as varchar) as customer_id,
    email,
    source_system,
    updated_at
from {{ ref('stg_customers') }}

Terminal (expected error)

$ dbt build -s dim_customers
    Failed model main.dim_customers (table)

=================== Errors and Warnings ====================
[error] [JinjaError (dbt1501)]: Error executing materialization macro
  'dbt_duckdb.materialization_table_duckdb' for model model.shop.dim_customers: Failed to eval the
    compiled Jinja expression invalid operation: Compilation Error for model.shop.dim_customers from
      models/marts/dim_customers.sql: This model has an enforced contract that failed.
 Please ensure the name, data_type, and number of columns in your contract match the columns in your
   model's definition.

| column_name | definition_type | contract_type |    mismatch_reason |
| ----------- | --------------- | ------------- | ------------------ |
| customer_id | text            | integer       | data type mismatch |

(in run/shop/models/marts/dim_customers.sql)
(in dbt_internal_packages/dbt-duckdb/macros/materializations/table.sql:39:11)
(in dbt_internal_packages/dbt-adapters/macros/etc/statement.sql:7:29)
(in dbt_internal_packages/dbt-duckdb/macros/materializations/table.sql:40:5)
(in dbt_internal_packages/dbt-adapters/macros/relations/table/create.sql:17:5)
(in dbt_internal_packages/dbt-duckdb/macros/adapters.sql:69:7)
(in dbt_internal_packages/dbt-adapters/macros/relations/column/columns_spec_ddl.sql:28:3)
(in dbt_internal_packages/dbt-adapters/macros/relations/column/columns_spec_ddl.sql:33:13)
(in dbt_internal_packages/dbt-adapters/macros/relations/column/columns_spec_ddl.sql:76:24)
  --> run/shop/models/marts/dim_customers.sql

Finished 'build' with 1 error for target 'dev'
Processed: 1 model
Summary: 1 total | 1 error

The build fails before the table is replaced, with This model has an enforced contract that failed. The mismatch table says it all: for customer_id the definition type is text, the contract type is integer, and the reason is data type mismatch. The downstream team never sees the change.

A few limits to know:

• Contracts work on tables and incremental models (with on_schema_change set to fail or append_new_columns)
• On views they check only names and types, with no constraints
• They don't work on ephemeral models
• Most platforms only enforce not_null, so tests still check the content

Breaking changes

When you really must change the shape, add a new model version, so consumers can move over on their own schedule. We cover versions later in this series.

Question 6: Raw data stopped arriving and nobody noticed for two days. How do you set up source freshness?

Source freshness compares the newest load time in a source table with the current time, and warns or errors when the gap gets too big.

First, we need a feed that has gone quiet. A small model stands in for the loader: the last payments batch landed two days ago. We run it with dbt run.

models/loader/payments_feed.sql

-- simulates the loader: the last batch landed two days ago
select
    1 as payment_id,
    1 as order_id,
    100.00 as amount,
    current_timestamp - interval 2 day as _loaded_at

Terminal

$ dbt run -s payments_feed
 Succeeded model main.payments_feed (view)

Finished 'run' successfully for target 'dev'
Processed: 1 model
Summary: 1 total | 1 success

Then we declare the source. In the source YAML, under config, we set the loaded_at_field and the thresholds: warn after 12 hours, error after 1 day. Let's run dbt source freshness.

models/staging/sources.yml

sources:
  - name: shop_app
    schema: main
    tables:
      - name: payments_feed
        config:
          loaded_at_field: _loaded_at
          freshness:
            warn_after: {count: 12, period: hour}
            error_after: {count: 1, period: day}

Terminal (expected error)

$ dbt source freshness
     Stale source shop_app.payments_feed (last updated 2days ago)
Successfully wrote sources.json

=================== Errors and Warnings ====================
[error] [StaleSource (dbt1063)]: Stale source source.shop.shop_app.payments_feed
  --> models/staging/sources.yml

Finished 'freshness' with 1 error for target 'dev'

The payments feed shows Stale (last updated 2days ago), and the command ends with StaleSource (dbt1063) and 1 error.

Two details interviewers like:

• freshness moved under config in dbt v1.9, and loaded_at_field in v1.10. The dbt v2 parser rejects freshness at the old top level.
• On DuckDB you need a loaded_at_field or a loaded_at_query, because the adapter has no metadata-based freshness.

dbt v2 also adds model freshness and the dbt freshness command, for sources and models together. dbt source freshness still works.

Production tip

Run freshness as the first step of the job, and alert on it, so a stopped feed pages someone the same morning, not two days later.

Scenario 7: Your manager asks you to build a testing strategy for a dbt project that has no tests today. Where do you start?

Start with what breaks dashboards most: keys. Then build out in layers, and run the tests everywhere.

Here is the order I go in:

• Keys: unique and not_null on the primary key of every model, then not_null and accepted_values on the columns reports filter on
• Sources: freshness on every feed, and key tests on the raw tables, so bad data is caught at the door
• Business rules: singular and generic tests
• Logic and shape: unit tests for tricky logic, contracts on the models other teams use
• Run it everywhere: dbt build in CI and production

Then set project defaults in dbt_project.yml. Here store_failures is on for every test, and customer 107 now has an email. Let's build the seed and staging again.

dbt_project.yml

name: shop
version: '1.0'
profile: shop

models:
  shop:
    staging:
      +materialized: view
    marts:
      +materialized: table

data_tests:
  shop:
    +store_failures: true

seeds/raw_customers.csv

customer_id,email,source_system,updated_at
101,ali@shop.com,crm,2026-09-01 10:00:00
102,sara@shop.com,crm,2026-09-02 11:00:00
102,sara.lee@shop.com,crm,2026-10-01 09:30:00
103,,legacy,2019-03-04 00:00:00
104,,legacy,2019-05-11 00:00:00
105,,legacy,2019-08-20 00:00:00
106,omar@shop.com,crm,2026-10-02 14:00:00
107,dina@shop.com,crm,2026-10-03 08:00:00

Terminal

$ dbt build -s raw_customers stg_customers
 Succeeded seed  main.raw_customers (table)
 Succeeded model main.stg_customers (view)
    Passed test  main_dbt_test__audit.not_null_stg_customers_customer_id
    Warned test  main_dbt_test__audit.not_null_stg_customers_email
      (models/staging/stg_customers.yml:10:13)
    Passed test  main_dbt_test__audit.unique_stg_customers_customer_id

Finished 'build' with 1 warning for target 'dev'
Processed: 1 model | 3 tests | 1 seed
Summary: 5 total | 4 success | 1 warn

SQL - check the table (DuckDB)

select customer_id, source_system from main_dbt_test__audit.not_null_stg_customers_email order by customer_id

Result3 rows

customer_id	source_system
103	legacy
104	legacy
105	legacy

The build still passes, with 1 warning, and the failing rows are saved in a table in the main_dbt_test__audit schema. The query on that table returns the three legacy customers, 103, 104 and 105. That is the list you hand to the source owner.

Finally, in CI run dbt build -s state:modified+ on the changed models and their children, and run dbt build in production, so a failing test stops bad data.

Real project tip

After every incident, add the test that would have caught it.

Key points to remember

• unique + not_null on every key; dbt build skips downstream
• warn_if / error_if for known bad rows, never disable
• Singular tests for totals and joins, generic tests to reuse
• Unit tests for logic; contracts for the shape
• Source freshness first in the job, with an alert

The full script

Here is a Python script that writes the whole project and runs every dbt command above in order; run pip install dbt first.

Show the full script that builds this project and runs every step

Full script - pip install dbt, then run it with Python in an empty folder

# ========================================================================
# dbt Interview Q&A #4 - Testing and data quality
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run with dbt v2 (dbt OSS 2.0.8) on DuckDB; the outputs in the video and the
# blog post are the real outputs. To run it yourself, in an empty folder:
#   pip install dbt     (dbt v2; adapters are bundled, and DuckDB needs no account or server)
#   python dbt_Interview_QA_04_Code.py
# ========================================================================

import os, subprocess

def write(path, text):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(text.strip("\n") + "\n")

def remove(path):
    if os.path.exists(path):
        os.remove(path)

def dbt(cmd, expect_error=False):
    print("\n$ " + cmd, flush=True)
    rc = subprocess.run(cmd, shell=True).returncode
    if rc != 0 and not expect_error:
        raise SystemExit("stopped: this command was expected to succeed")

# Step 0: the project and its sample data
write('dbt_project.yml', """\
name: shop
version: '1.0'
profile: shop

models:
  shop:
    staging:
      +materialized: view
    marts:
      +materialized: table
""")
write('seeds/raw_customers.csv', """\
customer_id,email,source_system,updated_at
101,ali@shop.com,crm,2026-09-01 10:00:00
102,sara@shop.com,crm,2026-09-02 11:00:00
102,sara.lee@shop.com,crm,2026-10-01 09:30:00
103,,legacy,2019-03-04 00:00:00
104,,legacy,2019-05-11 00:00:00
105,,legacy,2019-08-20 00:00:00
106,omar@shop.com,crm,2026-10-02 14:00:00
""")
write('models/staging/stg_customers.sql', """\
select
    customer_id,
    email,
    source_system,
    cast(updated_at as timestamp) as updated_at
from {{ ref('raw_customers') }}
""")
write('models/marts/dim_customers.sql', """\
select customer_id, email, source_system, updated_at
from {{ ref('stg_customers') }}
""")
write('profiles.yml', """\
shop:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: dev.duckdb
      threads: 1
""")

# Step 1: Primary key tests on stg_customers: the duplicate customer fails the build
write('models/staging/stg_customers.yml', """\
models:
  - name: stg_customers
    columns:
      - name: customer_id
        data_tests:
          - unique
          - not_null
""")
dbt('dbt build -s +dim_customers', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 2: Fix the cause: keep the latest row per customer in staging
write('models/staging/stg_customers.sql', """\
select
    customer_id,
    email,
    source_system,
    cast(updated_at as timestamp) as updated_at
from {{ ref('raw_customers') }}
-- one row per customer: keep the latest version
qualify row_number() over (
    partition by customer_id order by updated_at desc
) = 1
""")
dbt('dbt build -s +dim_customers')

# Step 3: A not_null test on email fails on three legacy rows
write('models/staging/stg_customers.yml', """\
models:
  - name: stg_customers
    columns:
      - name: customer_id
        data_tests:
          - unique
          - not_null
      - name: email
        data_tests:
          - not_null
""")
dbt('dbt build -s +dim_customers', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 4: warn_if and error_if: the three known rows warn, more than three fail
write('models/staging/stg_customers.yml', """\
models:
  - name: stg_customers
    columns:
      - name: customer_id
        data_tests:
          - unique
          - not_null
      - name: email
        data_tests:
          - not_null:
              config:
                severity: error
                warn_if: '>0'
                error_if: '>3'
""")
dbt('dbt build -s +dim_customers')

# Step 5: A fourth customer without an email arrives: the test fails again
write('seeds/raw_customers.csv', """\
customer_id,email,source_system,updated_at
101,ali@shop.com,crm,2026-09-01 10:00:00
102,sara@shop.com,crm,2026-09-02 11:00:00
102,sara.lee@shop.com,crm,2026-10-01 09:30:00
103,,legacy,2019-03-04 00:00:00
104,,legacy,2019-05-11 00:00:00
105,,legacy,2019-08-20 00:00:00
106,omar@shop.com,crm,2026-10-02 14:00:00
107,,crm,2026-10-03 08:00:00
""")
dbt('dbt build -s +dim_customers', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 6: Refunds: a custom generic test, not_greater_than
write('seeds/raw_orders.csv', """\
order_id,customer_id,amount,status
1,101,100.00,completed
2,102,50.00,shipped
3,106,75.00,cancelled
""")
write('seeds/raw_refunds.csv', """\
refund_id,order_id,refund_amount
1,1,30.00
2,1,80.00
3,2,60.00
4,3,20.00
""")
write('models/staging/stg_orders.sql', """\
select
    order_id,
    customer_id,
    cast(amount as decimal(10, 2)) as amount,
    status
from {{ ref('raw_orders') }}
""")
write('models/marts/fct_refunds.sql', """\
select
    r.refund_id,
    r.order_id,
    cast(r.refund_amount as decimal(10, 2)) as refund_amount,
    o.amount as order_amount
from {{ ref('raw_refunds') }} as r
join {{ ref('stg_orders') }} as o on r.order_id = o.order_id
""")
write('tests/generic/not_greater_than.sql', """\
{% test not_greater_than(model, column_name, other_column) %}

select *
from {{ model }}
where {{ column_name }} > {{ other_column }}

{% endtest %}
""")
write('models/marts/fct_refunds.yml', """\
models:
  - name: fct_refunds
    columns:
      - name: refund_amount
        data_tests:
          - not_greater_than:
              arguments:
                other_column: order_amount
""")
dbt('dbt build -s +fct_refunds', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 7: Refunds: a singular test that totals the refunds per order
write('tests/assert_refunds_not_more_than_order.sql', """\
-- all refunds of an order must not add up to more than the order
select
    order_id,
    sum(refund_amount) as total_refunded,
    max(order_amount) as order_amount
from {{ ref('fct_refunds') }}
group by order_id
having sum(refund_amount) > max(order_amount)
""")
dbt('dbt test -s assert_refunds_not_more_than_order', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 8: A unit test for the revenue CASE logic
write('models/marts/fct_orders.sql', """\
select
    order_id,
    customer_id,
    status,
    case
        when status in ('completed', 'shipped') then amount
        else 0
    end as revenue
from {{ ref('stg_orders') }}
""")
write('models/marts/fct_orders.yml', """\
models:
  - name: fct_orders
    columns:
      - name: revenue
        data_tests:
          - not_null

unit_tests:
  - name: revenue_counts_completed_and_shipped
    model: fct_orders
    given:
      - input: ref('stg_orders')
        rows:
          - {order_id: 1, amount: 100, status: completed}
          - {order_id: 2, amount: 50, status: shipped}
          - {order_id: 3, amount: 75, status: cancelled}
    expect:
      rows:
        - {order_id: 1, revenue: 100}
        - {order_id: 2, revenue: 50}
        - {order_id: 3, revenue: 0}
""")
dbt('dbt build -s +fct_orders')

# Step 9: A teammate changes the CASE: the unit test fails, the data test would still pass
write('models/marts/fct_orders.sql', """\
select
    order_id,
    customer_id,
    status,
    case
        when status = 'completed' then amount
        else 0
    end as revenue
from {{ ref('stg_orders') }}
""")
dbt('dbt build -s +fct_orders', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 10: An enforced contract on dim_customers
write('models/marts/dim_customers.yml', """\
models:
  - name: dim_customers
    config:
      contract:
        enforced: true
    columns:
      - name: customer_id
        data_type: integer
        constraints:
          - type: not_null
      - name: email
        data_type: varchar
      - name: source_system
        data_type: varchar
      - name: updated_at
        data_type: timestamp
""")
dbt('dbt build -s dim_customers')

# Step 11: A teammate casts customer_id to text: the contract stops the build
write('models/marts/dim_customers.sql', """\
select
    cast(customer_id as varchar) as customer_id,
    email,
    source_system,
    updated_at
from {{ ref('stg_customers') }}
""")
dbt('dbt build -s dim_customers', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 12: Simulate the loader: the payments feed last landed two days ago
write('models/loader/payments_feed.sql', """\
-- simulates the loader: the last batch landed two days ago
select
    1 as payment_id,
    1 as order_id,
    100.00 as amount,
    current_timestamp - interval 2 day as _loaded_at
""")
dbt('dbt run -s payments_feed')

# Step 13: Source freshness: warn after 12 hours, error after 1 day
write('models/staging/sources.yml', """\
sources:
  - name: shop_app
    schema: main
    tables:
      - name: payments_feed
        config:
          loaded_at_field: _loaded_at
          freshness:
            warn_after: {count: 12, period: hour}
            error_after: {count: 1, period: day}
""")
dbt('dbt source freshness', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 14: Project defaults: store_failures for every test, then rebuild
write('dbt_project.yml', """\
name: shop
version: '1.0'
profile: shop

models:
  shop:
    staging:
      +materialized: view
    marts:
      +materialized: table

data_tests:
  shop:
    +store_failures: true
""")
write('seeds/raw_customers.csv', """\
customer_id,email,source_system,updated_at
101,ali@shop.com,crm,2026-09-01 10:00:00
102,sara@shop.com,crm,2026-09-02 11:00:00
102,sara.lee@shop.com,crm,2026-10-01 09:30:00
103,,legacy,2019-03-04 00:00:00
104,,legacy,2019-05-11 00:00:00
105,,legacy,2019-08-20 00:00:00
106,omar@shop.com,crm,2026-10-02 14:00:00
107,dina@shop.com,crm,2026-10-03 08:00:00
""")
dbt('dbt build -s raw_customers stg_customers')

Next in this series

Part 5: Snapshots and SCD Type 2

The sample data on this page is invented for teaching. Every terminal output is the real output of the dbt command above it, run with dbt v2 (dbt OSS 2.0.8) on DuckDB; run times and node counters are left out, and node lines are listed in DAG order. Part of the Real Time dbt Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official dbt documentation (docs.getdbt.com) for anything you quote in an interview.