DEV Community: Ahmad Muhammad

Capture Slowly Changing Attributes in SQL - SCD Type 2

Ahmad Muhammad — Fri, 08 Aug 2025 15:50:48 +0000

SCD is a data modeling technique used to track how changes in entity attributes affect behavior over time.

Think:

A product’s price change vs sales impact
A button turns red, do clicks spike?
A post’s tone goes from formal to savage, what happens to engagement?

“What was true when this happened?”

Example

Metclicks, a streaming service startup, saw a views spike in “Spongebob Watches” , right after reclassifying it from "Cartoon" to "Realistic".

Slowly Changing Dimension Common Types

Type 0 - The Fanatic

No Change, Frozen in time

customer_id	name	tier
1	Sandy Cheeks	Gold

Type 1 - The Chad

New values overwrites old values Doesn't give a 🦆 about old values like it wasn't existed

customer_id	name	tier
1	Sandy Cheeks	Gold

Type 2 - The OCD

Keeps everything, marks old values with is_current flag, in addition to validity period for each change

customer_sk	customer_id	name	tier	valid_from	valid_to	is_current
abc123	1	Sandy Cheeks	Bronze	2024-01-01	2025-06-01	false
def456	1	Sandy Cheeks	Gold	2025-06-02	2999-12-31	true

Type 3 - The Short memory

It only remembers the previous values. barely!

customer_id	name	tier	prev_tier
1	Sandy Cheeks	Gold	Bronze

👉 Before starting, designing SCD 2 should be taken with a grain of salt, because:

Some newly-arrived attributes might have spelling issue or fidelity issues.

Know how would you handle NULLs, does NULL has any business meaning? should it be taken in consideration while designing the dimension or just ignore it?

Writer opinions:

The MDL should be extendable, meaning that tracking new attributes in the future should not be painful and doesn't require any huge refactoring and testing.
A lot of time you might go with a hybrid approach between SCD 2 and SCD 1, where applying strict tracking on some attributes and loosely track the others, for example: strictly track product's product_category but loosely track product_name (depending on business logic).

Getting Our Hands Dirty

let's have a look at this fake python-generated dataset, which loaded into duckdb into customers table.

Get data

create or replace table customers as
with cte_data as (
select *, now() as delivered_at from read_csv('./customers_example.csv')
)
select * from cte_data;

Source's Schema

column_name	column_type	null	key	default	extra
customer_id	BIGINT	YES	NULL	NULL	NULL
name	VARCHAR	YES	NULL	NULL	NULL
email	VARCHAR	YES	NULL	NULL	NULL
signup_date	DATE	YES	NULL	NULL	NULL
country	VARCHAR	YES	NULL	NULL	NULL
date	DATE	YES	NULL	NULL	NULL
tier	VARCHAR	YES	NULL	NULL	NULL
interaction_type	VARCHAR	YES	NULL	NULL	NULL
interaction_value	DOUBLE	YES	NULL	NULL	NULL
delivered_at	TIMESTAMP WITH TIME ZONE	YES	NULL	NULL	NULL

The goal is to track the following attributes:

Tracking Type	SCD Type	Attribute	Reason
Restrict	2	`country`	Impacts customer segmentation and behavior analysis.
Restrict	2	`tier`	Indicates customer status; affects engagement, offers, and analytics.
Restrict	2	`email`	Primary communication channel; needs to be historically tracked for auditing and outreach.
Loosy	1	`name`	Might change (e.g. after marriage), but we treat it as static. Changes reflect corrections, not business-impacting updates.
Loosy	1	`signup_date`	Treated as a factual point in time. Fixes may occur (e.g. migration issues), but don't require historical tracking.

SQL Implementation

Compress Tracked Attributes

By giving each version of attributes combination a hash, concatenate, or any type of idempotent process, it would be easier to identify when something changes in the next couple of steps.

Make sure to include your business key in the hash, just to embed to whom this piece of information belong.

WITH cte_scd_columns AS (
  SELECT
    MD5(
      COALESCE(CAST(customer_id AS TEXT), '__null__') || COALESCE(CAST(country AS TEXT), '__null__') || COALESCE(CAST(email AS TEXT), '__null__') || COALESCE(CAST(tier AS TEXT), '__null__')
    ) AS scd_id,
    *
  FROM events
)
SELECT
  *
FROM cte_scd_columns
LIMIT 5

scd_id	column0	customer_id	name	email	signup_date	country	timestamp	tier	interaction_type	interaction_value	delivered_at
7ebaf657dc78359b4030489069de9f7d	0	1	Allison Hill	donaldgarcia@example.net	2025-04-30 00:00:00	Benin	2025-05-22 09:44:21.840779	Gold	login	nan	2025-08-02 15:06:47.289000+03:00
7ebaf657dc78359b4030489069de9f7d	1	1	Allison Hill	donaldgarcia@example.net	2025-04-30 00:00:00	Benin	2025-05-22 05:40:45.840779	Gold	support_ticket	nan	2025-08-02 15:06:47.289000+03:00
c67d1706d04fce3f93ec891827804882	2	1	Allison Hill	donaldgarcia@example.net	2025-04-30 00:00:00	Benin	2025-05-23 15:20:28.840779	Silver	support_ticket	nan	2025-08-02 15:06:47.289000+03:00
c67d1706d04fce3f93ec891827804882	3	1	Allison Hill	donaldgarcia@example.net	2025-04-30 00:00:00	Benin	2025-05-23 10:04:58.840779	Silver	support_ticket	nan	2025-08-02 15:06:47.289000+03:00
c67d1706d04fce3f93ec891827804882	4	1	Allison Hill	donaldgarcia@example.net	2025-04-30 00:00:00	Benin	2025-05-23 17:27:35.840779	Silver	newsletter_click	nan	2025-08-02 15:06:47.289000+03:00

Track When Exactly The Marker Changed

With window functions, it's possible to get the previous combination of attributes per business key in order to check whether something changed or not.

WITH cte_scd_columns AS (
  SELECT
    MD5(
      COALESCE(CAST(customer_id AS TEXT), '__null__') || COALESCE(CAST(country AS TEXT), '__null__') || COALESCE(CAST(email AS TEXT), '__null__') || COALESCE(CAST(tier AS TEXT), '__null__')
    ) AS scd_id,
    *
  FROM events
), cte_change_indicator AS (
  SELECT
    scd_id,
    COALESCE(scd_id <> LAG(scd_id) OVER (PARTITION BY customer_id ORDER BY timestamp), TRUE) AS change_indicator, /* coalesce to handle the first appearance of the entity */ /* as null <> [value] is NULL */
    *
  FROM cte_scd_columns
)
SELECT
  *
FROM cte_change_indicator
LIMIT 5

scd_id	change_indicator	scd_id_1	column0	customer_id	name	email	signup_date	country	timestamp	tier	interaction_type	interaction_value	delivered_at
ced54b448991710b857be121958a99a2	True	ced54b448991710b857be121958a99a2	3177	53	Sylvia Martinez	lauriecontreras@example.net	2024-09-25 00:00:00	Zambia	2025-05-21 13:01:56.840779	Platinum	purchase	363.9	2025-08-02 15:06:47.289000+03:00
ced54b448991710b857be121958a99a2	False	ced54b448991710b857be121958a99a2	3178	53	Sylvia Martinez	lauriecontreras@example.net	2024-09-25 00:00:00	Zambia	2025-05-25 06:26:37.840779	Platinum	support_ticket	nan	2025-08-02 15:06:47.289000+03:00
8088f8ba9d0c567eb22e8e090a78789e	True	8088f8ba9d0c567eb22e8e090a78789e	3179	53	Sylvia Martinez	lauriecontreras@example.net	2024-09-25 00:00:00	Zambia	2025-05-25 20:17:17.840779	Gold	login	nan	2025-08-02 15:06:47.289000+03:00
8088f8ba9d0c567eb22e8e090a78789e	False	8088f8ba9d0c567eb22e8e090a78789e	3180	53	Sylvia Martinez	lauriecontreras@example.net	2024-09-25 00:00:00	Zambia	2025-05-28 10:25:32.840779	Gold	purchase	362.82	2025-08-02 15:06:47.289000+03:00
8088f8ba9d0c567eb22e8e090a78789e	False	8088f8ba9d0c567eb22e8e090a78789e	3182	53	Sylvia Martinez	lauriecontreras@example.net	2024-09-25 00:00:00	Zambia	2025-05-29 02:13:36.840779	Gold	newsletter_click	nan	2025-08-02 15:06:47.289000+03:00

