The Problem of Naïve Incremental Models

The basic incremental model seen in module 04 - filter on MAX(timestamp) and merge for unique_key - is sufficient for many use cases. But when the volume grows beyond a billion rows, when the data arrives hours late or days, or when the source is a CDC system issuing operations of type INSERT/UPDATE/DELETE, more sophisticated patterns are needed.

This module explores three advanced scenarios: the microbatch strategy (introduced in dbt 1.9) for huge time-series datasets, integration with the Change Data Capture to faithfully replicate changes from source databases, and techniques to ensure theidempotence in the presence of late-arriving data.

Recap: Standard Incremental Strategies

Before proceeding with the advanced patterns, let's recap the complete picture of the strategies available in dbt 1.8+:

-- Le 4 strategie standard (+ microbatch in 1.9):

-- 1. append: solo INSERT, nessun UPDATE, nessuna deduplicazione
-- Casi d'uso: audit log, event stream immutabili, fact tables append-only
{{ config(materialized='incremental', incremental_strategy='append') }}

-- 2. merge: UPSERT su unique_key (default con unique_key definita)
-- Casi d'uso: dimensioni che cambiano, ordini aggiornabili, qualsiasi SCD-1
{{ config(materialized='incremental', unique_key='id', incremental_strategy='merge') }}

-- 3. delete+insert: elimina partizioni interessate e reinserisce
-- Casi d'uso: BigQuery partitioned tables, re-processing per data
{{ config(materialized='incremental', unique_key=['id', 'date'],
          incremental_strategy='delete+insert',
          partition_by={'field': 'date', 'data_type': 'date'}) }}

-- 4. insert_overwrite: sovrascrive intere partizioni (Spark, BigQuery)
-- Casi d'uso: dataset partizionati per data, Spark/Databricks
{{ config(materialized='incremental', incremental_strategy='insert_overwrite',
          partition_by={'field': 'date', 'data_type': 'date'}) }}

Microbatch: The Strategy for Huge Time-Series Datasets

The strategy microbatches was introduced in dbt 1.9 to fix a specific problem: incremental time-series models where the volume per single execution is still too large, or where re-processing of specific time windows must be simple and atomic.

How Microbatch Works

Instead of running one large incremental query, microbatch splits the execution into time lots (batches) defined by the field event_time and from batch_size. Each batch is processed atomically and independently.

-- models/marts/events/fct_events_microbatch.sql
{{ config(
    materialized='incremental',
    incremental_strategy='microbatch',
    event_time='event_timestamp',    -- colonna timestamp per la suddivisione in batch
    begin='2024-01-01',              -- data di inizio per il full refresh iniziale
    batch_size='day',                -- granularità di ogni batch: hour | day | month | year
    lookback=1                       -- ri-processa anche N batch precedenti (late-arriving data)
) }}

SELECT
    event_id,
    user_id,
    session_id,
    event_type,
    page_url,
    referrer_url,
    device_type,
    country_code,
    event_timestamp,
    DATE_TRUNC('day', event_timestamp)  AS event_date,
    DATE_TRUNC('hour', event_timestamp) AS event_hour
FROM {{ ref('stg_events') }}

-- NOTA IMPORTANTE con microbatch:
-- NON usare is_incremental() con microbatch.
-- dbt gestisce automaticamente il filtro temporale su event_time.
-- La query deve essere scritta come se processasse TUTTI i dati --
-- dbt aggiunge automaticamente il filtro: WHERE event_timestamp >= batch_start AND event_timestamp < batch_end

Execution and Re-Processing with Microbatch

-- Esecuzione normale: processa i batch mancanti dall'ultima run
dbt run --select fct_events_microbatch

-- Re-processare un giorno specifico (molto utile per correzioni dati):
dbt run \
  --select fct_events_microbatch \
  --event-time-start "2024-06-15" \
  --event-time-end "2024-06-16"

-- Re-processare un intervallo (es. dopo una correzione alla sorgente):
dbt run \
  --select fct_events_microbatch \
  --event-time-start "2024-06-01" \
  --event-time-end "2024-07-01"

-- Full refresh dall'inizio (usa il 'begin' configurato):
dbt run --select fct_events_microbatch --full-refresh

