vglobal-0001-WIP-sequence-reworks.patch

From 1c81414bc3b118b8b26489ec8731e8590e3b4806 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 vglobal 1/2] WIP: sequence reworks

---
 src/backend/replication/logical/decode.c      |  34 +-
 .../replication/logical/reorderbuffer.c       | 475 +++++++-----------
 src/include/replication/reorderbuffer.h       |  36 +-
 3 files changed, 223 insertions(+), 322 deletions(-)

diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 77bc7aea7a0..77db984f594 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)
@@ -1351,33 +1349,15 @@ sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	if(!datalen || !tupledata)
 		return;
 
+	/* sets the snapshot etc. */
+	if (!SnapBuildProcessChange(builder, xid, buf->origptr))
+		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)))
-		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,
+	/* 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..3113076a690 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -156,6 +156,8 @@ typedef struct ReorderBufferSequenceEnt
 {
 	RelFileNode		rnode;
 	TransactionId	xid;
+	bool			transactional;
+	dlist_head		changes;
 } ReorderBufferSequenceEnt;
 
 /* data structures for (relfilenode, ctid) => (cmin, cmax) mapping */
@@ -365,6 +367,11 @@ ReorderBufferAllocate(void)
 											   SLAB_DEFAULT_BLOCK_SIZE,
 											   sizeof(ReorderBufferChange));
 
+	buffer->sequence_context = SlabContextCreate(new_ctx,
+												 "Sequence",
+												 SLAB_DEFAULT_BLOCK_SIZE,
+												 sizeof(ReorderBufferSequenceChange));
+
 	buffer->txn_context = SlabContextCreate(new_ctx,
 											"TXN",
 											SLAB_DEFAULT_BLOCK_SIZE,
@@ -516,6 +523,21 @@ ReorderBufferGetChange(ReorderBuffer *rb)
 	return change;
 }
 
+/*
+ * Get a fresh ReorderBufferSequenceChange.
+ */
+static ReorderBufferSequenceChange *
+ReorderBufferGetSequenceChange(ReorderBuffer *rb)
+{
+	ReorderBufferSequenceChange *change;
+
+	change = (ReorderBufferSequenceChange *)
+		MemoryContextAlloc(rb->sequence_context, sizeof(ReorderBufferSequenceChange));
+
+	memset(change, 0, sizeof(ReorderBufferSequenceChange));
+	return change;
+}
+
 /*
  * Free a ReorderBufferChange and update memory accounting, if requested.
  */
@@ -575,13 +597,6 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change,
 				change->data.truncate.relids = NULL;
 			}
 			break;
-		case REORDER_BUFFER_CHANGE_SEQUENCE:
-			if (change->data.sequence.tuple)
-			{
-				ReorderBufferReturnTupleBuf(rb, change->data.sequence.tuple);
-				change->data.sequence.tuple = NULL;
-			}
-			break;
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
 		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
@@ -923,48 +938,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);
-
-	return found;
-}
+	ent = hash_search(rb->sequences,
+					  (void *) &rnode,
+					  HASH_FIND,
+					  &found);
 
-/*
- * Cleanup sequences created in in-progress transactions.
- *
- * There's no way to search by XID, so we simply do a seqscan of all
- * the entries in the hash table. Hopefully there are only a couple
- * entries in most cases - people generally don't create many new
- * sequences over and over.
- */
-static void
-ReorderBufferSequenceCleanup(ReorderBuffer *rb, TransactionId xid)
-{
-	HASH_SEQ_STATUS scan_status;
-	ReorderBufferSequenceEnt *ent;
-
-	hash_seq_init(&scan_status, rb->sequences);
-	while ((ent = (ReorderBufferSequenceEnt *) hash_seq_search(&scan_status)) != NULL)
-	{
-		/* skip sequences not from this transaction */
-		if (ent->xid != xid)
-			continue;
+	/* if not found, it's definitely non-transactional */
+	if (!found)
+		return false;
 
-		(void) hash_search(rb->sequences,
-					   (void *) &(ent->rnode),
-					   HASH_REMOVE, NULL);
-	}
+	return ent->transactional;
 }
 
 /*
@@ -978,166 +971,97 @@ 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));
-
-		/* But we must have either created or found an existing entry. */
-		Assert(created || found);
-
-		/*
-		 * When creating the sequence, remember the XID of the transaction
-		 * that created id.
-		 */
-		if (created)
-			ent->xid = xid;
+	bool transactional;
+	MemoryContext oldcontext;
+	ReorderBufferSequenceChange *change;
 
