Chapter 18 · Capstone

Capstone: a Metadata-driven Dimension Loader

This chapter brings everything together. You'll turn the hand-written, 80-line dimemployee.sql into a model that declares its SCD2 rules in config and lets macros generate the rest. Along the way you'll use validated inputs, generated SQL, generic hooks, a generic test, a graph-walking audit, invocation logging and a safe rollout. The next Type 2 dimension then takes about fifteen lines.

The design

Requirements

1. The generated SQL must be logically identical to today's dimemployee, so existing keys and history are untouched.
2. A missing or wrong setting must fail with a clear message during execution, and must never break parsing.
3. No hard-coded schema names, and no silent run_id default.
4. Every SCD2 dimension must have the integrity test, enforced automatically.

Step 1: the metadata contract

dimension:      scd2                                   # marks the model for the audit operation
surrogate_key:  employeekey                            # filled from a sequence
business_key:   employeenationalidalternatekey         # identifies the entity
version_key:    startdate                              # identifies the version
hash_column:    row_hash                               # default row_hash
status_column:  status                                 # default status
current_value:  Current                                # default Current
sequence:       mart.dimemployee_key_seq               # default <schema>.<table>_key_seq
parent_key:                                            # optional self-reference
  column:       parentemployeekey
  business_key: parentemployeenationalidalternatekey

Step 2: reading and validating it

