0008-try-adding-fake-transaction-with-sequence-c-20230730.patch

From 594c4dcb609434fa417a0871212400bede4af44a Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@2ndquadrant.com>
Date: Sat, 29 Jul 2023 23:45:35 +0200
Subject: [PATCH 8/8] try adding fake transaction with sequence changes

---
 .../replication/logical/reorderbuffer.c       | 194 ++++++++++--------
 src/include/replication/reorderbuffer.h       |   5 +
 2 files changed, 111 insertions(+), 88 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 83ac3a6a4c..e68da32a92 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -411,6 +411,7 @@ ReorderBufferAllocate(void)
 
 	dlist_init(&buffer->toplevel_by_lsn);
 	dlist_init(&buffer->txns_by_base_snapshot_lsn);
+	dlist_init(&buffer->sequence_changes);
 	dclist_init(&buffer->catchange_txns);
 
 	/*
@@ -1231,99 +1232,25 @@ ReorderBufferQueueSequence(ReorderBuffer *rb, TransactionId xid,
 	}
 	else
 	{
-		/*
-		 * This increment is for a sequence that was not created in any
-		 * running transaction, so we treat it as non-transactional and
-		 * just send it to the output plugin directly.
-		 */
-		ReorderBufferTXN *txn = NULL;
-		volatile Snapshot snapshot_now = snapshot;
-		bool	using_subtxn;
-
-		/* non-transactional changes require a valid snapshot */
-		Assert(snapshot_now);
-
-		/* Make sure the sequence is not in any of the hash tables */
-		Assert(!TransactionIdIsValid(ReorderBufferSequenceGetXid(rb, rlocator)));
-
-		if (xid != InvalidTransactionId)
-			txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
-
-		/* setup snapshot to allow catalog access */
-		SetupHistoricSnapshot(snapshot_now, NULL);
-
-		/*
-		 * Decoding needs access to syscaches et al., which in turn use
-		 * heavyweight locks and such. Thus we need to have enough state around to
-		 * keep track of those.  The easiest way is to simply use a transaction
-		 * internally.  That also allows us to easily enforce that nothing writes
-		 * to the database by checking for xid assignments.
-		 *
-		 * When we're called via the SQL SRF there's already a transaction
-		 * started, so start an explicit subtransaction there.
-		 */
-		using_subtxn = IsTransactionOrTransactionBlock();
-
-		PG_TRY();
-		{
-			Relation	relation;
-			HeapTuple	tuple;
-			Form_pg_sequence_data seq;
-			Oid			reloid;
-			int64		value;
-
-			if (using_subtxn)
-				BeginInternalSubTransaction("sequence");
-			else
-				StartTransactionCommand();
-
-			reloid = RelidByRelfilenumber(rlocator.spcOid, rlocator.relNumber);
-
-			if (reloid == InvalidOid)
-				elog(ERROR, "could not map filenode \"%s\" to relation OID",
-					 relpathperm(rlocator,
-								 MAIN_FORKNUM));
-
-			relation = RelationIdGetRelation(reloid);
-
-			if (!RelationIsValid(relation))
-				elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
-					 reloid,
-					 relpathperm(rlocator, MAIN_FORKNUM));
-
-			tuple = &tuplebuf->tuple;
-			seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
-
-			/*
-			 * Calculate the first value of the next batch (at which point we
-			 * generate and decode another WAL record.
-			 */
-			value = seq->last_value;
-			value += (seq->is_called) ? seq->log_cnt : 0;
-
-			rb->sequence(rb, txn, lsn, relation, transactional, value);
+		MemoryContext oldcontext;
+		ReorderBufferChange *change;
 
-			RelationClose(relation);
+		/* allocate and queue the transactional sequence change */
+		oldcontext = MemoryContextSwitchTo(rb->context);
 
-			TeardownHistoricSnapshot(false);
+		change = ReorderBufferGetChange(rb);
 
-			AbortCurrentTransaction();
+		change->action = REORDER_BUFFER_CHANGE_SEQUENCE;
+		change->origin_id = origin_id;
+		change->lsn = lsn;
 
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
-		}
-		PG_CATCH();
-		{
-			TeardownHistoricSnapshot(true);
+		memcpy(&change->data.sequence.locator, &rlocator, sizeof(RelFileLocator));
 
-			AbortCurrentTransaction();
+		change->data.sequence.tuple = tuplebuf;
 
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
+		dlist_push_tail(&rb->sequence_changes, &change->node);
 
-			PG_RE_THROW();
-		}
-		PG_END_TRY();
+		MemoryContextSwitchTo(oldcontext);
 	}
 }
 
