Incremental Models and Scheduling on Snowflake

The pipeline works. Every model builds, the tests pass, and the DAG is green. It’s also quietly expensive in a way a demo never shows you: every run reprocesses all of TPCH from scratch. On your laptop that’s free. On a Snowflake warehouse, every full rebuild is a create table as select that scans the whole source and bills you for the compute — and most of what it recomputes hasn’t changed since yesterday. Orders from three years ago don’t get re-placed overnight. Rebuilding them anyway is money you’re setting on fire on a schedule.

Two changes fix the two halves of that waste. Incremental models stop you from reprocessing rows that haven’t moved. A schedule that follows the data stops you from running at all when there’s nothing new. Together they’re the shape almost every production pipeline lands on.

Incremental: process only what’s new

A normal dbt model rebuilds its whole table every run. An incremental model rebuilds it once, then on every subsequent run only processes the rows you tell it are new. The natural candidate is the orders_enriched mart from the last chapter — the one that ended with the promise that this post would turn it incremental. As it stands it’s a full rebuild of every order ever placed, every night:

-- models/marts/orders_enriched.sql
{{ config(
    materialized='incremental',
    unique_key='order_id'
) }}

select
    order_id,
    customer_id,
    customer_name,
    status,
    order_date,
    nation,
    region,
    order_revenue,
    line_item_count
from {{ ref('int_orders_enriched') }}

{% if is_incremental() %}
where order_date > (select max(order_date) from {{ this }})
{% endif %}

Notice what didn’t change: it still selects from int_orders_enriched, and it still carries order_revenue and line_item_count. That matters more than it looks. The intermediate model is where the last chapter collapsed line items to order grain, and reaching past it to re-join stg_orders and stg_lineitem here — which is the tempting shortcut, since you’re already writing a where clause — would quietly reintroduce the fan-out bug the whole intermediate layer exists to prevent, and this time inside an incremental model where the inflated rows get merged in and persist. The rule that falls out: making a model incremental changes when its rows are computed, never where they come from. If you find yourself rewriting the from clause to add a filter, stop — filter the model you already had.

Two things are doing the work. materialized='incremental' tells dbt to build the table once and append to it after that. The is_incremental() block is a filter that only applies on runs where the table already exists — it reads the latest order_date in the table ({{ this }} is the model’s own relation) and pulls only orders newer than that. On the first run, is_incremental() is false, the where clause disappears, and you get a full build. On every run after, it’s the delta and nothing else.

unique_key='order_id' is what makes this safe when a row can change, not just arrive. With a unique key set, dbt doesn’t blindly append — it MERGEs. New keys insert; existing keys update in place. Without it you’d get duplicates the first time an order’s status changed and the row came through twice.

Four strategies, and why Snowflake makes you choose

MERGE is the default because it’s correct, not because it’s cheap. dbt-snowflake exposes four incremental_strategy values, and the whole game is picking the one that matches how your data actually mutates:

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

merge (the default) runs a single MERGE INTO … USING (your select) ON target.order_id = source.order_id — matched rows update, unmatched insert. It handles the general case: rows that both arrive and change. The cost is that a MERGE has to locate every matched row in the target, and Snowflake locks the target table’s affected micro-partitions for the duration. On a wide, unclustered table that’s a lot of partitions to touch.

delete+insert does exactly what it says: delete from target where order_id in (select order_id from source), then a plain insert. Two statements instead of one atomic merge. It’s often cheaper than merge when your new batch is a well-defined slice — a day’s partition — because the delete can prune to that slice and the insert is an append with no per-row matching logic. The tradeoff is a brief window between the delete and the insert where those rows don’t exist; inside dbt’s transaction that’s invisible to other sessions, but it’s why delete+insert wants a unique_key that cleanly identifies the batch you’re replacing.

append skips matching entirely — no unique_key, no delete, no merge. Every run just inserts the new rows. It’s the cheapest strategy that exists and the easiest to get wrong. append is only safe when a row, once written, is immutable and can never arrive twice: append-only event logs, immutable fact rows keyed on an event id you’ve already deduplicated upstream. Point it at TPCH orders — where o_orderstatus flips from O to F as an order fills — and you’ll accumulate one row per status change per order, silently, forever. Use it for facts that are born final.

