Redshift中遍历记录更新状态的技术实现求助

问题场景 #

要处理250万条2022年3月1日起的潜在营销线索交互记录，按规则标记VALID（有效）或DUPE（重复）：

• Q类型线索：和同一手机号上一条有效记录间隔超30天 → 标记VALID，否则DUPE
• C类型线索：间隔阈值为60天
• 同一手机号可以同时有Q、C两种类型的有效记录
• 每个手机号的第一条记录已经标为VALID，现在要处理后续记录

之前在SQL Server里用游标+循环搞定了，现在要迁移到Redshift用存储过程的while loop实现，初始写的代码报语法错误（提示AS位置不对），调整细节后解决了问题。

初始报错信息 #

提示AS位置错误

初始报错代码 #

-- Redshift存储过程初始版本（报错）
CREATE OR REPLACE PROCEDURE mark_lead_status()
AS $$
DECLARE
    v_phone VARCHAR(20);
    v_last_valid_q DATE;
    v_last_valid_c DATE;
    cur CURSOR FOR SELECT DISTINCT phone FROM leads WHERE created_at >= '2022-03-01' ORDER BY phone;
BEGIN
    OPEN cur;
    LOOP
        FETCH cur INTO v_phone;
        EXIT WHEN NOT FOUND;

        -- 初始化该手机号各类型的上一条有效记录日期
        SELECT MAX(CASE WHEN type = 'Q' AND status = 'VALID' THEN created_at END) INTO v_last_valid_q
        FROM leads WHERE phone = v_phone;
        SELECT MAX(CASE WHEN type = 'C' AND status = 'VALID' THEN created_at END) INTO v_last_valid_c
        FROM leads WHERE phone = v_phone;

        -- 批量更新该手机号的未标记记录
        UPDATE leads
        SET status = CASE
            WHEN type = 'Q' AND created_at > v_last_valid_q + INTERVAL '30 days' THEN 'VALID'
            WHEN type = 'C' AND created_at > v_last_valid_c + INTERVAL '60 days' THEN 'VALID'
            ELSE 'DUPE'
        END AS New_Status  -- 这里的AS是语法错误的根源
        WHERE phone = v_phone AND created_at >= '2022-03-01' AND status IS NULL;

        -- 更新上一条有效记录日期
        SELECT MAX(CASE WHEN type = 'Q' AND status = 'VALID' THEN created_at END) INTO v_last_valid_q
        FROM leads WHERE phone = v_phone;
        SELECT MAX(CASE WHEN type = 'C' AND status = 'VALID' THEN created_at END) INTO v_last_valid_c
        FROM leads WHERE phone = v_phone;
    END LOOP;
    CLOSE cur;
END;
$$ LANGUAGE plpgsql;

示例数据 #

预期处理结果 #

最终修正后的存储过程代码 #

CREATE OR REPLACE PROCEDURE mark_lead_status()
AS $$
DECLARE
    v_phone VARCHAR(20);
    v_last_valid_q DATE;
    v_last_valid_c DATE;
    cur CURSOR FOR SELECT DISTINCT phone FROM leads WHERE created_at >= '2022-03-01' ORDER BY phone;
BEGIN
    OPEN cur;
    LOOP
        FETCH cur INTO v_phone;
        EXIT WHEN NOT FOUND;

        -- 初始化有效记录日期，用COALESCE避免NULL导致的逻辑错误
        SELECT COALESCE(MAX(CASE WHEN type = 'Q' AND status = 'VALID' THEN created_at END), '1900-01-01') INTO v_last_valid_q
        FROM leads WHERE phone = v_phone;
        SELECT COALESCE(MAX(CASE WHEN type = 'C' AND status = 'VALID' THEN created_at END), '1900-01-01') INTO v_last_valid_c
        FROM leads WHERE phone = v_phone;

        -- 按时间顺序逐条处理该手机号的未标记记录，确保逻辑准确
        FOR rec IN SELECT id, type, created_at FROM leads WHERE phone = v_phone AND created_at >= '2022-03-01' AND status IS NULL ORDER BY created_at LOOP
            DECLARE
                new_status VARCHAR(10);
            BEGIN
                -- 判断当前记录的状态
                new_status := CASE
                    WHEN rec.type = 'Q' AND rec.created_at > v_last_valid_q + INTERVAL '30 days' THEN 'VALID'
                    WHEN rec.type = 'C' AND rec.created_at > v_last_valid_c + INTERVAL '60 days' THEN 'VALID'
                    ELSE 'DUPE'
                END;

                -- 更新当前记录的状态
                UPDATE leads SET status = new_status WHERE id = rec.id;

                -- 如果是有效记录，实时更新对应类型的上一条有效日期
                IF new_status = 'VALID' THEN
                    IF rec.type = 'Q' THEN
                        v_last_valid_q := rec.created_at;
                    ELSE
                        v_last_valid_c := rec.created_at;
                    END IF;
                END IF;
            END;
        END LOOP;
    END LOOP;
    CLOSE cur;
END;
$$ LANGUAGE plpgsql;

关键修正点 #

1. 删掉多余的AS New_Status：Redshift的UPDATE语法不允许在SET子句后给列加别名，这就是初始报错的直接原因
2. 用COALESCE处理初始值：防止某类型还没有有效记录时，变量为NULL导致的日期计算错误
3. 改为逐条按时间处理：原批量更新会因为未按时间顺序判断，导致后续记录的有效日期判断出错；现在按created_at顺序处理，实时更新上一条有效日期，逻辑更准确
4. 单条记录精准更新：通过游标遍历单条未标记记录，用id定位更新，避免批量更新的逻辑漏洞

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