-- Output tipico durante l'esecuzione:
-- Running microbatch model fct_events_microbatch
--   Batch 2024-06-13 00:00:00 to 2024-06-14 00:00:00 [1 of 3]: OK (123,456 rows)
--   Batch 2024-06-14 00:00:00 to 2024-06-15 00:00:00 [2 of 3]: OK (98,234 rows)
--   Batch 2024-06-15 00:00:00 to 2024-06-16 00:00:00 [3 of 3]: OK (45,678 rows)

Advantages of Microbatch vs Incremental Standard

Change Data Capture: Replicate Changes from the Source Database

Il Change Data Capture (CDC) captures every INSERT, UPDATE and DELETE that occurs in the source database and propagates it into the pipeline. It is the standard for replicating data in near real time without impacting the transactional system.

CDC pattern with Debezium + Kafka

The most common stack for CDC with dbt: Debezium reads PostgreSQL WAL (Write-Ahead Log) or MySQL and publish changes to Kafka. A Kafka connector (Fivetran, Airbyte, or custom) loads the messages into the warehouse raw table. dbt then consolidates the change history.

-- La raw table CDC tipica dal connettore ha questa struttura:
-- Ogni riga rappresenta un'operazione (INSERT/UPDATE/DELETE)

-- raw.orders_cdc_stream:
-- _op          VARCHAR  -- 'r' (read/snapshot), 'c' (create/insert), 'u' (update), 'd' (delete)
-- _ts_ms       BIGINT   -- timestamp dell'operazione in milliseconds
-- _after       VARIANT  -- stato DOPO la modifica (JSON) -- NULL per DELETE
-- _before      VARIANT  -- stato PRIMA della modifica (JSON) -- NULL per INSERT
-- order_id     VARCHAR  -- chiave primaria (estratta dal messaggio)
-- customer_id  VARCHAR  -- ...altri campi estratti...

-- models/staging/stg_orders_cdc.sql
-- Normalizza i messaggi CDC in una struttura flat

{{ config(
    materialized='incremental',
    unique_key=['order_id', '_ts_ms'],
    incremental_strategy='append'   -- append-only: ogni operazione è una riga
) }}

SELECT
    order_id,
    _op                                          AS operation_type,
    TO_TIMESTAMP(_ts_ms / 1000)                  AS operation_timestamp,

    -- Estrai i campi dall'JSON _after (per INSERT e UPDATE)
    CASE WHEN _op != 'd'
         THEN _after:customer_id::VARCHAR
         ELSE _before:customer_id::VARCHAR
    END                                          AS customer_id,

    CASE WHEN _op != 'd'
         THEN _after:status::VARCHAR
         ELSE NULL
    END                                          AS status,

    CASE WHEN _op != 'd'
         THEN _after:total_amount::DECIMAL(10,2)
         ELSE NULL
    END                                          AS total_amount,

    CASE WHEN _op != 'd'
         THEN _after:updated_at::TIMESTAMP
         ELSE NULL
    END                                          AS updated_at,

    _op = 'd'                                    AS is_deleted

FROM {{ source('raw', 'orders_cdc_stream') }}

{% if is_incremental() %}
WHERE TO_TIMESTAMP(_ts_ms / 1000) > (
    SELECT MAX(operation_timestamp) FROM {{ this }}
)
{% endif %}

-- models/marts/dim_orders_from_cdc.sql
-- Consolida la storia CDC: prende solo l'ultimo stato di ogni ordine

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    merge_update_columns=['customer_id', 'status', 'total_amount', 'updated_at', 'is_deleted', 'last_operation_at']
) }}

WITH latest_per_order AS (
    -- Prendi solo l'operazione più recente per ogni ordine
    SELECT
        order_id,
        customer_id,
        status,
        total_amount,
        updated_at,
        is_deleted,
        operation_timestamp AS last_operation_at,
        -- Row number per prendere solo il record più recente
        ROW_NUMBER() OVER (
            PARTITION BY order_id
            ORDER BY operation_timestamp DESC
        ) AS rn
    FROM {{ ref('stg_orders_cdc') }}

    {% if is_incremental() %}
    -- In modalità incrementale: considera le operazioni recenti + lookback
    WHERE operation_timestamp > (
        SELECT DATEADD('hour', -2, MAX(last_operation_at))
        FROM {{ this }}
    )
    {% endif %}
)
SELECT
    order_id,
    customer_id,
    status,
    total_amount,
    updated_at,
    is_deleted,
    last_operation_at