microbatch is the strategy that earns its own section below, because it changes how the window itself is computed.

Here’s the same model as delete+insert, which for a daily partition of orders is usually the better call:

{{ config(
    materialized='incremental',
    unique_key='order_date',
    incremental_strategy='delete+insert'
) }}

Note the unique_key changed meaning. For merge it’s the grain of a row (order_id). For delete+insert it’s the grain of the batch you replace — set it to order_date and each run deletes the whole day and reinserts it, which makes reruns idempotent: run the same day twice and you get the same table, not doubled rows. That idempotency is worth more than it looks; it’s what lets a backfill re-run a window without you cleaning up first.

incremental_predicates: bounding the MERGE scan

The single largest cost lever on a Snowflake merge isn’t the strategy — it’s how much of the target the merge has to scan to find its matches. By default, MERGE … ON target.order_id = source.order_id gives Snowflake license to consider the entire target table. If nothing tells the optimizer that matches only live in the last few days, it may scan micro-partitions going back years.

incremental_predicates bolts an extra condition onto the merge’s ON clause, pruning the target before matching begins:

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    incremental_predicates=[
        "DBT_INTERNAL_DEST.order_date >= dateadd(day, -7, current_date)"
    ]
) }}

DBT_INTERNAL_DEST is dbt’s alias for the target table inside the compiled merge. That predicate becomes part of the join condition, so Snowflake prunes to the last seven days of target partitions before it looks for a single match. If your incoming batch only ever touches recent orders — which for late-arriving data it does — you’ve turned a full-history scan into a seven-day scan. On a large fact table this is frequently the difference between a merge that costs cents and one that costs dollars, on every single run.

The predicate has to be true for every row your source might update, or you’ll skip a legitimate match and leave a stale row behind. Size the window to your worst realistic lateness, not your average one.

Clustering so the predicate can prune

incremental_predicates only saves you if Snowflake can actually skip partitions using it — and it can only skip partitions that are physically organized by the predicate’s column. TPCH sample orders happen to load roughly in order_date order, so date pruning works out of the box. Real ingestion rarely lands that tidily. When a table’s rows are scattered across partitions by load order rather than by date, a where order_date >= … still has to open most partitions to check.

cluster_by tells Snowflake to co-locate rows by the columns you actually filter on:

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    cluster_by=['order_date'],
    incremental_predicates=[
        "DBT_INTERNAL_DEST.order_date >= dateadd(day, -7, current_date)"
    ]
) }}

Now the merge predicate and the physical layout agree: recent orders sit together, the seven-day predicate prunes to a handful of partitions, and the merge scans almost nothing. Clustering isn’t free — Snowflake maintains the order in the background and bills for the reclustering credits — so it pays off on large, frequently-merged tables and is overkill on a small one. The heuristic: cluster the incremental target on the same column your incremental_predicates bound, and the two features multiply. Cluster on a column you never filter on, and you’re paying maintenance for nothing.

on_schema_change: what happens when the SELECT grows a column

An incremental table already exists with a fixed set of columns. Add a column to the model’s select and the next incremental run has to decide what to do with a target that doesn’t have that column yet. on_schema_change is that decision:

{{ config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='append_new_columns'
) }}

ignore (the historical default) drops the new column silently on incremental runs — your model emits it, the target doesn’t have it, dbt just doesn’t write it, and you get a confusing gap until the next full-refresh. append_new_columns runs an alter table … add column and starts populating it going forward (old rows stay null, since the filter never revisited them). sync_all_columns adds new columns and drops ones you removed. fail aborts the run loudly, which is the right choice when a schema change should be a deliberate, reviewed event rather than something that happens on a Tuesday. Set it to append_new_columns for models under active development and fail for the ones you consider stable.

The boundary problem: > vs >=, and lateness

