Location via proxy:   [ UP ]  
[Report a bug]   [Manage cookies]                
Skip to content

Commit a271a1b

Browse files
author
Amit Kapila
committed
Allow decoding at prepare time in ReorderBuffer.
This patch allows PREPARE-time decoding of two-phase transactions (if the output plugin supports this capability), in which case the transactions are replayed at PREPARE and then committed later when COMMIT PREPARED arrives. Now that we decode the changes before the commit, the concurrent aborts may cause failures when the output plugin consults catalogs (both system and user-defined). We detect such failures with a special sqlerrcode ERRCODE_TRANSACTION_ROLLBACK introduced by commit 7259736 and stop decoding the remaining changes. Then we rollback the changes when rollback prepared is encountered. Author: Ajin Cherian and Amit Kapila based on previous work by Nikhil Sontakke and Stas Kelvich Reviewed-by: Amit Kapila, Peter Smith, Sawada Masahiko, Arseny Sher, and Dilip Kumar Tested-by: Takamichi Osumi Discussion: https://postgr.es/m/02DA5F5E-CECE-4D9C-8B4B-418077E2C010@postgrespro.ru https://postgr.es/m/CAMGcDxeqEpWj3fTXwqhSwBdXd2RS9jzwWscO-XbeCfso6ts3+Q@mail.gmail.com
1 parent ca3b374 commit a271a1b

File tree

11 files changed

+1296
-116
lines changed

11 files changed

+1296
-116
lines changed

‎contrib/test_decoding/Makefile

+1-1
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ PGFILEDESC = "test_decoding - example of a logical decoding output plugin"
55

66
REGRESS = ddl xact rewrite toast permissions decoding_in_xact \
77
decoding_into_rel binary prepared replorigin time messages \
8-
spill slot truncate stream stats
8+
spill slot truncate stream stats twophase twophase_stream
99
ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \
1010
oldest_xmin snapshot_transfer subxact_without_top concurrent_stream
1111

‎contrib/test_decoding/expected/twophase.out