Check out the prev_scd_id column in the output. When it's NULL, that means it's the very first entry for that customer_id. When prev_scd_id is different from scd_id, that's our signal that one of the strictly tracked attributes has changed.

At this stage, we'll tweak this CTE a bit to add a boolean column to identify changes. This change_indicator will be TRUE when a tracked attribute combination is new or has changed, and FALSE otherwise. The COALESCE is a trick to make sure the very first entry for a customer_id is always marked as a change, since LAG() would return NULL for it.

Add Valid From Valid To

valid_from: This is the date when a new combination of attributes became active. It's pulled directly from the date column for rows where change_indicator is TRUE.
valid_to: This is the date when that specific combination of attributes stopped being valid. We use LEAD() here to peek at the date of the next change for the same customer_id, then subtract one day to mark the end of the current record's validity. If there's no next change (meaning it's the most current record), LEAD() returns NULL, which we'll handle in the final step. We only care about the rows where change_indicator is TRUE because those are the actual points in time when a relevant attribute shifted.

WITH cte_scd_columns AS (
  SELECT
    MD5(
      COALESCE(CAST(customer_id AS TEXT), '__null__') || COALESCE(CAST(country AS TEXT), '__null__') || COALESCE(CAST(email AS TEXT), '__null__') || COALESCE(CAST(tier AS TEXT), '__null__')
    ) AS scd_id,
    *
  FROM events
), cte_change_indicator AS (
  SELECT
    scd_id,
    COALESCE(scd_id <> LAG(scd_id) OVER (PARTITION BY customer_id ORDER BY timestamp), TRUE) AS change_indicator, /* coalesce to handle the first appearance of the entity */ /* as null <> [value] is NULL */
    *
  FROM cte_scd_columns
), cte_valid_from_valid_to AS (
  SELECT
    *,
    timestamp AS valid_from, /* when a change occurred */
    LEAD(timestamp) OVER (PARTITION BY customer_id ORDER BY timestamp) /* when the next change occurred */ - CAST('1 microsecond' AS INTERVAL) AS valid_to
  FROM (
    SELECT
      *
    FROM cte_change_indicator
    WHERE
      change_indicator IS TRUE
  )
)
SELECT
  *
FROM cte_valid_from_valid_to
LIMIT 5

scd_id	change_indicator	scd_id_1	column0	customer_id	name	email	signup_date	country	timestamp	tier	interaction_type	interaction_value	delivered_at	valid_from	valid_to
ff39bc29510492f665cae5e23bdaaa5a	True	ff39bc29510492f665cae5e23bdaaa5a	1272	21	Laura Sanchez	maguirre@example.org	2023-09-27 00:00:00	Jersey	2025-05-20 23:26:46.840779	Gold	support_ticket	nan	2025-08-02 15:06:47.289000+03:00	2025-05-20 23:26:46.840779	2025-05-27 06:23:35.840778
ad3df688bca726e92e47974ba0e384e8	True	ad3df688bca726e92e47974ba0e384e8	1277	21	Laura Sanchez	maguirre@example.org	2023-09-27 00:00:00	Jersey	2025-05-27 06:23:35.840779	Bronze	support_ticket	nan	2025-08-02 15:06:47.289000+03:00	2025-05-27 06:23:35.840779	2025-06-03 01:00:23.840778
ff39bc29510492f665cae5e23bdaaa5a	True	ff39bc29510492f665cae5e23bdaaa5a	1287	21	Laura Sanchez	maguirre@example.org	2023-09-27 00:00:00	Jersey	2025-06-03 01:00:23.840779	Gold	login	nan	2025-08-02 15:06:47.289000+03:00	2025-06-03 01:00:23.840779	2025-06-12 21:08:30.840778
dc5be9887d65c60946b791b6e3bbda49	True	dc5be9887d65c60946b791b6e3bbda49	1292	21	Laura Sanchez	maguirre@example.org	2023-09-27 00:00:00	Jersey	2025-06-12 21:08:30.840779	Platinum	purchase	51.82	2025-08-02 15:06:47.289000+03:00	2025-06-12 21:08:30.840779	2025-06-28 21:01:02.840778
ad3df688bca726e92e47974ba0e384e8	True	ad3df688bca726e92e47974ba0e384e8	1302	21	Laura Sanchez	maguirre@example.org	2023-09-27 00:00:00	Jersey	2025-06-28 21:01:02.840779	Bronze	support_ticket	nan	2025-08-02 15:06:47.289000+03:00	2025-06-28 21:01:02.840779	2025-06-30 21:34:29.840778

Final Touch

Observe the valid_from and valid_to columns. Each row now represents a specific period during which the country, email, and tier attributes were stable for that customer_id. Notice some valid_to values are NULL for now; that's because they represent the current active record.

WITH cte_scd_columns AS (
  SELECT
    MD5(
      COALESCE(CAST(customer_id AS TEXT), '__null__') || COALESCE(CAST(country AS TEXT), '__null__') || COALESCE(CAST(email AS TEXT), '__null__') || COALESCE(CAST(tier AS TEXT), '__null__')
    ) AS scd_id,
    *
  FROM events
), cte_change_indicator AS (
  SELECT
    scd_id,
    COALESCE(scd_id <> LAG(scd_id) OVER (PARTITION BY customer_id ORDER BY timestamp), TRUE) AS change_indicator, /* coalesce to handle the first appearance of the entity */ /* as null <> [value] is NULL */
    *
  FROM cte_scd_columns
), cte_valid_from_valid_to AS (
  SELECT
    *,
    timestamp AS valid_from, /* when a change occurred */
    LEAD(timestamp) OVER (PARTITION BY customer_id ORDER BY timestamp) /* when the next change occurred */ - CAST('1 microsecond' AS INTERVAL) AS valid_to
  FROM (
    SELECT
      *
    FROM cte_change_indicator
    WHERE
      change_indicator IS TRUE
  )
), cte_scd2_scd1 AS (
  SELECT
    MD5(scd_id || CAST(valid_from AS TEXT)) AS customer_sk,
    customer_id,
    name, /* scd1 attributes */
    signup_date,
    country, /* scd2 attributes */
    email,
    tier,
    valid_from,
    CAST(COALESCE(valid_to, '9999-12-31T23:59:59') AS TIMESTAMP) AS valid_to,
    valid_to IS NULL AS is_effective_row
  FROM cte_valid_from_valid_to
)
SELECT
  *
FROM cte_scd2_scd1
LIMIT 5

customer_sk	customer_id	name	signup_date	country	email	tier	valid_from	valid_to	is_effective_row
4aa26cc24e8bfeb92462e7b77ddd6e5e	12	Kyle Mcdonald	2025-04-24 00:00:00	Sao Tome and Principe	tasha01@example.net	Platinum	2025-05-21 06:41:00.840779	2025-05-25 20:46:32.840778	False
509016b089696afc5f7177a4a45d38c3	12	Kyle Mcdonald	2025-04-24 00:00:00	Sao Tome and Principe	tasha01@example.net	Bronze	2025-05-25 20:46:32.840779	2025-06-14 07:11:09.840778	False
e929d1ae5e0d2deabb9a5e7ae42d1219	53	Sylvia Martinez	2024-09-25 00:00:00	Zambia	lauriecontreras@example.net	Platinum	2025-05-21 13:01:56.840779	2025-05-25 20:17:17.840778	False
fbf1b35f2d65ef63d1f24541a6d18088	53	Sylvia Martinez	2024-09-25 00:00:00	Zambia	lauriecontreras@example.net	Gold	2025-05-25 20:17:17.840779	2025-06-24 17:19:57.840778	False
41ce1a3fe926554585bb70b5302e48f4	53	Sylvia Martinez	2024-09-25 00:00:00	Zambia	lauriecontreras@example.net	Bronze	2025-06-24 17:19:57.840779	9999-12-31 23:59:59	True

After creating the table, it's essential to merge it and use it in another fact table.
That process is beyond the scope of this article, but you can check it here

The Myth of Sisyphus in Data Engineering

Ahmad Muhammad — Fri, 27 Jun 2025 15:27:29 +0000

Camus's reading of The Myth of Sisyphus is not just about existentialism; it fits well in the world of engineering where the absence of an "ideal" solution is a constant reality.

Like in real life, suicide, as well as collapsing data infrastructure, is a rejection of the freedom we actually possess.

In Greek methodology, Sisyphus was tormented by the Gods, not by being shredded into pieces or by death, but by being forced to roll a boulder uphill just to watch it fall, eternally!

Albert Camus uses this simple story to describe the struggle with life in a chaotic world, with a lack of meaning, which Albert described as absurd.

The struggle itself toward the heights is enough to fill a man’s heart. One must imagine Sisyphus happy. — Albert Camus, The Myth of Sisyphus.

Camus invited us to imagine Sisyphus happy with his eternal fate simply by lucidly embracing it.

Lack Of Meaning and the Engineering Curse

A company without a clear data vision—no direction like increasing revenue, expanding its user base, or solving real human problems—is going blindly. Engineering is a tool. It must serve business goals, which ultimately should serve humanity. It is not meant to serve itself.

Just as the gods condemned Sisyphus to an eternal, pointless task to punish him, many engineering teams find themselves trapped in cycles of building for the sake of building. Endless refactoring. New frameworks. New infra. But no impact. No purpose.

Engineers have become divorced from meaning. We obsess over technical novelty while ignoring whether what we build actually matters. We drown in abstraction and forget the people we’re supposed to serve.

Data work is so absurd: every schema will mutate, every dashboard request is a moving target, and every source of truth will eventually lie. Let's imagine Sisyphus happy and a revolution in this absurd world.

Sisyphus Being Happy

Camus's imagination of Sisyphus's happiness comes when Sisyphus acknowledges the futility of his task and the certainty of his fate, because the struggle itself was enough to fill his heart. From Camus's perspective, the only reasonable response to the absurd is to revolt, admitting that everything will break, and we have the freedom to respond.

Camus's Happy Sisyphus as A Data Engineer

Camus’s Sisyphus finds meaning not in arrival, but in the ascent itself. In data engineering, each best practice becomes a revolt against chaos:

Sisyphus would build pipelines to fail

Instead of hoping it will just work, he would assume it won't. Sisyphus would revolt against schema drift, in addition to data quality issues, to see failures coming without illusion.

Sisyphus would use Git

The absurd man prefers the ambiguity of life; less knowledge is better than the full knowledge illusion.

Lucidity is the clarity and courage of mind which refuses all comforting illusions.

Git is a tool for Lucidity, it would provide a lucid history of the struggle, whenever the boulder fall, allowing Sisyphus to learn from every fall and to prove that he was in control of his response to it.

Sisyphus would document his struggle

Sisyphus would document his README.md revolution on data infrastructure absurdity for the sake of scorn of the fate he has been dealt.

The daily work of an engineer, the failing pipelines, the shifting requirements, the decaying sources of truth, can feel like a pointless, eternal task. It is our boulder.
We can be tormented by its weight, or we can follow Camus's Sisyphus and find liberation in the act of pushing.

Meaning is not found in a mythical "finished" state, but in the conscious and defiant struggle against chaos. It is in the resilient pipeline, the lucid commit history, and the clear documentation left for the next person to take up the climb. The struggle itself toward the heights is enough to fill an engineer's heart.
One must imagine the data engineer happy.

dbt Data Quality Audit, But on Steroid

Ahmad Muhammad — Thu, 26 Jun 2025 13:50:13 +0000

Data quality auditing is a process of ensuring that data is fit for a given purpose. Poor data quality results in wrong business decisions and might decline corporate performance.

some of these metrics are:

Metric	Description	Formula
Completeness	Percentage of complete (non-null) values in a column.	`number of non-null values / total rows`
Consistency	Percentage of values that follow a defined format (e.g., date, phone number).	`number of consistent values / total rows`
Data Timeliness	Describes how up-to-date or fresh the data is relative to the current time.	Context-dependent; may use timestamp difference
Uniqueness	Describes how many values are unique (non-duplicate) in a column.	`number of distinct values / total rows`

The Problem With dbt tests

In dbt, you get some out-of-the-box data tests, in addition to user-defined tests on data, and you get results along with severity by setting its behavior to error or warn and threshold.

However, this is not enough for a good auditing strategy, as we don't have the ability to store these failures historically, even when using --store-failures.

A test's results will always replace previous failures for the same test. according to dbt docs

But, Why to store errors historically at the first place?

Trend Analysis: Historical failures help identify persistent issues and measure improvement over time.
Root Cause Tracking: Recurring errors often stem from upstream source systems or processes; historical data enables RCA.
Audit & Compliance: In regulated environments, it's essential to track and prove how data quality evolved.
Business Impact Assessment: Helps correlate poor data quality with operational or financial consequences.

Without historical data audit, you are flying blindly!

🤗
The upcomming assumes familiarity with concepts such as data quality, dbt tests, and dbt macros.

Proposed Solution

Additional work is needed to move stored-failures data to a secure location and mark each batch with a timestamp (hence the term "historical").
Calculate some KPIs (Key Performance Indicators) based on this data, with detailed information at the column level.

Here, I will define a simple core table schema for the proposed table that will store our test data.

Field Name	Data Type	Description
batch_id	TEXT	Batch unique identifier
model_id	TEXT	The tested model
column_name	TEXT	The tested column
kpi_name	TEXT	Type of test was performed on this table
failed_rows	INT	Number of rows which failed the test
total_rows	INT	Number of total rows (including nulls)
success_pct	DECIMAL(5, 2)	Passed_rows * 100 / total_rows
failure_pct	DECIMAL(5, 2)	Failed_rows * 100 / total_rows
created_at	TIMESTAMP	The timestamp at which stored failures moved to this table

A good start, but it needs additional fields, such as a failed sample and the query that loaded this data! Let's add these extra fields to our table.

Field Name	Data Type	Description
failed_sample	JSON	100 rows Sample of the failed population.
query_text	TEXT[]	The query used to insert data into this table.

Despite failure tables being expected to follow the naming convention <test_name>_<model_name>_<column_name>, you might encounter names like audit__consistency_mart_order__dd46b64be0a4d8ce4b4eb322540fd7fd, which makes extracting this data challenging.

Fortunately, there is a way to extract these information from graph context directly inside dbt.

The graph context variable is a dictionary which maps node ids onto dictionary representations of those nodes. - dbt docs

🥴
Clone this repository for the demo code.

Some built-in tests in dbt don't return failed rows, so new tests need to be implemented. Note that every test starts with audit__, as this naming convention will be important later!

-- some test examples from my demo

-- ./macros/audit/consistency.sql
{% test audit__consistency(model, column_name, regex, case_sensitive=false) %}
    SELECT * FROM {{ model }}

    {% if case_sensitive %}
        {% set operator = '!~' %}
    {% else %}
        {% set operator = '!~*' %}
    {% endif %}

    WHERE {{ column_name }} {{operator}} '{{ regex }}'

{% endtest %}


-- ./macros/audit/greater_than_zero.sql
{% test audit__greater_than_zero(model, column_name) %}
    SELECT * FROM {{ model }}
    WHERE {{ column_name }} < 0
{% endtest %}



-- ./macros/audit/not_null.sql
{% test audit__not_null(model, column_name) %}
    SELECT * FROM {{ model }}
    WHERE {{ column_name }} IS NULL
{% endtest %}



-- ./macros/audit/unique.sql
{% test audit__unique(model, column_name) %}
    with duplicates as (
        select {{ column_name }}
        from {{ model }}
        group by {{ column_name }}
        having count(*) > 1
    )

    select *
    from {{ model }}
    where {{ column_name }} in (select {{ column_name }} from duplicates)
{% endtest %}

Exploring Graph Context

To access the graph context in dbt, all you need to do is call {{ graph }}. That's it! But what does it contain?

Let's begin by exploring the graph's top-level keys.

dbt compile --inline "{{ graph.keys() }}"        
17:00:27  Running with dbt=1.9.6
17:00:27  Registered adapter: postgres=1.9.0
17:00:28  Found 9 models, 3 seeds, 6 data tests, 1 sql operation, 440 macros
17:00:28  
17:00:28  Concurrency: 1 threads (target='dev')
17:00:28  
Compiled inline node is:
dict_keys(['exposures', 'groups', 'metrics', 'nodes', 'sources', 'semantic_models', 'saved_queries'])

Now, Let’s get a level deeper and check nodes keys, and surprise! We got all of our models, seeds, and tests

dbt compile --inline "{{ graph.nodes.keys() }}"
17:04:28  Running with dbt=1.9.6
17:04:29  Registered adapter: postgres=1.9.0
17:04:29  Found 9 models, 3 seeds, 6 data tests, 1 sql operation, 440 macros
17:04:29  
17:04:29  Concurrency: 1 threads (target='dev')
17:04:29  
Compiled inline node is:
dict_keys(['model.audit_on_steroid.stg_orders', 'model.audit_on_steroid.stg_products', 'model.audit_on_steroid.stg_customers', 'model.audit_on_steroid.int_products', 'model.audit_on_steroid.int_order_customer_joined', 'model.audit_on_steroid.int_customers', 'model.audit_on_steroid.int_orders', 'model.audit_on_steroid.mart_order_summary', 'seed.audit_on_steroid.raw_orders', 'seed.audit_on_steroid.raw_products', 'seed.audit_on_steroid.raw_customers', 'model.audit_on_steroid.mart_order_summary_wap', 'test.audit_on_steroid.audit__not_null_mart_order_summary_wap_customer_id.7adbd28cc6', 'test.audit_on_steroid.audit__unique_mart_order_summary_wap_customer_id.bb8cd4e984', 'test.audit_on_steroid.audit__not_null_mart_order_summary_wap_total_spent.d0468db7f7', 'test.audit_on_steroid.audit__greater_than_zero_mart_order_summary_wap_total_spent.15fc2665f0', 'test.audit_on_steroid.audit__not_null_mart_order_summary_wap_first_order.22b1d319a6', 'test.audit_on_steroid.audit__consistency_mart_order_summary_wap_email___A_Za_z0_9___A_Za_z0_9_A_Za_z_.5e07113309'])

Let’s pickup a test and see what it includes:

dbt compile --inline "{{ graph.nodes.get('test.audit_on_steroid.audit__greater_than_zero_mart_order_summary_wap_total_spent.15fc2665f0') }}"

{
    "database": "data_warehouse",
    "schema": "prefex__dbt_test__audit",
    "name": "audit__greater_than_zero_mart_order_summary_wap_total_spent",
    "resource_type": "test",
    "package_name": "audit_on_steroid",
    "path": "audit__greater_than_zero_mart_order_summary_wap_total_spent.sql",
    "original_file_path": "models/mart/_mart_order_summary.yml",
    "unique_id": "test.audit_on_steroid.audit__greater_than_zero_mart_order_summary_wap_total_spent.15fc2665f0",
    "fqn": [
        "audit_on_steroid",
        "mart",
        "audit__greater_than_zero_mart_order_summary_wap_total_spent",
    ],
    "alias": "audit__greater_than_zero_mart_order_summary_wap_total_spent",
    "checksum": {"name": "none", "checksum": ""},
    "config": {
        "enabled": True,
        "alias": None,
        "schema": "dbt_test__audit",
        "database": None,
        "tags": [],
        "meta": {},
        "group": None,
        "materialized": "test",
        "severity": "warn",
        "store_failures": True,
        "store_failures_as": "table",
        "where": None,
        "limit": None,
        "fail_calc": "count(*)",
        "warn_if": "!= 0",
        "error_if": "!= 0",
    },
    "tags": [],
    "description": "",
    "columns": {},
    "meta": {},
    "group": None,
    "docs": {"show": True, "node_color": None},
    "patch_path": None,
    "build_path": None,
    "unrendered_config": {"store_failures": True, "severity": "warn"},
    "created_at": 1749301264.3145826,
    "config_call_dict": {},
    "unrendered_config_call_dict": {},
    "relation_name": '"data_warehouse"."prefex__dbt_test__audit"."audit__greater_than_zero_mart_order_summary_wap_total_spent"',
    "raw_code": "{{ test_audit__greater_than_zero(**_dbt_generic_test_kwargs) }}",
    "language": "sql",
    "refs": [{"name": "mart_order_summary_wap", "package": None, "version": None}],
    "sources": [],
    "metrics": [],
    "depends_on": {
        "macros": [
            "macro.audit_on_steroid.test_audit__greater_than_zero",
            "macro.dbt.get_where_subquery",
        ],
        "nodes": ["model.audit_on_steroid.mart_order_summary_wap"],
    },
    "compiled_path": None,
    "contract": {"enforced": False, "alias_types": True, "checksum": None},
    "column_name": "total_spent",
    "file_key_name": "models.mart_order_summary_wap",
    "attached_node": "model.audit_on_steroid.mart_order_summary_wap",
    "test_metadata": {
        "name": "audit__greater_than_zero",
        "kwargs": {
            "column_name": "total_spent",
            "model": "{{ get_where_subquery(ref('mart_order_summary_wap')) }}",
        },
        "namespace": None,
    },
}

That’s awesome! you have the access to all related metadata to any dbt resource.

But in our case, the interest is drifted to this section of the previous dictionary

by extracting tests from graph , it’s possible to get the model which it tests in addition to the column it tests.

    "column_name": "total_spent",
    "file_key_name": "models.mart_order_summary_wap",
    "attached_node": "model.audit_on_steroid.mart_order_summary_wap",

Utilize Graph Context in Data Quality Auditing Automation.

Collect Test Nodes

Since every node has resource_type property, it’s possible to filter nodes based on type, then extract needed metadata from it.

{% macro __get_test_metadata() %}
    {% set test_metadata = [] %}
    {% if execute %}
        {% for test_node in graph.nodes.values()
            | selectattr("resource_type", "equalto", "test") %}
            {% set model_node = graph.nodes.get(test_node.attached_node) %}
            {% if model_node and
                test_node.test_metadata.name and
                test_node.test_metadata.name.startswith('audit_') %}
                {% do test_metadata.append({
                'test_name': test_node.test_metadata.name,
                'column_name': test_node.column_name | default(''),
                'test_relation': test_node.relation_name,
                'model_relation': model_node.relation_name,
                'model_id': model_node.unique_id
                }) %}
            {% endif %}
        {% endfor %}
    {% endif %}
    {{ return(test_metadata) }}
{% endmacro %}

Let’s compile __get_test_metadata() macro to see it’s result

dbt compile --inline "{{ __get_test_metadata() }}"

[
    {
        "test_name": "audit__not_null",
        "column_name": "customer_id",
        "test_relation": '"data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_customer_id"',
        "model_relation": '"data_warehouse"."prefex__mart"."mart_order_summary_wap"',
        "model_id": "model.audit_on_steroid.mart_order_summary_wap",
    },
    {
        "test_name": "audit__unique",
        "column_name": "customer_id",
        "test_relation": '"data_warehouse"."prefex__dbt_test__audit"."audit__unique_mart_order_summary_wap_customer_id"',
        "model_relation": '"data_warehouse"."prefex__mart"."mart_order_summary_wap"',
        "model_id": "model.audit_on_steroid.mart_order_summary_wap",
    },
    {
        "test_name": "audit__not_null",
        "column_name": "total_spent",
        "test_relation": '"data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_total_spent"',
        "model_relation": '"data_warehouse"."prefex__mart"."mart_order_summary_wap"',
        "model_id": "model.audit_on_steroid.mart_order_summary_wap",
    },
    {
        "test_name": "audit__greater_than_zero",
        "column_name": "total_spent",
        "test_relation": '"data_warehouse"."prefex__dbt_test__audit"."audit__greater_than_zero_mart_order_summary_wap_total_spent"',
        "model_relation": '"data_warehouse"."prefex__mart"."mart_order_summary_wap"',
        "model_id": "model.audit_on_steroid.mart_order_summary_wap",
    },
    {
        "test_name": "audit__not_null",
        "column_name": "first_order",
        "test_relation": '"data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_first_order"',
        "model_relation": '"data_warehouse"."prefex__mart"."mart_order_summary_wap"',
        "model_id": "model.audit_on_steroid.mart_order_summary_wap",
    },
    {
        "test_name": "audit__consistency",
        "column_name": "email",
        "test_relation": '"data_warehouse"."prefex__dbt_test__audit"."audit__consistency_mart_order__dd46b64be0a4d8ce4b4eb322540fd7fd"',
        "model_relation": '"data_warehouse"."prefex__mart"."mart_order_summary_wap"',
        "model_id": "model.audit_on_steroid.mart_order_summary_wap",
    },
]

Constructing The Audit Table

In the previous array of dictionaries, each element can represent a select statement that will be inserted into the audit table.

🤯
A batch ID can be represented as a string that uniquely identifies a batch. This could include formats such as timestamps, integer UUIDs, or dbt's invocation_id, among others.

💡
sql_text field value will be injected after constructing the query in jinja micro

SELECT 
    '{{ batch_id }}'::TEXT as batch_id,
    '{{ test.model_id }}'::TEXT as model_id,
    '{{ test.model_relation }}'::TEXT as relation_name,
    '{{ test.column_name }}'::TEXT as column_name,
    '{{ test.test_name }}'::TEXT as kpi_name,
    ft.failed_count::INT as failed_rows,
    tt.total_count::INT as total_rows,

    CASE 
        WHEN tt.total_count = 0 THEN 0 
        ELSE ROUND((ft.failed_count * 100.0) / tt.total_count, 2)
    END as failure_pct,

    CASE 
        WHEN tt.total_count = 0 THEN 100 
        ELSE ROUND(100 - (ft.failed_count * 100.0) / tt.total_count, 2)
    END as success_pct,

    fs.failed_sample,

    CURRENT_TIMESTAMP as created_at
FROM 
    -- Failed rows count for this specific batch
    (
        SELECT COUNT(*) as failed_count
        FROM {{ test.test_relation }}
        WHERE {{ batch_id_column }} = '{{ batch_id }}'
    ) ft,

    -- Total rows count for this batch from the model
    (
        SELECT COUNT(*) as total_count 
        FROM {{ test.model_relation }}
        WHERE {{ batch_id_column }} = '{{ batch_id }}'
    ) tt,

    -- Sample of failed records for this batch
    (
        SELECT JSON_AGG(sample_data) as failed_sample
        FROM (
            SELECT row_to_json(t) AS sample_data
            FROM {{ test.test_relation }} t
            WHERE t.{{ batch_id_column }} = '{{ batch_id }}'
            LIMIT 100
        ) sample
    ) AS fs

The next step is to extract batches based on a given column that could be a batch identifier.

Test results will always be replaced after running the test, but:

If batch could pass the test, the stored failures table will not be emptied or replaced.
So, to avoid duplicate records in our historical audit table, it's crucial to verify if a given batch_id for a specific test and model has already been recorded.

SELECT DISTINCT 
    tr.{{ batch_id_column }}
FROM 
    {{ test.test_relation }} tr -- where stored failures occure
WHERE NOT EXISTS (
    SELECT 1
    FROM {{ audit_schema }}.{{ audit_table_name }} ar
    WHERE ar.batch_id = tr.{{ batch_id_column }}::TEXT
        AND ar.model_id = '{{ test.model_id }}'
        AND ar.kpi_name = '{{ test.test_name }}'
        AND ar.column_name = '{{ test.column_name }}'
)
ORDER BY tr.{{ batch_id_column }}

The workflow will look similar to this flowchart.

flowchart TD     
A[Start macro: audit__post_hook] --> B[Create audit schema and table if not exists]     
B --> C[Call macro: __get_test_metadata]     
C --> D[Loop through each test in metadata]     
D --> E{test_relation & model_relation exist?}     
E -- No --> D     
E -- Yes --> F[Run query to find unprocessed batch_ids]     
F --> G[Log found batch_ids]      
G --> H[Get first and only batch_id]     
H --> I[Build SQL to calculate KPIs and failed sample]     
I --> J[Wrap SQL with query_text and add to audit_select_queries]                          
J --> M[Construct INSERT INTO audit_report]     
M --> L{More tests}
L -- yes --> D
L -- no --> N[Run audit_insert_query]     
N --> O[Log insert query and result status]     
O --> P[End]

Demo

Setup The Project

docker compose up -d # run the postgres container
cd audit_on_steroid
dbt seed --profiles-dir . # this loads sample data into the db

Exploring The Demo

In our models directory, you'll find the three familiar layers: stg, int, and mart. Our goal is to use the Write Audit Publish (WAP) pattern. Check out the model mart_order_summary_wap.sql, which will store our audited data before it's published to production.

❯ tree ./models                                                 
./models
├── int
│   ├── int_customers.sql
│   ├── int_order_customer_joined.sql
│   ├── int_orders.sql
│   └── int_products.sql
├── mart
│   ├── mart_order_summary.sql
│   ├── mart_order_summary_wap.sql
│   └── _mart_order_summary.yml
└── stg
    ├── stg_customers.sql
    ├── stg_orders.sql
    └── stg_products.sql

3 directories, 10 files

Let’s check how _mart_order_summary.yml was written. Remember the goal of using custom generic tests is to return the entire failed row instead of failures summary.

version: 2

models:
  - name: mart_order_summary_wap
    config:
      tags: ['wap'] # Notice the tag here!
    description: "Customer-level order aggregation WAP table"
    columns:
      - name: customer_id
        data_tests: [audit__not_null, audit__unique]

      - name: total_spent
        data_tests: [audit__not_null, audit__greater_than_zero]

      - name: first_order
        data_tests:
          - audit__not_null

      - name: email
        data_tests: 
          - audit__consistency:
              regex: "^[A-Za-z0-9._%-]+@[A-Za-z0-9.-]+[.][A-Za-z]+$"

Now, it’s time to run the workflow!

dbt run -s +tag:wap && \ # run wap models upstream -including wap model-
dbt test -s tag:wap && \ # run tests step
dbt run-operation audit__post_hook \ # move test results to our audit table!
--args '{audit_schema: "audit__reports", audit_table_name: "audit_report", batch_id_column: "created_at"}' && \
dbt run -s tag:wap+ --exclude tag:wap # publish data to prod table

This table is what should be expected in the mart_order_summary_wap table.

customer_id	customer_name	email	total_orders	first_order	last_order	created_at
101	ali	invalid_email.com	1	2024-06-01	2024-06-01	2025-06-08 16:47:13.171396+00
102	sara	sara@example.com	1			2025-06-08 16:47:13.171396+00
103	ali	ali@example.com	1	2024-06-03	2024-06-03	2025-06-08 16:47:13.171396+00
104			1			2025-06-08 16:47:13.171396+00

In this demo, the severity was set to 'warn' for demonstration purposes, ensuring that it does not break the pipeline. You can experiment with it to establish data quality rules and observe the results. Let’s check tables created by the run:

❯ docker ps 
CONTAINER ID   IMAGE      COMMAND                  CREATED        STATUS       PORTS                                         NAMES
d49fd5fbe329   postgres   "docker-entrypoint.s…"   27 hours ago   Up 5 hours   0.0.0.0:5432->5432/tcp, [::]:5432->5432/tcp   audit_on_steroid-db-1

❯ docker exec -it audit_on_steroid-db-1 psql -U postgres -d data_warehouse   
psql (17.5 (Debian 17.5-1.pgdg120+1))
Type "help" for help.

data_warehouse=# SELECT * FROM pg_catalog.pg_tables where tablename not like 'pg_%';
       schemaname        |                            tablename                            | tableowner | tablespace | hasindexes | hasrules | hastriggers | rowsecurity 
-------------------------+-----------------------------------------------------------------+------------+------------+------------+----------+-------------+-------------
 prefex__stg             | stg_orders                                                      | postgres   |            | f          | f        | f           | f
 prefex_                 | raw_customers                                                   | postgres   |            | f          | f        | f           | f
 prefex__mart            | mart_order_summary_wap                                          | postgres   |            | f          | f        | f           | f
 prefex__dbt_test__audit | audit__not_null_mart_order_summary_wap_first_order              | postgres   |            | f          | f        | f           | f
 prefex_                 | raw_orders                                                      | postgres   |            | f          | f        | f           | f
 prefex__int             | int_customers                                                   | postgres   |            | f          | f        | f           | f
 prefex__dbt_test__audit | audit__consistency_mart_order__dd46b64be0a4d8ce4b4eb322540fd7fd | postgres   |            | f          | f        | f           | f
 prefex__dbt_test__audit | audit__not_null_mart_order_summary_wap_total_spent              | postgres   |            | f          | f        | f           | f
 prefex_                 | raw_products                                                    | postgres   |            | f          | f        | f           | f
 prefex__int             | int_orders                                                      | postgres   |            | f          | f        | f           | f
 prefex__dbt_test__audit | audit__greater_than_zero_mart_order_summary_wap_total_spent     | postgres   |            | f          | f        | f           | f
 prefex__dbt_test__audit | audit__unique_mart_order_summary_wap_customer_id                | postgres   |            | f          | f        | f           | f
 prefex__mart            | mart_order_summary                                              | postgres   |            | f          | f        | f           | f
 prefex__stg             | stg_customers                                                   | postgres   |            | f          | f        | f           | f
 prefex__int             | int_order_customer_joined                                       | postgres   |            | f          | f        | f           | f
 prefex__dbt_test__audit | audit__not_null_mart_order_summary_wap_customer_id              | postgres   |            | f          | f        | f           | f
 information_schema      | sql_parts                                                       | postgres   |            | f          | f        | f           | f
 information_schema      | sql_features                                                    | postgres   |            | f          | f        | f           | f
 information_schema      | sql_implementation_info                                         | postgres   |            | f          | f        | f           | f
 information_schema      | sql_sizing                                                      | postgres   |            | f          | f        | f           | f
 audit__reports          | audit_report                                                    | postgres   |            | f          | f        | f           | f

Let's examine the "audit__reports"."audit_report" table we have created.

batch_id	model_id	relation_name	column_name	kpi_name	failed_rows	total_rows	failure_pct	success_pct	failed_sample	query_text	created_at
2025-06-08 16:47:13.171396+00:00	model.audit_on_steroid.mart_order_summary_wap	data_warehouse."prefex__mart"."mart_order_summary_wap"	total_spent	audit__not_null	4	4	100	0	[{"customer_id":"101","customer_name":"ali","email":"invalid_email.com","total_orders":1,"total_spent":null,"first_order":"2024-06-01","last_order":"2024-06-01","created_at":"2025-06-08T16:47:13.171396+00:00"}, {"customer_id":"102","customer_name":"sara","email":"sara@example.com","total_orders":1,"total_spent":null,"first_order":null,"last_order":null,"created_at":"2025-06-08T16:47:13.171396+00:00"}, {"customer_id":"103","customer_name":"ali","email":"ali@example.com","total_orders":1,"total_spent":null,"first_order":"2024-06-03","last_order":"2024-06-03","created_at":"2025-06-08T16:47:13.171396+00:00"}, {"customer_id":"104","customer_name":null,"email":null,"total_orders":1,"total_spent":null,"first_order":null,"last_order":null,"created_at":"2025-06-08T16:47:13.171396+00:00"}]	{  SELECT   '2025-06-08 16:47:13.171396+00:00'::TEXT as batch_id,  'model.audit_on_steroid.mart_order_summary_wap'::TEXT as model_id,  '"data_warehouse"."prefex__mart"."mart_order_summary_wap"'::TEXT as relation_name,  'total_spent'::TEXT as column_name,  'audit__not_null'::TEXT as kpi_name,  ft.failed_count::INT as failed_rows,  tt.total_count::INT as total_rows,    CASE   WHEN tt.total_count = 0 THEN 0   ELSE ROUND((ft.failed_count 100.0) / tt.total_count, 2)  END as failure_pct,    CASE   WHEN tt.total_count = 0 THEN 100   ELSE ROUND(100 - (ft.failed_count 100.0) / tt.total_count, 2)  END as success_pct,    fs.failed_sample,  CURRENT_TIMESTAMP as dbt_created_at  FROM     (  SELECT COUNT() as failed_count  FROM "data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_total_spent"  WHERE created_at = '2025-06-08 16:47:13.171396+00:00'  ) ft,      (  SELECT COUNT() as total_count   FROM "data_warehouse"."prefex__mart"."mart_order_summary_wap"  WHERE created_at = '2025-06-08 16:47:13.171396+00:00'  ) tt,      (  SELECT JSON_AGG(sample_data) as failed_sample  FROM (  SELECT row_to_json(t) AS sample_data  FROM "data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_total_spent" t  WHERE t.created_at = '2025-06-08 16:47:13.171396+00:00'  LIMIT 100  ) sample  ) AS fs }	2025-06-08 16:47:21
2025-06-08 16:47:13.171396+00:00	model.audit_on_steroid.mart_order_summary_wap	data_warehouse."prefex__mart"."mart_order_summary_wap"	first_order	audit__not_null	2	4	50	50	[{"customer_id":"102","customer_name":"sara","email":"sara@example.com","total_orders":1,"total_spent":null,"first_order":null,"last_order":null,"created_at":"2025-06-08T16:47:13.171396+00:00"}, {"customer_id":"104","customer_name":null,"email":null,"total_orders":1,"total_spent":null,"first_order":null,"last_order":null,"created_at":"2025-06-08T16:47:13.171396+00:00"}]	{  SELECT   '2025-06-08 16:47:13.171396+00:00'::TEXT as batch_id,  'model.audit_on_steroid.mart_order_summary_wap'::TEXT as model_id,  '"data_warehouse"."prefex__mart"."mart_order_summary_wap"'::TEXT as relation_name,  'first_order'::TEXT as column_name,  'audit__not_null'::TEXT as kpi_name,  ft.failed_count::INT as failed_rows,  tt.total_count::INT as total_rows,    CASE   WHEN tt.total_count = 0 THEN 0   ELSE ROUND((ft.failed_count 100.0) / tt.total_count, 2)  END as failure_pct,    CASE   WHEN tt.total_count = 0 THEN 100   ELSE ROUND(100 - (ft.failed_count 100.0) / tt.total_count, 2)  END as success_pct,    fs.failed_sample,  CURRENT_TIMESTAMP as dbt_created_at  FROM     (  SELECT COUNT() as failed_count  FROM "data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_first_order"  WHERE created_at = '2025-06-08 16:47:13.171396+00:00'  ) ft,      (  SELECT COUNT() as total_count   FROM "data_warehouse"."prefex__mart"."mart_order_summary_wap"  WHERE created_at = '2025-06-08 16:47:13.171396+00:00'  ) tt,      (  SELECT JSON_AGG(sample_data) as failed_sample  FROM (  SELECT row_to_json(t) AS sample_data  FROM "data_warehouse"."prefex__dbt_test__audit"."audit__not_null_mart_order_summary_wap_first_order" t  WHERE t.created_at = '2025-06-08 16:47:13.171396+00:00'  LIMIT 100  ) sample  ) AS fs }	2025-06-08 16:47:21
2025-06-08 16:47:13.171396+00:00	model.audit_on_steroid.mart_order_summary_wap	data_warehouse."prefex__mart"."mart_order_summary_wap"	email	audit__consistency	1	4	25	75	[{"customer_id":"101","customer_name":"ali","email":"invalid_email.com","total_orders":1,"total_spent":null,"first_order":"2024-06-01","last_order":"2024-06-01","created_at":"2025-06-08T16:47:13.171396+00:00"}]	{  SELECT   '2025-06-08 16:47:13.171396+00:00'::TEXT as batch_id,  'model.audit_on_steroid.mart_order_summary_wap'::TEXT as model_id,  '"data_warehouse"."prefex__mart"."mart_order_summary_wap"'::TEXT as relation_name,  'email'::TEXT as column_name,  'audit__consistency'::TEXT as kpi_name,  ft.failed_count::INT as failed_rows,  tt.total_count::INT as total_rows,    CASE   WHEN tt.total_count = 0 THEN 0   ELSE ROUND((ft.failed_count 100.0) / tt.total_count, 2)  END as failure_pct,    CASE   WHEN tt.total_count = 0 THEN 100   ELSE ROUND(100 - (ft.failed_count 100.0) / tt.total_count, 2)  END as success_pct,    fs.failed_sample,  CURRENT_TIMESTAMP as dbt_created_at  FROM     (  SELECT COUNT() as failed_count  FROM "data_warehouse"."prefex__dbt_test__audit"."audit__consistency_mart_order__dd46b64be0a4d8ce4b4eb322540fd7fd"  WHERE created_at = '2025-06-08 16:47:13.171396+00:00'  ) ft,      (  SELECT COUNT() as total_count   FROM "data_warehouse"."prefex__mart"."mart_order_summary_wap"  WHERE created_at = '2025-06-08 16:47:13.171396+00:00'  ) tt,      (  SELECT JSON_AGG(sample_data) as failed_sample  FROM (  SELECT row_to_json(t) AS sample_data  FROM "data_warehouse"."prefex__dbt_test__audit"."audit__consistency_mart_order__dd46b64be0a4d8ce4b4eb322540fd7fd" t  WHERE t.created_at = '2025-06-08 16:47:13.171396+00:00'  LIMIT 100  ) sample  ) AS fs }	2025-06-08 16:47:21

What the Heck is Data Build Tool (dbt)

Ahmad Muhammad — Thu, 26 Jun 2025 13:41:44 +0000

When I first started with dbt, I couldn't understand the idea behind running SQL on top of {Jinja} templates or why they would reinvent the wheel when we already have Python, Jinja, and other tools to orchestrate SQL files.

In reality, dbt is designed to offload many low-level complexities, allowing us to focus solely on modeling and understanding business transformations. It also encapsulates best practices, such as: testing, documentation, and version control without worrying about the intricacies of SQL execution across different platforms.

According to dbt documentations, it claims that dbt optimize the engineering workflows by:

Avoid writing boilerplate DML and DDL by managing transactions, dropping tables, and managing schema changes. Write business logic with just a SQL select statement, or a Python DataFrame, that returns the dataset you need, and dbt takes care of materialization.
Build up reusable, or modular, data models that can be referenced in subsequent work instead of starting at the raw data with every analysis.
Dramatically reduce the time your queries take to run: Leverage metadata to find long-running models that you want to optimize and use incremental models which dbt makes easy to configure and use.
Write DRYer code by leveraging macros, hooks, and package management.

What is dbt

Data Build Tool (dbt) allows teams to develop and manage code-based transformations in SQL and Python. It centralizes business logic, provides modular analytics components, and offers open-source flexibility.

dbt was developed to encapsulate and partially automate best practices such as testing, documentation, and version control.

├── macros  
└── cent_to_dollars.sql  
├── models  
│ ├── intermediate  
│ │ └── finance  
│ │ ├── _int_finance__models.yml  
│ │ └── int_payments_pivoted_to_orders.sql  
│ ├── marts  
│ │ ├── finance  
│ │ │ ├── _finance__models.yml  
│ │ │ ├── orders.sql  
│ │ │ └── payments.sql  
│ │ └── marketing  
│ │ ├── _marketing__models.yml  
│ │ └── customers.sql  
│ ├── staging  
│ │ ├── jaffle_shop  
│ │ │ ├── _jaffle_shop__docs.md  
│ │ │ ├── _jaffle_shop__models.yml  
│ │ │ ├── _jaffle_shop__sources.yml  
│ │ │ ├── base  
│ │ │ │ ├── base_jaffle_shop__customers.sql  
│ │ │ │ └── base_jaffle_shop__deleted_customers.sql  
│ │ │ ├── stg_jaffle_shop__customers.sql  
│ │ │ └── stg_jaffle_shop__orders.sql  
│ │ └── stripe  
│ │ ├── _stripe__models.yml  
│ │ ├── _stripe__sources.yml  
│ │ └── stg_stripe__payments.sql  
│ └── utilities  
│ └── all_dates.sql
├── packages.yml  
├── snapshots  
└── tests  
└── assert_positive_value_for_total_amount.sql

dbt Models

A dbt model is a Jinja augmented SQL files inside models directory that defines desired transformations -typically a SELECT statement. which will automatically built into a complete regular SQL files that materialize it as:

Table.
View.
Incremental, where each select output will be inserted into the model.
Ephemeral, where your query will be used as a Common Table Expression (CTE), enhancing modularity across your project while keeping your warehouse clean.
Materialized view.

In other words, a dbt model can be:

A table (either standard or incremental)
A view
A materialized view
A Common Table Expression (CTE)

In this example from dbt docs, the customers model depends on stg_customers and stg_orders models to be existed.

-- ./models/customers.sql
{{ config(
materialized="view",
schema="marketing"
) }}

with customers as (

select * from {{ ref('stg_customers') }}

),

orders as (

select * from {{ ref('stg_orders') }}

),
customer_orders as (
select
customer_id,
min(order_date) as first_order_date,
max(order_date) as most_recent_order_date,
count(order_id) as number_of_orders

from orders

group by 1
)

select
customers.customer_id,
customers.first_name,
customers.last_name,
customer_orders.first_order_date,
customer_orders.most_recent_order_date,
coalesce(customer_orders.number_of_orders, 0) as number_of_orders

from customers

left join customer_orders using (customer_id)

This code is compiled into a CREATE statement (if the table or view does not exist). In the case of an incremental model, it is compiled into an INSERT statement.

create view analytics.customers as (
with customers as (

select * from analytics.stg_customers

),

orders as (

select * from analytics.stg_orders

),

...
)

...

dbt configuration can be defined either within the SQL file or externally in a .yml file:

name: jaffle_shop
config-version: 2
...

models:
jaffle_shop: # this matches the `name:`` config
+materialized: view # this applies to all models in the current project
marts:
+materialized: table # this applies to all models in the `marts/` directory
marketing:
+schema: marketing # this applies to all models in the `marts/marketing/`` directory
+materialized: view

dbt Modularity

Don't Repeat Yourself (DRY) is a software engineering principle that ensures complexity reduction, maintainability, and systems' scalability.

"every piece of knowledge must have a single, unambiguous, authoritative representation within a system" - Andy Hunt and Dave Thomas, The Pragmatic Programmer

dbt integrates Jinja with SQL code, addressing standard SQL’s limitations.

Jinja

Jinja is a templating language for python developers that is very close to python syntax, it is often used to dynamically generate SQL queries using known string manipulation techniques. in this example, we will use Jinja to generate this query:

-- /models/order_payment_method_amounts.sql
select
order_id,
sum(case when payment_method = 'bank_transfer' then amount end) as bank_transfer_amount,
sum(case when payment_method = 'credit_card' then amount end) as credit_card_amount,
sum(case when payment_method = 'gift_card' then amount end) as gift_card_amount,
sum(amount) as total_amount
from app_data.payments
group by 1

Notice how Jinja have reduced redundancy and maintainability, it's easy to add, remove, and modify your select statement only by modifying payment_methods variable

-- /models/order_payment_method_amounts.sql
{% set payment_methods = ["bank_transfer", "credit_card", "gift_card"] %}

select
order_id,
{% for payment_method in payment_methods %}
sum(case when payment_method = '{{payment_method}}' then amount end) as {{payment_method}}_amount,
{% endfor %}
sum(amount) as total_amount
from app_data.payments
group by 1

Macros

Macros are reusable blocks of code that analogous to functions in other programming languages, they are defined in .sql files.

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, scale=2) %}
({{ column_name }} / 100)::numeric(16, {{ scale }})
{% endmacro %}

