dbt incremental_predicates の使い所: 一部の行だけを更新対象に限定する実務パターン

dbt のインクリメンタルモデルは is_incremental() で「取り込むソース行」を絞れますが、ターゲット側（既存テーブル）への更新対象までは自動では絞り込みません。ここで効くのが incremental_predicates です。

本稿では、MERGE 時に対象行をさらに限定する incremental_predicates の設計ポイント、典型ユースケース、アダプター差分、試験で問われやすい観点をまとめます。

incremental_predicates とは何か、なぜ必要か

incremental_predicates は、インクリメンタルモデルの MERGE（またはアダプターが対応する同等戦略）で、ターゲット側に追加の絞り込み条件を与えるための設定です。実行時にはアダプターのマクロ内で、MERGE の対象（多くの場合、ターゲット表のエイリアス）に対する AND 条件として適用され、更新・削除の候補を限定します。

is_incremental() によるソース側の絞り込みだけでは、ターゲット側は全体をスキャンしがちです。特に巨大なパーティション/クラスターを持つ事実テーブルでは、MERGE のマッチ探索が高コスト・高ロックになりやすい。incremental_predicates により「影響し得るターゲット行」にまで範囲を狭めると、スキャン量・課金・ロック時間のいずれも下げられます。

• 使い所: 日次/時間分割の巨大 Fact テーブル、CDC で一部キーのみが変化するテーブル
• 効果: MERGE の対象縮小（テーブルスキャン回避、パーティション/クラスターの絞り込み）
• 注意: 条件が狭すぎると取りこぼしが起きる。遅延到着データや再処理設計とセットで検討する

MERGE 時の対象縮小イメージ

基本形: リストで条件を渡す（アダプターの内部エイリアスを利用）

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    -- 例: 影響し得るパーティションに限定（内部エイリアス名はアダプターで固定化されていることが多い）
    incremental_predicates=[
      "DBT_INTERNAL_DEST.order_date >= (select min(order_date) from DBT_INTERNAL_SOURCE)",
      "DBT_INTERNAL_DEST.order_date <= (select max(order_date) from DBT_INTERNAL_SOURCE)"
    ]
) }}

select ... from {{ ref('stg_orders') }}
{% if is_incremental() %}
  where updated_at >= (select max(updated_at) from {{ this }})
{% endif %}

ユースケース1: パーティション範囲でターゲットを限定

日付や時間で分割された巨大テーブルでは、増分ソースが触れる期間にだけターゲットを限定するのが定石です。ソース側の最小・最大の分割値を使い、ターゲットの同一範囲にだけ MERGE をかけます。

BigQuery/Snowflake/Databricks いずれも、分割列に対する述語は強力にプルーニングされます。is_incremental() で「新着のみ」を読みつつ、incremental_predicates で「対象パーティションのみ」を更新。二段構えにすることで、読み取りも書き込みも縮小できます。

• 分割列は日付（DATE）や日付時刻（TIMESTAMP）で定義済みであることが前提
• ソース側の min/max を使うと、遅延到着を多少含めても安全側の範囲になる
• 長期の再処理が必要な場合は、対象期間を可変にできるようパラメータ化しておく

例: BigQuery の日付パーティションを範囲指定

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    partition_by={'field': 'order_date', 'data_type': 'date'},
    incremental_predicates=[
      -- SOURCE の最小/最大日付で TARGET 側パーティションを限定
      "DBT_INTERNAL_DEST.order_date between (select min(order_date) from DBT_INTERNAL_SOURCE)
                                       and (select max(order_date) from DBT_INTERNAL_SOURCE)"
    ]
) }}

select ... from {{ ref('stg_orders') }}
{% if is_incremental() %}
  where order_date >= date_sub(current_date(), interval 7 day)
{% endif %}

ユースケース2: 変更キー群でターゲットを限定（CDC キー絞り込み）

CDC（変更データキャプチャ）では、1 回の取り込みで変わるキーはごく一部です。そこで「今回触れたキー集合」を別モデル（エフェメラルでも可）で作り、ターゲット側をその集合に限定します。

これにより、大規模なターゲットの全スキャンを避けつつ、ピンポイントで MERGE が可能になります。

• キー集合は重複排除（distinct）しておく
• エフェメラルモデルは SQL にインライン展開されるため、参照コストが低い
• 複合キーの場合は連結や構造体化で結合キーを一致させる

例: 今回触れた customer_id のみに限定

