From: Markus Wanner <markus.wanner@enterprisedb.com>
Date: Fri, 19 Feb 2021 13:34:22 +0100
Subject: Present all committed transactions to the output plugin

Drop the filtering of some empty transactions and more
consistently present all committed transactions to the output
plugin.  Changes behavior for empty transactions that do not even
have a base_snapshot.

While clearly not making any substantial changes to the database,
an empty transactions still increments the local TransactionId
and possibly carries a gid (in the case of two-phase
transactions).

Adjust the test_decoder to show the difference between empty
transactions that do not carry a snapshot and those that
do. Adjust tests to cover decoding of committed empty
transactions with and without snapshots.

Add a minor clarification in the docs.
---
 .../replication/logical/reorderbuffer.c       | 41 ++++++++-----------
 doc/src/sgml/logicaldecoding.sgml             | 10 +++--
 contrib/test_decoding/expected/mxact.out      |  2 +-
 .../test_decoding/expected/ondisk_startup.out |  2 +-
 contrib/test_decoding/expected/prepared.out   | 33 +++++++++++++--
 contrib/test_decoding/expected/xact.out       | 39 ++++++++++++++++++
 contrib/test_decoding/sql/prepared.sql        | 23 +++++++++--
 contrib/test_decoding/sql/xact.sql            | 21 ++++++++++
 contrib/test_decoding/test_decoding.c         | 12 ++++++
 9 files changed, 147 insertions(+), 36 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 5a62ab8bbc1..1e9ed81c3c1 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2016,7 +2016,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	ReorderBufferBuildTupleCidHash(rb, txn);
 
 	/* setup the initial snapshot */
-	SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
+	if (snapshot_now)
+		SetupHistoricSnapshot(snapshot_now, txn->tuplecid_hash);
 
 	/*
 	 * Decoding needs access to syscaches et al., which in turn use
@@ -2289,6 +2290,8 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 					break;
 
 				case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
+					Assert(snapshot_now);
+
 					/* get rid of the old */
 					TeardownHistoricSnapshot(false);
 
@@ -2321,6 +2324,7 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 					break;
 
 				case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
+					Assert(snapshot_now);
 					Assert(change->data.command_id != InvalidCommandId);
 
 					if (command_id < change->data.command_id)
@@ -2397,8 +2401,11 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		 * streaming mode.
 		 */
 		if (streaming)
+		{
+			Assert(snapshot_now);
 			ReorderBufferSaveTXNSnapshot(rb, txn, snapshot_now, command_id);
-		else if (snapshot_now->copied)
+		}
+		else if (snapshot_now && snapshot_now->copied)
 			ReorderBufferFreeSnap(rb, snapshot_now);
 
 		/* cleanup */