-- models/stg_payments.sql
select
id as payment_id,
{{ cents_to_dollars('amount') }} as amount_usd,
...
from app_data.payments

-- compiled query
select
id as payment_id,
(amount / 100)::numeric(16, 2) as amount_usd,
...
from app_data.payments

Contracts

For some models, a change in the upstream system's schema is a big deal, as it might break downstream models and systems. In such a case, it's better to define a set of rules that establish the model's shape, including column names, column types, and constraints. This ensures that the model transformations produce the expected model shape.

models:
  - name: dim_customers
    config:
      materialized: table
      contract:
        enforced: true
    columns:
      - name: customer_id
        data_type: int
        constraints:
          - type: not_null
      - name: customer_name
        data_type: string
      - name: non_integer
        data_type: numeric(38,3)

In incremental models, on_schema_change parameter can be used to determine the action of schema change, available options are: ignore, fail, append_new_columns, sync_all_columns.

{{
config(
materialized='incremental',
unique_key='date_day',
on_schema_change='fail'
)
}}

Model Dependency Management

Model dependencies are defined using the ref and source functions inside the model file, allowing dbt to create a Directed Acyclic Graph (DAG) to determine:

The execution order of models.
The lineage diagram in dbt's auto-generated documentation.

source Function

The source function is used to extract data from source databases. It looks similar to ref, but it is specifically for sources.