-		/* XXX Maybe check that we're still in the same top-level xact? */
+	/* lookup sequence by relfilenode */
+	ReorderBufferSequenceEnt   *ent;
+	bool						found;
 
-		/* OK, allocate and queue the change */
-		oldcontext = MemoryContextSwitchTo(rb->context);
+	transactional = ReorderBufferSequenceIsTransactional(rb, rnode, created);
 
-		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);
-	}
-	else
+	if (!found)
 	{
-		/*
-		 * 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);
+		ent->transactional = transactional;
+		dlist_init(&ent->changes);
+	}
 
-			if (reloid == InvalidOid)
-				elog(ERROR, "could not map filenode \"%s\" to relation OID",
-					 relpathperm(rnode,
-								 MAIN_FORKNUM));
+	/*
+	 * When creating the sequence, remember the XID of the transaction
+	 * that created id.
+	 */
+	if (created)
+		ent->xid = xid;
 
-			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 = ReorderBufferGetSequenceChange(rb);
 
-			RelationClose(relation);
+	change->origin_id = origin_id;
 
-			TeardownHistoricSnapshot(false);
+	memcpy(&change->relnode, &rnode, sizeof(RelFileNode));
 
-			AbortCurrentTransaction();
+	change->tuple = tuplebuf;
+	change->transactional = transactional;
 
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
-		}
-		PG_CATCH();
-		{
-			TeardownHistoricSnapshot(true);
+	Assert(InvalidXLogRecPtr != lsn);
+	change->lsn = lsn;
 
-			AbortCurrentTransaction();
+	if (xid != InvalidTransactionId)
+		change->txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 
-			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.
+	 */
+	/* add it to the list in the global hash table (to the tail, to keep
+	 * the right ordering of changes */
+	dlist_push_tail(&ent->changes, &change->node);
 
-			PG_RE_THROW();
-		}
-		PG_END_TRY();
-	}
+	MemoryContextSwitchTo(oldcontext);
 }
 
 /*
@@ -1816,9 +1740,6 @@ ReorderBufferCleanupTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
 				&found);
 	Assert(found);
 
-	/* Remove sequences created in this transaction (if any). */
-	ReorderBufferSequenceCleanup(rb, txn->xid);
-
 	/* remove entries spilled to disk */
 	if (rbtxn_is_serialized(txn))
 		ReorderBufferRestoreCleanup(rb, txn);
