dbt+Jinja中如何在CTE内执行动态查询并引用CTE表

问题描述 #

我在使用dbt搭配Snowflake时编写了如下含CTE的SQL脚本：

with bank_str_table as (
    select *
    from {{ ref('bank_str_dbt') }}
),

first_table as (
    select
        bk.id,
        bk.metadata$row_id,
        bk.label,
        ts_del.merge_key,
    from bank_str_table bk
    inner join bank_str_table ts_del on bk.METADATA$ROW_ID = ts_del.METADATA$ROW_ID
    where
        ts_del.METADATA$ACTION = 'DELETE' AND
        ts_del.METADATA$ISUPDATE = 'TRUE' AND
        bk.METADATA$ACTION = 'INSERT' AND
        bk.METADATA$ISUPDATE = 'TRUE'
),

bank_str_table_first as (
    select *
    from bank_str_table
    where label not in (select listagg(label, ', ') from first_table)
    union
    select *
    from first_table
)

select *
from first_table

问题在于bank_str_table_first的WHERE子句中，select listagg(label, ', ') from first_table需要动态执行，而非作为原生SQL直接嵌入。此前用开头的Jinja代码解决过类似问题，但当前脚本需要引用CTE生成的表，且要重复使用动态查询结果，无法解决两个核心问题：

• 如何在Jinja代码中引用CTE中的表
• 如何在脚本中间执行Jinja查询逻辑

解决方案 #

核心原理 #

Jinja的run_query是在dbt编译阶段执行的，而CTE是在SQL运行阶段才会被Snowflake解析生成，所以无法直接在Jinja中引用CTE。必须把需要动态计算的逻辑提前落地，或者直接嵌入到Jinja的查询语句中，先获取结果再注入主SQL。

方法一：将CTE逻辑抽为独立dbt模型 #

把first_table的逻辑单独做成一个dbt模型，让Jinja可以通过ref引用这个模型的结果，提前计算出动态的label列表。

1. 创建独立模型stg_first_table.sql：

with bank_str_table as (
    select *
    from {{ ref('bank_str_dbt') }}
)
select
    bk.id,
    bk.metadata$row_id,
    bk.label,
    ts_del.merge_key
from bank_str_table bk
inner join bank_str_table ts_del on bk.METADATA$ROW_ID = ts_del.METADATA$ROW_ID
where
    ts_del.METADATA$ACTION = 'DELETE' AND
    ts_del.METADATA$ISUPDATE = 'TRUE' AND
    bk.METADATA$ACTION = 'INSERT' AND
    bk.METADATA$ISUPDATE = 'TRUE'

2. 在主模型中用Jinja计算动态变量并引用：

{%- set get_label_list -%}
    select listagg(label, ', ') as label_list from {{ ref('stg_first_table') }}
{% endset %}

{% set results = run_query(get_label_list) %}

{% if execute %}
{# 将查询结果转为带引号的字符串列表，适配SQL的IN语法 #}
{% set label_list = results.columns[0].values()[0] | default('', true) %}
{% set quoted_labels = label_list.split(',') | map('trim') | map('quote') | join(', ') %}
{% else %}
{# 编译阶段默认值，避免报错 #}
{% set quoted_labels = '' %}
{% endif %}

with bank_str_table as (
    select *
    from {{ ref('bank_str_dbt') }}
),

first_table as (
    select * from {{ ref('stg_first_table') }}
),

bank_str_table_first as (
    select *
    from bank_str_table
    where label not in ({{ quoted_labels }})
    union
    select *
    from first_table
)

select * from first_table

方法二：直接在Jinja中嵌入CTE逻辑 #

如果不想新增独立模型，可以把first_table的CTE逻辑直接嵌入到Jinja的run_query语句中，一次性计算出动态结果：

{%- set get_label_list -%}
    with bank_str_table as (
        select *
        from {{ ref('bank_str_dbt') }}
    ),
    first_table as (
        select
            bk.id,
            bk.metadata$row_id,
            bk.label,
            ts_del.merge_key
        from bank_str_table bk
        inner join bank_str_table ts_del on bk.METADATA$ROW_ID = ts_del.METADATA$ROW_ID
        where
            ts_del.METADATA$ACTION = 'DELETE' AND
            ts_del.METADATA$ISUPDATE = 'TRUE' AND
            bk.METADATA$ACTION = 'INSERT' AND
            bk.METADATA$ISUPDATE = 'TRUE'
    )
    select listagg(label, ', ') as label_list from first_table
{% endset %}

{% set results = run_query(get_label_list) %}

{% if execute %}
{% set label_list = results.columns[0].values()[0] | default('', true) %}
{% set quoted_labels = label_list.split(',') | map('trim') | map('quote') | join(', ') %}
{% else %}
{% set quoted_labels = '' %}
{% endif %}

with bank_str_table as (
    select *
    from {{ ref('bank_str_dbt') }}
),

first_table as (
    select
        bk.id,
        bk.metadata$row_id,
        bk.label,
        ts_del.merge_key
    from bank_str_table bk
    inner join bank_str_table ts_del on bk.METADATA$ROW_ID = ts_del.METADATA$ROW_ID
    where
        ts_del.METADATA$ACTION = 'DELETE' AND
        ts_del.METADATA$ISUPDATE = 'TRUE' AND
        bk.METADATA$ACTION = 'INSERT' AND
        bk.METADATA$ISUPDATE = 'TRUE'
),

bank_str_table_first as (
    select *
    from bank_str_table
    where label not in ({{ quoted_labels }})
    union
    select *
    from first_table
)

select * from first_table

核心问题解答 #

1. 如何在Jinja中引用CTE中的表：
   CTE仅在SQL运行阶段存在，Jinja编译时无法直接访问。解决方式是：

   • 把CTE逻辑转为独立dbt模型，Jinja通过ref引用该模型的物理表/视图
   • 把CTE逻辑直接嵌入到Jinja的run_query语句中，让run_query直接执行完整SQL获取结果

2. 如何在脚本中间执行Jinja查询逻辑：
   Jinja代码是从上到下编译执行的，无法在脚本中间“动态插入”查询逻辑。正确做法是把所有动态计算逻辑放在脚本开头，计算出变量后，在脚本中间的SQL中直接引用这些变量，重复使用也只需调用变量即可。

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