如何避免PL/pgSQL循环？大表数据迁移高性能替代方案

核心问题分析 #

你当前的嵌套循环脚本逐行处理1.8亿条记录（3000万订单+1.5亿状态历史），每次循环都执行两次单条INSERT，事务开销极大，导致耗时数小时。批量插入的关键是一次性处理多行数据，并正确关联两个新表的依赖关系（history_info的ID作为order_history的rev字段）。

正确的批量插入脚本 #

下面的脚本通过CTE（公共表表达式）实现批量插入，同时保证history_info生成的ID能准确对应到order_history的记录：

WITH OSHO_DATA AS (
    -- 先获取所有需要迁移的状态历史数据，添加唯一行标识用于后续关联
    SELECT 
        row_number() OVER () AS row_id,
        osho.order_id,
        osho.new_status,
        osho.modified_date,
        osho.status_changer  -- 旧表中负责修改状态的字段，需确保存在
    FROM order_status_history_old osho
    -- 关联orders_old过滤无效订单（如果允许无对应订单的状态历史，可改为LEFT JOIN）
    JOIN orders_old oo ON osho.order_id = oo.id
),
HI_INSERT AS (
    -- 批量插入history_info，返回生成的rev ID和对应的行标识
    INSERT INTO history_info (id, date, updated_by)
    SELECT 
        nextval('HISTORY_INFO_SEQ'),
        modified_date,
        status_changer
    FROM OSHO_DATA
    ORDER BY row_id  -- 保证插入顺序与OSHO_DATA一致，确保关联准确
    RETURNING id AS rev, row_id
)
-- 批量插入order_history，通过row_id关联HI_INSERT的rev ID
INSERT INTO order_history (id, rev, rev_type, status)
SELECT 
    od.order_id::BIGINT,
    hi.rev,
    1,
    GET_STATUS_CODE(od.new_status)  -- 替换为你的状态转换函数
FROM OSHO_DATA od
JOIN HI_INSERT hi ON od.row_id = hi.row_id;

性能优化措施 #

1. 临时禁用约束与索引 #

迁移前关闭目标表的非必要约束和索引，大幅提升插入速度：

• 禁用触发器：ALTER TABLE history_info DISABLE TRIGGER ALL; ALTER TABLE order_history DISABLE TRIGGER ALL;
• 禁用外键约束：ALTER TABLE order_history DISABLE CONSTRAINT <外键名称>;（如果有）
• 删除目标表索引：DROP INDEX <索引名称>;（迁移完成后重建）

2. 分批次处理超大表 #

如果一次性处理1.5亿行内存不足，可按订单ID分批次处理：

DO $$
DECLARE
    batch_size INT := 100000;  -- 每批次处理10万条状态历史，可根据服务器配置调整
    max_order_id INT;
    current_min_id INT := (SELECT MIN(id) FROM orders_old);
BEGIN
    SELECT MAX(id) INTO max_order_id FROM orders_old;
    
    WHILE current_min_id <= max_order_id LOOP
        WITH OSHO_DATA AS (
            SELECT 
                row_number() OVER () AS row_id,
                osho.order_id,
                osho.new_status,
                osho.modified_date,
                osho.status_changer
            FROM order_status_history_old osho
            JOIN orders_old oo ON osho.order_id = oo.id
            WHERE oo.id BETWEEN current_min_id AND current_min_id + batch_size - 1
        ),
        HI_INSERT AS (
            INSERT INTO history_info (id, date, updated_by)
            SELECT nextval('HISTORY_INFO_SEQ'), modified_date, status_changer
            FROM OSHO_DATA
            ORDER BY row_id
            RETURNING id AS rev, row_id
        )
        INSERT INTO order_history (id, rev, rev_type, status)
        SELECT od.order_id::BIGINT, hi.rev, 1, GET_STATUS_CODE(od.new_status)
        FROM OSHO_DATA od
        JOIN HI_INSERT hi ON od.row_id = hi.row_id;
        
        current_min_id := current_min_id + batch_size;
        COMMIT;  -- 每批次提交事务，避免事务过大导致内存溢出
    END LOOP;
END $$;

3. 调整PostgreSQL配置（临时） #

修改postgresql.conf后重启服务（迁移完成后恢复原值）：

• maintenance_work_mem = 64MB → 改为512MB（提升排序和索引重建性能）
• work_mem = 4MB → 改为32MB（提升批量操作的内存分配）
• max_wal_size = 1GB → 改为4GB（减少检查点频率，提升写入速度）
• fsync = on → 改为off（仅迁移期间使用，风险：服务器断电可能丢失数据）

4. 精简查询字段 #

避免SELECT *，只查询需要的字段：

• 原循环脚本中SELECT * FROM orders_old → 改为SELECT id FROM orders_old
• SELECT * FROM order_status_history_old → 改为SELECT order_id, modified_date, status_changer, new_status FROM order_status_history_old

数据一致性验证 #

迁移完成后，务必验证数据正确性：

1. 检查记录数匹配：

   -- 旧状态历史记录数应等于新表的记录数
   SELECT COUNT(*) FROM order_status_history_old;
   SELECT COUNT(*) FROM history_info;
   SELECT COUNT(*) FROM order_history;
   

2. 抽样验证关联：

   -- 随机选一个订单，检查新旧表数据是否对应
   WITH sample_order AS (SELECT id FROM orders_old ORDER BY RANDOM() LIMIT 1)
   SELECT 
       osho.modified_date, osho.new_status,
       hi.date, oh.status, oh.rev
   FROM order_status_history_old osho
   JOIN sample_order so ON osho.order_id = so.id
   JOIN order_history oh ON oh.id = so.id
   JOIN history_info hi ON hi.id = oh.rev;
   

恢复约束与索引 #

迁移完成后，恢复目标表的约束和索引：

• 启用触发器：ALTER TABLE history_info ENABLE TRIGGER ALL; ALTER TABLE order_history ENABLE TRIGGER ALL;
• 启用外键约束：ALTER TABLE order_history ENABLE CONSTRAINT <外键名称>;
• 重建索引：CREATE INDEX <索引名称> ON <表名>(<字段>);

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