Skip to content

Commit 6a640de

Browse files
arssherarssher
committed
[PGPRO-4074] Port 2PC decoding patch.
This is an edited port of 3687dc02 -- 2pc decoding patch from ee-10, which is based on 29.01.18 version from mail lists. I've decided to keep it instead of picking latest patch on hackers because - I'm still not satisfied by the quality of the latter; - It is strongly mixed with streaming patch, there would be more conflicts; However the older patch isn't a honey as well, in particular I - removed weird filter_decode hook; - removed pgoutput support, which is most probably broken (at least tablesync, but probably more) and not needed; - removed wrong SnapBuildCommitTxn during PREPARE decoding. - the last fix led to another problem: since rbtxns are purged after PREPARE replay, on COMMIT PREPARED we don't know which subxids had catalog changers and which hadn't. To cope with it, mark all of them as potentially catalog modifying now. Reproduced by regression tests (they have DDL in subxacts). This also required not asserting strict LSN monotonicity of rbtxns. tags: multimaster (cherry picked from commit d38b55fc36ae5ae54bacbf37956ce24ab8632986)
1 parent 043ce02 commit 6a640de

File tree

12 files changed

+1329
-62
lines changed

12 files changed

+1329
-62
lines changed

‎contrib/test_decoding/Makefile‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,3 +32,8 @@ endif
3232
installcheck-force:
3333
$(pg_regress_installcheck) $(REGRESS)
3434
$(pg_isolation_regress_installcheck) $(ISOLATION)
35+
36+
temp-install: EXTRA_INSTALL=contrib/test_decoding
37+
38+
2pc-check: temp-install
39+
$(prove_check)

‎contrib/test_decoding/expected/prepared.out‎

Lines changed: 317 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -6,69 +6,366 @@ SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_d
66
init
77
(1 row)
88

