Testing

dbt Snapshot Testing: Validate Slowly Changing Dimensions

09 Sep 2026 — 5 min read

dbt snapshots implement Slowly Changing Dimensions (SCD Type 2) — tracking how a record changes over time by keeping the full history. A bug in snapshot configuration silently corrupts your history, making past-dated reports wrong. Testing snapshots requires validating that history is captured correctly, records are updated (not duplicated), and the dbt_valid_from/dbt_valid_to columns are set correctly.

How dbt Snapshots Work

-- snapshots/orders_snapshot.sql
{% snapshot orders_snapshot %}
{{
    config(
        target_database='analytics',
        target_schema='snapshots',
        unique_key='order_id',
        strategy='timestamp',
        updated_at='updated_at',
    )
}}

SELECT
    order_id,
    customer_id,
    status,
    amount,
    updated_at
FROM {{ ref('stg_orders') }}

{% endsnapshot %}

dbt adds four columns:

dbt_scd_id — surrogate key for the snapshot row
dbt_updated_at — timestamp from source
dbt_valid_from — when this version became active
dbt_valid_to — when this version was superseded (NULL = current)

What to Test

Current records are current — only one active row (dbt_valid_to IS NULL) per unique key
History rows are closed — superseded rows have dbt_valid_to set
Timestamps are ordered — dbt_valid_from < dbt_valid_to
Changes are captured — when a record changes, a new row is created
Unchanged records are not duplicated — re-running snapshot on unchanged data doesn't add rows
Deletes are handled — hard deletes follow your configured strategy

dbt Tests for Snapshot Tables

# models/schema.yml
models:
  - name: orders_snapshot
    description: "SCD Type 2 history of order records"
    columns:
      - name: dbt_scd_id
        tests:
          - unique
          - not_null
      
      - name: order_id
        tests:
          - not_null
      
      - name: dbt_valid_from
        tests:
          - not_null
      
      - name: dbt_valid_to
        # NULL = current record, so not_null test doesn't apply

Built-in dbt tests are too limited for snapshots. Write custom tests:

-- tests/assert_one_active_snapshot_per_key.sql
-- Fails if any unique key has more than one active row (dbt_valid_to IS NULL)
SELECT
    order_id,
    COUNT(*) AS active_count
FROM {{ ref('orders_snapshot') }}
WHERE dbt_valid_to IS NULL
GROUP BY order_id
HAVING COUNT(*) > 1

-- tests/assert_snapshot_timestamps_ordered.sql
-- Fails if any row has dbt_valid_from >= dbt_valid_to
SELECT *
FROM {{ ref('orders_snapshot') }}
WHERE
    dbt_valid_to IS NOT NULL
    AND dbt_valid_from >= dbt_valid_to

-- tests/assert_no_overlapping_snapshot_periods.sql
-- Fails if history periods for a key overlap
SELECT
    a.order_id,
    a.dbt_scd_id AS scd_id_a,
    b.dbt_scd_id AS scd_id_b
FROM {{ ref('orders_snapshot') }} a
JOIN {{ ref('orders_snapshot') }} b
    ON a.order_id = b.order_id
    AND a.dbt_scd_id != b.dbt_scd_id
    AND a.dbt_valid_from < COALESCE(b.dbt_valid_to, '9999-12-31')
    AND b.dbt_valid_from < COALESCE(a.dbt_valid_to, '9999-12-31')

-- tests/assert_history_complete.sql
-- Verifies that closed records cover all time between versions
-- (no gaps in history for a given order_id)
WITH ordered_history AS (
    SELECT
        order_id,
        dbt_valid_from,
        dbt_valid_to,
        LEAD(dbt_valid_from) OVER (
            PARTITION BY order_id ORDER BY dbt_valid_from
        ) AS next_valid_from
    FROM {{ ref('orders_snapshot') }}
),
gaps AS (
    SELECT *
    FROM ordered_history
    WHERE
        dbt_valid_to IS NOT NULL
        AND dbt_valid_to != next_valid_from
)
SELECT * FROM gaps

Integration Testing Snapshot Behavior

To test that history is actually captured when data changes, you need to:

Load initial data
Run snapshot
Change the source data
Run snapshot again
Verify history is correct

This is hard to do with dbt's built-in test infrastructure alone. Use a test script:

# tests/test_snapshot_behavior.py
import subprocess
import pytest
import sqlalchemy as sa
import pandas as pd

# These tests require a real (or ephemeral) database
@pytest.fixture(scope="module")
def engine():
    return sa.create_engine("postgresql://test:test@localhost/testdb")

