vchanges-0001-WIP-sequence-reworks.patch

From 294faa547cf8347881831d2137b22b6c25ffa445 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas.vondra@postgresql.org>
Date: Sat, 2 Apr 2022 20:44:15 +0200
Subject: [PATCH vchanges 1/2] WIP: sequence reworks

---
 src/backend/replication/logical/decode.c      |  32 +--
 .../replication/logical/reorderbuffer.c       | 236 ++++++------------
 src/include/replication/reorderbuffer.h       |   8 +-
 3 files changed, 92 insertions(+), 184 deletions(-)

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 77bc7aea7a0..bd78a122b03 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -1313,9 +1313,7 @@ sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	TransactionId xid = XLogRecGetXid(r);
 	uint8		info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
 	xl_seq_rec *xlrec;
-	Snapshot	snapshot;
 	RepOriginId origin_id = XLogRecGetOrigin(r);
-	bool		transactional;
 
 	/* only decode changes flagged with XLOG_SEQ_LOG */
 	if (info != XLOG_SEQ_LOG)
@@ -1352,32 +1350,14 @@ sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		return;
 
 	tuplebuf = ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
-	DecodeSeqTuple(tupledata, datalen, tuplebuf);
 
-	/*
-	 * Should we handle the sequence increment as transactional or not?
-	 *
-	 * If the sequence was created in a still-running transaction, treat
-	 * it as transactional and queue the increments. Otherwise it needs
-	 * to be treated as non-transactional, in which case we send it to
-	 * the plugin right away.
-	 */
-	transactional = ReorderBufferSequenceIsTransactional(ctx->reorder,
-														 target_node,
-														 xlrec->created);
-
-	/* Skip the change if already processed (per the snapshot). */
-	if (transactional &&
-		!SnapBuildProcessChange(builder, xid, buf->origptr))
-		return;
-	else if (!transactional &&
-			 (SnapBuildCurrentState(builder) != SNAPBUILD_CONSISTENT ||
-			  SnapBuildXactNeedsSkip(builder, buf->origptr)))
+	if (!SnapBuildProcessChange(builder, xid, buf->origptr))
 		return;
 
-	/* Queue the increment (or send immediately if not transactional). */
-	snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
-	ReorderBufferQueueSequence(ctx->reorder, xid, snapshot, buf->endptr,
-							   origin_id, target_node, transactional,
+	DecodeSeqTuple(tupledata, datalen, tuplebuf);
+
+	/* queue the sequence increment */
+	ReorderBufferQueueSequence(ctx->reorder, xid, buf->endptr,
+							   origin_id, target_node,
 							   xlrec->created, tuplebuf);
 }
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index c2d9be81fae..0c4fb9cb561 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -155,6 +155,7 @@ typedef struct ReorderBufferTXNByIdEnt
 typedef struct ReorderBufferSequenceEnt
 {
 	RelFileNode		rnode;
+	XLogRecPtr		lsn;
 	TransactionId	xid;
 } ReorderBufferSequenceEnt;
 
@@ -923,21 +924,26 @@ ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
  * so we simply do a lookup (the sequence is identified by relfilende). If
  * we find a match, the increment should be handled as transactional.
  */
-bool
+static bool
 ReorderBufferSequenceIsTransactional(ReorderBuffer *rb,
 									 RelFileNode rnode, bool created)
 {
 	bool	found = false;
+	ReorderBufferSequenceEnt *ent;
 
 	if (created)
 		return true;
 
-	hash_search(rb->sequences,
-				(void *) &rnode,
-				HASH_FIND,
-				&found);
+	ent = hash_search(rb->sequences,
+					  (void *) &rnode,
+					  HASH_FIND,
+					  &found);
+
+	/* if not found, it's definitely non-transactional */
+	if (!found)
+		return false;
 
-	return found;
+	return (ent->xid != InvalidTransactionId);
 }
 
 /*
@@ -958,12 +964,10 @@ ReorderBufferSequenceCleanup(ReorderBuffer *rb, TransactionId xid)
 	while ((ent = (ReorderBufferSequenceEnt *) hash_seq_search(&scan_status)) != NULL)
 	{
 		/* skip sequences not from this transaction */
-		if (ent->xid != xid)
-			continue;
+		if (ent->xid == xid)
+			ent->xid = InvalidTransactionId;
 
-		(void) hash_search(rb->sequences,
-					   (void *) &(ent->rnode),
-					   HASH_REMOVE, NULL);
+		/* FIXME maybe we could remove the entries at some point */
 	}
 }
 