+235
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,235 @@
1+
-- Test prepared transactions. When two-phase-commit is enabled, transactions are
2+
-- decoded at PREPARE time rather than at COMMIT PREPARED time.
3+
SET synchronous_commit = on;
4+
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
5+
?column?
6+
----------
7+
init
8+
(1 row)
9+
10+
CREATE TABLE test_prepared1(id integer primary key);
11+
CREATE TABLE test_prepared2(id integer primary key);
12+
-- Test that decoding happens at PREPARE time when two-phase-commit is enabled.
13+
-- Decoding after COMMIT PREPARED must have all the commands in the transaction.
14+
BEGIN;
15+
INSERT INTO test_prepared1 VALUES (1);
16+
INSERT INTO test_prepared1 VALUES (2);
17+
-- should show nothing because the xact has not been prepared yet.
18+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
19+
data
20+
------
21+
(0 rows)
22+
23+
PREPARE TRANSACTION 'test_prepared#1';
24+
-- should show both the above inserts and the PREPARE TRANSACTION.
25+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
26+
data
27+
----------------------------------------------------
28+
BEGIN
29+
table public.test_prepared1: INSERT: id[integer]:1
30+
table public.test_prepared1: INSERT: id[integer]:2
31+
PREPARE TRANSACTION 'test_prepared#1'
32+
(4 rows)
33+
34+
COMMIT PREPARED 'test_prepared#1';
35+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
36+
data
37+
----------------------------------------------------
38+
BEGIN
39+
table public.test_prepared1: INSERT: id[integer]:1
40+
table public.test_prepared1: INSERT: id[integer]:2
41+
PREPARE TRANSACTION 'test_prepared#1'
42+
COMMIT PREPARED 'test_prepared#1'
43+
(5 rows)
44+
45+
-- Test that rollback of a prepared xact is decoded.
46+
BEGIN;
47+
INSERT INTO test_prepared1 VALUES (3);
48+
PREPARE TRANSACTION 'test_prepared#2';
49+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
50+
data
51+
----------------------------------------------------
52+
BEGIN
53+
table public.test_prepared1: INSERT: id[integer]:3
54+
PREPARE TRANSACTION 'test_prepared#2'
55+
(3 rows)
56+
57+
ROLLBACK PREPARED 'test_prepared#2';
58+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
59+
data
60+
-------------------------------------
61+
ROLLBACK PREPARED 'test_prepared#2'
62+
(1 row)
63+
64+
-- Test prepare of a xact containing ddl. Leaving xact uncommitted for next test.
65+
BEGIN;
66+
ALTER TABLE test_prepared1 ADD COLUMN data text;
67+
INSERT INTO test_prepared1 VALUES (4, 'frakbar');
68+
PREPARE TRANSACTION 'test_prepared#3';
69+
-- confirm that exclusive lock from the ALTER command is held on test_prepared1 table
70+
SELECT 'test_prepared_1' AS relation, locktype, mode
71+
FROM pg_locks
72+
WHERE locktype = 'relation'
73+
AND relation = 'test_prepared1'::regclass;
74+
relation | locktype | mode
75+
-----------------+----------+---------------------
76+
test_prepared_1 | relation | RowExclusiveLock
77+
test_prepared_1 | relation | AccessExclusiveLock
78+
(2 rows)
79+
80+
-- The insert should show the newly altered column but not the DDL.
81+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
82+
data
83+
-------------------------------------------------------------------------
84+
BEGIN
85+
table public.test_prepared1: INSERT: id[integer]:4 data[text]:'frakbar'
86+
PREPARE TRANSACTION 'test_prepared#3'
87+
(3 rows)
88+
89+
-- Test that we decode correctly while an uncommitted prepared xact
90+
-- with ddl exists.
91+
--
92+
-- Use a separate table for the concurrent transaction because the lock from
93+
-- the ALTER will stop us inserting into the other one.
94+
--
95+
INSERT INTO test_prepared2 VALUES (5);
96+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
97+
data
98+
----------------------------------------------------
99+
BEGIN
100+
table public.test_prepared2: INSERT: id[integer]:5
101+
COMMIT
102+
(3 rows)
103+
104+
COMMIT PREPARED 'test_prepared#3';
105+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
106+
data
107+
-------------------------------------------------------------------------
108+
BEGIN
109+
table public.test_prepared1: INSERT: id[integer]:4 data[text]:'frakbar'
110+
PREPARE TRANSACTION 'test_prepared#3'
111+
COMMIT PREPARED 'test_prepared#3'
112+
(4 rows)
113+
114+
-- make sure stuff still works
115+
INSERT INTO test_prepared1 VALUES (6);
116+
INSERT INTO test_prepared2 VALUES (7);
117+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
118+
data
119+
--------------------------------------------------------------------
120+
BEGIN
121+
table public.test_prepared1: INSERT: id[integer]:6 data[text]:null
122+
COMMIT
123+
BEGIN
124+
table public.test_prepared2: INSERT: id[integer]:7
125+
COMMIT
126+
(6 rows)
127+
128+
-- Check 'CLUSTER' (as operation that hold exclusive lock) doesn't block
129+
-- logical decoding.
130+
BEGIN;
131+
INSERT INTO test_prepared1 VALUES (8, 'othercol');
132+
CLUSTER test_prepared1 USING test_prepared1_pkey;
133+
INSERT INTO test_prepared1 VALUES (9, 'othercol2');
134+
PREPARE TRANSACTION 'test_prepared_lock';
135+
SELECT 'test_prepared1' AS relation, locktype, mode
136+
FROM pg_locks
137+
WHERE locktype = 'relation'
138+
AND relation = 'test_prepared1'::regclass;
139+
relation | locktype | mode
140+
----------------+----------+---------------------
141+
test_prepared1 | relation | RowExclusiveLock
142+
test_prepared1 | relation | ShareLock
143+
test_prepared1 | relation | AccessExclusiveLock
144+
(3 rows)
145+
146+
-- The above CLUSTER command shouldn't cause a timeout on 2pc decoding. The
147+
-- call should return within a second.
148+
SET statement_timeout = '1s';
149+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
150+
data
151+
---------------------------------------------------------------------------
152+
BEGIN
153+
table public.test_prepared1: INSERT: id[integer]:8 data[text]:'othercol'
154+
table public.test_prepared1: INSERT: id[integer]:9 data[text]:'othercol2'
155+
PREPARE TRANSACTION 'test_prepared_lock'
156+
(4 rows)
157+
158+
RESET statement_timeout;
159+
COMMIT PREPARED 'test_prepared_lock';
160+
-- consume the commit
161+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
162+
data
163+
---------------------------------------------------------------------------
164+
BEGIN
165+
table public.test_prepared1: INSERT: id[integer]:8 data[text]:'othercol'
166+
table public.test_prepared1: INSERT: id[integer]:9 data[text]:'othercol2'
167+
PREPARE TRANSACTION 'test_prepared_lock'
168+
COMMIT PREPARED 'test_prepared_lock'
169+
(5 rows)
170+
171+
-- Test savepoints and sub-xacts. Creating savepoints will create
172+
-- sub-xacts implicitly.
173+
BEGIN;
174+
CREATE TABLE test_prepared_savepoint (a int);
175+
INSERT INTO test_prepared_savepoint VALUES (1);
176+
SAVEPOINT test_savepoint;
177+
INSERT INTO test_prepared_savepoint VALUES (2);
178+
ROLLBACK TO SAVEPOINT test_savepoint;
179+
PREPARE TRANSACTION 'test_prepared_savepoint';
180+
-- should show only 1, not 2
181+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
182+
data
183+
------------------------------------------------------------
184+
BEGIN
185+
table public.test_prepared_savepoint: INSERT: a[integer]:1
186+
PREPARE TRANSACTION 'test_prepared_savepoint'
187+
(3 rows)
188+
189+
COMMIT PREPARED 'test_prepared_savepoint';
190+
-- consume the commit
191+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
192+
data
193+
------------------------------------------------------------
194+
BEGIN
195+
table public.test_prepared_savepoint: INSERT: a[integer]:1
196+
PREPARE TRANSACTION 'test_prepared_savepoint'
197+
COMMIT PREPARED 'test_prepared_savepoint'
198+
(4 rows)
199+
200+
-- Test that a GID containing "_nodecode" gets decoded at commit prepared time.
201+
BEGIN;
202+
INSERT INTO test_prepared1 VALUES (20);
203+
PREPARE TRANSACTION 'test_prepared_nodecode';
204+
-- should show nothing
205+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
206+
data
207+
------
208+
(0 rows)
209+
210+
COMMIT PREPARED 'test_prepared_nodecode';
211+
-- should be decoded now
212+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
213+
data
214+
---------------------------------------------------------------------
215+
BEGIN
216+
table public.test_prepared1: INSERT: id[integer]:20 data[text]:null
217+
COMMIT
218+
(3 rows)
219+
220+
-- Test 8:
221+
-- cleanup and make sure results are also empty
222+
DROP TABLE test_prepared1;
223+
DROP TABLE test_prepared2;
224+
-- show results. There should be nothing to show
225+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1');
226+
data
227+
------
228+
(0 rows)
229+
230+
SELECT pg_drop_replication_slot('regression_slot');
231+
pg_drop_replication_slot
232+
--------------------------
233+
234+
(1 row)
235+

