如何让dbt宏返回表？快照调用宏时遇语法错误

问题背景 #

尝试通过dbt宏创建临时表用于快照，但宏未返回有效内容，快照执行时出现语法错误。相关代码及错误信息如下：

原宏代码 #

{% macro notifications_clean() %}
{% if execute %}

    {% set nc_sql %}
        
        CREATE TEMPORARY TABLE temp_notifications AS
        {% if var('full_load') == true %}

            {{ log('Dropping existing snapshot for conforming_notifications.notifications and derived current table for notifications_current', info=True) }}
            {% set dropsnapshot %}
                DROP TABLE IF EXISTS conforming_notifications.notifications CASCADE;
            {% endset %}
            {% do run_query(dropsnapshot) %}

            {{ log('Getting all data after last full load operation', info=True) }}

            SELECT
                *
            FROM
                raw_notifications.dbo_notifications
            WHERE
                etl_loaddate >= cast ((select max(etl_loaddate) from raw_notifications.dbo_notifications WHERE op = 'H') as date)

        {% else %}
        
            {{ log('Getting all cdc data after last snapshot operation', info=True) }}

            SELECT
                *
            FROM
                raw_notifications.dbo_notifications
            WHERE
                cdcoperationtimestamp >= (SELECT MAX(etl_loaddate) max_etl_loaddate FROM raw_notifications.dbo_notifications WHERE op = 'H')

        {% endif %}

    {% endset %}
    
   -- {% do run_query(nc_sql) %}
    {{ return(temp_notifications) }}

{% endif %}

{% endmacro %}

原快照代码 #

{% snapshot notifications %}

    {{ 
        config(
            target_database='dwh',
            target_schema='conforming_notifications',
            strategy='timestamp',
            unique_key='id',
            updated_at='cdcoperationtimestamp'
        )
    }}

    SELECT
        *
    FROM
        {{ notifications_clean() }}

{% endsnapshot %}

错误信息 #

00:45:38  Database Error in snapshot notifications (snapshots/conforming_notifications.sql)
00:45:38    syntax error at or near ")"
00:45:38    LINE 32:     ) sbq
00:45:38                 ^
00:45:38    compiled SQL at target/run/data_platform/snapshots/conforming_notifications.sql

错误原因 #

1. 宏返回值错误：return(temp_notifications)试图返回SQL中的表对象，但dbt宏需要返回可编译的字符串（SQL片段或表名），而非内存对象。
2. 临时表未创建：注释了{% do run_query(nc_sql) %}，导致创建临时表的SQL未执行，即使返回表名也不存在对应的表。
3. 快照编译异常：宏未返回有效内容，导致快照编译后的SQL中FROM子句为空，出现FROM )的非法语法。

修复方案 #

方案一：使用临时表（适合大数据量场景） #

修改宏代码，确保临时表被创建并返回正确的表名：

{% macro notifications_clean() %}
{% if execute %}
    {% set nc_sql %}
        CREATE TEMPORARY TABLE temp_notifications AS
        {% if var('full_load') == true %}
            {{ log('删除conforming_notifications.notifications快照及衍生表', info=True) }}
            {% set dropsnapshot %}
                DROP TABLE IF EXISTS conforming_notifications.notifications CASCADE;
            {% endset %}
            {% do run_query(dropsnapshot) %}

            {{ log('获取上次全量加载后的所有数据', info=True) }}
            SELECT
                *
            FROM
                raw_notifications.dbo_notifications
            WHERE
                etl_loaddate >= cast ((select max(etl_loaddate) from raw_notifications.dbo_notifications WHERE op = 'H') as date)
        {% else %}
            {{ log('获取上次快照后的所有CDC数据', info=True) }}
            SELECT
                *
            FROM
                raw_notifications.dbo_notifications
            WHERE
                cdcoperationtimestamp >= (SELECT MAX(etl_loaddate) FROM raw_notifications.dbo_notifications WHERE op = 'H')
        {% endif %}
    {% endset %}
    
    {% do run_query(nc_sql) %}
    {{ return('temp_notifications') }}
{% endif %}
{% endmacro %}

修改点说明：

• 取消注释{% do run_query(nc_sql) %}，实际执行创建临时表的SQL
• 将return(temp_notifications)改为return('temp_notifications')，返回字符串形式的临时表名，确保dbt编译时能正确替换

方案二：直接返回SQL子查询（更符合dbt声明式风格） #

无需创建临时表，让宏直接返回查询语句，快照中以子查询形式引用：

修改后的宏代码

{% macro notifications_clean() %}
    {% if var('full_load') == true %}
        {{ log('删除conforming_notifications.notifications快照及衍生表', info=True) }}
        {% if execute %}
            {% set dropsnapshot %}
                DROP TABLE IF EXISTS conforming_notifications.notifications CASCADE;
            {% endset %}
            {% do run_query(dropsnapshot) %}
        {% endif %}

        {{ log('获取上次全量加载后的所有数据', info=True) }}
        SELECT
            *
        FROM
            raw_notifications.dbo_notifications
        WHERE
            etl_loaddate >= cast ((select max(etl_loaddate) from raw_notifications.dbo_notifications WHERE op = 'H') as date)
    {% else %}
        {{ log('获取上次快照后的所有CDC数据', info=True) }}
        SELECT
            *
        FROM
            raw_notifications.dbo_notifications
        WHERE
            cdcoperationtimestamp >= (SELECT MAX(etl_loaddate) FROM raw_notifications.dbo_notifications WHERE op = 'H')
    {% endif %}
{% endmacro %}

修改后的快照代码

{% snapshot notifications %}
    {{ 
        config(
            target_database='dwh',
            target_schema='conforming_notifications',
            strategy='timestamp',
            unique_key='id',
            updated_at='cdcoperationtimestamp'
        )
    }}

    SELECT * FROM ({{ notifications_clean() }}) sbq
{% endsnapshot %}

修改点说明：

• 宏直接返回查询的SQL片段，无需创建临时表
• 快照中把宏返回的SQL作为子查询，避免空FROM子句导致的语法错误

验证方法 #

执行dbt快照命令：

dbt snapshot

检查编译后的SQL（target/run/.../conforming_notifications.sql），确认FROM子句有合法的表或子查询，且无语法错误。

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