Redshift/PostgreSQL递归查询多重复计算问题求助

问题背景 #

现有递归查询对前12个周期的重复患者（rep_pat）计算正确，但第13个周期及之后的计算逻辑不符合业务要求：

• 第13个周期：需关联12个周期前的新患者数据计算rep_pat
• 第14个周期及以后：需同时关联上月和12个周期前的新患者数据计算rep_pat

核心业务规则 #

• 剂量规则：治疗第1、2、13、14个月服用6片，第3-12个月无药物
• TPE（Total Patient Equals）= 当月SU值 / 6
• 新患者（new_pat）= TPE - 重复患者（rep_pat）
• 前12个周期：rep_pat等于上月的new_pat
• 第13个周期：rep_pat为12个周期前的new_pat（不超过当月TPE）
• 第14个周期及以后：rep_pat为上月new_pat与12个周期前new_pat之和（不超过当月TPE）
• PEQ为最近14个月new_pat的累计和

源数据表 #

CREATE TABLE products_su(country, intprd, "period", su)AS VALUES
 ('GL', 'Medicine', '2019-04-01'::date, 57)
,('GL', 'Medicine', '2019-05-01', 298)
,('GL', 'Medicine', '2019-06-01', 860)
,('GL', 'Medicine', '2019-07-01', 1649)
,('GL', 'Medicine', '2019-08-01', 2227)
,('GL', 'Medicine', '2019-09-01', 1914)
,('GL', 'Medicine', '2019-10-01', 1751)
,('GL', 'Medicine', '2019-11-01', 2007)
,('GL', 'Medicine', '2019-12-01', 2649)
,('GL', 'Medicine', '2020-01-01', 2452)
,('GL', 'Medicine', '2020-02-01', 2733)
,('GL', 'Medicine', '2020-03-01', 3185)
,('GL', 'Medicine', '2020-04-01', 1768)
,('GL', 'Medicine', '2020-05-01', 1779)
,('GL', 'Medicine', '2020-06-01', 3030)
,('GL', 'Medicine', '2020-07-01', 3133)
,('GL', 'Medicine', '2020-08-01', 3373)
,('GL', 'Medicine', '2020-09-01', 4953)
,('GL', 'Medicine', '2020-10-01', 4478)
,('GL', 'Medicine', '2020-11-01', 4471)
,('GL', 'Medicine', '2020-12-01', 5212);

原递归查询（仅前12周期正确） #

with recursive build as (
select country,intprd,period,su,tpe,rp,rep_pat,cur_rn,max_rn
from(select country, intprd, period, su
          , su / 6 as tpe
          , 0::float as rp
          , 0::float as rep_pat
          , row_number()over(partition by country order by period) as rn
          , 2::int as cur_rn
          , count(1)over() as max_rn
     from(select country, intprd, period, su::float
               , row_number()over(partition by country order by period) as rn
          from products_su)_)_
where rn = 1
union all
select t.country, t.intprd, t.period, t.su
     , t.su/6 as tpe, b.tpe as rp
     , least((b.tpe - b.rep_pat), t.tpe) as rep_pat
     , b.cur_rn + 1 as cur_rn
     , b.max_rn
from build b
join(select country, intprd, period, su::float, tpe
          , 0::float as rep_pat
          , lag(period)over(partition by country order by period) prev_period
          , row_number()over(partition by country order by period) as rn
     from(select country, intprd, period, SU
               , SU/6 as tpe
               , row_number()over(partition by country order by period) as rn
          from products_su)_)as t
  on t.prev_period = b.period
 and t.country = b.country
 and t.intprd = b.intprd
where t.rn = b.cur_rn
  and b.cur_rn <= b.max_rn
)
select country, intprd, period, su
     , round(tpe, 1) as tpe
     , round(rep_pat, 1) as rep_pat
     , round((tpe - rep_pat), 1) as new_pat
     , round(sum(rep_pat+new_pat)over(partition by country order by period rows unbounded preceding), 1) as peq
