干货！PostgreSQL普通表转换为原生分区表的五种方法

CREATE TABLE tb_test
(id int primary key,
 info varchar,
 crt_time timestamp
);


INSERT INTO tb_test (id, info, crt_time)
SELECT generate_series(1, 1000000) AS id,
       md5(random()::text) AS info,
       '2023-01-01'::timestamp + random() * ('2025-12-31'::timestamp - '2023-01-01'::timestamp) AS crt_time;

CREATE TABLE tb_test_part
(id int,
 info varchar,
 crt_time timestamp,
 PRIMARY KEY (id, crt_time))
PARTITION BY RANGE(crt_time);
CREATE TABLE tb_test_before2023 PARTITION OF tb_test_part FOR VALUES FROM (MINVALUE) TO ('2023-01-01');
CREATE TABLE tb_test_2023 PARTITION OF tb_test_part FOR VALUES FROM ('2023-01-01') TO ('2024-01-01');
CREATE TABLE tb_test_2024 PARTITION OF tb_test_part FOR VALUES FROM ('2024-01-01') TO ('2025-01-01');
CREATE TABLE tb_test_2025 PARTITION OF tb_test_part FOR VALUES FROM ('2025-01-01') TO ('2026-01-01');
CREATE TABLE tb_test_2026 PARTITION OF tb_test_part FOR VALUES FROM ('2026-01-01') TO ('2027-01-01');

1. 使用INSERT INTO ... SELECT ...将源表数据插入分区表中。
2. 然后对源表和分区表进行重命名，让分区表接管业务。

• 操作十分简单。

• 迁移操作时，源表上不能有DML操作，迁移时需要业务系统停机。

适用场景：

• 允许业务系统短期停机的场景（例如凌晨的维护窗口期）。

-- 数据迁移
INSERT INTO tb_test_part SELECT * FROM tb_test;
-- 重命名库表
BEGIN;
ALTER TABLE tb_test RENAME TO tb_test_old;
ALTER TABLE tb_test_part RENAME TO tb_test;
END;

1. 使用COPY语句将源表中的输出导出成csv文件。
2. 通过COPY语句将csv文件中的内容导入分区表。
3. 对源表和分区表进行重命名，让分区表接管业务。

• 数据迁移效率高（COPY语句的效率远高于INSERT）。

• 可以通过脚本实现并发迁移（根据条件导出到多个csv文件）。

• 仍需要业务系统停机进行迁移。

• 需要使用具有SUPERUSER权限的用户来操作。

• 导出的csv文件需要占用额外的存储空间。

适用场景：

• 允许业务系统短期停机，但数据量较大的场景。

-- 导出数据
COPY tb_test TO '/backup/csv/tb_test.csv';
-- 导入数据
COPY tb_test_part FROM '/backup/csv/tb_test.csv';
-- 重命名库表
BEGIN;
ALTER TABLE tb_test RENAME TO tb_test_old;
ALTER TABLE tb_test_part RENAME TO tb_test;
END;

1. 创建视图作为业务表的入口，同时使用触发器作为路由，使得业务系统对库表结构无感知。
2. 批量迁移旧数据。
3. 删除视图、触发器，将分区表重命名为业务表。

• 业务无感知、零停机。

• 不需要第三方依赖。

• 操作相比前两个方案较复杂。

• 由于触发器的存在，可能会有一定的性能下降。

适用场景：

• 业务系统不能接受停机，能够接受少量性能下降。

1. 创建触发器函数。
   CREATE OR REPLACE FUNCTION fun_route_tb_test()
   RETURNS TRIGGER AS $$
   DECLARE flag_count int;
   BEGIN
       CASE TG_OP
           --  插入时先判断记录是否与现有数据冲突，无冲突则将记录插入分区表中。
           WHEN 'INSERT'
           THEN SELECT count(*) 
                  FROM (SELECT * FROM tb_test_old t1 WHERE t1.id=NEW.id
                        UNION ALL
                        SELECT * FROM tb_test_part t2 WHERE t2.id=NEW.id) t
                  INTO flag_count;
                IF flag_count>0 THEN
                    RETURN NULL;
                END IF;
                INSERT INTO tb_test_part VALUES (NEW.*);
                RETURN NEW;
          -- 删除时从原表和分区表中均执行一次DELETE操作。
           WHEN 'DELETE'
           THEN DELETE FROM tb_test_part t WHERE ROW(t.*) = ROW(OLD.*);
                DELETE FROM tb_test_old t WHERE ROW(t.*) = ROW(OLD.*);
                RETURN OLD;
           -- 更新时，先判断数据是否存在，再进行操作。
           WHEN 'UPDATE'
           THEN SELECT count(*) 
                  FROM (SELECT * FROM tb_test_old t1 WHERE ROW(t1.*) = ROW(OLD.*)
                        UNION ALL
                        SELECT * FROM tb_test_part t2 WHERE ROW(t2.*) = ROW(OLD.*)) t
                  INTO flag_count;
                IF flag_count=0 THEN
                    RETURN NULL;
                END IF;
                DELETE FROM tb_test_part t WHERE ROW(t.*) = ROW(OLD.*);
                INSERT INTO tb_test_part VALUES (NEW.*);
                DELETE FROM tb_test_old t WHERE ROW(t.*) = ROW(OLD.*);
                RETURN NEW;
        END CASE;
   END;
   $$ LANGUAGE plpgsql;
   

   
   