Sources are defined in YAML (.yml) files because dbt connects to these sources and ingests data from them.

# ./models/<file name>.yml
version: 2

sources:
- name: jaffle_shop # this is the source name
database: raw

tables:
- name: customers # this is the table name
- name: orders

customers source can be ingested into stg_customers model with this query.

-- ./models/stg_customers.sql
select * from {{ source("jaffle_shop", "customers") }}

ref Function

The ref function is used to reference a model in another model. Models are stacked on each other in all data engineering projects.

-- ./models/customers.sql
with customers as (

select * from {{ ref('stg_customers') }}

),

orders as (

select * from {{ ref('stg_orders') }}

),
...

The expected DAG (or lineage diagram) from this model looks like this. This makes it easy to define the execution order of models without any doubts.

source: https://docs.getdbt.com/docs/build/sql-models

Note that dbt quickly identifies circular dependencies in models, thanks ref function!

Model Versioning

Data contracts change over time—this could be renaming a column, adding or removing a column, or even changing data types. The idea is to force every downstream consumer to handle these breaking changes as soon as they’re deployed to production. By default, dbt uses the latest version in downstream models, making migration to the new upstream model easier. It’s also possible to reference a specific model version using the ref function, for example:

with dim_customers AS (
SELECT col1, col2, col3 FROM {{ ref('dim_customers', v=1) }} -- choses a specific version. file name: dim_customers_v1.sql
)
,
with dim_orders AS (
SELECT col1, col2, col3 FROM {{ ref('dim_orders') }} -- choses the latest version, file name: dim_orders_v<X>.sql .. X is an integer represents version number
)
...