‎contrib/test_decoding/expected/twophase_stream.out

+147
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
-- Test streaming of two-phase commits
2+
SET synchronous_commit = on;
3+
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
4+
?column?
5+
----------
6+
init
7+
(1 row)
8+
9+
CREATE TABLE stream_test(data text);
10+
-- consume DDL
11+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
12+
data
13+
------
14+
(0 rows)
15+
16+
-- streaming test with sub-transaction and PREPARE/COMMIT PREPARED
17+
BEGIN;
18+
SAVEPOINT s1;
19+
SELECT 'msg5' FROM pg_logical_emit_message(true, 'test', repeat('a', 50));
20+
?column?
21+
----------
22+
msg5
23+
(1 row)
24+
25+
INSERT INTO stream_test SELECT repeat('a', 2000) || g.i FROM generate_series(1, 35) g(i);
26+
TRUNCATE table stream_test;
27+
ROLLBACK TO s1;
28+
INSERT INTO stream_test SELECT repeat('a', 10) || g.i FROM generate_series(1, 20) g(i);
29+
PREPARE TRANSACTION 'test1';
30+
-- should show the inserts after a ROLLBACK
31+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
32+
data
33+
----------------------------------------------------------
34+
streaming message: transactional: 1 prefix: test, sz: 50
35+
opening a streamed block for transaction
36+
streaming change for transaction
37+
streaming change for transaction
38+
streaming change for transaction
39+
streaming change for transaction
40+
streaming change for transaction
41+
streaming change for transaction
42+
streaming change for transaction
43+
streaming change for transaction
44+
streaming change for transaction
45+
streaming change for transaction
46+
streaming change for transaction
47+
streaming change for transaction
48+
streaming change for transaction
49+
streaming change for transaction
50+
streaming change for transaction
51+
streaming change for transaction
52+
streaming change for transaction
53+
streaming change for transaction
54+
streaming change for transaction
55+
streaming change for transaction
56+
closing a streamed block for transaction
57+
preparing streamed transaction 'test1'
58+
(24 rows)
59+
60+
COMMIT PREPARED 'test1';
61+
--should show the COMMIT PREPARED and the other changes in the transaction
62+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
63+
data
64+
-------------------------------------------------------------
65+
BEGIN
66+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa1'
67+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa2'
68+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa3'
69+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa4'
70+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa5'
71+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa6'
72+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa7'
73+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa8'
74+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa9'
75+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa10'
76+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa11'
77+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa12'
78+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa13'
79+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa14'
80+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa15'
81+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa16'
82+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa17'
83+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa18'
84+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa19'
85+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa20'
86+
PREPARE TRANSACTION 'test1'
87+
COMMIT PREPARED 'test1'
88+
(23 rows)
89+
90+
-- streaming test with sub-transaction and PREPARE/COMMIT PREPARED but with
91+
-- filtered gid. gids with '_nodecode' will not be decoded at prepare time.
92+
BEGIN;
93+
SAVEPOINT s1;
94+
SELECT 'msg5' FROM pg_logical_emit_message(true, 'test', repeat('a', 50));
95+
?column?
96+
----------
97+
msg5
98+
(1 row)
99+
100+
INSERT INTO stream_test SELECT repeat('a', 2000) || g.i FROM generate_series(1, 35) g(i);
101+
TRUNCATE table stream_test;
102+
ROLLBACK to s1;
103+
INSERT INTO stream_test SELECT repeat('a', 10) || g.i FROM generate_series(1, 20) g(i);
104+
PREPARE TRANSACTION 'test1_nodecode';
105+
-- should NOT show inserts after a ROLLBACK
106+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
107+
data
108+
----------------------------------------------------------
109+
streaming message: transactional: 1 prefix: test, sz: 50
110+
(1 row)
111+
112+
COMMIT PREPARED 'test1_nodecode';
113+
-- should show the inserts but not show a COMMIT PREPARED but a COMMIT
114+
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'two-phase-commit', '1', 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
115+
data
116+
-------------------------------------------------------------
117+
BEGIN
118+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa1'
119+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa2'
120+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa3'
121+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa4'
122+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa5'
123+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa6'
124+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa7'
125+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa8'
126+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa9'
127+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa10'
128+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa11'
129+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa12'
130+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa13'
131+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa14'
132+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa15'
133+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa16'
134+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa17'
135+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa18'
136+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa19'
137+
table public.stream_test: INSERT: data[text]:'aaaaaaaaaa20'
138+
COMMIT
139+
(22 rows)
140+
141+
DROP TABLE stream_test;
142+
SELECT pg_drop_replication_slot('regression_slot');
143+
pg_drop_replication_slot
144+
--------------------------
145+
146+
(1 row)
147+

0 commit comments

Comments
 (0)