PostgreSQL 逻辑复制中的表DDL变更执行步骤

 
PostgreSQL的逻辑复制不会自动同步DDL（第三方插件另说），因此在逻辑复制环境中，当表发布的表执行DDL的时候，如果不停止业务的话，publication（发布节点，主节点）和subscription（订阅节点，从节点）如何执行相关的DDL，才能确保复制正常运行，不受DDL的影响？
结论是：首先在subscription从节点执行DDL，然后再publication主节点执行相关DDL，这样的话，复制不会受到任何影响。
在不停止业务的情况下，如果先在主节点DDL，那么就可能会往表的新增字段中写入数据，当WAL Sender进程解码出来新增的字段，通过apply worker进程在subscription上执行的时候，会因为找不到相关的字段报错。相反，如果先在subscription上执行相关DDL，修改表结构，此时apply worker中的数据并不包含新增字段的数据，是不影响复制的，然后在publication上执行相关DDL，这样就算是新增字段的数据立马写入数据，通过WAL Sender传递到subsacrption上可以完全正确地执行。

以下以t3表为例，t3表是一个发布订阅的表，现在要增加字段，也即执行DDL，可以先从subscription上执行，然后再publication执行，这样整个复制过程不会收到任何影响。
1，主节点逻辑复制槽以及publication信息

 
2，从节点subscription信息

 
3，主节点上写入数据

 
4，从节点正常复制

 
5，subscription从节点先执行DDL，与此同时，主节点执行DDL之前，t3上再次写入一条测试数据，同时可以正常复制主节点数据

 
6，publication主节点执行DDL，执行完往新字段写入数据

 
7，subscription从节点复制正常

 
主节点测试脚本

1. db01=# \x
2. Expanded display is on.
3. db01=#
4. db01=# select * from pg_replication_slots;
5. -[ RECORD 1 ]-------+------------------------------
6. slot_name           | pgstandby_slave01
7. plugin              |
8. slot_type           | physical
9. datoid              |
10. database            |
11. temporary           | f
12. active              | f
13. active_pid          |
14. xmin                |
15. catalog_xmin        |
16. restart_lsn         | 1/20518D30
17. confirmed_flush_lsn |
18. wal_status          | reserved
19. safe_wal_size       |
20. two_phase           | f
21. conflicting         |
22. -[ RECORD 2 ]-------+------------------------------
23. slot_name           | db01_logic_replication_slot01
24. plugin              | pgoutput
25. slot_type           | logical
26. datoid              | 16400
27. database            | db01
28. temporary           | f
29. active              | t
30. active_pid          | 33412
31. xmin                |
32. catalog_xmin        | 1171
33. restart_lsn         | 1/214B8040
34. confirmed_flush_lsn | 1/214B8078
35. wal_status          | reserved
36. safe_wal_size       |
37. two_phase           | f
38. conflicting         | f
40. db01=#
41. db01=# \x off
42. Expanded display is off.
43. db01=#
44. db01=#  select * from pg_catalog.pg_publication;
45.   oid  |        pubname         | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot
46. -------+------------------------+----------+--------------+-----------+-----------+-----------+-------------+------------
47. 41149 | master_db01_pulication |       10 | f            | t         | t         | f         | f           | f
48. (1 row)
50. db01=#
51. db01=# SELECT oid, prpubid, prrelid::regclass FROM pg_publication_rel;
52.   oid  | prpubid | prrelid
53. -------+---------+---------
54. 41150 |   41149 | t1
55. 41151 |   41149 | t2
56. 41158 |   41149 | t3
57. (3 rows)
59. db01=#
60. db01=# \d t3
61.                                      Table "public.t3"
62. Column |            Type             | Collation | Nullable |           Default
63. --------+-----------------------------+-----------+----------+------------------------------
64. c1     | integer                     |           | not null | generated always as identity
65. c2     | character varying(100)      |           |          |
66. c3     | timestamp without time zone |           |          |
67. Indexes:
68.     "t3_pkey" PRIMARY KEY, btree (c1)
69. Publications:
70.     "master_db01_pulication"
72. db01=#
73. db01=# insert into t3(c1,c2,c3) values (1,'aaa',now());
74. ERROR:  cannot insert a non-DEFAULT value into column "c1"
75. DETAIL:  Column "c1" is an identity column defined as GENERATED ALWAYS.
76. HINT:  Use OVERRIDING SYSTEM VALUE to override.
77. db01=#
78. db01=#
79. db01=#
80. db01=#
81. db01=# \d t3
82.                                      Table "public.t3"
83. Column |            Type             | Collation | Nullable |           Default
84. --------+-----------------------------+-----------+----------+------------------------------
85. c1     | integer                     |           | not null | generated always as identity
86. c2     | character varying(100)      |           |          |
87. c3     | timestamp without time zone |           |          |
88. Indexes:
89.     "t3_pkey" PRIMARY KEY, btree (c1)
90. Publications:
91.     "master_db01_pulication"
93. db01=#
94. db01=#
95. db01=# insert into t3(c2,c3) values ('aaa',now());
96. INSERT 0 1
97. db01=#
98. db01=# select * from t3
99. db01-# ;
100. c1 | c2  |            c3
101. ----+-----+--------------------------
102.   1 | aaa | 2025-06-26 14:31:50.1698
103. (1 row)
105. db01=#
106. db01=#
107. db01=# insert into t3(c2,c3) values ('bbb',now());
108. INSERT 0 1
109. db01=#
110. db01=# select * from t3;
111. c1 | c2  |             c3
112. ----+-----+----------------------------
113.   1 | aaa | 2025-06-26 14:31:50.1698
114.   2 | bbb | 2025-06-26 14:35:15.171404
115. (2 rows)
117. db01=# alter table t3 add c4 timestamp;
118. ALTER TABLE
119. db01=#
120. db01=# insert into t3(c2,c3) values ('ccc',now(),now());
121. ERROR:  INSERT has more expressions than target columns
122. LINE 1: insert into t3(c2,c3) values ('ccc',now(),now());
123.                                                   ^
124. db01=# insert into t3(c2,c3,c4) values ('ccc',now(),now());
125. INSERT 0 1
126. db01=#
127. db01=# select * from t3;
128. c1 | c2  |             c3             |             c4
129. ----+-----+----------------------------+----------------------------
130.   1 | aaa | 2025-06-26 14:31:50.1698   |
131.   2 | bbb | 2025-06-26 14:35:15.171404 |
132.   3 | ccc | 2025-06-26 14:37:43.571553 | 2025-06-26 14:37:43.571553
133. (3 rows)
135. db01=#
136. db01=#

