如何在Snowflake中让LAG递归查找非浮动transaction_id

我的数据表中存在一类浮动行（floating rows）：这类行满足type = mod、关联customer_id，但未关联表中有效的transaction_id。我需要为这些浮动行关联有效的transaction_id，创建一个映射表记录浮动transaction_id及其对应的关联previous_transaction_id。

目前我用LAG函数获取previous_transaction_id，但只能处理部分场景。我需要实现：当LAG找到的是浮动transaction_id时，持续回溯直至找到非浮动的transaction_id。

我使用Snowflake，同时可借助jinja和dbt工具。以下是相关模拟数据、当前查询语句、当前输出及期望输出：

模拟数据与当前查询 #

with data as (
select 
    transaction_id,
    customer_id,
    transaction_date,
    amount,
    type
from (values 
    (1, 'a','03/24/2022'::date, 10, 'cat'),
    (1, 'a','03/24/2022'::date, 15, 'dog'),
    (1, 'a','03/24/2022'::date, 20, 'mouse'),
    (1, 'a','03/24/2022'::date, 30, 'rabbit'),
    (1, 'a','03/24/2022'::date, 5, 'squirrel'),
    (2, 'a','03/24/2022'::date, 4, 'mod'), -- floater
    (3, 'b','05/20/2022'::date, 100, 'cat'),
    (3, 'b','05/20/2022'::date, 150, 'dog'),
    (3, 'b','05/20/2022'::date, 200, 'mouse'),
    (3, 'b','05/20/2022'::date, 300, 'rabbit'),
    (3, 'b','05/20/2022'::date, 50, 'squirrel'),
    (4, 'b','07/20/2022'::date, 40, 'mod'), -- floater
    (5, 'c','02/02/2020'::date, 100, 'cat'),
    (5, 'c','02/02/2020'::date, 150, 'dog'),
    (5, 'c','02/02/2020'::date, 200, 'mouse'),
    (5, 'c','02/02/2020'::date, 300, 'rabbit'),
    (6, 'c','08/01/2020'::date, 50, 'mod'), -- floater
    (7, 'c','12/25/2020'::date, 40, 'mod'), -- floater
    (8, 'd','01/15/2021'::date, 10, 'cat'),
    (8, 'd','01/15/2021'::date, 15, 'dog'),
    (8, 'd','01/15/2021'::date, 20, 'mouse'),
    (8, 'd','01/15/2021'::date, 30, 'rabbit'),
    (8, 'd','01/15/2021'::date, 5, 'squirrel'),
    (8, 'd','01/15/2021'::date, 4, 'mod'),
    (9, 'e','02/10/2020'::date, 100, 'cat'),
    (9, 'e','02/10/2020'::date, 150, 'dog'),
    (9, 'e','02/10/2020'::date, 200, 'mouse'),
    (9, 'e','02/10/2020'::date, 300, 'rabbit'),
    (10, 'e','08/17/2020'::date, 50, 'mod'), -- floater
    (11, 'e','12/15/2020'::date, 40, 'mod'), -- floater
    (12, 'e','02/14/2021'::date, 40, 'mod'), -- floater
    (13, 'c','04/09/2022'::date, 0, 'mouse'),
    (13, 'c','04/09/2022'::date, 0, 'rabbit'),
    (13, 'c','04/09/2022'::date, 50, 'mod') -- floater because other values for transaction_id sum to 0
    ) as tbl (transaction_id, customer_id, transaction_date, amount, type)
),
previous_transaction_id as (
select 
    transaction_id,
    customer_id,
    lag(transaction_id, 1, null) over (partition by customer_id order by transaction_date) as previous_transaction_id
from data
   qualify transaction_id != previous_transaction_id
),

floating_mods as (
    select 
        transaction_id,
        sum(iff(type = 'mod', amount, 0)) as mod_amount,
        sum(amount) - mod_amount as non_mod_amount
    from data
    group by 1
    having non_mod_amount = 0
)

select 
    gp.transaction_id,
    gp.previous_transaction_id
from previous_transaction_id gp
    inner join floating_mods fm on gp.transaction_id = fm.transaction_id
order by gp.transaction_id

当前查询输出 #

期望输出 #

要实现持续回溯到非浮动transaction_id，可以使用递归CTE遍历每个浮动ID的前置ID，直到找到非浮动的有效ID。以下是修改后的查询语句：