2. 重命名源表、创建视图、触发器。
   BEGIN;
   SET lock_timeout ='3s';
   --修改原表名称
   ALTER TABLE tb_test RENAME TO tb_test_old;
   --创建视图作为业务入口
   CREATE OR REPLACE VIEW tb_test AS
     SELECT * FROM tb_test_old
     UNION ALL
     SELECT * FROM tb_test_part;
   --创建触发器
   CREATE TRIGGER trg_route_tb_test
       INSTEAD OF INSERT OR UPDATE OR DELETE
       ON tb_test
       FOR EACH ROW
       EXECUTE FUNCTION fun_route_tb_test();
   END;
   

   
   

3. 增删改查测试
   SELECT测试
   db_test=# SELECT count(1) FROM tb_test;
     count
   ---------
    1000000
   (1 row)
   db_test=# SELECT count(1) FROM tb_test_old;
     count
   ---------
    1000000
   (1 row)
   db_test=# SELECT count(1) FROM tb_test_part;
    count
   -------
        0
   (1 row)
   INSERT测试
   db_test=# INSERT INTO tb_test VALUES (1,'test',now());
   INSERT 0 0
   db_test=# INSERT INTO tb_test VALUES (8888888,'test',now());
   INSERT 0 1
   db_test=# SELECT * FROM tb_test WHERE id=8888888;
      id    | info |          crt_time
   ---------+------+----------------------------
    8888888 | test | 2025-10-11 20:41:26.714864
   (1 row)
   db_test=# SELECT * FROM tb_test_part WHERE id=8888888;
      id    | info |          crt_time
   ---------+------+----------------------------
    8888888 | test | 2025-10-11 20:41:26.714864
   (1 row)
   db_test=# SELECT * FROM tb_test_old WHERE id=8888888;
    id | info | crt_time
   ----+------+----------
   (0 rows)
   DELETE测试
   db_test=# DELETE FROM tb_test WHERE id=6666666;
   DELETE 0
   db_test=# DELETE FROM tb_test WHERE id=9999;
   DELETE 1
   UPDATE测试
   db_test=# UPDATE tb_test SET info='test' WHERE id=9999999;
   UPDATE 0
   db_test=# SELECT * FROM tb_test limit 1;
    id |               info               |          crt_time
   ----+----------------------------------+----------------------------
     1 | 3869acfe33f9f7786ecc81d0437f7a0b | 2023-04-09 11:21:41.544323
   (1 row)
   db_test=# UPDATE tb_test SET info='test' WHERE id=1;
   UPDATE 1
   db_test=# SELECT * FROM tb_test WHERE id=1;
    id | info |          crt_time
   ----+------+----------------------------
     1 | test | 2023-04-09 11:21:41.544323
   (1 row)
   db_test=# SELECT * FROM tb_test_old WHERE id=1;
    id | info | crt_time
   ----+------+----------
   (0 rows)
   db_test=# SELECT * FROM tb_test_part WHERE id=1;
    id | info |          crt_time
   ----+------+----------------------------
     1 | test | 2023-04-09 11:21:41.544323
   (1 row)
   

   
   

4. 批量迁移数据，每次10000行。
   DO
   $$
   DECLARE flag_count int; -- 每批实际迁移的行数
           total_count int; --迁移的总行数
           begin_time timestamp; --开始迁移的时间
   BEGIN
       flag_count := 0;
       total_count := 0;
       begin_time := clock_timestamp();
       LOOP
           WITH t AS (                
               DELETE FROM tb_test_old WHERE ctid = ANY(ARRAY (SELECT ctid FROM tb_test_old LIMIT 10000 FOR UPDATE SKIP LOCKED)) RETURNING *  
           )
           INSERT INTO tb_test_part SELECT * FROM t;
           GET DIAGNOSTICS flag_count = row_count;
           EXIT WHEN flag_count = 0;
           total_count := total_count + flag_count;
           RAISE INFO 'move % rows, total % rows, after % .', flag_count, total_count,(clock_timestamp() - begin_time);
       END LOOP;
   END;
   $$;
   

   
   