FROM latest_per_order
WHERE rn = 1
  AND NOT is_deleted  -- esclude i record cancellati dalla vista finale

SCD-2 Pattern from CDC: Keeping the History Complete

-- models/marts/dim_orders_history.sql
-- SCD Type 2 completo da stream CDC: ogni cambiamento di stato viene storicizzato

{{ config(materialized='incremental', unique_key='order_state_id') }}

WITH cdc_with_prev AS (
    SELECT
        order_id,
        operation_type,
        operation_timestamp,
        status,
        total_amount,
        customer_id,
        -- Stato precedente per rilevare cambiamenti reali
        LAG(status) OVER (PARTITION BY order_id ORDER BY operation_timestamp)        AS prev_status,
        LAG(total_amount) OVER (PARTITION BY order_id ORDER BY operation_timestamp)  AS prev_total,
        -- Prossimo timestamp per calcolare valid_to
        LEAD(operation_timestamp) OVER (PARTITION BY order_id ORDER BY operation_timestamp) AS next_op_at
    FROM {{ ref('stg_orders_cdc') }}
    WHERE NOT is_deleted

    {% if is_incremental() %}
    AND operation_timestamp > (
        SELECT DATEADD('hour', -1, MAX(valid_from)) FROM {{ this }}
    )
    {% endif %}
),
filtered_changes AS (
    -- Includi solo righe dove qualcosa è effettivamente cambiato
    SELECT *
    FROM cdc_with_prev
    WHERE operation_type = 'c'                  -- nuovi inserimenti
       OR status != prev_status                  -- cambio di stato
       OR ABS(total_amount - prev_total) > 0.01  -- cambio importo
)
SELECT
    -- Chiave surrogata per questa versione del record
    {{ dbt_utils.generate_surrogate_key(['order_id', 'operation_timestamp']) }} AS order_state_id,
    order_id,
    customer_id,
    status,
    total_amount,
    operation_timestamp                          AS valid_from,
    COALESCE(next_op_at, '9999-12-31'::TIMESTAMP) AS valid_to,
    next_op_at IS NULL                           AS is_current
FROM filtered_changes

Idempotence: The Fundamental Property

An incremental model is idempotent whether to run it multiple times with the same data it always produces the same result. It is a critical guarantee in production: automatic retries, re-processing for corrections, and full refreshes do not have to produce different results.

Anti-pattern: Non-Idempotent Patterns

-- ANTI-PATTERN: questo modello NON è idempotente

{{ config(materialized='incremental') }}

SELECT
    order_id,
    customer_id,
    total_amount,
    CURRENT_TIMESTAMP AS inserted_at  -- PROBLEMA: ogni re-run produce timestamp diversi
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
WHERE order_id NOT IN (SELECT order_id FROM {{ this }})  -- PROBLEMA: lento su tabelle grandi
{% endif %}

-- Il problema: se la pipeline ri-processa lo stesso giorno per una correzione,
-- CURRENT_TIMESTAMP cambierà e avrai righe duplicate o inconsistenti.

-- CORRETTO: modello idempotente

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

SELECT
    order_id,
    customer_id,
    total_amount,
    created_at,           -- usa il timestamp dalla sorgente, non CURRENT_TIMESTAMP
    updated_at            -- timestamp di aggiornamento della sorgente
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
-- Filtro basato su updated_at (non su NOT IN, che è O(n*m)):
WHERE updated_at > (
    SELECT MAX(updated_at) FROM {{ this }}
)
{% endif %}

-- Con merge su unique_key='order_id':
-- * Re-run dello stesso range di dati: MERGE aggiornerà le righe esistenti
--   con gli stessi valori → nessun cambiamento netto (idempotente)
-- * Nuovi dati: MERGE inserirà le nuove righe
-- * Dati corretti retroattivamente: MERGE aggiornerà correttamente

Idempotence Testing

-- tests/assert_incremental_idempotent.sql
-- Verifica che eseguire il modello due volte non cambi il conteggio delle righe

-- Questo test è un placeholder: l'idempotenza vera si verifica in CI
-- confrontando il conteggio prima e dopo un secondo run sugli stessi dati.

