mirror of
https://git.postgresql.org/git/postgresql.git
synced 2025-02-11 19:20:40 +08:00
19890a064e
Commit 0aa8a01d04 extends the output plugin API to allow decoding of
prepared xacts and allowed the user to enable/disable the two-phase option
via pg_logical_slot_get_changes(). This can lead to a problem such that
the first time when it gets changes via pg_logical_slot_get_changes()
without two_phase option enabled it will not get the prepared even though
prepare is after consistent snapshot. Now next time during getting changes,
if the two_phase option is enabled it can skip prepare because by that
time start decoding point has been moved. So the user will only get commit
prepared.
Allow to enable/disable this option at the create slot time and default
will be false. It will break the existing slots which is fine in a major
release.
Author: Ajin Cherian
Reviewed-by: Amit Kapila and Vignesh C
Discussion: https://postgr.es/m/d0f60d60-133d-bf8d-bd70-47784d8fabf3@enterprisedb.com
221 lines
8.6 KiB
Plaintext
221 lines
8.6 KiB
Plaintext
-- Test prepared transactions. When two-phase-commit is enabled, transactions are
|
|
-- decoded at PREPARE time rather than at COMMIT PREPARED time.
|
|
SET synchronous_commit = on;
|
|
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding', false, true);
|
|
?column?
|
|
----------
|
|
init
|
|
(1 row)
|
|
|
|
CREATE TABLE test_prepared1(id integer primary key);
|
|
CREATE TABLE test_prepared2(id integer primary key);
|
|
-- Test that decoding happens at PREPARE time when two-phase-commit is enabled.
|
|
-- Decoding after COMMIT PREPARED must have all the commands in the transaction.
|
|
BEGIN;
|
|
INSERT INTO test_prepared1 VALUES (1);
|
|
INSERT INTO test_prepared1 VALUES (2);
|
|
-- should show nothing because the xact has not been prepared yet.
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
------
|
|
(0 rows)
|
|
|
|
PREPARE TRANSACTION 'test_prepared#1';
|
|
-- should show both the above inserts and the PREPARE TRANSACTION.
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
----------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared1: INSERT: id[integer]:1
|
|
table public.test_prepared1: INSERT: id[integer]:2
|
|
PREPARE TRANSACTION 'test_prepared#1'
|
|
(4 rows)
|
|
|
|
COMMIT PREPARED 'test_prepared#1';
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
-----------------------------------
|
|
COMMIT PREPARED 'test_prepared#1'
|
|
(1 row)
|
|
|
|
-- Test that rollback of a prepared xact is decoded.
|
|
BEGIN;
|
|
INSERT INTO test_prepared1 VALUES (3);
|
|
PREPARE TRANSACTION 'test_prepared#2';
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
----------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared1: INSERT: id[integer]:3
|
|
PREPARE TRANSACTION 'test_prepared#2'
|
|
(3 rows)
|
|
|
|
ROLLBACK PREPARED 'test_prepared#2';
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
-------------------------------------
|
|
ROLLBACK PREPARED 'test_prepared#2'
|
|
(1 row)
|
|
|
|
-- Test prepare of a xact containing ddl. Leaving xact uncommitted for next test.
|
|
BEGIN;
|
|
ALTER TABLE test_prepared1 ADD COLUMN data text;
|
|
INSERT INTO test_prepared1 VALUES (4, 'frakbar');
|
|
PREPARE TRANSACTION 'test_prepared#3';
|
|
-- confirm that exclusive lock from the ALTER command is held on test_prepared1 table
|
|
SELECT 'test_prepared_1' AS relation, locktype, mode
|
|
FROM pg_locks
|
|
WHERE locktype = 'relation'
|
|
AND relation = 'test_prepared1'::regclass;
|
|
relation | locktype | mode
|
|
-----------------+----------+---------------------
|
|
test_prepared_1 | relation | RowExclusiveLock
|
|
test_prepared_1 | relation | AccessExclusiveLock
|
|
(2 rows)
|
|
|
|
-- The insert should show the newly altered column but not the DDL.
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
-------------------------------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared1: INSERT: id[integer]:4 data[text]:'frakbar'
|
|
PREPARE TRANSACTION 'test_prepared#3'
|
|
(3 rows)
|
|
|
|
-- Test that we decode correctly while an uncommitted prepared xact
|
|
-- with ddl exists.
|
|
--
|
|
-- Use a separate table for the concurrent transaction because the lock from
|
|
-- the ALTER will stop us inserting into the other one.
|
|
--
|
|
INSERT INTO test_prepared2 VALUES (5);
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
----------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared2: INSERT: id[integer]:5
|
|
COMMIT
|
|
(3 rows)
|
|
|
|
COMMIT PREPARED 'test_prepared#3';
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
-----------------------------------
|
|
COMMIT PREPARED 'test_prepared#3'
|
|
(1 row)
|
|
|
|
-- make sure stuff still works
|
|
INSERT INTO test_prepared1 VALUES (6);
|
|
INSERT INTO test_prepared2 VALUES (7);
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
--------------------------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared1: INSERT: id[integer]:6 data[text]:null
|
|
COMMIT
|
|
BEGIN
|
|
table public.test_prepared2: INSERT: id[integer]:7
|
|
COMMIT
|
|
(6 rows)
|
|
|
|
-- Check 'CLUSTER' (as operation that hold exclusive lock) doesn't block
|
|
-- logical decoding.
|
|
BEGIN;
|
|
INSERT INTO test_prepared1 VALUES (8, 'othercol');
|
|
CLUSTER test_prepared1 USING test_prepared1_pkey;
|
|
INSERT INTO test_prepared1 VALUES (9, 'othercol2');
|
|
PREPARE TRANSACTION 'test_prepared_lock';
|
|
SELECT 'test_prepared1' AS relation, locktype, mode
|
|
FROM pg_locks
|
|
WHERE locktype = 'relation'
|
|
AND relation = 'test_prepared1'::regclass;
|
|
relation | locktype | mode
|
|
----------------+----------+---------------------
|
|
test_prepared1 | relation | RowExclusiveLock
|
|
test_prepared1 | relation | ShareLock
|
|
test_prepared1 | relation | AccessExclusiveLock
|
|
(3 rows)
|
|
|
|
-- The above CLUSTER command shouldn't cause a timeout on 2pc decoding.
|
|
SET statement_timeout = '180s';
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
---------------------------------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared1: INSERT: id[integer]:8 data[text]:'othercol'
|
|
table public.test_prepared1: INSERT: id[integer]:9 data[text]:'othercol2'
|
|
PREPARE TRANSACTION 'test_prepared_lock'
|
|
(4 rows)
|
|
|
|
RESET statement_timeout;
|
|
COMMIT PREPARED 'test_prepared_lock';
|
|
-- consume the commit
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
--------------------------------------
|
|
COMMIT PREPARED 'test_prepared_lock'
|
|
(1 row)
|
|
|
|
-- Test savepoints and sub-xacts. Creating savepoints will create
|
|
-- sub-xacts implicitly.
|
|
BEGIN;
|
|
CREATE TABLE test_prepared_savepoint (a int);
|
|
INSERT INTO test_prepared_savepoint VALUES (1);
|
|
SAVEPOINT test_savepoint;
|
|
INSERT INTO test_prepared_savepoint VALUES (2);
|
|
ROLLBACK TO SAVEPOINT test_savepoint;
|
|
PREPARE TRANSACTION 'test_prepared_savepoint';
|
|
-- should show only 1, not 2
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
------------------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared_savepoint: INSERT: a[integer]:1
|
|
PREPARE TRANSACTION 'test_prepared_savepoint'
|
|
(3 rows)
|
|
|
|
COMMIT PREPARED 'test_prepared_savepoint';
|
|
-- consume the commit
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
-------------------------------------------
|
|
COMMIT PREPARED 'test_prepared_savepoint'
|
|
(1 row)
|
|
|
|
-- Test that a GID containing "_nodecode" gets decoded at commit prepared time.
|
|
BEGIN;
|
|
INSERT INTO test_prepared1 VALUES (20);
|
|
PREPARE TRANSACTION 'test_prepared_nodecode';
|
|
-- should show nothing
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
------
|
|
(0 rows)
|
|
|
|
COMMIT PREPARED 'test_prepared_nodecode';
|
|
-- should be decoded now
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
---------------------------------------------------------------------
|
|
BEGIN
|
|
table public.test_prepared1: INSERT: id[integer]:20 data[text]:null
|
|
COMMIT
|
|
(3 rows)
|
|
|
|
-- Test 8:
|
|
-- cleanup and make sure results are also empty
|
|
DROP TABLE test_prepared1;
|
|
DROP TABLE test_prepared2;
|
|
-- show results. There should be nothing to show
|
|
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
|
|
data
|
|
------
|
|
(0 rows)
|
|
|
|
SELECT pg_drop_replication_slot('regression_slot');
|
|
pg_drop_replication_slot
|
|
--------------------------
|
|
|
|
(1 row)
|
|
|