from build  
order by period;

修改后的查询（支持全周期正确计算） #

递归查询需要跟踪历史的new_pat值，以下是调整后的递归实现，适配全周期的计算逻辑：

WITH RECURSIVE base_data AS (
    SELECT 
        country,
        intprd,
        period,
        su,
        su::FLOAT / 6 AS tpe,
        ROW_NUMBER() OVER (PARTITION BY country ORDER BY period) AS rn,
        COUNT(*) OVER (PARTITION BY country) AS max_rn
    FROM products_su
),
pat_calc AS (
    -- 初始行：第一个周期
    SELECT 
        country,
        intprd,
        period,
        su,
        tpe,
        rn,
        0::FLOAT AS rep_pat,
        tpe AS new_pat,
        ARRAY[tpe] AS new_pat_history  -- 存储历史new_pat，用于后续周期调用
    FROM base_data
    WHERE rn = 1

    UNION ALL

    SELECT 
        bd.country,
        bd.intprd,
        bd.period,
        bd.su,
        bd.tpe,
        bd.rn,
        -- 分阶段计算rep_pat，取最小值避免超过当月TPE
        CASE
            WHEN bd.rn <= 12 THEN LEAST(pc.new_pat, bd.tpe)
            WHEN bd.rn = 13 THEN LEAST(pc.new_pat_history[1], bd.tpe)
            ELSE LEAST(pc.new_pat + pc.new_pat_history[1], bd.tpe)
        END AS rep_pat,
        -- 计算当前new_pat
        bd.tpe - CASE
            WHEN bd.rn <= 12 THEN LEAST(pc.new_pat, bd.tpe)
            WHEN bd.rn = 13 THEN LEAST(pc.new_pat_history[1], bd.tpe)
            ELSE LEAST(pc.new_pat + pc.new_pat_history[1], bd.tpe)
        END AS new_pat,
        -- 更新历史数组：保留最近12个周期的new_pat
        CASE
            WHEN bd.rn < 12 THEN pc.new_pat_history || (bd.tpe - CASE WHEN bd.rn <=12 THEN LEAST(pc.new_pat, bd.tpe) ELSE 0 END)
            ELSE (pc.new_pat_history[2:]) || (bd.tpe - CASE WHEN bd.rn <=12 THEN LEAST(pc.new_pat, bd.tpe) ELSE LEAST(pc.new_pat + pc.new_pat_history[1], bd.tpe) END)
        END AS new_pat_history
    FROM pat_calc pc
    JOIN base_data bd ON bd.country = pc.country AND bd.rn = pc.rn + 1
)
SELECT 
    country,
    intprd,
    period,
    su,
    ROUND(tpe, 1) AS tpe,
    ROUND(rep_pat, 1) AS rep_pat,
    ROUND(new_pat, 1) AS new_pat,
    -- 计算最近14个月new_pat的滑动累计和作为PEQ
    ROUND(SUM(new_pat) OVER (PARTITION BY country ORDER BY period ROWS BETWEEN 13 PRECEDING AND CURRENT ROW), 1) AS peq
FROM pat_calc
ORDER BY period;

逻辑说明 #

1. base_data：预处理数据，计算每个周期的TPE、周期序号rn及总周期数
2. pat_calc递归CTE：
   • 初始行：第一个周期无重复患者，new_pat等于TPE，初始化历史数组存储该值
   • 递归步骤：
     • 前12个周期：rep_pat取上月new_pat，且不超过当月TPE
     • 第13个周期：从历史数组中取出12个周期前的new_pat作为rep_pat，且不超过当月TPE
     • 第14个周期及以后：rep_pat取上月new_pat与12个周期前new_pat之和，且不超过当月TPE
     • 维护历史数组，始终保留最近12个周期的new_pat，用于后续周期调用
3. 最终查询：计算最近14个月new_pat的滑动和作为PEQ，按周期排序输出

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