@@ -2520,7 +2527,6 @@ ReorderBufferReplay(ReorderBufferTXN *txn,
 					TimestampTz commit_time,
 					RepOriginId origin_id, XLogRecPtr origin_lsn)
 {
-	Snapshot	snapshot_now;
 	CommandId	command_id = FirstCommandId;
 
 	txn->final_lsn = commit_lsn;
@@ -2544,28 +2550,15 @@ ReorderBufferReplay(ReorderBufferTXN *txn,
 	}
 
 	/*
-	 * If this transaction has no snapshot, it didn't make any changes to the
-	 * database, so there's nothing to decode.  Note that
-	 * ReorderBufferCommitChild will have transferred any snapshots from
-	 * subtransactions if there were any.
+	 * Process and send the changes to output plugin.
+	 *
+	 * Note that for empty transactions, txn->base_snapshot may well be
+	 * NULL. The corresponding callbacks will still be invoked, as even an
+	 * empty transaction carries information (LSN increments, the gid in
+	 * case of a two-phase transaction).  This is unlike versions prior to
+	 * 14 which optimized away transactions without a snapshot.
 	 */
-	if (txn->base_snapshot == NULL)
-	{
-		Assert(txn->ninvalidations == 0);
-
-		/*
-		 * Removing this txn before a commit might result in the computation
-		 * of an incorrect restart_lsn. See SnapBuildProcessRunningXacts.
-		 */
-		if (!rbtxn_prepared(txn))
-			ReorderBufferCleanupTXN(rb, txn);
-		return;
-	}
-
-	snapshot_now = txn->base_snapshot;
-
-	/* Process and send the changes to output plugin. */
-	ReorderBufferProcessTXN(rb, txn, commit_lsn, snapshot_now,
+	ReorderBufferProcessTXN(rb, txn, commit_lsn, txn->base_snapshot,
 							command_id, false);
 }
 
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index cf705ed9cda..832acee8270 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -540,9 +540,8 @@ CREATE TABLE another_catalog_table(data text) WITH (user_catalog_table = true);
      Concurrent transactions are decoded in commit order, and only changes
      belonging to a specific transaction are decoded between
      the <literal>begin</literal> and <literal>commit</literal>
-     callbacks. Transactions that were rolled back explicitly or implicitly
-     never get
-     decoded. Successful savepoints are
+     callbacks. All transactions are presented to the output plugin, whether
+     they made changes or not. Successful savepoints are
      folded into the transaction containing them in the order they were
      executed within that transaction. A transaction that is prepared for
      a two-phase commit using <command>PREPARE TRANSACTION</command> will
@@ -632,7 +631,10 @@ typedef void (*LogicalDecodeBeginCB) (struct LogicalDecodingContext *ctx,
 </programlisting>
       The <parameter>txn</parameter> parameter contains meta information about
       the transaction, like the time stamp at which it has been committed and
-      its XID.
+      its XID. A transaction without a snapshot in
+      <literal>txn-&gt;base_snapshot</literal> will certainly not contain any
+      changes.  Presence of a snapshot does not guarantee that a change will
+      follow prior to the commit, though.
      </para>
     </sect3>
 
diff --git a/contrib/test_decoding/expected/mxact.out b/contrib/test_decoding/expected/mxact.out
index f0d96cc67d0..9a3e3f21bc7 100644
--- a/contrib/test_decoding/expected/mxact.out
+++ b/contrib/test_decoding/expected/mxact.out
@@ -55,7 +55,7 @@ step s0start: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NUL
 data           
 
 BEGIN          
-COMMIT         
+COMMIT (empty with snapshot)
 BEGIN          
 table public.do_write: INSERT: id[integer]:1 ts[timestamp with time zone]:null
 COMMIT         
diff --git a/contrib/test_decoding/expected/ondisk_startup.out b/contrib/test_decoding/expected/ondisk_startup.out
index 586b03d75db..d390d90fd26 100644
--- a/contrib/test_decoding/expected/ondisk_startup.out
+++ b/contrib/test_decoding/expected/ondisk_startup.out
@@ -43,7 +43,7 @@ BEGIN
 table public.do_write: INSERT: id[integer]:2 addedbys2[integer]:null
 COMMIT         
 BEGIN          
-COMMIT         
+COMMIT (empty with snapshot)
 BEGIN          
 table public.do_write: INSERT: id[integer]:3 addedbys2[integer]:null addedbys1[integer]:null
 COMMIT         
diff --git a/contrib/test_decoding/expected/prepared.out b/contrib/test_decoding/expected/prepared.out
index 46e915d4ffa..6656d132069 100644
--- a/contrib/test_decoding/expected/prepared.out
+++ b/contrib/test_decoding/expected/prepared.out
@@ -35,9 +35,6 @@ COMMIT PREPARED 'test_prepared#3';
 -- make sure stuff still works
 INSERT INTO test_prepared1 VALUES (8);
 INSERT INTO test_prepared2 VALUES (9);
--- cleanup
-DROP TABLE test_prepared1;
-DROP TABLE test_prepared2;
 -- show results
 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
                                   data                                   
@@ -66,6 +63,36 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc
  COMMIT
 (22 rows)
 
+-- test decoding of an empty two-phase transaction
+BEGIN;
+SELECT 1 FROM test_prepared1 FOR UPDATE LIMIT 1;
+ ?column? 
+----------
+        1
+(1 row)
+
+PREPARE TRANSACTION 'test_prepared#4';
+COMMIT PREPARED 'test_prepared#4';
+-- test decoding of an empty transaction for which the reorderbuffer
+-- had to create a snapshot
+BEGIN;
+SAVEPOINT barf;
+INSERT INTO test_prepared1 VALUES (10);
+ROLLBACK TO SAVEPOINT barf;
+PREPARE TRANSACTION 'test_prepared#5';
+COMMIT PREPARED 'test_prepared#5';
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+             data             
+------------------------------
+ BEGIN (w/o snapshot)
+ COMMIT (empty w/o snapshot)
+ BEGIN
+ COMMIT (empty with snapshot)
+(4 rows)
+
+-- cleanup
+DROP TABLE test_prepared1;
+DROP TABLE test_prepared2;
 SELECT pg_drop_replication_slot('regression_slot');
  pg_drop_replication_slot 
 --------------------------
diff --git a/contrib/test_decoding/expected/xact.out b/contrib/test_decoding/expected/xact.out
index ec4745005d7..3b50f8d073b 100644
--- a/contrib/test_decoding/expected/xact.out
+++ b/contrib/test_decoding/expected/xact.out
@@ -55,6 +55,45 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc
  COMMIT
 (3 rows)
 
+-- test decoding of an aborted transaction
+BEGIN;
+INSERT INTO xact_test VALUES ('aborted-txn');
+ROLLBACK;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+ data 
+------
+(0 rows)
+
+-- test decoding of an empty but not aborted transaction
+BEGIN;
+SELECT 1 FROM xact_test FOR UPDATE LIMIT 1;
+ ?column? 
+----------
+        1
+(1 row)
+
+COMMIT;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+            data             
+-----------------------------
+ BEGIN (w/o snapshot)
+ COMMIT (empty w/o snapshot)
+(2 rows)
+
+-- test decoding of an empty transaction for which the reorderbuffer
+-- had to create a snapshot
+BEGIN;
+SAVEPOINT barf;
+INSERT INTO xact_test VALUES ('change-to-roll-back');
+ROLLBACK TO SAVEPOINT barf;
+COMMIT;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+             data             
+------------------------------
+ BEGIN
+ COMMIT (empty with snapshot)
+(2 rows)
+
 DROP TABLE xact_test;
 SELECT pg_drop_replication_slot('regression_slot');
  pg_drop_replication_slot 
diff --git a/contrib/test_decoding/sql/prepared.sql b/contrib/test_decoding/sql/prepared.sql
index e72639767ee..dcb18ecc80d 100644
--- a/contrib/test_decoding/sql/prepared.sql
+++ b/contrib/test_decoding/sql/prepared.sql
@@ -40,11 +40,28 @@ COMMIT PREPARED 'test_prepared#3';
 INSERT INTO test_prepared1 VALUES (8);
 INSERT INTO test_prepared2 VALUES (9);
 
+-- show results
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+
+-- test decoding of an empty two-phase transaction
+BEGIN;
+SELECT 1 FROM test_prepared1 FOR UPDATE LIMIT 1;
+PREPARE TRANSACTION 'test_prepared#4';
+COMMIT PREPARED 'test_prepared#4';
+
+-- test decoding of an empty transaction for which the reorderbuffer
+-- had to create a snapshot
+BEGIN;
+SAVEPOINT barf;
+INSERT INTO test_prepared1 VALUES (10);
+ROLLBACK TO SAVEPOINT barf;
+PREPARE TRANSACTION 'test_prepared#5';
+COMMIT PREPARED 'test_prepared#5';
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+
 -- cleanup
 DROP TABLE test_prepared1;
 DROP TABLE test_prepared2;
 
--- show results
-SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
-
 SELECT pg_drop_replication_slot('regression_slot');
diff --git a/contrib/test_decoding/sql/xact.sql b/contrib/test_decoding/sql/xact.sql
index aa555911e86..419b26f9b38 100644
--- a/contrib/test_decoding/sql/xact.sql
+++ b/contrib/test_decoding/sql/xact.sql
@@ -28,6 +28,27 @@ COMMIT;
 -- and now show those changes
 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
 
+-- test decoding of an aborted transaction
+BEGIN;
+INSERT INTO xact_test VALUES ('aborted-txn');
+ROLLBACK;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+
+-- test decoding of an empty but not aborted transaction
+BEGIN;
+SELECT 1 FROM xact_test FOR UPDATE LIMIT 1;
+COMMIT;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+
+-- test decoding of an empty transaction for which the reorderbuffer
+-- had to create a snapshot
+BEGIN;
+SAVEPOINT barf;
+INSERT INTO xact_test VALUES ('change-to-roll-back');
+ROLLBACK TO SAVEPOINT barf;
+COMMIT;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '0');
+
 DROP TABLE xact_test;
 
 SELECT pg_drop_replication_slot('regression_slot');
diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index 929255eac74..c9dabaf60ae 100644
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -324,6 +324,10 @@ pg_output_begin(LogicalDecodingContext *ctx, TestDecodingData *data, ReorderBuff
 		appendStringInfo(ctx->out, "BEGIN %u", txn->xid);
 	else
 		appendStringInfoString(ctx->out, "BEGIN");
+
+	if (txn->base_snapshot == NULL)
+		appendStringInfo(ctx->out, " (w/o snapshot)");
+
 	OutputPluginWrite(ctx, last_write);
 }
 
@@ -348,6 +352,14 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	else
 		appendStringInfoString(ctx->out, "COMMIT");
 
+	if (!xact_wrote_changes)
+	{
+		if (txn->base_snapshot == NULL)
+			appendStringInfo(ctx->out, " (empty w/o snapshot)");
+		else
+			appendStringInfo(ctx->out, " (empty with snapshot)");
+	}
+
 	if (data->include_timestamp)
 		appendStringInfo(ctx->out, " (at %s)",
 						 timestamptz_to_str(txn->commit_time));
-- 
2.30.0