def run_dbt(command: list[str]) -> subprocess.CompletedProcess:
    result = subprocess.run(
        ["dbt"] + command + ["--profiles-dir", ".", "--target", "test"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        pytest.fail(f"dbt {command} failed:\n{result.stdout}\n{result.stderr}")
    return result

def test_snapshot_captures_initial_state(engine):
    # Seed initial data
    initial_orders = pd.DataFrame([
        {'order_id': 'o1', 'customer_id': 'c1', 'status': 'pending',
         'amount': 100.0, 'updated_at': '2024-01-01 10:00:00'},
        {'order_id': 'o2', 'customer_id': 'c2', 'status': 'pending',
         'amount': 200.0, 'updated_at': '2024-01-01 10:00:00'},
    ])
    initial_orders.to_sql('stg_orders', engine, if_exists='replace', index=False)
    
    run_dbt(['snapshot', '--select', 'orders_snapshot'])
    
    snapshot = pd.read_sql("SELECT * FROM snapshots.orders_snapshot", engine)
    
    assert len(snapshot) == 2
    assert snapshot['dbt_valid_to'].isna().all()  # All current

def test_snapshot_tracks_status_change(engine):
    # Update one order's status
    with engine.connect() as conn:
        conn.execute(sa.text("""
            UPDATE stg_orders
            SET status = 'completed', updated_at = '2024-01-02 10:00:00'
            WHERE order_id = 'o1'
        """))
        conn.commit()
    
    run_dbt(['snapshot', '--select', 'orders_snapshot'])
    
    snapshot = pd.read_sql("""
        SELECT * FROM snapshots.orders_snapshot
        WHERE order_id = 'o1'
        ORDER BY dbt_valid_from
    """, engine)
    
    # Should have 2 rows for o1
    assert len(snapshot) == 2
    
    # First row (historical) should be closed
    historical = snapshot.iloc[0]
    assert historical['status'] == 'pending'
    assert historical['dbt_valid_to'] is not None
    assert historical['dbt_valid_from'].strftime('%Y-%m-%d') == '2024-01-01'
    
    # Second row (current) should be active
    current = snapshot.iloc[1]
    assert current['status'] == 'completed'
    assert pd.isna(current['dbt_valid_to'])

def test_unchanged_record_not_duplicated(engine):
    """Re-running snapshot on unchanged data should not add rows."""
    before_count = pd.read_sql(
        "SELECT COUNT(*) AS n FROM snapshots.orders_snapshot WHERE order_id = 'o2'",
        engine
    ).iloc[0]['n']
    
    run_dbt(['snapshot', '--select', 'orders_snapshot'])
    
    after_count = pd.read_sql(
        "SELECT COUNT(*) AS n FROM snapshots.orders_snapshot WHERE order_id = 'o2'",
        engine
    ).iloc[0]['n']
    
    assert after_count == before_count, \
        f"Snapshot added {after_count - before_count} duplicate rows for unchanged record"

def test_snapshot_handles_multiple_changes(engine):
    """Each change should create a new history row."""
    # Change o2 twice
    for status, ts in [('processing', '2024-01-03'), ('completed', '2024-01-04')]:
        with engine.connect() as conn:
            conn.execute(sa.text(f"""
                UPDATE stg_orders
                SET status = '{status}', updated_at = '{ts} 10:00:00'
                WHERE order_id = 'o2'
            """))
            conn.commit()
        
        run_dbt(['snapshot', '--select', 'orders_snapshot'])
    
    history = pd.read_sql("""
        SELECT status, dbt_valid_from, dbt_valid_to
        FROM snapshots.orders_snapshot
        WHERE order_id = 'o2'
        ORDER BY dbt_valid_from
    """, engine)
    
    # Should have 3 rows: pending → processing → completed
    assert len(history) == 3
    assert list(history['status']) == ['pending', 'processing', 'completed']
    
    # Only last row should be current
    assert history.iloc[-1]['dbt_valid_to'] is None
    assert history.iloc[0]['dbt_valid_to'] is not None
    assert history.iloc[1]['dbt_valid_to'] is not None

Snapshot Configuration Testing

Test that your snapshot configuration handles edge cases:

-- tests/assert_check_strategy_detects_any_change.sql
-- For 'check' strategy: verify that changing any monitored column creates a new row
-- Run after seeding data where a non-updated_at column changed

SELECT
    order_id,
    dbt_valid_from,
    dbt_valid_to,
    status
FROM {{ ref('orders_snapshot') }}
WHERE order_id = 'test-check-strategy-order'
ORDER BY dbt_valid_from

Automating Snapshot Tests in CI

# .github/workflows/dbt-snapshots.yml
name: dbt Snapshot Tests

on:
  push:
    paths:
      - 'snapshots/**'
      - 'models/**'
      - 'tests/**'

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: test
          POSTGRES_DB: testdb
        ports: ['5432:5432']
        options: --health-cmd pg_isready
    
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dbt
        run: pip install dbt-postgres pytest
      
      - name: Setup dbt profile
        run: |
          mkdir -p ~/.dbt
          cat > ~/.dbt/profiles.yml << 'EOF'
          my_project:
            target: test
            outputs:
              test:
                type: postgres
                host: localhost
                port: 5432
                user: postgres
                password: test
                dbname: testdb
                schema: public
                threads: 1
          EOF
      
      - name: Run dbt build
        run: dbt build --full-refresh
      
      - name: Run custom snapshot tests
        run: dbt test --select snapshots
      
      - name: Run snapshot behavior tests
        run: pytest tests/test_snapshot_behavior.py -v
        env:
          DATABASE_URL: postgresql://postgres:test@localhost/testdb

Monitoring Snapshots in Production

Beyond CI tests, monitor snapshots in production:

-- Monitor snapshot row growth (alert if suspiciously high)
SELECT
    CURRENT_DATE AS check_date,
    COUNT(*) AS total_rows,
    COUNT(DISTINCT order_id) AS unique_keys,
    SUM(CASE WHEN dbt_valid_to IS NULL THEN 1 ELSE 0 END) AS active_rows,
    SUM(CASE WHEN dbt_valid_to IS NOT NULL THEN 1 ELSE 0 END) AS historical_rows,
    AVG(CASE WHEN dbt_valid_to IS NOT NULL
        THEN EXTRACT(EPOCH FROM (dbt_valid_to - dbt_valid_from)) / 86400
        ELSE NULL END) AS avg_version_lifetime_days
FROM orders_snapshot

If total_rows grows much faster than unique_keys, your snapshot may be creating too many versions (check for source data with frequently updated timestamps but unchanged business values — use the check strategy with explicit column lists instead of timestamp).

dbt snapshots are invisible when working correctly but catastrophically wrong when misconfigured. Testing them explicitly — both with dbt's built-in tests for column-level invariants and integration tests for behavioral correctness — is the only way to trust your historical data.