复制代码

 
从节点测试脚本

1. mydb01=#  select * from pg_subscription;
2. -[ RECORD 1 ]-------+---------------------------------------------------------------------------------
3. oid                 | 16407
4. subdbid             | 16388
5. subskiplsn          | 0/0
6. subname             | slave_db01_subscription
7. subowner            | 10
8. subenabled          | t
9. subbinary           | f
10. substream           | f
11. subtwophasestate    | d
12. subdisableonerr     | f
13. subpasswordrequired | t
14. subrunasowner       | f
15. subconninfo         | host=8.*.*.129 port=9000 dbname=db01 user=replica_user  password=******
16. subslotname         | db01_logic_replication_slot01
17. subsynccommit       | off
18. subpublications     | {master_db01_pulication}
19. suborigin           | any
21. mydb01=# SELECT *, srrelid::regclass FROM pg_subscription_rel;
22. -[ RECORD 1 ]----------
23. srsubid    | 16407
24. srrelid    | 16402
25. srsubstate | r
26. srsublsn   | 1/21166E58
27. srrelid    | t3
28. -[ RECORD 2 ]----------
29. srsubid    | 16407
30. srrelid    | 16390
31. srsubstate | r
32. srsublsn   | 1/2116F7E0
33. srrelid    | t1
34. -[ RECORD 3 ]----------
35. srsubid    | 16407
36. srrelid    | 16396
37. srsubstate | r
38. srsublsn   | 1/21171BF0
39. srrelid    | t2
41. mydb01=#
42. mydb01=# \d t3
43.                                      Table "public.t3"
44. Column |            Type             | Collation | Nullable |           Default
45. --------+-----------------------------+-----------+----------+------------------------------
46. c1     | integer                     |           | not null | generated always as identity
47. c2     | character varying(100)      |           |          |
48. c3     | timestamp without time zone |           |          |
49. Indexes:
50.     "t3_pkey" PRIMARY KEY, btree (c1)
52. mydb01=#
53. mydb01=#
54. mydb01=# select * from t3;
55. -[ RECORD 1 ]----------------
56. c1 | 1
57. c2 | aaa
58. c3 | 2025-06-26 14:31:50.1698
60. mydb01=#
61. mydb01=# \x off
62. Expanded display is off.
63. mydb01=#
64. mydb01=#
65. mydb01=# select * from t3;
66. c1 | c2  |            c3
67. ----+-----+--------------------------
68.   1 | aaa | 2025-06-26 14:31:50.1698
69. (1 row)
71. mydb01=#
72. mydb01=#
73. mydb01=#
74. mydb01=# \d t3
75.                                      Table "public.t3"
76. Column |            Type             | Collation | Nullable |           Default
77. --------+-----------------------------+-----------+----------+------------------------------
78. c1     | integer                     |           | not null | generated always as identity
79. c2     | character varying(100)      |           |          |
80. c3     | timestamp without time zone |           |          |
81. Indexes:
82.     "t3_pkey" PRIMARY KEY, btree (c1)
84. mydb01=#
85. mydb01=#
86. mydb01=# select * from t3;
87. c1 | c2  |            c3
88. ----+-----+--------------------------
89.   1 | aaa | 2025-06-26 14:31:50.1698
90. (1 row)
92. mydb01=#
93. mydb01=#
94. mydb01=# alter table t3 add c4 timestamp;
95. ALTER TABLE
96. mydb01=#
97. mydb01=# select * from t3;
98. c1 | c2  |             c3             | c4
99. ----+-----+----------------------------+----
100.   1 | aaa | 2025-06-26 14:31:50.1698   |
101.   2 | bbb | 2025-06-26 14:35:15.171404 |
102. (2 rows)
104. mydb01=# select * from t3;
105. c1 | c2  |             c3             |             c4
106. ----+-----+----------------------------+----------------------------
107.   1 | aaa | 2025-06-26 14:31:50.1698   |
108.   2 | bbb | 2025-06-26 14:35:15.171404 |
109.   3 | ccc | 2025-06-26 14:37:43.571553 | 2025-06-26 14:37:43.571553
110. (3 rows)
112. mydb01=#

复制代码

 

来源：豆瓜网用户自行投稿发布，如果侵权，请联系站长删除