@@ -2239,21 +2160,21 @@ ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn,
  */
 static inline void
 ReorderBufferApplySequence(ReorderBuffer *rb, ReorderBufferTXN *txn,
-						   Relation relation, ReorderBufferChange *change,
+						   Relation relation, ReorderBufferSequenceChange *change,
 						   bool streaming)
 {
 	HeapTuple	tuple;
 	Form_pg_sequence_data seq;
 
-	tuple = &change->data.sequence.tuple->tuple;
+	tuple = &change->tuple->tuple;
 	seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
 
 	/* 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->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->transactional,
 					 seq->last_value, seq->log_cnt, seq->is_called);
 }
 
@@ -2313,6 +2234,83 @@ ReorderBufferResetTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	}
 }
 
+/*
+ * Walk all the accumulated sequence increments, and apply all changes between
+ * the start/end LSNs (e.g. between two decoded changes).
+ */
+static void
+ReorderBufferProcessSequences(ReorderBuffer *rb, ReorderBufferTXN *txn,
+							  Snapshot snapshot,
+							  XLogRecPtr start_lsn, XLogRecPtr end_lsn,
+							  bool streaming)
+{
+	HASH_SEQ_STATUS scan_status;
+	ReorderBufferSequenceEnt *ent;
+
+	/* walk changes for all sequences, apply relevant ones */
+	hash_seq_init(&scan_status, rb->sequences);
+	while ((ent = (ReorderBufferSequenceEnt *) hash_seq_search(&scan_status)) != NULL)
+	{
+		dlist_mutable_iter iter;
+
+		/*
+		 * Walk changes for the sequence, apply those that are non-transactional
+		 * ones or transactional for this XID. Stop on the first transactional
+		 * one for a different XID (XXX that should get a different relnode, so
+		 * not sure it can even happen mid-list).
+		 */
+		dlist_foreach_modify(iter, &ent->changes)
+		{
+			Oid			reloid;
+			Relation	relation;
+			ReorderBufferSequenceChange *change;
+
+			change = dlist_container(ReorderBufferSequenceChange, node, iter.cur);
+
+			/*
+			 * ignore past/future changes
+			 *
+			 * FIXME not sure this is really correct
+			 */
+			if (change->lsn < start_lsn || change->lsn > end_lsn)
+				break;
+
+			/* stop if transactional for a different XID */
+			if (ent->transactional && change->txn != txn)
+				break;
+
+			/* gonna apply the change, so remove from list */
+			dlist_delete(&change->node);
+
+			Assert(snapshot);
+
+			reloid = RelidByRelfilenode(change->relnode.spcNode,
+										change->relnode.relNode);
+
+			if (reloid == InvalidOid)
+				elog(ERROR, "could not map filenode \"%s\" to relation OID",
+					 relpathperm(change->relnode, MAIN_FORKNUM));
+
+			relation = RelationIdGetRelation(reloid);
+
+			if (!RelationIsValid(relation))
+				elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
+					 reloid,
+					 relpathperm(change->relnode, MAIN_FORKNUM));
+
+			if (RelationIsLogicallyLogged(relation))
+				ReorderBufferApplySequence(rb, change->txn, relation, change, streaming);
+
+			RelationClose(relation);
+		}
+
+		if (dlist_is_empty(&ent->changes))
+			(void) hash_search(rb->sequences,
+							   (void *) &(ent->rnode),
+							   HASH_REMOVE, NULL);
+	}
+}
+
 /*
  * Helper function for ReorderBufferReplay and ReorderBufferStreamTXN.
  *
@@ -2419,6 +2417,13 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				SetupCheckXidLive(curtxn->xid);
 			}
 
+			/*
+			 * Emit sequences up to this change (might change snapshot, and
+			 * we need to do lookups by relfilenode)
+			 */
+			ReorderBufferProcessSequences(rb, txn, snapshot_now,
+										  txn->first_lsn, change->lsn, streaming);
+
 			switch (change->action)
 			{
 				case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
@@ -2700,30 +2705,6 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 					elog(ERROR, "tuplecid value in changequeue");
 					break;
 
-				case REORDER_BUFFER_CHANGE_SEQUENCE:
-					Assert(snapshot_now);
-
-					reloid = RelidByRelfilenode(change->data.sequence.relnode.spcNode,
-												change->data.sequence.relnode.relNode);
-
-					if (reloid == InvalidOid)
-						elog(ERROR, "could not map filenode \"%s\" to relation OID",
-							 relpathperm(change->data.sequence.relnode,
-										 MAIN_FORKNUM));
-
-					relation = RelationIdGetRelation(reloid);
-
-					if (!RelationIsValid(relation))
-						elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
-							 reloid,
-							 relpathperm(change->data.sequence.relnode,
-										 MAIN_FORKNUM));
-
-					if (RelationIsLogicallyLogged(relation))
-						ReorderBufferApplySequence(rb, txn, relation, change, streaming);
-
-					RelationClose(relation);
-					break;
 			}
 		}
 
@@ -2734,6 +2715,10 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 		ReorderBufferIterTXNFinish(rb, iterstate);
 		iterstate = NULL;
 