-- Test pratico: verifica che non ci siano duplicati sulla unique_key
SELECT
    order_id,
    COUNT(*) AS occurrences
FROM {{ ref('fct_orders') }}
GROUP BY order_id
HAVING COUNT(*) > 1

-- Se questo test passa, il merge sta funzionando correttamente.

Management of Late-Arriving Data

I late-arriving data they are timestamped events that arrive in the pipeline late. It is a common problem with mobile sources (offline apps), systems with retry, and batch pipelines that arrive with variable delay.

-- Pattern 1: Lookback window fissa
-- Semplice e robusto per la maggior parte dei casi

{{ config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge'
) }}

SELECT
    event_id,
    user_id,
    event_type,
    event_timestamp,
    received_at  -- quando è arrivato nel warehouse
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_timestamp >= (
    -- Lookback di 48 ore per catturare late-arriving data
    SELECT DATEADD('hour', -48, MAX(event_timestamp))
    FROM {{ this }}
)
{% endif %}

-- Pattern 2: Lookback adattivo basato sul lag storico
-- Più sofisticato: calcola il lookback in base al ritardo osservato

{{ config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge'
) }}

{% set lookback_hours = var('event_lookback_hours', 48) %}

WITH late_data_analysis AS (
    -- Calcola il ritardo medio degli ultimi 7 giorni
    -- (solo in produzione e quando la tabella esiste)
    {% if is_incremental() %}
    SELECT
        AVG(DATEDIFF('hour', event_timestamp, received_at)) AS avg_lag_hours,
        MAX(DATEDIFF('hour', event_timestamp, received_at)) AS max_lag_hours
    FROM {{ this }}
    WHERE received_at > DATEADD('day', -7, CURRENT_TIMESTAMP)
    {% else %}
    SELECT {{ lookback_hours }} AS avg_lag_hours, {{ lookback_hours }} AS max_lag_hours
    {% endif %}
),
dynamic_lookback AS (
    -- Usa il 95° percentile del ritardo storico + 20% di buffer
    SELECT GREATEST(
        {{ lookback_hours }},
        CEIL(max_lag_hours * 1.2)
    ) AS lookback_hours
    FROM late_data_analysis
)
SELECT
    e.event_id,
    e.user_id,
    e.event_type,
    e.event_timestamp,
    e.received_at
FROM {{ ref('stg_events') }} e
CROSS JOIN dynamic_lookback dl

{% if is_incremental() %}
WHERE e.event_timestamp >= DATEADD('hour', -dl.lookback_hours,
    (SELECT MAX(event_timestamp) FROM {{ this }}))
{% endif %}

Incremental Model Monitoring

-- models/admin/incremental_health_check.sql
-- Modello di monitoraggio: traccia metriche degli incrementali in produzione

{{ config(materialized='table') }}

SELECT
    'fct_orders'                                         AS model_name,
    COUNT(*)                                             AS total_rows,
    MIN(created_at)                                      AS earliest_record,
    MAX(updated_at)                                      AS latest_record,
    DATEDIFF('hour', MAX(updated_at), CURRENT_TIMESTAMP) AS hours_since_last_update,
    COUNT(DISTINCT DATE_TRUNC('day', created_at))        AS distinct_days_covered,
    CURRENT_TIMESTAMP                                    AS check_timestamp
FROM {{ ref('fct_orders') }}

UNION ALL

SELECT
    'fct_events_hourly',
    COUNT(*),
    MIN(event_hour),
    MAX(event_hour),
    DATEDIFF('hour', MAX(event_hour), CURRENT_TIMESTAMP),
    COUNT(DISTINCT DATE_TRUNC('day', event_hour)),
    CURRENT_TIMESTAMP
FROM {{ ref('fct_events_hourly') }}

Conclusions and Next Steps

Advanced incremental models are where data engineering on dbt requires reasoning careful about trade-offs: microbatch simplifies re-processing but adds complexity configuration; CDC requires deep understanding of message format; idempotence it is non-negotiable in systems that manage data retries and corrections.

The unifying criterion: each incremental model must answer the question “what happens if I run this twice on the same data?” If the answer is no is “the same result”, the model has a bug.

The final module of the series closes the circle with dbt in enterprise contexts: data governance, Semantic Layer with MetricFlow, and orchestration with Airflow.