5. 删除视图、触发器函数，重命名分区表。

BEGIN;  
SET lock_timeout ='3s';
DROP VIEW tb_test CASCADE;
DROP FUNCTION fun_route_tb_test();
ALTER TABLE tb_test_part RENAME TO tb_test;  
END;

1. 创建视图作为业务表的入口，同时使用规则（RULE）作为路由，使得业务系统对库表结构无感知。
2. 批量迁移旧数据。
3. 删除视图、触发器，将分区表重命名为业务表。

• 业务无感知、零停机。

• 不需要第三方依赖。

• 操作相比方案一、方案二较复杂。

• 相比触发器，规则不能使用plpgsql，有局限性。

适用场景：

• 业务系统不能接受停机，库表操作的逻辑不是特别复杂。

1. 重命名源表，创建视图，创建规则。

   BEGIN;
   SET lock_timeout ='3s';
   --修改原表名称
   ALTER TABLE tb_test RENAME TO tb_test_old;
   --创建视图作为业务入口
   CREATE OR REPLACE VIEW tb_test AS
     SELECT * FROM tb_test_old
     UNION ALL
     SELECT * FROM tb_test_part;
   --创建插入规则
   CREATE OR REPLACE RULE rul_ins_tb_test
       AS ON INSERT TO tb_test
       DO INSTEAD (
           INSERT INTO tb_test_part
                SELECT NEW.*
                 WHERE (SELECT count(*)
                          FROM tb_test_old t
                         WHERE t.id = new.id) = 0;
       );
   --创建删除规则
   CREATE OR REPLACE RULE rul_del_tb_test
       AS ON DELETE TO tb_test
       DO INSTEAD (
           DELETE FROM tb_test_part t WHERE ROW(t.*) = ROW(OLD.*);
           DELETE FROM tb_test_old t WHERE ROW(t.*) = ROW(OLD.*);
       );
   --创建更新规则
   CREATE OR REPLACE RULE rul_upd_tb_test
       AS ON UPDATE TO tb_test
       DO INSTEAD (
           UPDATE tb_test_old
              SET id = NEW.id,
                  info = NEW.info,
                  crt_time = NEW.crt_time
            WHERE id = OLD.id
              AND info = OLD.info
              AND crt_time = OLD.crt_time;
           UPDATE tb_test_part
              SET id = NEW.id,
                  info = NEW.info,
                  crt_time = NEW.crt_time
            WHERE id = OLD.id
              AND info = OLD.info
              AND crt_time = OLD.crt_time;
       );
   END;
   

   
   

2. 增删改查测试。

   SELECT测试
   db_test=# SELECT count(1) FROM tb_test;
     count
   ---------
    1000000
   (1 row)
   db_test=# SELECT count(1) FROM tb_test_old;
     count
   ---------
    1000000
   (1 row)
   db_test=# SELECT count(1) FROM tb_test_part;
    count
   -------
        0
   (1 row)
   INSERT测试
   db_test=# INSERT INTO tb_test VALUES (1,'test',now());
   INSERT 0 0
   db_test=# INSERT INTO tb_test VALUES (8888888,'test',now());
   INSERT 0 1
   db_test=# SELECT * FROM tb_test WHERE id=8888888;
      id    | info |          crt_time
   ---------+------+----------------------------
    8888888 | test | 2025-10-11 21:54:18.169915
   (1 row)
   db_test=# SELECT * FROM tb_test_old WHERE id=8888888;
    id | info | crt_time
   ----+------+----------
   (0 rows)
   db_test=# SELECT * FROM tb_test_part WHERE id=8888888;
      id    | info |          crt_time
   ---------+------+----------------------------
    8888888 | test | 2025-10-11 21:54:18.169915
   (1 row)
   DELETE测试
   db_test=# DELETE FROM tb_test WHERE id=6666666;
   DELETE 0
   db_test=# DELETE FROM tb_test WHERE id=9999;
   DELETE 1
   UPDATE测试
   db_test=# UPDATE tb_test SET info='test' WHERE id=9999999;
   UPDATE 0
   db_test=# SELECT * FROM tb_test limit 1;
    id |               info               |          crt_time
   ----+----------------------------------+----------------------------
     1 | 39538637a24d5d45f531626c12c14aa6 | 2023-03-25 09:53:37.405079
   (1 row)
   db_test=# UPDATE tb_test SET info='test' WHERE id=1;
   UPDATE 1
   db_test=# SELECT * FROM tb_test WHERE id=1;
    id | info |          crt_time
   ----+------+----------------------------
     1 | test | 2023-04-09 11:21:41.544323
   (1 row)
   db_test=# SELECT * FROM tb_test_old WHERE id=1;
    id | info |          crt_time
   ----+------+----------------------------
     1 | test | 2023-03-25 09:53:37.405079
   (1 row)
   db_test=# SELECT * FROM tb_test_part WHERE id=1;
    id | info | crt_time
   ----+------+----------
   (0 rows)
   

   
   