@@ -978,166 +982,90 @@ ReorderBufferSequenceCleanup(ReorderBuffer *rb, TransactionId xid)
  */
 void
 ReorderBufferQueueSequence(ReorderBuffer *rb, TransactionId xid,
-						   Snapshot snapshot, XLogRecPtr lsn, RepOriginId origin_id,
-						   RelFileNode rnode, bool transactional, bool created,
+						   XLogRecPtr lsn, RepOriginId origin_id,
+						   RelFileNode rnode, bool created,
 						   ReorderBufferTupleBuf *tuplebuf)
 {
-	/*
-	 * Change needs to be handled as transactional, because the sequence was
-	 * created in a transaction that is still running. In that case all the
-	 * changes need to be queued in that transaction, we must not send them
-	 * to the downstream until the transaction commits.
-	 *
-	 * There's a bit of a trouble with subtransactions - we can't queue it
-	 * into the subxact, because it might be rolled back and we'd lose the
-	 * increment. We need to queue it into the same (sub)xact that created
-	 * the sequence, which is why we track the XID in the hash table.
-	 */
-	if (transactional)
-	{
-		MemoryContext oldcontext;
-		ReorderBufferChange *change;
-
-		/* lookup sequence by relfilenode */
-		ReorderBufferSequenceEnt   *ent;
-		bool						found;
-
-		/* transactional changes require a transaction */
-		Assert(xid != InvalidTransactionId);
-
-		/* search the lookup table (we ignore the return value, found is enough) */
-		ent = hash_search(rb->sequences,
-						  (void *) &rnode,
-						  created ? HASH_ENTER : HASH_FIND,
-						  &found);
-
-		/*
-		 * If this is the "create" increment, we must not have found any
-		 * pre-existing entry in the hash table (i.e. there must not be
-		 * any conflicting sequence).
-		 */
-		Assert(!(created && found));
+	bool transactional;
+	MemoryContext oldcontext;
+	ReorderBufferChange *change;
 
-		/* But we must have either created or found an existing entry. */
-		Assert(created || found);
+	/* lookup sequence by relfilenode */
+	ReorderBufferSequenceEnt   *ent;
+	bool						found;
 
-		/*
-		 * When creating the sequence, remember the XID of the transaction
-		 * that created id.
-		 */
-		if (created)
-			ent->xid = xid;
-
-		/* XXX Maybe check that we're still in the same top-level xact? */
+	transactional = ReorderBufferSequenceIsTransactional(rb, rnode, created);
 
-		/* OK, allocate and queue the change */
-		oldcontext = MemoryContextSwitchTo(rb->context);
-
-		change = ReorderBufferGetChange(rb);
+	/*
+	 * Maintenance of the sequences hash, so that we can distinguish which
+	 * additional increments are transactional and non-transactional.
+	 */
 
-		change->action = REORDER_BUFFER_CHANGE_SEQUENCE;
-		change->origin_id = origin_id;
+	/* transactional changes require a transaction */
+	Assert(!(transactional && (xid == InvalidTransactionId)));
 
-		memcpy(&change->data.sequence.relnode, &rnode, sizeof(RelFileNode));
+	/* search the lookup table */
+	ent = hash_search(rb->sequences,
+					  (void *) &rnode,
+					  HASH_ENTER,
+					  &found);
 
-		change->data.sequence.tuple = tuplebuf;
+	/*
+	 * If this is the "create" increment, we must not have found any entry in
+	 * the hash table (i.e. there must not be any conflicting sequence with the
+	 * same relfilenode).
+	 */
+	Assert(!(created && found));
 
-		/* add it to the same subxact that created the sequence */
-		ReorderBufferQueueChange(rb, ent->xid, lsn, change, false);
+	/*
+	 * It's either non-transactioanl change, creation of a relfilenode, or we
+	 * should have found an existing entry.
+	 */
+	Assert(!transactional || created || found);
 
-		MemoryContextSwitchTo(oldcontext);
-	}
+	/*
+	 * When creating the sequence, remember the XID of the transaction
+	 * that created id.
+	 */
+	if (transactional)
+		ent->xid = 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;
-
-#ifdef USE_ASSERT_CHECKING
-		/* All "creates" have to be handled as transactional. */
-		Assert(!created);
-
-		/* Make sure the sequence is not in the hash table. */
-		{
-			bool	found;
-			hash_search(rb->sequences,
-						(void *) &rnode,
-						HASH_FIND, &found);
-			Assert(!found);
-		}
-#endif
-
-		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;
-
-			if (using_subtxn)
-				BeginInternalSubTransaction("sequence");
-			else
-				StartTransactionCommand();
-
-			reloid = RelidByRelfilenode(rnode.spcNode, rnode.relNode);
-
-			if (reloid == InvalidOid)
-				elog(ERROR, "could not map filenode \"%s\" to relation OID",
-					 relpathperm(rnode,
-								 MAIN_FORKNUM));
+		ent->xid = InvalidTransactionId;
 
-			relation = RelationIdGetRelation(reloid);
-			tuple = &tuplebuf->tuple;
-			seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
+	/*
+	 * Queue the sequence change, into a global list.
+	 *
+	 * There's a bit of a trouble with subtransactions - we can't queue it
+	 * into the subxact, because it might be rolled back and we'd lose the
+	 * increment. We need to queue it into the same (sub)xact that created
+	 * the sequence, which is why we track the XID in the hash table.
+	 */
+	oldcontext = MemoryContextSwitchTo(rb->context);
 
-			rb->sequence(rb, txn, lsn, relation, transactional,
-						 seq->last_value, seq->log_cnt, seq->is_called);
+	change = ReorderBufferGetChange(rb);
 
-			RelationClose(relation);
+	change->action = REORDER_BUFFER_CHANGE_SEQUENCE;
+	change->origin_id = origin_id;
 
-			TeardownHistoricSnapshot(false);
+	memcpy(&change->data.sequence.relnode, &rnode, sizeof(RelFileNode));
 
-			AbortCurrentTransaction();
+	change->data.sequence.tuple = tuplebuf;
+	change->data.sequence.transactional = transactional;
 
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
-		}
-		PG_CATCH();
-		{
-			TeardownHistoricSnapshot(true);
+	Assert(InvalidXLogRecPtr != lsn);
+	change->lsn = lsn;
 
-			AbortCurrentTransaction();
+	ReorderBufferQueueChange(rb, xid, lsn, change, false);
 
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
+	/*
+	 * FIXME What to do about concurrent_abort for the transaction? Maybe
+	 * should depend on transactional vs. non-transactional change.
+	 *
+	 * FIXME Also what to do about memory accounting? If not treated as
+	 * regular changes, it's not part of memory accounting.
+	 */
 
-			PG_RE_THROW();
-		}
-		PG_END_TRY();
-	}
+	MemoryContextSwitchTo(oldcontext);
 }
 
 /*
@@ -2250,10 +2178,12 @@ ReorderBufferApplySequence(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
 	/* Only ever called from ReorderBufferApplySequence, so transational. */
 	if (streaming)
