如何让增量dbt模型同步源数据的删除记录至临时表？

现有dbt增量同步逻辑用于生成临时表emp__dbt_tmp，出现以下问题：

• 3月20日，源表raw.emp的一条记录成功同步到临时表和目标表；
• 3月22日，该记录在源表中被标记为删除（cdc_operation='D'、is_deleted=true），但这条删除变更未同步到临时表，导致目标表中对应记录始终存在。

需求：如何让删除标记的记录同步到临时表，并最终在目标表中完成删除操作？是否需要用post-hook或者其他增量策略？

不需要依赖post-hook，核心是调整增量过滤规则和merge语句逻辑，让删除变更能流入临时表，并触发目标表的删除动作，具体操作如下：

1. 放开增量同步的过滤限制 #

当前增量逻辑里加了is_deleted = false，直接把删除标记的记录拦在了外面。要删掉这个条件，只保留时间过滤（用目标表的最新更新时间做基准，而非硬写死时间），这样所有新增/更新的变更（包括删除）都能进入临时表。

2. 给merge语句加删除分支 #

默认的merge只有更新和插入逻辑，需要新增WHEN MATCHED AND 源表是删除标记 THEN DELETE的分支，当临时表传来删除记录时，直接删掉目标表的对应行。

3. 保留最新版本记录的逻辑 #

原有的ROW_NUMBER()按mandt+loc分区、按时间倒序取最新记录的逻辑可以保留，确保每个主键只取最新的变更（包括最新的删除标记）。

dbt模型核心代码 #

{{ config(
    materialized='incremental',
    unique_key=['mandt', 'loc'],
    on_schema_change='append_new_columns',
    partition_by={
        "field": "gcp_updated_ts",
        "data_type": "timestamp",
        "granularity": "day"  
    }
) }}

WITH source_data AS (
    SELECT 
        COALESCE(SAFE_CAST(mandt AS STRING), '') AS mandt, 
        COALESCE(SAFE_CAST(loc AS STRING), '') AS loc,
        SAFE_CAST(RECORDSTAMP AS TIMESTAMP) AS gcp_updated_ts, 
        SAFE_CAST(OPERATION_FLAG AS STRING) AS cdc_operation, 
        SAFE_CAST(IS_DELETED AS BOOLEAN) AS is_deleted,
        -- 按主键分区，取最新的变更记录
        ROW_NUMBER() OVER (PARTITION BY mandt, loc ORDER BY recordstamp DESC) AS row_num
    FROM {{ source('raw', 'emp') }}
   
    {% if is_incremental() %}
        -- 用目标表的最新时间做过滤，而非硬写死时间
        WHERE recordstamp > (SELECT MAX(gcp_updated_ts) FROM {{ this }})
        -- 移除is_deleted = false，允许删除标记的记录进入
    {% endif %}
)

SELECT * EXCEPT(row_num) FROM source_data WHERE row_num = 1

自定义merge语句（通过post-hook实现） #

如果dbt默认生成的merge不支持删除分支，可以通过post-hook自定义完整的merge逻辑：

{{ config(
    materialized='incremental',
    unique_key=['mandt', 'loc'],
    on_schema_change='append_new_columns',
    partition_by={
        "field": "gcp_updated_ts",
        "data_type": "timestamp",
        "granularity": "day"  
    },
    post_hook=[
        """
        MERGE INTO `vk`.`transient`.`emp` AS DBT_INTERNAL_DEST
        USING (SELECT * FROM `vk`.`transient`.`emp__dbt_tmp`) AS DBT_INTERNAL_SOURCE
        ON (
            DBT_INTERNAL_SOURCE.mandt = DBT_INTERNAL_DEST.mandt
            AND DBT_INTERNAL_SOURCE.loc = DBT_INTERNAL_DEST.loc
        )
        -- 匹配到删除标记记录时，删除目标表行
        WHEN MATCHED AND DBT_INTERNAL_SOURCE.is_deleted = TRUE THEN DELETE
        -- 匹配到正常更新记录时，更新目标表行
        WHEN MATCHED THEN UPDATE SET
            `mandt` = DBT_INTERNAL_SOURCE.`mandt`,
            `loc` = DBT_INTERNAL_SOURCE.`loc`,
            `posn` = DBT_INTERNAL_SOURCE.`posn`,
            `gcp_updated_ts` = DBT_INTERNAL_SOURCE.`gcp_updated_ts`,
            `is_deleted` = DBT_INTERNAL_SOURCE.`is_deleted`,
            `cdc_operation` = DBT_INTERNAL_SOURCE.`cdc_operation`
        -- 未匹配到记录时，插入新行
        WHEN NOT MATCHED THEN INSERT
            (`mandt`, `loc`, `posn`,`gcp_updated_ts`,`is_deleted`,`cdc_operation`)
        VALUES
            (`mandt`, `loc`, `posn`,`gcp_updated_ts`,`is_deleted`,`cdc_operation`)
        """
    ]
) }}

-- 下面是原有的source_data逻辑，和上面的模型代码一致
WITH source_data AS (
    SELECT 
        COALESCE(SAFE_CAST(mandt AS STRING), '') AS mandt, 
        COALESCE(SAFE_CAST(loc AS STRING), '') AS loc,
        SAFE_CAST(RECORDSTAMP AS TIMESTAMP) AS gcp_updated_ts, 
        SAFE_CAST(OPERATION_FLAG AS STRING) AS cdc_operation, 
        SAFE_CAST(IS_DELETED AS BOOLEAN) AS is_deleted,
        ROW_NUMBER() OVER (PARTITION BY mandt, loc ORDER BY recordstamp DESC) AS row_num
    FROM {{ source('raw', 'emp') }}
   
    {% if is_incremental() %}
        WHERE recordstamp > (SELECT MAX(gcp_updated_ts) FROM {{ this }})
    {% endif %}
)

SELECT * EXCEPT(row_num) FROM source_data WHERE row_num = 1

内容的提问来源于stack exchange，提问作者Vaishak Nambiar