1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
|
# Copyright (c) 2021, PostgreSQL Global Development Group
# Test TRUNCATE
use strict;
use warnings;
use PostgresNode;
use TestLib;
use Test::More tests => 14;
# setup
my $node_publisher = get_new_node('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->start;
my $node_subscriber = get_new_node('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->append_conf('postgresql.conf',
qq(max_logical_replication_workers = 6));
$node_subscriber->start;
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_publisher->safe_psql('postgres',
"CREATE TABLE tab1 (a int PRIMARY KEY)");
$node_subscriber->safe_psql('postgres',
"CREATE TABLE tab1 (a int PRIMARY KEY)");
$node_publisher->safe_psql('postgres',
"CREATE TABLE tab2 (a int PRIMARY KEY)");
$node_subscriber->safe_psql('postgres',
"CREATE TABLE tab2 (a int PRIMARY KEY)");
$node_publisher->safe_psql('postgres',
"CREATE TABLE tab3 (a int PRIMARY KEY)");
$node_subscriber->safe_psql('postgres',
"CREATE TABLE tab3 (a int PRIMARY KEY)");
$node_publisher->safe_psql('postgres',
"CREATE TABLE tab4 (x int PRIMARY KEY, y int REFERENCES tab3)");
$node_subscriber->safe_psql('postgres',
"CREATE TABLE tab4 (x int PRIMARY KEY, y int REFERENCES tab3)");
$node_subscriber->safe_psql('postgres',
"CREATE SEQUENCE seq1 OWNED BY tab1.a");
$node_subscriber->safe_psql('postgres', "ALTER SEQUENCE seq1 START 101");
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION pub1 FOR TABLE tab1");
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION pub2 FOR TABLE tab2 WITH (publish = insert)");
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION pub3 FOR TABLE tab3, tab4");
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
);
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub2 CONNECTION '$publisher_connstr' PUBLICATION pub2"
);
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub3 CONNECTION '$publisher_connstr' PUBLICATION pub3"
);
# Wait for initial sync of all subscriptions
my $synced_query =
"SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
$node_subscriber->poll_query_until('postgres', $synced_query)
or die "Timed out while waiting for subscriber to synchronize data";
# insert data to truncate
$node_subscriber->safe_psql('postgres',
"INSERT INTO tab1 VALUES (1), (2), (3)");
$node_publisher->wait_for_catchup('sub1');
# truncate and check
$node_publisher->safe_psql('postgres', "TRUNCATE tab1");
$node_publisher->wait_for_catchup('sub1');
my $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab1");
is($result, qq(0||), 'truncate replicated');
$result = $node_subscriber->safe_psql('postgres', "SELECT nextval('seq1')");
is($result, qq(1), 'sequence not restarted');
# truncate with restart identity
$node_publisher->safe_psql('postgres', "TRUNCATE tab1 RESTART IDENTITY");
$node_publisher->wait_for_catchup('sub1');
$result = $node_subscriber->safe_psql('postgres', "SELECT nextval('seq1')");
is($result, qq(101), 'truncate restarted identities');
# test publication that does not replicate truncate
$node_subscriber->safe_psql('postgres',
"INSERT INTO tab2 VALUES (1), (2), (3)");
$node_publisher->safe_psql('postgres', "TRUNCATE tab2");
$node_publisher->wait_for_catchup('sub2');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab2");
is($result, qq(3|1|3), 'truncate not replicated');
$node_publisher->safe_psql('postgres',
"ALTER PUBLICATION pub2 SET (publish = 'insert, truncate')");
$node_publisher->safe_psql('postgres', "TRUNCATE tab2");
$node_publisher->wait_for_catchup('sub2');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab2");
is($result, qq(0||), 'truncate replicated after publication change');
# test multiple tables connected by foreign keys
$node_subscriber->safe_psql('postgres',
"INSERT INTO tab3 VALUES (1), (2), (3)");
$node_subscriber->safe_psql('postgres',
"INSERT INTO tab4 VALUES (11, 1), (111, 1), (22, 2)");
$node_publisher->safe_psql('postgres', "TRUNCATE tab3, tab4");
$node_publisher->wait_for_catchup('sub3');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab3");
is($result, qq(0||), 'truncate of multiple tables replicated');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(x), max(x) FROM tab4");
is($result, qq(0||), 'truncate of multiple tables replicated');
# test truncate of multiple tables, some of which are not published
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub2");
$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub2");
$node_subscriber->safe_psql('postgres',
"INSERT INTO tab1 VALUES (1), (2), (3)");
$node_subscriber->safe_psql('postgres',
"INSERT INTO tab2 VALUES (1), (2), (3)");
$node_publisher->safe_psql('postgres', "TRUNCATE tab1, tab2");
$node_publisher->wait_for_catchup('sub1');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab1");
is($result, qq(0||), 'truncate of multiple tables some not published');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab2");
is($result, qq(3|1|3), 'truncate of multiple tables some not published');
# Test that truncate works for synchronous logical replication
$node_publisher->safe_psql('postgres',
"ALTER SYSTEM SET synchronous_standby_names TO 'sub1'");
$node_publisher->safe_psql('postgres', "SELECT pg_reload_conf()");
# insert data to truncate
$node_publisher->safe_psql('postgres',
"INSERT INTO tab1 VALUES (1), (2), (3)");
$node_publisher->wait_for_catchup('sub1');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab1");
is($result, qq(3|1|3), 'check synchronous logical replication');
$node_publisher->safe_psql('postgres', "TRUNCATE tab1");
$node_publisher->wait_for_catchup('sub1');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab1");
is($result, qq(0||),
'truncate replicated in synchronous logical replication');
$node_publisher->safe_psql('postgres',
"ALTER SYSTEM RESET synchronous_standby_names");
$node_publisher->safe_psql('postgres', "SELECT pg_reload_conf()");
# test that truncate works for logical replication when there are multiple
# subscriptions for a single table
$node_publisher->safe_psql('postgres', "CREATE TABLE tab5 (a int)");
$node_subscriber->safe_psql('postgres', "CREATE TABLE tab5 (a int)");
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION pub5 FOR TABLE tab5");
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub5_1 CONNECTION '$publisher_connstr' PUBLICATION pub5"
);
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub5_2 CONNECTION '$publisher_connstr' PUBLICATION pub5"
);
# wait for initial data sync
$node_subscriber->poll_query_until('postgres', $synced_query)
or die "Timed out while waiting for subscriber to synchronize data";
# insert data to truncate
$node_publisher->safe_psql('postgres',
"INSERT INTO tab5 VALUES (1), (2), (3)");
$node_publisher->wait_for_catchup('sub5_1');
$node_publisher->wait_for_catchup('sub5_2');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab5");
is($result, qq(6|1|3), 'insert replicated for multiple subscriptions');
$node_publisher->safe_psql('postgres', "TRUNCATE tab5");
$node_publisher->wait_for_catchup('sub5_1');
$node_publisher->wait_for_catchup('sub5_2');
$result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), min(a), max(a) FROM tab5");
is($result, qq(0||), 'truncate replicated for multiple subscriptions');
# check deadlocks
$result = $node_subscriber->safe_psql('postgres',
"SELECT deadlocks FROM pg_stat_database WHERE datname='postgres'");
is($result, qq(0), 'no deadlocks detected');
|