9-
CREATE TABLE test_prepared1(id int);
10-
CREATE TABLE test_prepared2(id int);
9+
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_2pc', 'test_decoding');
10+
?column?
11+
----------
12+
init
13+
(1 row)
14+
15+
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_2pc_nofilter', 'test_decoding');
16+
?column?
17+
----------
18+
init
19+
(1 row)
20+
21+
CREATE TABLE test_prepared1(id integer primary key);
22+
CREATE TABLE test_prepared2(id integer primary key);
23+
-- Reused queries
24+
\set get_no2pc 'SELECT data FROM pg_logical_slot_get_changes(''regression_slot'', NULL, NULL, ''include-xids'', ''0'', ''skip-empty-xacts'', ''1'');'
25+
\set get_with2pc 'SELECT data FROM pg_logical_slot_get_changes(''regression_slot_2pc'', NULL, NULL, ''include-xids'', ''0'', ''skip-empty-xacts'', ''1'', ''twophase-decoding'', ''1'');'
26+
\set get_with2pc_nofilter 'SELECT data FROM pg_logical_slot_get_changes(''regression_slot_2pc_nofilter'', NULL, NULL, ''include-xids'', ''0'', ''skip-empty-xacts'', ''1'', ''twophase-decoding'', ''1'', ''twophase-decode-with-catalog-changes'', ''1'');'
1127
-- test simple successful use of a prepared xact
1228
BEGIN;
1329
INSERT INTO test_prepared1 VALUES (1);
1430
PREPARE TRANSACTION 'test_prepared#1';
31+
:get_no2pc
32+
data
33+
------
34+
(0 rows)
35+
36+
:get_with2pc
37+
data
38+
----------------------------------------------------
39+
BEGIN
40+
table public.test_prepared1: INSERT: id[integer]:1
41+
PREPARE TRANSACTION 'test_prepared#1'
42+
(3 rows)
43+
44+
:get_with2pc_nofilter
45+
data
46+
----------------------------------------------------
47+
BEGIN
48+
table public.test_prepared1: INSERT: id[integer]:1
49+
PREPARE TRANSACTION 'test_prepared#1'
50+
(3 rows)
51+
1552
COMMIT PREPARED 'test_prepared#1';
53+
:get_no2pc
54+
data
55+
----------------------------------------------------
56+
BEGIN
57+
table public.test_prepared1: INSERT: id[integer]:1
58+
COMMIT
59+
(3 rows)
60+
61+
:get_with2pc
62+
data
63+
-----------------------------------
64+
COMMIT PREPARED 'test_prepared#1'
65+
(1 row)
66+
67+
:get_with2pc_nofilter
68+
data
69+
-----------------------------------
70+
COMMIT PREPARED 'test_prepared#1'
71+
(1 row)
72+
1673
INSERT INTO test_prepared1 VALUES (2);
1774
-- test abort of a prepared xact
1875
BEGIN;
1976
INSERT INTO test_prepared1 VALUES (3);
2077
PREPARE TRANSACTION 'test_prepared#2';
78+
:get_no2pc
79+
data
80+
----------------------------------------------------
81+
BEGIN
82+
table public.test_prepared1: INSERT: id[integer]:2
83+
COMMIT
84+
(3 rows)
85+
86+
:get_with2pc
87+
data
88+
----------------------------------------------------
89+
BEGIN
90+
table public.test_prepared1: INSERT: id[integer]:2
91+
COMMIT
92+
BEGIN
93+
table public.test_prepared1: INSERT: id[integer]:3
94+
PREPARE TRANSACTION 'test_prepared#2'
95+
(6 rows)
96+
97+
:get_with2pc_nofilter
98+
data
99+
----------------------------------------------------
100+
BEGIN
101+
table public.test_prepared1: INSERT: id[integer]:2
102+
COMMIT
103+
BEGIN
104+
table public.test_prepared1: INSERT: id[integer]:3
105+
PREPARE TRANSACTION 'test_prepared#2'
106+
(6 rows)
107+
21108
ROLLBACK PREPARED 'test_prepared#2';
109+
:get_no2pc
110+
data
111+
------
112+
(0 rows)
113+
114+
:get_with2pc
115+
data
116+
-------------------------------------
117+
ROLLBACK PREPARED 'test_prepared#2'
118+
(1 row)
119+
120+
:get_with2pc_nofilter
121+
data
122+
-------------------------------------
123+
ROLLBACK PREPARED 'test_prepared#2'
124+
(1 row)
125+
22126
INSERT INTO test_prepared1 VALUES (4);
23127
-- test prepared xact containing ddl
24128
BEGIN;
25129
INSERT INTO test_prepared1 VALUES (5);
26130
ALTER TABLE test_prepared1 ADD COLUMN data text;
27131
INSERT INTO test_prepared1 VALUES (6, 'frakbar');
28132
PREPARE TRANSACTION 'test_prepared#3';
29-
-- test that we decode correctly while an uncommitted prepared xact
30-
-- with ddl exists.
31-
-- separate table because of the lock from the ALTER
32-
-- this will come before the '5' row above, as this commits before it.
33-
INSERT INTO test_prepared2 VALUES (7);
34-
COMMIT PREPARED 'test_prepared#3';
35-
-- make sure stuff still works
36-
INSERT INTO test_prepared1 VALUES (8);
37-
INSERT INTO test_prepared2 VALUES (9);
38-
-- cleanup
39-
DROP TABLE test_prepared1;
40-
DROP TABLE test_prepared2;
41-
-- show results
42-
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
133+
SELECT 'test_prepared_1' AS relation, locktype, mode
134+
FROM pg_locks
135+
WHERE locktype = 'relation'
136+
AND relation = 'test_prepared1'::regclass;
137+
relation | locktype | mode
138+
-----------------+----------+---------------------
139+
test_prepared_1 | relation | RowExclusiveLock
140+
test_prepared_1 | relation | AccessExclusiveLock
141+
(2 rows)
142+
143+
:get_no2pc
144+
data
145+
----------------------------------------------------
146+
BEGIN
147+
table public.test_prepared1: INSERT: id[integer]:4
148+
COMMIT
149+
(3 rows)
150+
151+
:get_with2pc
152+
data
153+
----------------------------------------------------
154+
BEGIN
155+
table public.test_prepared1: INSERT: id[integer]:4
156+
COMMIT
157+
(3 rows)
158+
159+
:get_with2pc_nofilter
43160
data
44161
-------------------------------------------------------------------------
45162
BEGIN
46-
table public.test_prepared1: INSERT: id[integer]:1
163+
table public.test_prepared1: INSERT: id[integer]:4
47164
COMMIT
48165
BEGIN
49-
table public.test_prepared1: INSERT: id[integer]:2
166+
table public.test_prepared1: INSERT: id[integer]:5
167+
table public.test_prepared1: INSERT: id[integer]:6 data[text]:'frakbar'
168+
PREPARE TRANSACTION 'test_prepared#3'
169+
(7 rows)
170+
171+
-- Test that we decode correctly while an uncommitted prepared xact
172+
-- with ddl exists. Our 2pc filter callback will skip decoding of xacts
173+
-- with catalog changes at PREPARE time, so we don't decode it now.
174+
--
175+
-- Use a separate table for the concurrent transaction because the lock from
176+
-- the ALTER will stop us inserting into the other one.
177+
--
178+
-- We should see '7' before '5' in our results since it commits first.
179+
--
180+
INSERT INTO test_prepared2 VALUES (7);
181+
:get_no2pc
182+
data
183+
----------------------------------------------------
184+
BEGIN
185+
table public.test_prepared2: INSERT: id[integer]:7
50186
COMMIT
187+
(3 rows)
188+
189+
:get_with2pc
190+
data
191+
----------------------------------------------------
51192
BEGIN
52-
table public.test_prepared1: INSERT: id[integer]:4
193+
table public.test_prepared2: INSERT: id[integer]:7
53194
COMMIT
195+
(3 rows)
196+
197+
:get_with2pc_nofilter
198+
data
199+
----------------------------------------------------
54200
BEGIN
55201
table public.test_prepared2: INSERT: id[integer]:7
56202
COMMIT
203+
(3 rows)
204+
205+
COMMIT PREPARED 'test_prepared#3';
206+
:get_no2pc
207+
data
208+
-------------------------------------------------------------------------
209+
BEGIN
210+
table public.test_prepared1: INSERT: id[integer]:5
211+
table public.test_prepared1: INSERT: id[integer]:6 data[text]:'frakbar'
212+
COMMIT
213+
(4 rows)
214+
215+
:get_with2pc
216+
data
217+
-------------------------------------------------------------------------
57218
BEGIN
58219
table public.test_prepared1: INSERT: id[integer]:5
59220
table public.test_prepared1: INSERT: id[integer]:6 data[text]:'frakbar'
60221
COMMIT
222+
(4 rows)
223+
224+
:get_with2pc_nofilter
225+
data
226+
-----------------------------------
227+
COMMIT PREPARED 'test_prepared#3'
228+
(1 row)
229+
230+
-- make sure stuff still works
231+
INSERT INTO test_prepared1 VALUES (8);
232+
INSERT INTO test_prepared2 VALUES (9);
233+
:get_no2pc
234+
data
235+
--------------------------------------------------------------------
236+
BEGIN
237+
table public.test_prepared1: INSERT: id[integer]:8 data[text]:null
238+
COMMIT
239+
BEGIN
240+
table public.test_prepared2: INSERT: id[integer]:9
241+
COMMIT
242+
(6 rows)
243+
244+
:get_with2pc
245+
data
246+
--------------------------------------------------------------------
247+
BEGIN
248+
table public.test_prepared1: INSERT: id[integer]:8 data[text]:null
249+
COMMIT
250+
BEGIN
251+
table public.test_prepared2: INSERT: id[integer]:9
252+
COMMIT
253+
(6 rows)
254+
255+
:get_with2pc_nofilter
256+
data
257+
--------------------------------------------------------------------
61258
BEGIN
62259
table public.test_prepared1: INSERT: id[integer]:8 data[text]:null
63260
COMMIT
64261
BEGIN
65262
table public.test_prepared2: INSERT: id[integer]:9
66263
COMMIT
67-
(22 rows)
264+
(6 rows)
265+
266+
-- Check `CLUSTER` (as operation that hold exclusive lock) doesn't block
267+
-- logical decoding.
268+
BEGIN;
269+
INSERT INTO test_prepared1 VALUES (10, 'othercol');
270+
CLUSTER test_prepared1 USING test_prepared1_pkey;
271+
INSERT INTO test_prepared1 VALUES (11, 'othercol2');
272+
PREPARE TRANSACTION 'test_prepared_lock';
273+
BEGIN;
274+
insert into test_prepared2 values (12);
275+
PREPARE TRANSACTION 'test_prepared_lock2';
276+
COMMIT PREPARED 'test_prepared_lock2';
277+
SELECT 'pg_class' AS relation, locktype, mode
278+
FROM pg_locks
279+
WHERE locktype = 'relation'
280+
AND relation = 'pg_class'::regclass;
281+
relation | locktype | mode
282+
----------+----------+------
283+
(0 rows)
284+
285+
-- Shouldn't see anything with 2pc decoding off
286+
:get_no2pc
287+
data
288+
-----------------------------------------------------
289+
BEGIN
290+
table public.test_prepared2: INSERT: id[integer]:12
291+
COMMIT
292+
(3 rows)
293+
294+
-- Shouldn't timeout on 2pc decoding.
295+
SET statement_timeout = '1s';
296+
:get_with2pc
297+
data
298+
-----------------------------------------------------
299+
BEGIN
300+
table public.test_prepared2: INSERT: id[integer]:12
301+
PREPARE TRANSACTION 'test_prepared_lock2'
302+
COMMIT PREPARED 'test_prepared_lock2'
303+
(4 rows)
304+
305+
:get_with2pc_nofilter
306+
data
307+
----------------------------------------------------------------------------
308+
BEGIN
309+
table public.test_prepared1: INSERT: id[integer]:10 data[text]:'othercol'
310+
table public.test_prepared1: INSERT: id[integer]:11 data[text]:'othercol2'
311+
PREPARE TRANSACTION 'test_prepared_lock'
312+
BEGIN
313+
table public.test_prepared2: INSERT: id[integer]:12
314+
PREPARE TRANSACTION 'test_prepared_lock2'
315+
COMMIT PREPARED 'test_prepared_lock2'
316+
(8 rows)
317+
318+
RESET statement_timeout;
319+
COMMIT PREPARED 'test_prepared_lock';
320+
-- Both will work normally after we commit
321+
:get_no2pc
322+
data
323+
----------------------------------------------------------------------------
324+
BEGIN
325+
table public.test_prepared1: INSERT: id[integer]:10 data[text]:'othercol'
326+
table public.test_prepared1: INSERT: id[integer]:11 data[text]:'othercol2'
327+
COMMIT
328+
(4 rows)
329+
330+
:get_with2pc
331+
data
332+
----------------------------------------------------------------------------
333+
BEGIN
334+
table public.test_prepared1: INSERT: id[integer]:10 data[text]:'othercol'
335+
table public.test_prepared1: INSERT: id[integer]:11 data[text]:'othercol2'
336+
COMMIT
337+
(4 rows)
338+
339+
:get_with2pc_nofilter
340+
data
341+
--------------------------------------
342+
COMMIT PREPARED 'test_prepared_lock'
343+
(1 row)
344+
345+
-- cleanup
346+
DROP TABLE test_prepared1;
347+
DROP TABLE test_prepared2;
348+
-- show results
349+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
350+
data
351+
------
352+
(0 rows)
68353

69354
SELECT pg_drop_replication_slot('regression_slot');
70355
pg_drop_replication_slot
71356
--------------------------
72357

73358
(1 row)
74359

360+
SELECT pg_drop_replication_slot('regression_slot_2pc');
361+
pg_drop_replication_slot
362+
--------------------------
363+
364+
(1 row)
365+
366+
SELECT pg_drop_replication_slot('regression_slot_2pc_nofilter');
367+
pg_drop_replication_slot
368+
--------------------------
369+
370+
(1 row)
371+

0 commit comments

Comments
 (0)