Postgres双Schema数据迁移后无法截断落地区表的解决方案咨询

报错原因 #

你遇到的cannot TRUNCATE "table1" because it is being used by active queries in this session错误，核心原因是TRUNCATE需要获取表的ACCESS EXCLUSIVE独占锁，但触发器运行在插入事务的上下文内，当前会话正持有落地区表的共享锁（插入操作未完成），锁冲突导致截断失败。

解决方案1：批量迁移+定时异步执行（推荐） #

鉴于你的场景是每3小时批量插入数据，完全不需要行级触发器实时迁移，改用批量操作+定时任务更高效可靠：

1. 删除原有行级触发器
   先移除之前创建的所有迁移、截断相关触发器，避免冲突。

2. 创建批量迁移+截断的存储过程
   编写存储过程完成数据迁移与截断，关键是拆分事务，先提交迁移操作释放锁，再执行截断：

   CREATE OR REPLACE PROCEDURE migrate_and_truncate()
   LANGUAGE plpgsql
   AS $$
   BEGIN
       -- 批量迁移15张表（示例以3张为例，需补全所有表）
       INSERT INTO schema2.table1 SELECT * FROM schema1.table1 ON CONFLICT DO NOTHING;
       INSERT INTO schema2.table2 SELECT * FROM schema1.table2 ON CONFLICT DO NOTHING;
       INSERT INTO schema2.table3 SELECT * FROM schema1.table3 ON CONFLICT DO NOTHING;
       -- ... 补全剩余12张表的迁移语句
   
       -- 提交迁移事务，释放落地区表的锁
       COMMIT;
   
       -- 启动新事务执行截断
       BEGIN
           TRUNCATE TABLE schema1.table1, schema1.table2, schema1.table3, ..., schema1.table15;
           COMMIT;
       EXCEPTION
           WHEN OTHERS THEN
               ROLLBACK;
               RAISE;
       END;
   END;
   $$;
   

   注：ON CONFLICT子句根据你的业务需求调整，如需覆盖重复数据可改为DO UPDATE SET ...

3. 用pg_cron定时执行存储过程
   安装并配置pg_cron扩展，设置每3小时执行一次（可调整时间确保在插入任务完成后运行）：

   -- 启用pg_cron（首次使用需执行）
   ALTER SYSTEM SET cron.database_name = '你的数据库名';
   SELECT pg_reload_conf();
   
   -- 创建每3小时执行的定时任务（从整点开始）
   SELECT cron.schedule('migrate-and-truncate', '0 */3 * * *', 'CALL migrate_and_truncate();');
   

解决方案2：实时迁移+异步截断（适合必须实时同步的场景） #

如果需要插入后立即迁移数据，可通过LISTEN/NOTIFY机制实现异步截断：

1. 修改迁移触发器，仅做数据迁移并发送通知

   CREATE OR REPLACE FUNCTION schema1.migrate_single_row()
   RETURNS TRIGGER AS $$
   BEGIN
       INSERT INTO schema2."table1" VALUES (NEW.*);
       -- 发送迁移完成通知
       PERFORM pg_notify('migrate_done', 'table1');
       RETURN NEW;
   END;
   $$ LANGUAGE plpgsql;
   

   为落地区所有表绑定该触发器。

2. 编写后台监听脚本
   用Python/Shell等编写独立脚本，持续监听migrate_done通知，当确认所有15张表都完成迁移后，在独立会话中执行TRUNCATE命令。示例Python脚本片段：

   import psycopg2
   from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
   
   conn = psycopg2.connect("dbname=你的数据库名 user=用户名")
   conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
   cur = conn.cursor()
   cur.execute("LISTEN migrate_done;")
   
   migrated_tables = set()
   all_tables = {"table1", "table2", ..., "table15"}
   
   while True:
       conn.poll()
       while conn.notifies:
           notify = conn.notifies.pop()
           migrated_tables.add(notify.payload)
           # 所有表完成迁移后执行截断
           if migrated_tables == all_tables:
               cur.execute("TRUNCATE TABLE schema1.table1, schema1.table2, ..., schema1.table15;")
               migrated_tables.clear()
   

关键注意事项 #

• 禁止在触发器内部执行TRUNCATE：触发器处于插入事务上下文，无法获取截断所需的独占锁
• 批量操作比行级触发器效率高10倍以上，尤其适合大批次数据场景
• 多表TRUNCATE尽量合并为一个命令，减少锁的获取次数
• 若落地区表存在外键，需添加CASCADE参数，或按依赖顺序截断表

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