The where order_date > (select max(order_date) from {{ this }}) filter has a bug hiding in the comparison operator, and it’s the classic incremental footgun. Orders don’t arrive perfectly ordered by date. If your max(order_date) in the table is 2026-06-08, and after that run three more 2026-06-08 orders land in the source (a late file, a slow upstream), the next run’s > '2026-06-08' filter excludes them — they’re not strictly greater than the max you already saw. You’ve silently dropped a day’s tail.

Switch to >= and you re-scan the boundary day, catching the stragglers — but now you reprocess every row from that day, and unless your unique_key/MERGE dedupes them, you double-count. >= with a merge on order_id is correct: you re-read the boundary day, but the merge overwrites the rows you’d already written instead of appending duplicates. That’s the safe combination — a small deliberate overlap that the merge reconciles.

For data that can arrive genuinely late — hours or days after its business date — a one-row boundary isn’t enough. You want a lookback window:

{% if is_incremental() %}
where order_date >= (
    select dateadd(day, -3, max(order_date)) from {{ this }}
)
{% endif %}

This reprocesses the last three days on every run. Combined with a merge, late orders up to three days stale get corrected, and the merge keeps the table from doubling. The cost is three days of rescan per run instead of one, which is exactly why the incremental_predicates and clustering above matter — the lookback defines how much you must read, and clustering makes reading it cheap.

microbatch: let Airflow define the window

Every filter so far derives its window from the data — max(order_date) — which means the model’s idea of “new” depends on the current state of the table. microbatch, dbt’s newer incremental materialization, inverts that: the window comes from time, sliced into fixed batches, and each batch is processed independently.

{{ config(
    materialized='incremental',
    incremental_strategy='microbatch',
    event_time='order_date',
    batch_size='day',
    lookback=3,
    begin='2024-01-01'
) }}

select
    order_id,
    customer_id,
    customer_name,
    status,
    order_date,
    nation,
    region,
    order_revenue,
    line_item_count
from {{ ref('int_orders_enriched') }}

There’s no is_incremental() block at all. You declare event_time (the column that timestamps each row), batch_size (day here), and lookback (reprocess the last N batches to absorb late data), and dbt slices the run into one filtered statement per day. Each batch does an atomic replace of its own slice — so it’s idempotent by construction, and a failed batch doesn’t poison its neighbors. lookback=3 means every run also re-runs the previous three days’ batches, which is the boundary/late-arrival fix from above, expressed as configuration instead of hand-written SQL.

The reason microbatch belongs in a scheduling chapter is the payoff: dbt derives the batch window from dbt run’s --event-time-start/--event-time-end, and Airflow already hands every run a data interval. Wire the DAG’s interval to those flags and a scheduled run processes exactly the slice it was scheduled for — nothing more, nothing less — and a backfill of last March processes March’s batches with March’s boundaries, not “everything since max(order_date)”. The window stops depending on table state and starts depending on the run’s identity, which is precisely what makes backfills correct.

Full-refresh: cost, locks, and triggering it through Cosmos

Every incremental strategy shares one hazard: the table drifts from what a clean rebuild would produce. Change the join, fix a bug in the logic, add a column that on_schema_change can’t backfill into old rows — the existing rows still hold the old output, because your filter never revisited them. --full-refresh is the reset:

dbt build --select orders_enriched --full-refresh

It ignores the incremental logic, issues a create or replace table … as select over the entire history, and gives you a table that matches the current SQL exactly. Two things to respect about it on Snowflake: it costs — you’re scanning and rewriting the whole thing, the one bill incremental was built to avoid — and create or replace swaps the table atomically, so readers see the old table until the instant it’s replaced rather than a lock that blocks them. The expensive part is compute, not contention.

Because full-refresh is occasional and deliberate, you don’t want it on the daily schedule. You want a way to trigger it on demand. Cosmos lets you pass dbt flags straight through, so the cleanest pattern is a separate param-driven DAG whose whole job is a full-refresh:

from airflow.sdk import dag, Param
from cosmos import DbtTaskGroup, ProjectConfig, RenderConfig
from datetime import datetime