-		rb->stream_sequence(rb, txn, change->lsn, relation, true,
+		rb->stream_sequence(rb, txn, change->lsn, relation,
+							change->data.sequence.transactional,
 							seq->last_value, seq->log_cnt, seq->is_called);
 	else
-		rb->sequence(rb, txn, change->lsn, relation, true,
+		rb->sequence(rb, txn, change->lsn, relation,
+					 change->data.sequence.transactional,
 					 seq->last_value, seq->log_cnt, seq->is_called);
 }
 
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 0bcc150b331..6e939e27a15 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -165,6 +165,7 @@ typedef struct ReorderBufferChange
 		{
 			RelFileNode relnode;
 			ReorderBufferTupleBuf *tuple;
+			bool		transactional;
 		}			sequence;
 	}			data;
 
@@ -670,8 +671,8 @@ void		ReorderBufferQueueMessage(ReorderBuffer *, TransactionId, Snapshot snapsho
 									  bool transactional, const char *prefix,
 									  Size message_size, const char *message);
 void		ReorderBufferQueueSequence(ReorderBuffer *rb, TransactionId xid,
-									   Snapshot snapshot, XLogRecPtr lsn, RepOriginId origin_id,
-									   RelFileNode rnode, bool transactional, bool created,
+									   XLogRecPtr lsn, RepOriginId origin_id,
+									   RelFileNode rnode, bool created,
 									   ReorderBufferTupleBuf *tuplebuf);
 void		ReorderBufferCommit(ReorderBuffer *, TransactionId,
 								XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
@@ -720,7 +721,4 @@ void		ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);
 
 void		StartupReorderBuffer(void);
 
-bool		ReorderBufferSequenceIsTransactional(ReorderBuffer *rb,
-												 RelFileNode rnode, bool created);
-
 #endif
-- 
2.34.1