dbtを用いてソースデータに削除レコードが発生した場合にsnapshotで管理テーブルを作成しincremental modelsの既存データを削除する仕組みを考えてみた

【課題のイメージ】
ソース: [A, B, C] → [A, C]  (Bを削除)
既存データ: [A, B, C] → [A, B, C]  (Bが残る)

【処理フロー】
1. snapshot: ソースの変更を追跡（削除も検知）
2. snapshotのhard_deletes: 削除対象のORDER_IDを特定
3. orders_incremental: post-hookで削除を実行

dbt_snapshot_test/
├── dbt_project.yml
├── models/
│   ├── sources.yml
│   └── orders_incremental.sql  # post-hookで削除実行
└── snapshots/
    ├── orders_snapshot.sql     # hard_deletes='new_record'
    └── schema.yml     #snapshotの設定

version: 2

sources:
  - name: raw
    database: KAWABATA_MART_DB
    schema: DBT_TKAWABATA
    tables:
      - name: orders_row_ts_test
        description: "注文テーブル（削除・更新が発生する）"

version: 2

snapshots:
  - name: orders_snapshot
    config:
      unique_key: ORDER_ID
      strategy: check
      check_cols: all
      hard_deletes: new_record

{% snapshot orders_snapshot %}

SELECT * FROM {{ source('raw', 'orders_row_ts_test') }}

{% endsnapshot %}

{{
    config(
        materialized='incremental',
        unique_key='ORDER_ID',
        incremental_strategy='merge',
        post_hook=[
            "DELETE FROM {{ this }} WHERE ORDER_ID IN (SELECT ORDER_ID FROM {{ target.database }}.{{ target.schema }}.ORDERS_SNAPSHOT WHERE DBT_IS_DELETED = 'True')"
        ]
    )
}}

SELECT
    ORDER_ID,
    LOCATION_ID,
    CUSTOMER_ID,
    SUBTOTAL,
    TAX_PAID,
    ORDER_TOTAL,
    ORDER_DATE,
    ORDER_COST,
    COUNT_ORDER_ITEMS,
    IS_FOOD_ORDER,
    IS_DRINK_ORDER,
    CUSTOMER_ORDER_NUMBER,
    {# Row timestamps: ソースのコミット時刻を保持 #}
    METADATA$ROW_LAST_COMMIT_TIME AS ROW_TIMESTAMP
FROM {{ source('raw', 'orders_row_ts_test') }}

{% if is_incremental() %}
WHERE 
    {# 
      条件1: 前回実行以降にコミットされたレコード（新規・更新）
      - METADATA$ROW_LAST_COMMIT_TIME: ソースの各行のコミット時刻
      - ROW_TIMESTAMP: ターゲットに保存済みの最新コミット時刻
    #}
    METADATA$ROW_LAST_COMMIT_TIME > (
        SELECT COALESCE(MAX(ROW_TIMESTAMP), '1900-01-01'::TIMESTAMP_LTZ)
        FROM {{ this }}
    )

{% endif %}

{#
  【処理フロー】
  1. 初回: フルロード（全レコード + ROW_TIMESTAMP）
  2. 2回目以降:
     - Row timestamps比較で新規・更新を検知
     - snapshotのDBT_IS_DELETEDで削除を検知
  3. MERGE: ORDER_IDで既存データと照合・更新
  4. POST-HOOK: 削除対象を物理削除
#}

SELECT COUNT(*) AS source_count 
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST

dbt snapshot

dbt run --select orders_incremental

DELETE FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST 
WHERE ORDER_ID IN (
    '528273e8-4c5b-4f98-91c2-31dfb99d3e00',
    '5dcd1dc2-bf9e-485a-86aa-3e78dee5a41a'
);

dbt snapshot

dbt run --select orders_incremental

-- 3テーブルの件数確認
SELECT 
    'ORDERS_INCREMENTAL' AS table_name,
    COUNT(*) AS total_count,
    NULL AS deleted_count
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_INCREMENTAL
UNION ALL
SELECT 
    'ORDERS_ROW_TS_TEST',
    COUNT(*),
    NULL
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST
UNION ALL
SELECT 
    'ORDERS_SNAPSHOT',
    COUNT(*),
    SUM(CASE WHEN DBT_IS_DELETED = 'True' THEN 1 ELSE 0 END)
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_SNAPSHOT;