PostgreSQL 15批量插入时实现自定义Upsert的方案咨询

针对你的需求，有两种高效的实现方式，无需逐行从应用端插入，兼顾性能和业务逻辑：

方案一：批量操作+冲突行检测（推荐，性能最优） #

这种方式通过临时表批量导入数据，分三步完成逻辑，全程批量操作，性能最佳：

1. 创建临时表并导入批量数据
   先把要插入的所有数据存入临时表：

   CREATE TEMP TABLE temp_mytab (r char(38), h char(2), p bytea) ON COMMIT DROP;
   INSERT INTO temp_mytab VALUES 
   ('val_r', 'a0', 'pr_1'), ('r1', 'a1', 'p1'), -- 替换为你的批量数据
   ('val_r2', 'a2', 'pr_2');
   

2. 插入无冲突的行
   仅插入(r,h)主键不存在的行：

   INSERT INTO mytab (r, h, p)
   SELECT t.r, t.h, t.p
   FROM temp_mytab t
   LEFT JOIN mytab m ON t.r = m.r AND t.h = m.h
   WHERE m.r IS NULL;
   

3. 检测并返回冲突且p值不同的行
   查询出主键存在但p值与新值不一致的行，将结果返回给应用：

   SELECT 
       t.r, 
       t.h, 
       t.p AS 新p值, 
       m.p AS 现有p值
   FROM temp_mytab t
   JOIN mytab m ON t.r = m.r AND t.h = m.h
   WHERE t.p != m.p;
   

应用端（psycopg2）执行这三步后，就能拿到所有冲突行的信息，同时无冲突行已成功插入，相同p值的冲突行自动忽略。

方案二：存储过程循环处理（更灵活，适合复杂逻辑） #

如果需要在数据库端直接处理异常提示，可以用存储过程遍历临时表数据，捕获每一行的冲突异常，判断后抛出提示：

CREATE OR REPLACE PROCEDURE bulk_insert_mytab(p_data IN TABLE(r char(38), h char(2), p bytea))
LANGUAGE plpgsql
AS $$
DECLARE
    rec RECORD;
    existing_p bytea;
BEGIN
    -- 初始化临时表存储批量数据
    CREATE TEMP TABLE IF NOT EXISTS temp_mytab (LIKE mytab) ON COMMIT DROP;
    TRUNCATE temp_mytab;
    INSERT INTO temp_mytab SELECT * FROM p_data;

    -- 遍历每一行处理
    FOR rec IN SELECT * FROM temp_mytab LOOP
        BEGIN
            INSERT INTO mytab (r, h, p) VALUES (rec.r, rec.h, rec.p);
        EXCEPTION
            WHEN unique_violation THEN
                -- 查询现有行的p值
                SELECT p INTO existing_p FROM mytab WHERE r = rec.r AND h = rec.h;
                IF existing_p != rec.p THEN
                    -- 抛出提示信息，应用端可捕获NOTICE日志
                    RAISE NOTICE '冲突行：r=%, h=%，现有p值与新值不一致', rec.r, rec.h;
                END IF;
                -- p值相同时不做任何操作
        END;
    END LOOP;
END;
$$;

在psycopg2中调用存储过程时，可以传入批量数据（比如用execute_batch或构造表参数），同时监听数据库的NOTICE信息，获取冲突行的提示。

注意事项 #

• 两种方案都避免了应用端逐行插入的性能问题，数百条数据的处理效率很高；
• 方案一的批量操作性能优于方案二的循环，优先推荐；
• 如果需要强制终止事务并抛出异常（而非仅提示），可将方案二中的RAISE NOTICE改为RAISE EXCEPTION，但这样会终止整个批量操作，不符合你"继续处理其他行"的需求，所以不建议。

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