@dag(
    schedule=None,                       # manual trigger only
    start_date=datetime(2026, 6, 1),
    catchup=False,
    params={"select": Param("marts", type="string")},
)
def tpch_full_refresh():
    DbtTaskGroup(
        group_id="dbt_full_refresh",
        project_config=ProjectConfig("/opt/airflow/dbt/tpch"),
        render_config=RenderConfig(
            select=["{{ params.select }}"],
            dbt_cmd_flags=["--full-refresh"],
        ),
    )

tpch_full_refresh()

schedule=None means it only runs when a human triggers it. The select param lets you point it at one model or a whole layer from the trigger form, and dbt_cmd_flags=["--full-refresh"] is what turns every rendered dbt run/dbt build command in the group into its full-refresh variant. Ship a logic change, trigger this DAG for the affected model, and the daily incremental DAG goes back to processing deltas the next morning — clean slate, no code edit to the scheduled pipeline.

Scheduling: a clock, then the data

The Cosmos-rendered DAG from earlier in this series turns your dbt project into Airflow tasks — one per model, wired in dependency order (that’s the Cosmos series in a sentence). Giving it a schedule is a single argument:

from airflow.sdk import DAG
from cosmos import DbtDag

dag = DbtDag(
    schedule="@daily",
    catchup=False,
    max_active_runs=1,
    ...
)

In Airflow 3.x the argument is the unified schedule= — @daily fires the whole pipeline once a day. catchup=False is the setting you almost always want: without it, a DAG created today with a start date last month will immediately try to run every day it “missed,” which for an incremental pipeline is a stampede of runs you never asked for. Leave catchup off unless you specifically want to backfill.

max_active_runs=1 matters more for an incremental pipeline than for a normal one, and it’s easy to forget. If two runs overlap — a slow run still merging while the next one starts — you have two MERGEs racing on the same target table. At best they serialize on Snowflake’s partition locks and one stalls; at worst, with a max(order_date) filter, the second run reads a max the first hasn’t finished writing and picks the wrong window. Pinning to one active run makes the merges strictly sequential, which is the only safe way to run stateful incremental logic on a schedule.

@daily is a guess, though — a clock hoping the data landed before it fired. The better trigger is the data itself. Airflow 3.x renames Datasets to Assets, and a DAG can be scheduled on one:

from airflow.sdk import Asset

raw_orders = Asset("snowflake://analytics_prod/raw/orders")

transform_dag = DbtDag(
    schedule=[raw_orders],
    max_active_runs=1,
    ...
)

This DAG is the consumer side of the asset. It doesn’t run at midnight and hope; it runs when whatever produces raw_orders finishes and updates it. And that producer isn’t hypothetical — it’s the ingestion DAG this trilogy builds in the ingestion chapter, whose COPY INTO/Snowpipe/EL task declares the same asset in its outlets:

# in the ingest DAG (chapter 08) — the PRODUCER
from airflow.sdk import Asset

@task(outlets=[Asset("snowflake://analytics_prod/raw/orders")])
def load_raw_orders():
    ...  # COPY INTO ANALYTICS_PROD.RAW.orders

The URIs match, so the loop closes: ingest lands the day’s orders and updates the asset, the update triggers the transform, and nobody wired a timer between them. The pipeline runs because the data arrived. If ingest is late, the transform waits; if ingest fails, the transform never fires on stale input. That’s the event-driven scheduling the Airflow series covers, applied to the whole dbt layer at once.

When the transform needs more than one input

Orders alone rarely make a mart. revenue_by_region needs line items and orders and the customer/nation dimensions. Airflow lets you express “wait for all of these” or “wake on any of these” with AssetAll and AssetAny:

from airflow.sdk import Asset
from airflow.sdk.definitions.asset import AssetAll, AssetAny

raw_orders    = Asset("snowflake://analytics_prod/raw/orders")
raw_lineitem  = Asset("snowflake://analytics_prod/raw/lineitem")
raw_customer  = Asset("snowflake://analytics_prod/raw/customer")