Some changes like adding a column, or fixing calculation bugs don't require creating a new model, other breaking changes like column removal, column renaming, or changing datatypes do require model versioning.

Creating Model Version

inside a YML file inside models directory, write down your models specifications, note how contract enforcing option is marked as true . Note that file names must be versioned, like: dim_customers_v1.sql, and dim_customers_v2.sql

models:
  - name: dim_customers
    latest_version: 1

    # declare the versions, and fully specify them
    versions:
      - v: 2
        config:
          materialized: table
          contract: {enforced: true}
        columns:
          - name: customer_id
            description: This is the primary key
            data_type: int
          # no country_name column

      - v: 1
        config:
          materialized: table
          contract: {enforced: true}
        columns:
          - name: customer_id
            description: This is the primary key
            data_type: int
          - name: country_name
            description: Where this customer lives
            data_type: varchar

Tests

dbt makes it damn simple to ensure your data models are doing what they're supposed to by letting you integrate tests right into your workflow. Whether you lean on built-in schema tests or craft your own custom ones, these tests enforce data quality and catch anomalies early in the transformation process.

Singular Data Test

The easiest way to write a data test is to simply create a SQL query that returns records breaking your rules. We call these "singular" tests—one-off assertions aimed at a single purpose. For example:

-- tests/assert_total_payment_amount_is_positive.sql
-- Refunds have a negative amount, so total amount should always be >= 0.
-- Return records where total_amount < 0 to fail the test.
select
    order_id,
    sum(amount) as total_amount
from {{ ref('fct_payments') }}
group by 1
having total_amount < 0

Generic Data Test

A generic data test is defined in a special test block (just like a macro). Once set up, you can reference it by name in your .yml files for models, columns, sources, snapshots, and seeds. dbt comes with four built-in generic tests—and you should definitely take advantage of them!

{% test only_letters(model, column_name) %}
    select *
    from {{ model }}
    where {{ column_name }} !~ '^[A-Za-z]+$'
{% endtest %}

Applying previous custom tests to the schema

# models/schema.yml
version: 2

models:
  - name: fct_payments
    columns:
      - name: customer_name
        tests:
          - only_letters
      - name: email
        tests:
          - not_null
    .
    .
    .

Built-in Tests

Out of the box, dbt includes four generic data tests: unique, not_null, accepted_values, and relationships. Here’s how you can integrate them into your project:

version: 2

models:
  - name: orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned']
      - name: customer_id
        tests:
          - relationships:
              to: ref('customers')
              field: id

Normally, a data test query calculates failures on the fly. If you set the optional --store-failures flag or use the store_failures / store_failures_as configs, dbt will first save the test query results to a table in your database, then query that table to count the number of failures.

# dbt_project.yml
data_tests:  
  +store_failures: true