+		/* Replicate relevant changes for sequences. */
+		ReorderBufferProcessSequences(rb, txn, snapshot_now,
+									  txn->first_lsn, commit_lsn, streaming);
+
 		/*
 		 * Update total transaction count and total bytes processed by the
 		 * transaction and its subtransactions. Ensure to not count the
@@ -4108,39 +4093,6 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				memcpy(data, change->data.truncate.relids, size);
 				data += size;
 
-				break;
-			}
-		case REORDER_BUFFER_CHANGE_SEQUENCE:
-			{
-				char	   *data;
-				ReorderBufferTupleBuf *tup;
-				Size		len = 0;
-
-				tup = change->data.sequence.tuple;
-
-				if (tup)
-				{
-					sz += sizeof(HeapTupleData);
-					len = tup->tuple.t_len;
-					sz += len;
-				}
-
-				/* make sure we have enough space */
-				ReorderBufferSerializeReserve(rb, sz);
-
-				data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
-				/* might have been reallocated above */
-				ondisk = (ReorderBufferDiskChange *) rb->outbuf;
-
-				if (len)
-				{
-					memcpy(data, &tup->tuple, sizeof(HeapTupleData));
-					data += sizeof(HeapTupleData);
-
-					memcpy(data, tup->tuple.t_data, len);
-					data += len;
-				}
-
 				break;
 			}
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
@@ -4405,22 +4357,6 @@ ReorderBufferChangeSize(ReorderBufferChange *change)
 			{
 				sz += sizeof(Oid) * change->data.truncate.nrelids;
 
-				break;
-			}
-		case REORDER_BUFFER_CHANGE_SEQUENCE:
-			{
-				ReorderBufferTupleBuf *tup;
-				Size		len = 0;
-
-				tup = change->data.sequence.tuple;
-
-				if (tup)
-				{
-					sz += sizeof(HeapTupleData);
-					len = tup->tuple.t_len;
-					sz += len;
-				}
-
 				break;
 			}
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
@@ -4723,29 +4659,6 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				break;
 			}
 
-		case REORDER_BUFFER_CHANGE_SEQUENCE:
-			if (change->data.sequence.tuple)
-			{
-				uint32		tuplelen = ((HeapTuple) data)->t_len;
-
-				change->data.sequence.tuple =
-					ReorderBufferGetTupleBuf(rb, tuplelen - SizeofHeapTupleHeader);
-
-				/* restore ->tuple */
-				memcpy(&change->data.sequence.tuple->tuple, data,
-					   sizeof(HeapTupleData));
-				data += sizeof(HeapTupleData);
-
-				/* reset t_data pointer into the new tuplebuf */
-				change->data.sequence.tuple->tuple.t_data =
-					ReorderBufferTupleBufData(change->data.sequence.tuple);
-
-				/* restore tuple data itself */
-				memcpy(change->data.sequence.tuple->tuple.t_data, data, tuplelen);
-				data += tuplelen;
-			}
-			break;
-
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
 		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 0bcc150b331..d94bce8d39f 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -64,8 +64,7 @@ typedef enum ReorderBufferChangeType
 	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT,
 	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
 	REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
-	REORDER_BUFFER_CHANGE_TRUNCATE,
-	REORDER_BUFFER_CHANGE_SEQUENCE
+	REORDER_BUFFER_CHANGE_TRUNCATE
 } ReorderBufferChangeType;
 
 /* forward declaration */
@@ -159,13 +158,6 @@ typedef struct ReorderBufferChange
 			uint32		ninvalidations; /* Number of messages */
 			SharedInvalidationMessage *invalidations;	/* invalidation message */
 		}			inval;
-
-		/* Context data for Sequence changes */
-		struct
-		{
-			RelFileNode relnode;
-			ReorderBufferTupleBuf *tuple;
-		}			sequence;
 	}			data;
 
 	/*
@@ -175,6 +167,24 @@ typedef struct ReorderBufferChange
 	dlist_node	node;
 } ReorderBufferChange;
 
+typedef struct ReorderBufferSequenceChange
+{
+	XLogRecPtr	lsn;
+
+	/* Transaction this change belongs to. */
+	struct ReorderBufferTXN *txn;
+
+	RepOriginId origin_id;
+
+	/* sequence data */
+	bool	transactional;
+	RelFileNode relnode;
+	ReorderBufferTupleBuf *tuple;
+
+	/* list of sequence changes */
+	dlist_node	node;
+} ReorderBufferSequenceChange;
+
 /* ReorderBufferTXN txn_flags */
 #define RBTXN_HAS_CATALOG_CHANGES 0x0001
 #define RBTXN_IS_SUBXACT          0x0002
@@ -615,6 +625,7 @@ struct ReorderBuffer
 	 * Memory contexts for specific types objects
 	 */
 	MemoryContext change_context;
+	MemoryContext sequence_context;
 	MemoryContext txn_context;
 	MemoryContext tup_context;
 
@@ -670,8 +681,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 +731,4 @@ void		ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr);
 
 void		StartupReorderBuffer(void);
 
-bool		ReorderBufferSequenceIsTransactional(ReorderBuffer *rb,
-												 RelFileNode rnode, bool created);
-
 #endif
-- 
2.34.1