transform_dag = DbtDag(
    schedule=AssetAll(raw_orders, raw_lineitem, raw_customer),
    max_active_runs=1,
    ...
)

AssetAll fires only once all three raw tables have been refreshed since the last transform run — the right choice here, because building revenue_by_region off a fresh orders but a stale lineitem would produce a half-updated number. AssetAny fires when any one updates, which suits a mart that should refresh the moment any of its inputs moves. The two compose, so AssetAll(orders, AssetAny(dim_a, dim_b)) is a legal, readable condition. Conditional scheduling is how you stop a transform from running on a partially-arrived dataset without writing a single sensor.

Backfills, catchup, and incremental state

There’s a sharp interaction between backfills and incremental state worth naming before it bites you. A max(order_date)-filtered model has no real concept of “which run am I” — it reads whatever’s in the table and pulls everything newer. Turn on catchup=True (or launch a scheduler-managed backfill) against that model and every “missed” run computes the same max(order_date) and pulls the same delta, because the earlier backfill runs haven’t committed in the order you’d expect. The state-based filter and the multi-run backfill fight each other.

Two fixes, both already on the table. microbatch is the clean one: its window comes from the run’s data interval, so backfill run for March 3rd processes March 3rd’s batch and only that — the runs are independent by construction. The delete+insert with unique_key='order_date' is the other: each run idempotently replaces its own day, so re-running a window is safe even if the ordering is odd. Airflow 3.x runs backfills through the scheduler (airflow backfill create, or the UI) rather than the old detached CLI daemon, which means a backfill respects max_active_runs=1 and your merges stay serialized instead of stampeding. The combination — scheduler-managed backfill, max_active_runs=1, an interval-keyed or idempotent incremental strategy — is what makes “re-run last quarter” a safe operation instead of a data-corruption event.

Tag the runs so the bill is legible

One more line pays for itself the first time someone asks “what did the transform layer cost last month.” Snowflake attaches a QUERY_TAG to every statement in a session, and it surfaces in ACCOUNT_USAGE.QUERY_HISTORY — so tagging your incremental runs lets you slice credits by exactly this pipeline. Set it in the dbt profile or per-run:

# profiles.yml
tpch:
  target: prod
  outputs:
    prod:
      type: snowflake
      query_tag: "dbt_incremental_transform"
      # ...connection details from post 3

Every merge, every full-refresh, every create or replace from this project now carries dbt_incremental_transform in the query history, and you can attribute warehouse spend to the incremental layer down to the statement. That thread — turning tags into a per-model cost report — is the whole subject of the cost chapter; here it’s enough to know the tag is the hook everything downstream hangs on, and it costs one line to set.

The shape it converges to

Put the pieces together and you get the pattern most production pipelines end up at without planning to. Ingest lands the day’s new orders and updates an asset. That asset — with AssetAll if the mart needs several inputs — triggers the Cosmos DAG. max_active_runs=1 keeps merges from overlapping. The dbt run is incremental with a merge and a bounded incremental_predicate, reading only the recent, clustered slice. A lookback window (or microbatch, keyed to Airflow’s data interval) absorbs late arrivals without double-counting. A separate param DAG fires --full-refresh when logic changes. And every statement is query-tagged so the bill is legible.

Daily cadence, incremental compute, driven by the data rather than the clock. Every piece is a config change, and none of them touched a line of business SQL. That’s the through-line of this whole stack: the interesting work stays in the models, and the machinery around them is configuration.

Final thoughts

Incremental models and asset-driven scheduling look like two separate optimizations — one saves warehouse credits, one saves you from running on empty. They’re really the same instinct wearing two hats: do work in proportion to what actually changed. A full rebuild on a fixed clock does maximum work regardless of input, which is fine until the bill or the data volume makes it not fine. The version that merges the bounded delta when the delta lands scales with your data instead of with the calendar. Get this shape right early — the right strategy, a predicate that prunes, clustering that lets it, and a schedule that follows the asset — and “the pipeline got slow” becomes a problem you never have to have.

Next: Production: The Pipeline That Runs Without You