-- models/int_keys_this_run.sql (ephemeral)
{{ config(materialized='ephemeral') }}
select distinct customer_id
from {{ ref('stg_cdc_events') }}
{% if is_incremental() %}
  where _ingested_at >= (select coalesce(max(_ingested_at), '1970-01-01') from {{ this }})
{% endif %}

-- models/fct_customers.sql (incremental)
{{ config(
    materialized='incremental',
    unique_key='customer_id',
    incremental_strategy='merge',
    incremental_predicates=[
      "DBT_INTERNAL_DEST.customer_id in (select customer_id from {{ ref('int_keys_this_run') }})"
    ]
) }}
select ... from {{ ref('stg_customer_snapshot') }}

運用の注意点と落とし穴

取りこぼし防止は最優先です。遅延到着やアップサートの再送を見越し、incremental_predicates はソースの min/max など安全側の範囲で組むのが無難です。キー限定の場合も、再処理ウィンドウ（例: 過去 N 日のキーを常に含める）を設けると安心です。

スキーマ変更や列追加との相性にも注意します。MERGE で更新対象列を限定する場合は、merge_update_columns と on_schema_change を併用して整合性を保ちます。フルリフレッシュ時は incremental_predicates は無視されるため、全件再構築されます。

• 遅延到着対策として「過去 N 日まとめて再更新」か「キー集合に過去 N 日分追加」をどちらか必ず用意
• merge_update_columns は必要最小限に。不要列の更新は避ける
• 監視: 更新件数、スキャン量、対象パーティション数をメトリクス化

設定例: 更新列の限定とスキーマ変更方針

{{ config(
    materialized='incremental',
    unique_key='id',
    incremental_strategy='merge',
    merge_update_columns=['status', 'updated_at'],
    on_schema_change='append_new_columns',  -- アダプターにより選択肢は異なる
    incremental_predicates=[
      "DBT_INTERNAL_DEST.dt >= dateadd(day, -3, current_date)"  -- Snowflake 例
    ]
) }}
select id, status, updated_at, dt from {{ ref('stg_events') }}

アダプター差分と検証手順

incremental_predicates の適用位置や内部エイリアス名は、アダプター実装（BigQuery、Snowflake、Databricks／Spark など）に依存します。多くの実装では MERGE の WHEN MATCHED（あるいは対象テーブル別名）に AND 条件として差し込まれますが、戦略が insert_overwrite の場合などは無関係になることがあります。

本番前には必ず dbt compile で生成 SQL を確認し、該当述語が期待どおりにターゲット側に適用されているかを目視で検証します。さらに小規模データで A/B 実行して件数・スキャン量の差分を計測しておくと安全です。

• MERGE を使わない append 戦略では incremental_predicates は効果を持たない
• insert_overwrite（特に BigQuery のパーティション上書き）では別の最適化が有効になる
• コンパイル結果に DBT_INTERNAL_SOURCE/DEST がどう展開されるかを必ず確認

検証のための基本コマンドとチェック項目

# コンパイルして SQL を確認
$ dbt compile -s models/fact_orders.sql

# 生成 SQL 中の該当箇所（例）
-- WHEN MATCHED AND (DBT_INTERNAL_DEST.order_date between ...)
-- が想定どおりに入っているか

# 小規模サンプルでの比較
$ dbt run -s fact_orders --vars '{sample: true}'
$ dbt run -s fact_orders --full-refresh  # ベースライン比較

# メトリクス
-- 対象パーティション数、更新件数、スキャンバイト/時間

試験（Analytics Engineer）で問われやすいポイント

is_incremental() はソース側の行を絞る。incremental_predicates はターゲット側の更新対象を絞る。役割の違いを明確に説明できるようにしましょう。

append 戦略や insert_overwrite 戦略では挙動が異なる点、MERGE 戦略における効果が本質である点を押さえる。パーティション列やキー集合による限定の設計パターンを、短い SQL で書けるようにしておくと実務・試験ともに有利です。

• 定義: incremental_predicates は MERGE 対象の限定用の追加述語（アダプター適用）
• 適用シーン: 大規模パーティション表、CDC で一部キーのみ更新
• 非適用: append 戦略（更新がない）、insert_overwrite（上書き粒度の設計が要点）

頻出パターンのテンプレート

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    incremental_predicates=[
      "DBT_INTERNAL_DEST.partition_date between (select min(partition_date) from DBT_INTERNAL_SOURCE)
                                         and (select max(partition_date) from DBT_INTERNAL_SOURCE)"
    ]
) }}
select ... from {{ ref('stg_orders_incremental') }}
{% if is_incremental() %}
  where partition_date >= date_sub(current_date(), interval 3 day)
{% endif %}

問題で確認

よくある質問