@@ -2209,8 +2136,11 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 		dclist_delete_from(&rb->catchange_txns, &txn->catchange_node);
 
 	/* now remove reference from buffer */
-	hash_search(rb->by_txn, &txn->xid, HASH_REMOVE, &found);
-	Assert(found);
+	if (TransactionIdIsValid(txn->xid))	/* XXX fake sequence TXN has invalid XID */
+	{
+		hash_search(rb->by_txn, &txn->xid, HASH_REMOVE, &found);
+		Assert(found);
+	}
 
 	/* remove entries spilled to disk */
 	if (rbtxn_is_serialized(txn))
@@ -3372,6 +3302,81 @@ ReorderBufferReplay(ReorderBufferTXN *txn,
 							command_id, false);
 }
 
+/*
+ * Create a "fake" substransaction injecting the non-transactional sequence
+ * changes that happened until this commit.
+ *
+ * XXX This has issues with snapshots. Firstly, the "parent" transaction may
+ * be empty withoug a base_snapshot, so that ReorderBufferReplay() will treat
+ * is as empty transaction - and won't apply even the sequence change.
+ *
+ * Secondly, the sequence changes originate from other transactions, so it's
+ * not clear if we can simply transfer snapshots in some way. Maybe we can?
+ * sequence_decode() knows the *current* snapshot, but can we keep it for
+ * later, somehow? Or can we maybe just use some "fresh" snapshot (as these
+ * non-transactional changes should not change sequence-related catalogs).
+ */
+static void
+ReorderBufferAddSequenceTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, XLogRecPtr end_lsn)
+{
+	dlist_mutable_iter miter;
+	ReorderBufferTXN *subtxn = NULL;
+
+	dlist_foreach_modify(miter, &rb->sequence_changes)
+	{
+		ReorderBufferChange *change = dlist_container(ReorderBufferChange, node,
+													  miter.cur);
+
+		/* changes are ordered by LSN, so stop once we exceed end_lsn */
+		if (change->lsn > end_lsn)
+			break;
+
+		if (subtxn == NULL)
+		{
+			subtxn = ReorderBufferGetTXN(rb);
+
+			subtxn->first_lsn = change->lsn;
+
+			/* similar to ReorderBufferAssignChild, but we don't have XIDs */
+			subtxn->txn_flags |= RBTXN_IS_SUBXACT;
+			subtxn->toplevel_xid = txn->xid;
+
+			Assert(subtxn->nsubtxns == 0);
+
+			/* set the reference to top-level transaction */
+			subtxn->toptxn = txn;
+
+			/* add to subtransaction list */
+			dlist_push_tail(&txn->subtxns, &subtxn->node);
+			txn->nsubtxns++;
+		}
+
+		/* remove the change from the top-level list of changes */
+		dlist_delete(miter.cur);
+
+		change->txn = subtxn;
+
+		/* add it to the next subxact */
+		dlist_push_tail(&subtxn->changes, &change->node);
+		subtxn->nentries++;
+		subtxn->nentries_mem++;
+
+		/*
+		 * XXX should this do memory limit checks and partial changes similar to
+		 * ReorderBufferQueueChange?
+		 *
+		 * We have to do at least the accounting, so that the numbers match.
+		 */
+
+		/* update memory accounting information */
+		ReorderBufferChangeMemoryUpdate(rb, change, true,
+										ReorderBufferChangeSize(change));
+
+		/* check the memory limits and evict something if needed */
+		ReorderBufferCheckMemoryLimit(rb);
+	}
+}
+
 /*
  * Commit a transaction.
  *
@@ -3388,6 +3393,19 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
 								false);
 
+	/*
+	 * Maybe add a "fake" subtransaction with non-transactional sequence
+	 * changes that happened since the last commit.
+	 *
+	 * XXX I wonder if this might have issues with sequence changes preceding
+	 * the first LSN of the transaction. What snapshot would those changes use?
+	 * Although, maybe we should just skip those changes, at it has to be from
+	 * not-committed transaction.
+	 *
+	 * XXX Should this be looking at commit_lsn or end_lsn?
+	 */
+	ReorderBufferAddSequenceTXN(rb, txn, commit_lsn);
+
 	/* unknown transaction, nothing to replay */
 	if (txn == NULL)
 		return;
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index e5d8e1c0d1..97d6e56b4c 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -602,6 +602,11 @@ struct ReorderBuffer
 	 */
 	dlist_head	txns_by_base_snapshot_lsn;
 
+	/*
+	 * Non-transactional sequence changes, applied in the nearest commit.
+	 */
+	dlist_head sequence_changes;
+
 	/*
 	 * Transactions and subtransactions that have modified system catalogs.
 	 */
-- 
2.41.0