PostgreSQL多查询处理：优化含大量CTE操作的可读性与可调试性

单条SQL塞30个增删改CTE的写法虽然能保证原子性、结构看起来紧凑，但存在两个硬伤：一是CTE的执行顺序受优化器影响，和手写顺序不一定完全一致；二是中间执行过程完全黑盒，出问题时没法快速定位是哪段逻辑出错、影响了多少数据，调试成本极高。
以下是按改造成本、实用性排序的落地方案：

方案1：按操作粒度拆分SQL + 显式事务统一管控（优先推荐） #

这种方案和原有写法的原子性完全一致，改造成本最低，调试灵活度最高：

• 放弃单条巨型SQL的写法，按「单表单操作」的粒度拆分逻辑：每张表的delete、upsert动作单独写SQL，Python侧通过psycopg2开启显式事务，所有拆分后的SQL在同一个事务内执行，和原写法一样满足全成功才提交、出错全回滚的要求。
• 每段SQL末尾加RETURNING返回执行结果，Python侧直接统计每步的影响行数、操作类型，打日志留痕，不用猜执行状态。示例代码如下：

import psycopg2
from datetime import date

def run_daily_delsert(db_config):
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    sync_date = date.today()
    try:
        # 逐表逐操作执行，顺序和原CTE的依赖逻辑保持一致即可
        # 表1：删除过期数据
        cur.execute("""
            DELETE FROM table1
            WHERE sync_dt = %s AND expire_flag = true
            RETURNING pk_id
        """, (sync_date,))
        del_count = len(cur.fetchall())
        print(f"[table1] 删除过期数据{del_count}条")

        # 表1：执行upsert
        cur.execute("""
            INSERT INTO table1 (col_a, col_b, sync_dt)
            SELECT col_a, col_b, sync_dt FROM stg_table1
            ON CONFLICT (uk_col) DO UPDATE SET
                col_a = EXCLUDED.col_a,
                col_b = EXCLUDED.col_b,
                update_tm = now()
            RETURNING pk_id, (xmax = 0) AS is_new
        """)
        res = cur.fetchall()
        ins_count = sum(1 for row in res if row[1])
        upd_count = len(res) - ins_count
        print(f"[table1] 新增{ins_count}条，更新{upd_count}条")

        # 剩余9张表按相同逻辑编写即可
        # ...... 其余表操作逻辑

        conn.commit()
        print("全表同步任务执行完成")
    except Exception as e:
        conn.rollback()
        print(f"任务执行失败，已回滚，错误信息：{str(e)}")
        raise
    finally:
        cur.close()
        conn.close()

• 调试时可以单独拎出任意一段SQL执行，也可以加debug开关，需要时打印中间结果，不用在几十个CTE里翻找逻辑。

方案2：封装PL/pgSQL存储过程（适合逻辑稳定少变的场景） #

如果delsert逻辑基本不会随业务频繁调整，可以把所有操作收敛到数据库侧的存储过程中，用RAISE NOTICE打印每步执行日志，psycopg2可以直接接收数据库侧的notice输出，调试也很方便：

• 首先在数据库中创建存储过程：

CREATE OR REPLACE PROCEDURE sp_daily_delsert(IN p_sync_dt DATE)
LANGUAGE plpgsql
AS $$
DECLARE
    v_row_cnt INT;
    v_ins_cnt INT;
    v_upd_cnt INT;
BEGIN
    -- 表1删除逻辑
    DELETE FROM table1 WHERE sync_dt = p_sync_dt AND expire_flag = true;
    GET DIAGNOSTICS v_row_cnt = ROW_COUNT;
    RAISE NOTICE '[table1] 删除过期数据%条', v_row_cnt;

    -- 表1upsert逻辑
    WITH upsert_res AS (
        INSERT INTO table1 (col_a, col_b, sync_dt)
        SELECT col_a, col_b, sync_dt FROM stg_table1
        ON CONFLICT (uk_col) DO UPDATE SET
            col_a = EXCLUDED.col_a,
            col_b = EXCLUDED.col_b,
            update_tm = now()
        RETURNING (xmax = 0) AS is_new
    )
    SELECT
        COUNT(*) FILTER (WHERE is_new = true),
        COUNT(*) FILTER (WHERE is_new = false)
    INTO v_ins_cnt, v_upd_cnt
    FROM upsert_res;
    RAISE NOTICE '[table1] 新增%条，更新%条', v_ins_cnt, v_upd_cnt;

    -- 其余表按相同逻辑编写
    -- ......
END;
$$;

• Python侧调用时只需要配置notice接收，就能拿到数据库侧打印的每步日志，调用代码非常简洁：

conn = psycopg2.connect(**db_config)
# 配置日志接收，把数据库的notice输出到本地日志
def db_notice_logger(diag):
    print(f"[DB LOG] {diag.message_primary}")
conn.add_notice_handler(db_notice_logger)

cur = conn.cursor()
try:
    cur.callproc('sp_daily_delsert', (date.today(),))
    conn.commit()
except Exception as e:
    conn.rollback()
    raise
finally:
    cur.close()
    conn.close()

• 这个方案的缺点是存储过程的版本管理、逻辑调整需要直接操作数据库，迭代灵活度不如Python侧写SQL，如果业务规则经常变动不推荐用。

避坑提示 #

• 不要为了追求“单条SQL写完”硬堆CTE：PostgreSQL 12之前CTE是优化器栅栏，12之后虽然支持优化器下推，但几十层CTE嵌套很容易出现执行计划偏离预期的问题，性能往往不如拆分后的独立SQL。
• 无论选哪种方案，一定要保证所有操作在同一个事务内执行，不要开自动提交逐表跑，避免中途失败导致数据不一致。
• 可以加全局debug开关，开启后把每步的中间样本数据写到临时日志表，出问题时不用复现任务就能直接查中间状态。

内容的提问来源于stack exchange，提问作者Kar Keung Christopher Fok