3. 批量迁移数据，每次10000行。

   DO
   $$
   DECLARE flag_count int; -- 每批实际迁移的行数
           total_count int; --迁移的总行数
           begin_time timestamp; --开始迁移的时间
   BEGIN
       flag_count := 0;
       total_count := 0;
       begin_time := clock_timestamp();
       LOOP
           WITH t AS (                
               DELETE FROM tb_test_old WHERE ctid = ANY(ARRAY (SELECT ctid FROM tb_test_old LIMIT 10000 FOR UPDATE SKIP LOCKED)) RETURNING *  
           )
           INSERT INTO tb_test_part SELECT * FROM t;
           GET DIAGNOSTICS flag_count = row_count;
           EXIT WHEN flag_count = 0;
           total_count := total_count + flag_count;
           RAISE INFO 'move % rows, total % rows, after % .', flag_count, total_count,(clock_timestamp() - begin_time);
       END LOOP;
   END;
   $$;
   

   
   

4. 删除视图、规则，重命名分区表。

BEGIN;  
SET lock_timeout ='3s';
DROP VIEW tb_test CASCADE;
ALTER TABLE tb_test_part RENAME TO tb_test;  
END;

1. 安装pg_rewrite扩展。
2. 修改数据库参数，重启数据库。
3. 通过pg_rewrite扩展提供的函数进行库表转换。

• 迁移过程不需要业务系统停机。

• 迁移操作较方案三、四较简单。

• 需要安装第三方扩展。

• 修改参数需要重启数据库。

• 使用了逻辑复制来同步数据，迁移过程中会产生大量的WAL日志。

• 可能需要提前修改源表的主键。

适用场景：

• 有较高操作权限的场景（允许安装第三方扩展和重启数据库）。

1. 安装pg_rewrite扩展。

   扩展源码包安装下载地址：https://github.com/cybertec-postgresql/pg_rewrite

   编译安装过程

   make
   make install
   

   
   

2. 修改数据库参数。

   PostgreSQL中配置以下参数，并重启数据库。其中max_replication_slots参数的值需要大于当前在使用的复制槽数量即可。

   wal_level = 'logical'max_replication_slots = 10
shared_preload_libraries = 'pg_rewrite'
   

   
   

3. 创建插件。

   db_test=# CREATE EXTENSION pg_rewrite;
   CREATE EXTENSION
   

   
   

4. 修改源表主键。

   -- 迁移过程要求目标表和源表的主键完全一致，需要修改源表的主键
   db_test=# ALTER TABLE tb_test DROP CONSTRAINT tb_test_pkey;
   ALTER TABLE
   db_test=# ALTER TABLE tb_test ADD CONSTRAINT tb_test_pkey PRIMARY KEY (id, crt_time);
   ALTER TABLE
   

   
   

5. 执行转换过程。

db_test=# SELECT rewrite_table('tb_test', 'tb_test_part', 'tb_test_old');
 rewrite_table
---------------
(1 row)
db_test=# \dt tb_test*
                      List of relations
 Schema |        Name        |       Type        |   Owner
--------+--------------------+-------------------+-----------
 public | tb_test            | partitioned table | pgmanager
 public | tb_test_2023       | table             | pgmanager
 public | tb_test_2024       | table             | pgmanager
 public | tb_test_2025       | table             | pgmanager
 public | tb_test_2026       | table             | pgmanager
 public | tb_test_before2023 | table             | pgmanager
 public | tb_test_old        | table             | pgmanager
(7 rows)
-- 迁移结果
db_test=# SELECT count(1) FROM tb_test;
  count
---------
 1000000
(1 row)
db_test=# SELECT count(1) FROM tb_test_old;
  count
---------
 1000000
(1 row)

• 无论采用哪种方式进行分区表转换，迁移前一定要做好备份工作。

• 正式转换前，要先在测试环境对方案进行验证，确保业务兼容性和操作可回滚。

• 在迁移过程中，要实时监控迁移进度和数据库服务器的资源使用情况，避免影响业务。

最后，欢迎感兴趣的朋友加入DB演武场交流群，在这里可以交流PostgreSQL、Linux、Kingbase、openGauss等各类技术知识。关注公众号“张家辉的DB演武场”后，后台发送“加群“，即可获取最新群二维码。