From f9c1e899cbcec536fe23dffc7b1010503841169c Mon Sep 17 00:00:00 2001
From: Ajin Cherian <ajinc@fast.au.fujitsu.com>
Date: Wed, 16 Dec 2020 23:12:58 -0500
Subject: [PATCH] Support 2PC consistent snapshot isolation tests

Added isolation test-case to test that if a consistent snapshot is created
between a PREPARE and a COMMIT PREPARED, then the whole transaction is decoded
on COMMIT PREPARED.
---
 contrib/test_decoding/Makefile                     |  2 +-
 .../test_decoding/expected/twophase_snapshot.out   | 43 +++++++++++++++++++
 contrib/test_decoding/specs/twophase_snapshot.spec | 49 ++++++++++++++++++++++
 3 files changed, 93 insertions(+), 1 deletion(-)
 create mode 100644 contrib/test_decoding/expected/twophase_snapshot.out
 create mode 100644 contrib/test_decoding/specs/twophase_snapshot.spec

diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile
index 49523fe..ba6a2c2 100644
--- a/contrib/test_decoding/Makefile
+++ b/contrib/test_decoding/Makefile
@@ -7,7 +7,7 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \
 	decoding_into_rel binary prepared replorigin time two_phase two_phase_stream messages \
 	spill slot truncate stream stats
 ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \
-	oldest_xmin snapshot_transfer subxact_without_top concurrent_stream
+	oldest_xmin snapshot_transfer subxact_without_top concurrent_stream twophase_snapshot
 
 TAP_TESTS = 1
 
diff --git a/contrib/test_decoding/expected/twophase_snapshot.out b/contrib/test_decoding/expected/twophase_snapshot.out
new file mode 100644
index 0000000..53aaf01
--- /dev/null
+++ b/contrib/test_decoding/expected/twophase_snapshot.out
@@ -0,0 +1,43 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s2b s2txid s1init s3b s3txid s2alter s2c s4b s4insert s4prepare s3c s1insert s1checkpoint s1start s4commit s1start
+step s2b: BEGIN;
+step s2txid: SELECT pg_current_xact_id() IS NULL;
+?column?       
+
+f              
+step s1init: SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding'); <waiting ...>
+step s3b: BEGIN;
+step s3txid: SELECT pg_current_xact_id() IS NULL;
+?column?       
+
+f              
+step s2alter: ALTER TABLE do_write ADD COLUMN addedbys2 int;
+step s2c: COMMIT;
+step s4b: BEGIN;
+step s4insert: INSERT INTO do_write DEFAULT VALUES;
+step s4prepare: PREPARE TRANSACTION 'test1';
+step s3c: COMMIT;
+step s1init: <... completed>
+?column?       
+
+init           
+step s1insert: INSERT INTO do_write DEFAULT VALUES;
+step s1checkpoint: CHECKPOINT;
+step s1start: SELECT data  FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'two-phase-commit', '1');
+data           
+
+BEGIN          
+table public.do_write: INSERT: id[integer]:2 addedbys2[integer]:null
+COMMIT         
+step s4commit: COMMIT PREPARED 'test1';
+step s1start: SELECT data  FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'two-phase-commit', '1');
+data           
+
+BEGIN          
+table public.do_write: INSERT: id[integer]:1 addedbys2[integer]:null
+PREPARE TRANSACTION 'test1'
+COMMIT PREPARED 'test1'
+?column?       
+
+stop           
diff --git a/contrib/test_decoding/specs/twophase_snapshot.spec b/contrib/test_decoding/specs/twophase_snapshot.spec
new file mode 100644
index 0000000..505e5e3
--- /dev/null
+++ b/contrib/test_decoding/specs/twophase_snapshot.spec
@@ -0,0 +1,49 @@
+# Test decoding of two-phase transactions during the build of a consistent snapshot.
+setup
+{
+    DROP TABLE IF EXISTS do_write;
+    CREATE TABLE do_write(id serial primary key);
+}
+
+teardown
+{
+    DROP TABLE do_write;
+    SELECT 'stop' FROM pg_drop_replication_slot('isolation_slot');
+}
+
+
+session "s1"
+setup { SET synchronous_commit=on; }
+
+step "s1init" {SELECT 'init' FROM pg_create_logical_replication_slot('isolation_slot', 'test_decoding');}
+step "s1start" {SELECT data  FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', 'false', 'two-phase-commit', '1');}
+step "s1insert" { INSERT INTO do_write DEFAULT VALUES; }
+step "s1checkpoint" { CHECKPOINT; }
+
+session "s2"
+setup { SET synchronous_commit=on; }
+
+step "s2b" { BEGIN; }
+step "s2txid" { SELECT pg_current_xact_id() IS NULL; }
+step "s2alter" { ALTER TABLE do_write ADD COLUMN addedbys2 int; }
+step "s2c" { COMMIT; }
+
+
+session "s3"
+setup { SET synchronous_commit=on; }
+
+step "s3b" { BEGIN; }
+step "s3txid" { SELECT pg_current_xact_id() IS NULL; }
+step "s3c" { COMMIT; }
+
+session "s4"
+setup { SET synchronous_commit=on; }
+
+step "s4b" { BEGIN; }
+step "s4insert" { INSERT INTO do_write DEFAULT VALUES; }
+step "s4prepare" { PREPARE TRANSACTION 'test1'; }
+step "s4commit" { COMMIT PREPARED 'test1'; }
+
+# Force building of a consistent snapshot between a PREPARE and COMMIT PREPARED.
+# Ensure that the whole transaction is decoded fresh at the time of COMMIT PREPARED.
+permutation "s2b" "s2txid" "s1init" "s3b" "s3txid" "s2alter" "s2c" "s4b" "s4insert" "s4prepare" "s3c""s1insert" "s1checkpoint" "s1start" "s4commit" "s1start"
-- 
1.8.3.1