with data as (
select 
    transaction_id,
    customer_id,
    transaction_date,
    amount,
    type
from (values 
    (1, 'a','03/24/2022'::date, 10, 'cat'),
    (1, 'a','03/24/2022'::date, 15, 'dog'),
    (1, 'a','03/24/2022'::date, 20, 'mouse'),
    (1, 'a','03/24/2022'::date, 30, 'rabbit'),
    (1, 'a','03/24/2022'::date, 5, 'squirrel'),
    (2, 'a','03/24/2022'::date, 4, 'mod'), -- floater
    (3, 'b','05/20/2022'::date, 100, 'cat'),
    (3, 'b','05/20/2022'::date, 150, 'dog'),
    (3, 'b','05/20/2022'::date, 200, 'mouse'),
    (3, 'b','05/20/2022'::date, 300, 'rabbit'),
    (3, 'b','05/20/2022'::date, 50, 'squirrel'),
    (4, 'b','07/20/2022'::date, 40, 'mod'), -- floater
    (5, 'c','02/02/2020'::date, 100, 'cat'),
    (5, 'c','02/02/2020'::date, 150, 'dog'),
    (5, 'c','02/02/2020'::date, 200, 'mouse'),
    (5, 'c','02/02/2020'::date, 300, 'rabbit'),
    (6, 'c','08/01/2020'::date, 50, 'mod'), -- floater
    (7, 'c','12/25/2020'::date, 40, 'mod'), -- floater
    (8, 'd','01/15/2021'::date, 10, 'cat'),
    (8, 'd','01/15/2021'::date, 15, 'dog'),
    (8, 'd','01/15/2021'::date, 20, 'mouse'),
    (8, 'd','01/15/2021'::date, 30, 'rabbit'),
    (8, 'd','01/15/2021'::date, 5, 'squirrel'),
    (8, 'd','01/15/2021'::date, 4, 'mod'),
    (9, 'e','02/10/2020'::date, 100, 'cat'),
    (9, 'e','02/10/2020'::date, 150, 'dog'),
    (9, 'e','02/10/2020'::date, 200, 'mouse'),
    (9, 'e','02/10/2020'::date, 300, 'rabbit'),
    (10, 'e','08/17/2020'::date, 50, 'mod'), -- floater
    (11, 'e','12/15/2020'::date, 40, 'mod'), -- floater
    (12, 'e','02/14/2021'::date, 40, 'mod'), -- floater
    (13, 'c','04/09/2022'::date, 0, 'mouse'),
    (13, 'c','04/09/2022'::date, 0, 'rabbit'),
    (13, 'c','04/09/2022'::date, 50, 'mod') -- floater because other values for transaction_id sum to 0
    ) as tbl (transaction_id, customer_id, transaction_date, amount, type)
),
-- 标记每个transaction_id是否为浮动行
floating_mods as (
    select 
        transaction_id,
        customer_id,
        sum(iff(type = 'mod', amount, 0)) as mod_amount,
        sum(amount) - mod_amount as non_mod_amount,
        case when non_mod_amount = 0 then 1 else 0 end is_floating
    from data
    group by transaction_id, customer_id
),
-- 获取每个transaction_id的直接前置ID（按customer分组、日期排序）
transaction_prev_map as (
    select 
        distinct 
        d.transaction_id,
        d.customer_id,
        lag(d.transaction_id) over (partition by d.customer_id order by d.transaction_date) as previous_transaction_id
    from data d
),
-- 递归CTE：回溯找到最近的非浮动ID
recursive_backtrack as (
    -- 基准情况：先获取所有浮动行的直接前置ID
    select 
        fm.transaction_id,
        fm.customer_id,
        tpm.previous_transaction_id,
        fm.is_floating
    from floating_mods fm
    left join transaction_prev_map tpm 
        on fm.transaction_id = tpm.transaction_id
    where fm.is_floating = 1

    union all

    -- 递归情况：如果前置ID还是浮动行，继续回溯
    select 
        r.transaction_id,
        r.customer_id,
        tpm.previous_transaction_id,
        fm.is_floating
    from recursive_backtrack r
    left join floating_mods fm 
        on r.previous_transaction_id = fm.transaction_id
    left join transaction_prev_map tpm 
        on r.previous_transaction_id = tpm.transaction_id
    where fm.is_floating = 1 -- 当前置ID是浮动行时继续递归
)
-- 取每个浮动行最后一次回溯得到的非浮动ID
select 
    transaction_id,
    max(previous_transaction_id) as previous_transaction_id
from recursive_backtrack
where previous_transaction_id is not null
group by transaction_id
order by transaction_id;

思路说明 #

1. 标记浮动行：通过floating_modsCTE标记每个transaction_id是否为浮动行（non_mod_amount = 0）。
2. 构建前置ID映射：用transaction_prev_map获取每个transaction_id按customer_id和transaction_date排序后的直接前置ID。
3. 递归回溯：通过recursive_backtrack递归遍历，只要当前前置ID是浮动行，就继续回溯，直到找到非浮动的ID。
4. 聚合结果：对每个浮动行取最后一次回溯得到的有效ID，得到最终映射关系。

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