{#- Returns the dwlab dimension settings of the calling model, with defaults applied.
    Validates only while executing, so parsing never fails because of it. -#}
{% macro dim_meta() %}
    {%- set raw = (config.get('meta') or {}).get('dwlab') or {} -%}
    {%- set m = {
        'dimension':     raw.get('dimension'),
        'surrogate_key': raw.get('surrogate_key'),
        'business_key':  raw.get('business_key'),
        'version_key':   raw.get('version_key'),
        'hash_column':   raw.get('hash_column', 'row_hash'),
        'status_column': raw.get('status_column', 'status'),
        'current_value': raw.get('current_value', 'Current'),
        'sequence':      raw.get('sequence', this.schema ~ '.' ~ this.identifier ~ '_key_seq'),
        'parent_key':    raw.get('parent_key')
    } -%}

    {%- if execute -%}
        {%- for k in ['surrogate_key', 'business_key', 'version_key'] if not m[k] -%}
            {{ exceptions.raise_compiler_error(model.name ~ ": meta.dwlab." ~ k ~ " is required") }}
        {%- endfor -%}
        {%- set expected_key = [m.business_key, m.version_key] -%}
        {%- if config.get('unique_key') != expected_key -%}
            {{ exceptions.raise_compiler_error(model.name ~ ": unique_key must be " ~ expected_key
                                              ~ " (got " ~ config.get('unique_key') ~ ")") }}
        {%- endif -%}
        {%- if m.parent_key and not (m.parent_key.get('column') and m.parent_key.get('business_key')) -%}
            {{ exceptions.raise_compiler_error(model.name ~ ": parent_key needs 'column' and 'business_key'") }}
        {%- endif -%}
    {%- endif -%}

    {{ return(m) }}
{% endmacro %}

{#- Column names of the prepared source, in order. Empty while parsing. -#}
{% macro dim_source_columns(relation) %}
    {%- if not execute -%}{{ return([]) }}{%- endif -%}
    {{ return(adapter.get_columns_in_relation(relation) | map(attribute='name') | list) }}
{% endmacro %}

You'll also need require_run_id() from chapter 17.

Step 3: generating the SELECT

{#- The full SELECT of an incremental SCD2 dimension (merge on business_key + version_key).
    source: a Relation (ref/source) with one row per version, including the hash column. -#}
{% macro scd2_dimension(source) %}
    {%- set m        = dim_meta() -%}
    {%- set rid      = require_run_id() -%}
    {%- set incr     = is_incremental() -%}
    {%- set cols     = dim_source_columns(source) -%}
    {%- set sk, bk, vk = m.surrogate_key, m.business_key, m.version_key -%}
    {%- set st, cur  = m.status_column, m.current_value -%}

with src as (
    select * from {{ source }}
    order by {{ bk }}, {{ vk }}                 -- keys are handed out in this order
)
{%- if incr %}
, tgt as (
    select {{ sk }}, {{ bk }}, {{ vk }}, {{ m.hash_column }}, {{ st }},
           dw_first_run_id, dw_expired_run_id, dw_inserted_at
    from {{ this }}
)
{%- endif %}

select
    {%- if incr %}
    coalesce(t.{{ sk }}, nextval('{{ m.sequence }}'))   as {{ sk }},
    {%- else %}
    nextval('{{ m.sequence }}')                         as {{ sk }},
    {%- endif %}
    {%- if m.parent_key %}
    null::bigint                                        as {{ m.parent_key.column }},   -- set by the post-hook
    {%- endif %}
    {%- for c in cols %}
    s.{{ c }},
    {%- endfor %}
    -- audit columns
    {%- if incr %}
    coalesce(t.dw_first_run_id, {{ rid }})              as dw_first_run_id,
    {{ rid }}                                           as dw_last_run_id,
    case when s.{{ st }} is null and t.{{ st }} = '{{ cur }}' then {{ rid }}
         else t.dw_expired_run_id end                   as dw_expired_run_id,
    coalesce(t.dw_inserted_at, now())                   as dw_inserted_at,
    {%- else %}
    {{ rid }}                                           as dw_first_run_id,
    {{ rid }}                                           as dw_last_run_id,
    null::integer                                       as dw_expired_run_id,
    now()                                               as dw_inserted_at,
    {%- endif %}
    now()                                               as dw_updated_at
from src s
{%- if incr %}
left join tgt t
       on t.{{ bk }} = s.{{ bk }}
      and t.{{ vk }} = s.{{ vk }}
where t.{{ sk }} is null                                     -- new version
   or t.{{ m.hash_column }} is distinct from s.{{ m.hash_column }}   -- changed version
{%- endif %}
{% endmacro %}

The column order is the same as today: the surrogate key, the parent key, the prepared columns in their view order (ending with row_hash), then the five audit columns. That's requirement 1.

Step 4: generic hooks

{% macro dim_pre_hook() %}
    {%- set m = dim_meta() -%}
    create sequence if not exists {{ m.sequence }}
{% endmacro %}

{% macro dim_post_hook() %}
    {%- set m = dim_meta() -%}
    {%- if m.parent_key -%}
    update {{ this }} d
       set {{ m.parent_key.column }} = p.{{ m.surrogate_key }}
      from {{ this }} p
     where p.{{ m.business_key }} = d.{{ m.parent_key.business_key }}
       and p.{{ m.status_column }} = '{{ m.current_value }}'
       and d.{{ m.parent_key.column }} is distinct from p.{{ m.surrogate_key }};
    {%- endif %}
    {{ record_dim_stats(scd2=true) }}
{% endmacro %}

record_dim_stats is the hardened version from chapter 17 (it uses source('audit', 'dim_load_stats') and require_run_id()). Both hook statements still run inside the merge's transaction, so they're atomic with the data (ch. 09).

Step 5: the new model

{#
  mart.dimemployee — Type 2 on department periods, Type 1 on all other columns.
  Everything below the config is generated by scd2_dimension(); see macros/dims/.
#}
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key=['employeenationalidalternatekey', 'startdate'],
    on_schema_change='fail',
    full_refresh=false,
    meta={'dwlab': {
        'dimension':     'scd2',
        'surrogate_key': 'employeekey',
        'business_key':  'employeenationalidalternatekey',
        'version_key':   'startdate',
        'parent_key':    {'column': 'parentemployeekey',
                          'business_key': 'parentemployeenationalidalternatekey'}
    }},
    pre_hook="{{ dim_pre_hook() }}",
    post_hook="{{ dim_post_hook() }}"
) }}

{{ scd2_dimension(ref('int_dimemployee_prepared')) }}

Step 6: tests

  - name: dimemployee
    description: "Employee dimension - Type 2 on DepartmentName, Type 1 on all other columns."
    data_tests:
      - scd2_integrity:                       # chapter 13 — replaces both assert_dimemployee_* files
          arguments:
            business_key: employeenationalidalternatekey
          config:
            store_failures: true
      - dbt_utils.unique_combination_of_columns:
          arguments:
            combination_of_columns: [employeenationalidalternatekey, startdate]
    columns:
      - name: employeekey
        data_tests: [unique, not_null]
      # … existing column tests unchanged …

Step 7: enforce it across the project

{#- dbt run-operation audit_dimensions — fails if any meta.dwlab.dimension=scd2 model lacks scd2_integrity -#}
{% macro audit_dimensions() %}
    {%- if not execute -%}{{ return('') }}{%- endif -%}
    {%- set missing = [] -%}
    {%- for n in graph.nodes.values()
          if n.resource_type == 'model'
             and (((n.config.meta or {}).get('dwlab') or {}).get('dimension')) == 'scd2' -%}
        {%- set ns = namespace(tested=false) -%}
        {%- for t in graph.nodes.values()
              if t.resource_type == 'test'
                 and n.unique_id in t.depends_on.nodes
                 and (t.test_metadata or {}).get('name') == 'scd2_integrity' -%}
            {%- set ns.tested = true -%}
        {%- endfor -%}
        {{ log(('✓ ' if ns.tested else '✗ ') ~ n.name, info=True) }}
        {%- if not ns.tested -%}{%- do missing.append(n.name) -%}{%- endif -%}
    {%- endfor -%}
    {%- if missing -%}
        {{ exceptions.raise_compiler_error("SCD2 dimensions without scd2_integrity: " ~ missing | join(', ')) }}
    {%- endif -%}
{% endmacro %}

Add it as the first dbt task in the Airflow DAG, or in CI. It needs no database writes and no --vars.

Step 8: invocation logging

on-run-end:
  - "{{ log_dbt_results(results) }}"        # chapter 09

Step 9: roll out safely

1. Save today's compiled SQL. Before changing anything, run dbt compile -s dimemployee --vars '{"run_id": 999}' and copy target/compiled/dwlab/models/mart/dimemployee.sql to /tmp/before.sql.
2. Apply the change (the macros, the model and the YAML) and compile again with the same vars.
3. Compare with diff -w /tmp/before.sql target/compiled/…/dimemployee.sql. Only whitespace and comments should differ. Any other difference must be explained before you continue.
4. Check the hooks: dbt --debug run against a copy of the database, then compare the logged hook SQL with the old hooks.
5. Run the tests: dbt test -s dimemployee and dbt run-operation audit_dimensions.
6. Delete the replaced singular tests once scd2_integrity has passed in the real pipeline.

Step 10: the next dimension

Suppose a later pipeline prepares int_dimcustomer_prepared with one row per customer version and a row_hash. The whole dimension is then:

{{ config(
    materialized='incremental', incremental_strategy='merge', on_schema_change='fail', full_refresh=false,
    unique_key=['customeralternatekey', 'validfrom'],
    meta={'dwlab': {'dimension': 'scd2', 'surrogate_key': 'customerkey',
                    'business_key': 'customeralternatekey', 'version_key': 'validfrom'}},
    pre_hook="{{ dim_pre_hook() }}",
    post_hook="{{ dim_post_hook() }}"
) }}
{{ scd2_dimension(ref('int_dimcustomer_prepared')) }}

If you forget the scd2_integrity test, audit_dimensions fails the pipeline.

Where to go from here

• Move the macros into dwlab_macros (chapter 16) with an integration test project.
• Switch the hash in int_dimemployee_prepared to the dispatched, null-safe row_hash (chapters 08 and 10), with a planned one-time re-hash.
• Make schemas environment-aware (chapter 11) so dev runs stop touching the lab's mart.
• Collapse stages 3–5 into one dbt build with -- depends_on: hints on rpt_load_summary (internals ch. 05), and keep record_dbt_step plus the new on-run-end log for auditing.

Exercises