Chapter 12 · Metaprogramming

Custom Materializations

A materialization decides what "building a model" means. Writing one moves logic out of every model and into one reusable strategy. In this chapter you'll build scd2, a small but complete Type 2 materialization for PostgreSQL, and look at every line that dbt requires along the way.

When a materialization is the right tool

Snapshots are dbt's built-in SCD2, so why build your own? Snapshots are separate resources, they add dbt_valid_from/dbt_valid_to columns, and they work on their own schedule. A materialization lets an ordinary model keep history using your own column names and rules, and it's the clearest way to learn how materializations work.

Anatomy

{% materialization my_strategy, adapter='postgres' %}     {# or default=true for all adapters #}

  {#- 1. relations & config -#}
  {%- set target_relation   = this.incorporate(type='table') -%}
  {%- set existing_relation = load_cached_relation(this) -%}
  {%- set some_option       = config.get('some_option', 'x') -%}

  {#- 2. pre-hooks -#}
  {{ run_hooks(pre_hooks, inside_transaction=False) }}
  {{ run_hooks(pre_hooks, inside_transaction=True) }}

  {#- 3. the work — exactly one statement named 'main' -#}
  {% call statement('main') %}
      … uses {{ sql }} (the model's compiled SELECT) …
  {% endcall %}

  {#- 4. post-hooks & commit -#}
  {{ run_hooks(post_hooks, inside_transaction=True) }}
  {% do adapter.commit() %}
  {{ run_hooks(post_hooks, inside_transaction=False) }}

  {#- 5. tell dbt what now exists (updates the relation cache) -#}
  {{ return({'relations': [target_relation]}) }}
{% endmaterialization %}

• Name and adapter. dbt registers this as the macro materialization_my_strategy_postgres. A model uses it with config(materialized='my_strategy').
• sql is the model's compiled SELECT, and it's available only inside materializations.
• statement('main') is required. The runner reads its result for the run summary and run_results.json.
• The return value lists the relations the materialization created or replaced, so dbt's cache stays correct for later nodes.

Designing scd2

Contract with the model:

• The model selects the current state of each entity, one row per business key, including a row_hash.
• Config unique_key names the business key (one column, to keep the example readable).
• The materialization adds and maintains valid_from, valid_to and is_current.
• Keys that disappear from the source are closed when invalidate_hard_deletes=true.

The materialization

{% materialization scd2, adapter='postgres' %}

  {%- set unique_key      = config.require('unique_key') -%}
  {%- set hash_column     = config.get('hash_column', 'row_hash') -%}
  {%- set hard_deletes    = config.get('invalidate_hard_deletes', false) -%}
  {%- if unique_key is not string -%}
      {{ exceptions.raise_compiler_error("scd2: unique_key must be a single column name") }}
  {%- endif -%}

  {%- set target_relation   = this.incorporate(type='table') -%}
  {%- set existing_relation = load_cached_relation(this) -%}
  {%- set tmp_relation      = make_temp_relation(target_relation) -%}
  {%- set full_refresh      = should_full_refresh() -%}

  {{ run_hooks(pre_hooks, inside_transaction=False) }}
  {{ run_hooks(pre_hooks, inside_transaction=True) }}

  {%- if existing_relation is not none and existing_relation.type != 'table' -%}
      {#- a view (or other) with our name is in the way -#}
      {% do adapter.drop_relation(existing_relation) %}
      {%- set existing_relation = none -%}
  {%- endif -%}

  {% if existing_relation is none or full_refresh %}

      {%- if existing_relation is not none -%}
          {% do adapter.drop_relation(existing_relation) %}
      {%- endif -%}
      {% call statement('main') %}
          create table {{ target_relation }} as
          select src.*,
                 now()            as valid_from,
                 null::timestamptz as valid_to,
                 true             as is_current
          from ( {{ sql }} ) src
      {% endcall %}

  {% else %}

      {% call statement('build_tmp') %}
          {{ get_create_table_as_sql(True, tmp_relation, sql) }}
      {% endcall %}

      {% call statement('main') %}
          -- 1. close versions whose attributes changed
          update {{ target_relation }} t
             set valid_to = now(), is_current = false
            from {{ tmp_relation }} s
           where t.{{ unique_key }} = s.{{ unique_key }}
             and t.is_current
             and t.{{ hash_column }} is distinct from s.{{ hash_column }};

          {% if hard_deletes -%}
          -- 2. close versions whose key disappeared from the source
          update {{ target_relation }} t
             set valid_to = now(), is_current = false
           where t.is_current
             and not exists (select 1 from {{ tmp_relation }} s
                              where s.{{ unique_key }} = t.{{ unique_key }});
          {%- endif %}

          -- 3. insert a new current version for every key without one (new + just closed)
          insert into {{ target_relation }}
          select s.*, now(), null::timestamptz, true
          from {{ tmp_relation }} s
          where not exists (select 1 from {{ target_relation }} t
                             where t.{{ unique_key }} = s.{{ unique_key }}
                               and t.is_current);
      {% endcall %}

  {% endif %}

  {{ run_hooks(post_hooks, inside_transaction=True) }}
  {% do adapter.commit() %}
  {{ run_hooks(post_hooks, inside_transaction=False) }}

  {{ return({'relations': [target_relation]}) }}

{% endmaterialization %}

Line-by-line notes

Using it

{{ config(
    materialized='scd2',
    unique_key='salesterritorykey',
    invalidate_hard_deletes=true,
    post_hook="{{ record_dim_stats(scd2=false) }}"
) }}

select
    salesterritorykey,
    salesterritoryregion,
    salesterritorycountry,
    salesterritorygroup,
    {{ row_hash(['salesterritoryregion', 'salesterritorycountry', 'salesterritorygroup']) }} as row_hash
from {{ ref('dimsalesterritory') }}

$ dbt run -s dimterritory_history --vars '{"run_id": 1}'     # first run: create table
$ dbt run -s dimterritory_history --vars '{"run_id": 2}'     # no changes: 0 rows inserted
#   change a territory name in stg.salesterritory, rebuild dimsalesterritory, then:
$ dbt run -s dimsalesterritory dimterritory_history --vars '{"run_id": 3}'

Exercises

{% set src_cols = adapter.get_columns_in_relation(tmp_relation) | map(attribute='name') | list %}
{% set tgt_cols = adapter.get_columns_in_relation(target_relation) | map(attribute='name')
                  | reject('in', ['valid_from', 'valid_to', 'is_current']) | list %}
{% if src_cols != tgt_cols %}
  {{ exceptions.raise_compiler_error("scd2: columns changed: " ~ src_cols ~ " vs " ~ tgt_cols) }}
{% endif %}