dbt增量模型自引用问题：Redshift条件更新字段逻辑实现求助

核心思路 #

dbt增量模型无法直接自引用目标表，但可以借助Redshift原生的MERGE语句实现条件更新逻辑，结合dbt的is_incremental()宏区分全量初始化与增量更新场景。

调整后的模型代码 #

{{ config(
    materialized='incremental',
    unique_key='song_id',
    schema='mart'
) }}

WITH fact_intermediate AS (
    SELECT
        st.song_id,
        st.album_id,
        st.artist_id,
        -- 若业务必须从dim_dates取date_id，替换为原关联逻辑，否则直接用current_date转字符串
        current_date::varchar AS first_loaded,
        current_date::varchar AS last_loaded,
        st.song_duration_ms
    FROM stage.stg_chart_songs st
    -- 如需关联dim_dates获取date_id，取消注释以下语句
    -- INNER JOIN mart.dim_dates d1 
    --     ON current_date = TO_DATE(d1.year || '-' || d1.month || '-' || d1.day, 'yyyy-mm-dd')
)

{% if is_incremental() %}
-- 增量更新：通过MERGE实现条件更新逻辑
MERGE INTO mart.fact_charts AS target
USING fact_intermediate AS source
ON target.song_id = source.song_id
WHEN MATCHED THEN
    UPDATE SET
        last_loaded = source.last_loaded,
        first_loaded = CASE
            -- 判断目标表last_loaded距当前日期是否超过1天
            WHEN current_date - TO_DATE(target.last_loaded, 'yyyy-mm-dd') > 1 
            THEN source.first_loaded
            ELSE target.first_loaded
        END
WHEN NOT MATCHED THEN
    INSERT (song_id, album_id, artist_id, first_loaded, last_loaded, song_duration_ms)
    VALUES (source.song_id, source.album_id, source.artist_id, source.first_loaded, source.last_loaded, source.song_duration_ms);
{% else %}
-- 全量初始化：首次运行时直接插入所有数据
SELECT
    song_id,
    album_id,
    artist_id,
    first_loaded,
    last_loaded,
    song_duration_ms
FROM fact_intermediate;
{% endif %}

关键说明 #

1. 移除merge_update_columns：默认的静态更新列配置无法满足条件更新需求，需通过MERGE自定义更新逻辑。
2. MERGE语句的作用：直接关联源数据（stage表）与目标表（fact_charts），避免自引用问题，同时原生支持条件更新。
3. 分支逻辑说明：
   • 全量模式（首次运行）：从stage表加载所有数据，初始化first_loaded和last_loaded为当前日期。
   • 增量模式：匹配到已有song_id时，强制更新last_loaded；仅当目标表的last_loaded距当前日期超过1天时，才更新first_loaded，否则保留原值。
4. dim_dates关联可选：若业务要求必须从dim_dates获取日期格式，取消注释对应的JOIN语句即可，需确保dim_dates包含当前日期的记录。

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