0001-rework-sequence-decoding.patch

From 0ba154eef933a9b39bb7eaa0e339620a2f98a37b Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas.vondra@postgresql.org>
Date: Tue, 5 Apr 2022 00:25:02 +0200
Subject: [PATCH] rework sequence decoding

---
 contrib/test_decoding/test_decoding.c         |  24 +-
 src/backend/access/rmgrdesc/logicalmsgdesc.c  |  13 +
 src/backend/access/transam/xact.c             |  25 ++
 src/backend/commands/sequence.c               | 263 +++++++-----
 src/backend/replication/logical/decode.c      | 112 ++---
 src/backend/replication/logical/logical.c     |   9 +-
 src/backend/replication/logical/message.c     |   3 +-
 src/backend/replication/logical/proto.c       |   4 +-
 .../replication/logical/reorderbuffer.c       | 387 ++----------------
 src/backend/replication/logical/tablesync.c   |   3 +-
 src/backend/replication/logical/worker.c      |  16 +-
 src/backend/replication/pgoutput/pgoutput.c   |  21 +-
 src/include/access/rmgrlist.h                 |   2 +-
 src/include/commands/sequence.h               |   5 +-
 src/include/replication/logicalproto.h        |   2 -
 src/include/replication/message.h             |  15 +
 src/include/replication/output_plugin.h       |   2 -
 src/include/replication/reorderbuffer.h       |  16 +-
 18 files changed, 330 insertions(+), 592 deletions(-)

diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index c7a87f5fe5b..851117b4155 100644
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -79,7 +79,7 @@ static void pg_decode_message(LogicalDecodingContext *ctx,
 							  Size sz, const char *message);
 static void pg_decode_sequence(LogicalDecodingContext *ctx,
 							  ReorderBufferTXN *txn, XLogRecPtr sequence_lsn,
-							  Relation rel, bool transactional,
+							  Relation rel,
 							  int64 last_value, int64 log_cnt, bool is_called);
 static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
 									 TransactionId xid,
@@ -123,7 +123,7 @@ static void pg_decode_stream_message(LogicalDecodingContext *ctx,
 									 Size sz, const char *message);
 static void pg_decode_stream_sequence(LogicalDecodingContext *ctx,
 									  ReorderBufferTXN *txn, XLogRecPtr sequence_lsn,
-									  Relation rel, bool transactional,
+									  Relation rel,
 									  int64 last_value, int64 log_cnt, bool is_called);
 static void pg_decode_stream_truncate(LogicalDecodingContext *ctx,
 									  ReorderBufferTXN *txn,
@@ -782,7 +782,6 @@ pg_decode_message(LogicalDecodingContext *ctx,
 static void
 pg_decode_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 				   XLogRecPtr sequence_lsn, Relation rel,
-				   bool transactional,
 				   int64 last_value, int64 log_cnt, bool is_called)
 {
 	TestDecodingData *data = ctx->output_plugin_private;
@@ -792,22 +791,19 @@ pg_decode_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 		return;
 
 	/* output BEGIN if we haven't yet, but only for the transactional case */
-	if (transactional)
+	if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
 	{
-		if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
-		{
-			pg_output_begin(ctx, data, txn, false);
-		}
-		txndata->xact_wrote_changes = true;
+		pg_output_begin(ctx, data, txn, false);
 	}
+	txndata->xact_wrote_changes = true;
 
 	OutputPluginPrepareWrite(ctx, true);
 	appendStringInfoString(ctx->out, "sequence ");
 	appendStringInfoString(ctx->out,
 						   quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(rel))),
 													  RelationGetRelationName(rel)));
-	appendStringInfo(ctx->out, 	": transactional:%d last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d",
-					 transactional, last_value, log_cnt, is_called);
+	appendStringInfo(ctx->out, 	": last_value: %zu log_cnt: %zu is_called:%d",
+					 last_value, log_cnt, is_called);
 	OutputPluginWrite(ctx, true);
 }
 
@@ -1013,7 +1009,6 @@ pg_decode_stream_message(LogicalDecodingContext *ctx,
 static void
 pg_decode_stream_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 						  XLogRecPtr sequence_lsn, Relation rel,
-						  bool transactional,
 						  int64 last_value, int64 log_cnt, bool is_called)
 {
 	TestDecodingData *data = ctx->output_plugin_private;
@@ -1023,7 +1018,6 @@ pg_decode_stream_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 		return;
 
 	/* output BEGIN if we haven't yet, but only for the transactional case */
-	if (transactional)
 	{
 		if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
 		{
@@ -1037,8 +1031,8 @@ pg_decode_stream_sequence(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	appendStringInfoString(ctx->out,
 						   quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(rel))),
 													  RelationGetRelationName(rel)));
-	appendStringInfo(ctx->out, 	": transactional:%d last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d",
-					 transactional, last_value, log_cnt, is_called);
+	appendStringInfo(ctx->out, 	": last_value: " INT64_FORMAT " log_cnt: " INT64_FORMAT " is_called:%d",
+					 last_value, log_cnt, is_called);
 	OutputPluginWrite(ctx, true);
 }
 
diff --git a/src/backend/access/rmgrdesc/logicalmsgdesc.c b/src/backend/access/rmgrdesc/logicalmsgdesc.c
index 099e11a84e7..de77dcb0dc2 100644
--- a/src/backend/access/rmgrdesc/logicalmsgdesc.c
+++ b/src/backend/access/rmgrdesc/logicalmsgdesc.c
@@ -40,6 +40,16 @@ logicalmsg_desc(StringInfo buf, XLogReaderState *record)
 			sep = " ";
 		}
 	}
+	else if (info == XLOG_LOGICAL_SEQUENCE)
+	{
+		xl_logical_sequence *xlrec = (xl_logical_sequence *) rec;
+
+		appendStringInfo(buf, "rel %u/%u/%u last: %lu log_cnt: %lu is_called: %d",
+						 xlrec->node.spcNode,
+						 xlrec->node.dbNode,
+						 xlrec->node.relNode,
+						 xlrec->last, xlrec->log_cnt, xlrec->is_called);
+	}
 }
 
 const char *
@@ -48,5 +58,8 @@ logicalmsg_identify(uint8 info)
 	if ((info & ~XLR_INFO_MASK) == XLOG_LOGICAL_MESSAGE)
 		return "MESSAGE";
 
+	if ((info & ~XLR_INFO_MASK) == XLOG_LOGICAL_SEQUENCE)
+		return "SEQUENCE";
+
 	return NULL;
 }
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 3596a7d7345..5b6a07ca892 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -37,6 +37,7 @@
 #include "catalog/storage.h"
 #include "commands/async.h"
 #include "commands/tablecmds.h"
+#include "commands/sequence.h"
 #include "commands/trigger.h"
 #include "common/pg_prng.h"
 #include "executor/spi.h"
@@ -2223,6 +2224,15 @@ CommitTransaction(void)
 	/* Prevent cancel/die interrupt while cleaning up */
 	HOLD_INTERRUPTS();
 
+	/*
+	 * Write state of sequences to WAL. We need to do this early enough so
+	 * that we can still write stuff to WAL, but probably before changing
+	 * the transaction state.
+	 *
+	 * XXX Should this happen before/after holding the interrupts?
+	 */
+	AtEOXact_Sequences(true);
+
 	/* Commit updates to the relation map --- do this as late as possible */
 	AtEOXact_RelationMap(true, is_parallel_worker);
 
@@ -2499,6 +2509,15 @@ PrepareTransaction(void)
 	/* Prevent cancel/die interrupt while cleaning up */
 	HOLD_INTERRUPTS();
 
+	/*
+	 * Write state of sequences to WAL. We need to do this early enough so
+	 * that we can still write stuff to WAL, but probably before changing
+	 * the transaction state.
+	 *
+	 * XXX Should this happen before/after holding the interrupts?
+	 */
+	AtEOXact_Sequences(true);
+
 	/*
 	 * set the current transaction state information appropriately during
 	 * prepare processing
@@ -2734,6 +2753,12 @@ AbortTransaction(void)
 			 TransStateAsString(s->state));
 	Assert(s->parent == NULL);
 
+	/*
+	 * For transaction abort we don't need to write anything to WAL, we just
+	 * do cleanup of the hash table etc.
+	 */
+	AtEOXact_Sequences(false);
+
 	/*
 	 * set the current transaction state information appropriately during the
 	 * abort processing
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index 717bb0b2aa9..50c03b71dac 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -37,6 +37,7 @@
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
 #include "parser/parse_type.h"
+#include "replication/message.h"
 #include "storage/lmgr.h"
 #include "storage/proc.h"
 #include "storage/smgr.h"
@@ -75,6 +76,7 @@ typedef struct SeqTableData
 {
 	Oid			relid;			/* pg_class OID of this sequence (hash key) */
 	Oid			filenode;		/* last seen relfilenode of this sequence */
+	Oid			tablespace;		/* last seen tablespace of this sequence */
 	LocalTransactionId lxid;	/* xact in which we last did a seq op */
 	bool		last_valid;		/* do we have a valid "last" value? */
 	int64		last;			/* value last returned by nextval */
@@ -82,6 +84,7 @@ typedef struct SeqTableData
 	/* if last != cached, we have not used up all the cached values */
 	int64		increment;		/* copy of sequence's increment field */
 	/* note that increment is zero until we first do nextval_internal() */
+	bool		need_log;		/* should be written to WAL at commit? */
 } SeqTableData;
 
 typedef SeqTableData *SeqTable;
@@ -336,83 +339,13 @@ ResetSequence(Oid seq_relid)
 	relation_close(seq_rel, NoLock);
 }
 
-/*
- * Update the sequence state by modifying the existing sequence data row.
- *
- * This keeps the same relfilenode, so the behavior is non-transactional.
- */
-static void
-SetSequence_non_transactional(Oid seqrelid, int64 last_value, int64 log_cnt, bool is_called)
-{
-	SeqTable	elm;
-	Relation	seqrel;
-	Buffer		buf;
-	HeapTupleData seqdatatuple;
-	Form_pg_sequence_data seq;
-
-	/* open and lock sequence */
-	init_sequence(seqrelid, &elm, &seqrel);
-
-	/* lock page' buffer and read tuple */
-	seq = read_seq_tuple(seqrel, &buf, &seqdatatuple);
-
-	/* check the comment above nextval_internal()'s equivalent call. */
-	if (RelationNeedsWAL(seqrel))
-	{
-		GetTopTransactionId();
-
-		if (XLogLogicalInfoActive())
-			GetCurrentTransactionId();
-	}
-
-	/* ready to change the on-disk (or really, in-buffer) tuple */
-	START_CRIT_SECTION();
-
-	seq->last_value = last_value;
-	seq->is_called = is_called;
-	seq->log_cnt = log_cnt;
-
-	MarkBufferDirty(buf);
-
-	/* XLOG stuff */
-	if (RelationNeedsWAL(seqrel))
-	{
-		xl_seq_rec	xlrec;
-		XLogRecPtr	recptr;
-		Page		page = BufferGetPage(buf);
-
-		XLogBeginInsert();
-		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
-
-		xlrec.node = seqrel->rd_node;
-		xlrec.created = false;
-
-		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
-		XLogRegisterData((char *) seqdatatuple.t_data, seqdatatuple.t_len);
-
-		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
-
-		PageSetLSN(page, recptr);
-	}
-
-	END_CRIT_SECTION();
-
-	UnlockReleaseBuffer(buf);
-
-	/* Clear local cache so that we don't think we have cached numbers */
-	/* Note that we do not change the currval() state */
-	elm->cached = elm->last;
-
-	relation_close(seqrel, NoLock);
-}
-
 /*
  * Update the sequence state by creating a new relfilenode.
  *
  * This creates a new relfilenode, to allow transactional behavior.
  */
-static void
-SetSequence_transactional(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called)
+void
+SetSequence(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called)
 {
 	SeqTable	elm;
 	Relation	seqrel;
@@ -469,27 +402,6 @@ SetSequence_transactional(Oid seq_relid, int64 last_value, int64 log_cnt, bool i
 	relation_close(seqrel, NoLock);
 }
 
-/*
- * Set a sequence to a specified internal state.
- *
- * The change is made transactionally, so that on failure of the current
- * transaction, the sequence will be restored to its previous state.
- * We do that by creating a whole new relfilenode for the sequence; so this
- * works much like the rewriting forms of ALTER TABLE.
- *
- * Caller is assumed to have acquired AccessExclusiveLock on the sequence,
- * which must not be released until end of transaction.  Caller is also
- * responsible for permissions checking.
- */
-void
-SetSequence(Oid seq_relid, bool transactional, int64 last_value, int64 log_cnt, bool is_called)
-{
-	if (transactional)
-		SetSequence_transactional(seq_relid, last_value, log_cnt, is_called);
-	else
-		SetSequence_non_transactional(seq_relid, last_value, log_cnt, is_called);
-}
-
 /*
  * Initialize a sequence's relation with the specified tuple as content
  */
@@ -530,7 +442,13 @@ fill_seq_with_data(Relation rel, HeapTuple tuple)
 	tuple->t_data->t_infomask |= HEAP_XMAX_INVALID;
 	ItemPointerSet(&tuple->t_data->t_ctid, 0, FirstOffsetNumber);
 
-	/* check the comment above nextval_internal()'s equivalent call. */
+	/*
+	 * Remember we need to write this sequence to WAL at commit, and make sure
+	 * we have XID if we may need to decode this sequence.
+	 *
+	 * XXX Maybe we should not be doing this except when actually reading a
+	 * value from the sequence?
+	 */
 	if (RelationNeedsWAL(rel))
 	{
 		GetTopTransactionId();
@@ -558,7 +476,6 @@ fill_seq_with_data(Relation rel, HeapTuple tuple)
 		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
 
 		xlrec.node = rel->rd_node;
-		xlrec.created = true;
 
 		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 		XLogRegisterData((char *) tuple->t_data, tuple->t_len);
@@ -762,6 +679,23 @@ nextval_internal(Oid relid, bool check_permissions)
 	/* open and lock sequence */
 	init_sequence(relid, &elm, &seqrel);
 
+	/*
+	 * Remember we need to write this sequence to WAL at commit, and make sure
+	 * we have XID if we may need to decode this sequence.
+	 *
+	 * XXX Maybe we should not be doing this except when actually reading a
+	 * value from the sequence?
+	 */
+	if (RelationNeedsWAL(seqrel))
+	{
+		elm->need_log = true;
+
+		GetTopTransactionId();
+
+		if (XLogLogicalInfoActive())
+			GetCurrentTransactionId();
+	}
+
 	if (check_permissions &&
 		pg_class_aclcheck(elm->relid, GetUserId(),
 						  ACL_USAGE | ACL_UPDATE) != ACLCHECK_OK)
@@ -973,7 +907,6 @@ nextval_internal(Oid relid, bool check_permissions)
 		seq->log_cnt = 0;
 
 		xlrec.node = seqrel->rd_node;
-		xlrec.created = false;
 
 		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 		XLogRegisterData((char *) seqdatatuple.t_data, seqdatatuple.t_len);
@@ -1092,6 +1025,23 @@ do_setval(Oid relid, int64 next, bool iscalled)
 	/* open and lock sequence */
 	init_sequence(relid, &elm, &seqrel);
 
+	/*
+	 * Remember we need to write this sequence to WAL at commit, and make sure
+	 * we have XID if we may need to decode this sequence.
+	 *
+	 * XXX Maybe we should not be doing this except when actually reading a
+	 * value from the sequence?
+	 */
+	if (RelationNeedsWAL(seqrel))
+	{
+		elm->need_log = true;
+
+		GetTopTransactionId();
+
+		if (XLogLogicalInfoActive())
+			GetCurrentTransactionId();
+	}
+
 	if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
@@ -1166,8 +1116,6 @@ do_setval(Oid relid, int64 next, bool iscalled)
 		XLogRegisterBuffer(0, buf, REGBUF_WILL_INIT);
 
 		xlrec.node = seqrel->rd_node;
-		xlrec.created = false;
-
 		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 		XLogRegisterData((char *) seqdatatuple.t_data, seqdatatuple.t_len);
 
@@ -1316,6 +1264,7 @@ init_sequence(Oid relid, SeqTable *p_elm, Relation *p_rel)
 	if (seqrel->rd_rel->relfilenode != elm->filenode)
 	{
 		elm->filenode = seqrel->rd_rel->relfilenode;
+		elm->tablespace = seqrel->rd_rel->reltablespace;
 		elm->cached = elm->last;
 	}
 
@@ -2051,3 +2000,121 @@ seq_mask(char *page, BlockNumber blkno)
 
 	mask_unused_space(page);
 }
+
+
+static void
+read_sequence_info(Relation seqrel, int64 *last_value, int64 *log_cnt, bool *is_called)
+{
+	Buffer		buf;
+	Form_pg_sequence_data seq;
+	HeapTupleData	seqdatatuple;
+
+	seq = read_seq_tuple(seqrel, &buf, &seqdatatuple);
+
+	*last_value = seq->last_value;
+	*is_called = seq->is_called;
+	*log_cnt = seq->log_cnt;
+
+	UnlockReleaseBuffer(buf);
+}
+
+/* XXX Do this only for wal_level = logical, probably? */
+void
+AtEOXact_Sequences(bool isCommit)
+{
+	SeqTable		entry;
+	HASH_SEQ_STATUS	scan;
+
+	if (!seqhashtab)
+		return;
+
+	/* only do this with wal_level=logical */
+	if (!XLogLogicalInfoActive())
+		return;
+
+	/*
+	 * XXX Maybe we could enforce having XID here? Or is it too late?
+	 */
+	// Assert(GetTopTransactionId() != InvalidTransactionId);
+	// Assert(GetCurrentTransactionId() != InvalidTransactionId);
+
+	hash_seq_init(&scan, seqhashtab);
+
+	while ((entry = (SeqTable) hash_seq_search(&scan)))
+	{
+		Relation			rel;
+		RelFileNode			rnode;
+		xl_logical_sequence	xlrec;
+
+		/* not commit, we don't guarantee any data to be durable */
+		if (!isCommit)
+			entry->need_log = false;
+
+		/*
+		 * If not touched in the current transaction, don't log anything.
+		 * We leave needs_log set, so that if future transactions touch
+		 * the sequence we'll log it properly.
+		 */
+		if (!entry->need_log)
+			continue;
+
+		/* if this is commit, we'll log the */
+		entry->need_log = false;
+
+		/*
+		 * does the relation still exist?
+		 *
+		 * XXX We need to make sure another transaction can't jump ahead of
+		 * this one, otherwise the ordering of sequence changes could change.
+		 * Imagine T1 and T2, where T1 writes sequence state first, but then
+		 * T2 does it too and commits first:
+		 *
+		 * T1: log sequence state
+		 * T2: increment sequence
+		 * T2: log sequence state
+		 * T2: commit
+		 * T1: commit
+		 *
+		 * If we apply the sequences in this order, it'd be broken as the
+		 * value might go backwars. ShareUpdateExclusive protects against
+		 * that, but it's also restricting commit throughtput, probably.
+		 *
+		 * XXX This might be an issue for deadlocks, too, if two xacts try
+		 * to write sequences in different ordering. We may need to sort
+		 * the OIDs first, to enforce the same lock ordering.
+		 */
+		rel = try_relation_open(entry->relid, ShareUpdateExclusiveLock);
+
+		/* XXX relation might have been dropped, maybe we should have logged
+		 * the last change? */
+		if (!rel)
+			continue;
+
+		/* tablespace */
+		if (OidIsValid(entry->tablespace))
+			rnode.spcNode = entry->tablespace;
+		else
+			rnode.spcNode = MyDatabaseTableSpace;
+
+		rnode.dbNode = MyDatabaseId;			/* database */
+		rnode.relNode = entry->filenode;		/* relation */
+
+		xlrec.node = rnode;
+		xlrec.reloid = entry->relid;
+
+		/* XXX is it good enough to log values we have in cache? seems
+		 * wrong and we may need to re-read that. */
+		read_sequence_info(rel, &xlrec.last, &xlrec.log_cnt, &xlrec.is_called);
+
+		XLogBeginInsert();
+		XLogRegisterData((char *) &xlrec, SizeOfLogicalSequence);
+
+		/* allow origin filtering */
+		XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+		(void) XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_SEQUENCE);
+
+		/* hold the sequence lock until the end of the transaction */
+		relation_close(rel, NoLock);
+	}
+}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 77bc7aea7a0..a8548f9ac64 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -44,6 +44,9 @@
 #include "storage/standby.h"
 #include "commands/sequence.h"
 
+static void DecodeLogicalMessage(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+static void DecodeLogicalSequence(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+
 /* individual record(group)'s handlers */
 static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
 static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
@@ -64,7 +67,6 @@ static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 
 /* common function to decode tuples */
 static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup);
-static void DecodeSeqTuple(char *data, Size len, ReorderBufferTupleBuf *tuple);
 
 /* helper functions for decoding transactions */
 static inline bool FilterPrepare(LogicalDecodingContext *ctx,
@@ -560,6 +562,27 @@ FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
  */
 void
 logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+	XLogReaderState *r = buf->record;
+	uint8		info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
+
+	switch (info)
+	{
+		case XLOG_LOGICAL_MESSAGE:
+			DecodeLogicalMessage(ctx, buf);
+			break;
+
+		case XLOG_LOGICAL_SEQUENCE:
+			DecodeLogicalSequence(ctx, buf);
+			break;
+
+		default:
+			elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
+	}
+}
+
+static void
+DecodeLogicalMessage(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	SnapBuild  *builder = ctx->snapshot_builder;
 	XLogReaderState *r = buf->record;
@@ -1253,36 +1276,6 @@ DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 			ctx->fast_forward || FilterByOrigin(ctx, origin_id));
 }
 
-/*
- * DecodeSeqTuple
- *		decode tuple describing the sequence increment
- *
- * Sequences are represented as a table with a single row, which gets updated
- * by nextval(). The tuple is stored in WAL right after the xl_seq_rec, so we
- * simply copy it into the tuplebuf (similar to seq_redo).
- */
-static void
-DecodeSeqTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
-{
-	int			datalen = len - sizeof(xl_seq_rec) - SizeofHeapTupleHeader;
-
-	Assert(datalen >= 0);
-
-	tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
-
-	ItemPointerSetInvalid(&tuple->tuple.t_self);
-
-	tuple->tuple.t_tableOid = InvalidOid;
-
-	memcpy(((char *) tuple->tuple.t_data),
-		   data + sizeof(xl_seq_rec),
-		   SizeofHeapTupleHeader);
-
-	memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
-		   data + sizeof(xl_seq_rec) + SizeofHeapTupleHeader,
-		   datalen);
-}
-
 /*
  * Handle sequence decode
  *
@@ -1301,24 +1294,18 @@ DecodeSeqTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
  * plugin - it might get confused about which sequence it's related to etc.
  */
 void
-sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+DecodeLogicalSequence(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 {
 	SnapBuild  *builder = ctx->snapshot_builder;
-	ReorderBufferTupleBuf *tuplebuf;
 	RelFileNode target_node;
 	XLogReaderState *r = buf->record;
-	char	   *tupledata = NULL;
-	Size		tuplelen;
-	Size		datalen = 0;
 	TransactionId xid = XLogRecGetXid(r);
 	uint8		info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
-	xl_seq_rec *xlrec;
-	Snapshot	snapshot;
+	xl_logical_sequence *xlrec;
 	RepOriginId origin_id = XLogRecGetOrigin(r);
-	bool		transactional;
 
 	/* only decode changes flagged with XLOG_SEQ_LOG */
-	if (info != XLOG_SEQ_LOG)
+	if (info != XLOG_LOGICAL_SEQUENCE)
 		elog(ERROR, "unexpected RM_SEQ_ID record type: %u", info);
 
 	ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
@@ -1331,8 +1318,11 @@ sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 		ctx->fast_forward)
 		return;
 
+	/* extract the WAL record, with "created" flag */
+	xlrec = (xl_logical_sequence *) XLogRecGetData(r);
+
 	/* only interested in our database */
-	XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
+	target_node = xlrec->node;
 	if (target_node.dbNode != ctx->slot->data.database)
 		return;
 
@@ -1340,44 +1330,12 @@ sequence_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 	if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
 		return;
 
-	tupledata = XLogRecGetData(r);
-	datalen = XLogRecGetDataLen(r);
-	tuplelen = datalen - SizeOfHeapHeader - sizeof(xl_seq_rec);
-
-	/* extract the WAL record, with "created" flag */
-	xlrec = (xl_seq_rec *) XLogRecGetData(r);
-
-	/* XXX how could we have sequence change without data? */
-	if(!datalen || !tupledata)
-		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,
-							   xlrec->created, tuplebuf);
+	ReorderBufferQueueSequence(ctx->reorder, xid, buf->endptr,
+							   origin_id, xlrec->reloid, target_node,
+							   xlrec->last, xlrec->log_cnt, xlrec->is_called);
 }
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index e1f14aeecb5..d00cc08fb3a 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -75,7 +75,6 @@ static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 							   const char *prefix, Size message_size, const char *message);
 static void sequence_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 								XLogRecPtr sequence_lsn, Relation rel,
-								bool transactional,
 								int64 last_value, int64 log_cnt, bool is_called);
 
 /* streaming callbacks */
@@ -96,7 +95,6 @@ static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *tx
 									  const char *prefix, Size message_size, const char *message);
 static void stream_sequence_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 									   XLogRecPtr sequence_lsn, Relation rel,
-									   bool transactional,
 									   int64 last_value, int64 log_cnt, bool is_called);
 static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 									   int nrelations, Relation relations[], ReorderBufferChange *change);
@@ -1218,7 +1216,7 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 
 static void
 sequence_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
-					XLogRecPtr sequence_lsn, Relation rel, bool transactional,
+					XLogRecPtr sequence_lsn, Relation rel,
 					int64 last_value, int64 log_cnt, bool is_called)
 {
 	LogicalDecodingContext *ctx = cache->private_data;
@@ -1245,7 +1243,7 @@ sequence_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 	ctx->write_location = sequence_lsn;
 
 	/* do the actual work: call callback */
-	ctx->callbacks.sequence_cb(ctx, txn, sequence_lsn, rel, transactional,
+	ctx->callbacks.sequence_cb(ctx, txn, sequence_lsn, rel,
 							   last_value, log_cnt, is_called);
 
 	/* Pop the error context stack */
@@ -1560,7 +1558,6 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 static void
 stream_sequence_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 						   XLogRecPtr sequence_lsn, Relation rel,
-						   bool transactional,
 						   int64 last_value, int64 log_cnt, bool is_called)
 {
 	LogicalDecodingContext *ctx = cache->private_data;
@@ -1591,7 +1588,7 @@ stream_sequence_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 	ctx->write_location = sequence_lsn;
 
 	/* do the actual work: call callback */
-	ctx->callbacks.sequence_cb(ctx, txn, sequence_lsn, rel, transactional,
+	ctx->callbacks.sequence_cb(ctx, txn, sequence_lsn, rel,
 							   last_value, log_cnt, is_called);
 
 	/* Pop the error context stack */
diff --git a/src/backend/replication/logical/message.c b/src/backend/replication/logical/message.c
index 1c34912610e..69178dc1cfb 100644
--- a/src/backend/replication/logical/message.c
+++ b/src/backend/replication/logical/message.c
@@ -82,7 +82,8 @@ logicalmsg_redo(XLogReaderState *record)
 {
 	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
-	if (info != XLOG_LOGICAL_MESSAGE)
+	if (info != XLOG_LOGICAL_MESSAGE &&
+		info != XLOG_LOGICAL_SEQUENCE)
 		elog(PANIC, "logicalmsg_redo: unknown op code %u", info);
 
 	/* This is only interesting for logical decoding, see decode.c. */
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 18d3cbb9248..ae078eed048 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -667,7 +667,7 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
  */
 void
 logicalrep_write_sequence(StringInfo out, Relation rel, TransactionId xid,
-						  XLogRecPtr lsn, bool transactional,
+						  XLogRecPtr lsn,
 						  int64 last_value, int64 log_cnt, bool is_called)
 {
 	uint8		flags = 0;
@@ -686,7 +686,6 @@ logicalrep_write_sequence(StringInfo out, Relation rel, TransactionId xid,
 	relname = RelationGetRelationName(rel);
 	pq_sendstring(out, relname);
 
-	pq_sendint8(out, transactional);
 	pq_sendint64(out, last_value);
 	pq_sendint64(out, log_cnt);
 	pq_sendint8(out, is_called);
@@ -706,7 +705,6 @@ logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata)
 	seqdata->nspname = pstrdup(logicalrep_read_namespace(in));
 	seqdata->seqname = pstrdup(pq_getmsgstring(in));
 
-	seqdata->transactional = pq_getmsgint(in, 1);
 	seqdata->last_value = pq_getmsgint64(in);
 	seqdata->log_cnt = pq_getmsgint64(in);
 	seqdata->is_called = pq_getmsgint(in, 1);
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 4702750a2e7..0cc7c8d3bc4 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -77,39 +77,7 @@
  *	  a bit more memory to the oldest subtransactions, because it's likely
  *	  they are the source for the next sequence of changes.
  *
- *	  When decoding sequences, we differentiate between a sequences created
- *	  in a (running) transaction, and sequences created in other (already
- *	  committed) transactions. Changes for sequences created in the same
- *	  top-level transaction are treated as "transactional" i.e. just like
- *	  any other change from that transaction (and discarded in case of a
- *	  rollback). Changes for sequences created earlier are treated as not
- *	  transactional - are processed immediately, as if performed outside
- *	  any transaction (and thus not rolled back).
- *
- *	  This mixed behavior is necessary - sequences are non-transactional
- *	  (e.g. ROLLBACK does not undo the sequence increments). But for new
- *	  sequences, we need to handle them in a transactional way, because if
- *	  we ever get some DDL support, the sequence won't exist until the
- *	  transaction gets applied. So we need to ensure the increments don't
- *	  happen until the sequence gets created.
- *
- *	  To differentiate which sequences are "old" and which were created
- *	  in a still-running transaction, we track sequences created in running
- *	  transactions in a hash table. Sequences are identified by relfilenode,
- *	  and we track XID of the (sub)transaction that created it. This means
- *	  that if a transaction does something that changes the relfilenode
- *	  (like an alter / reset of a sequence), the new relfilenode will be
- *	  treated as if created in the transaction. The list of sequences gets
- *	  discarded when the transaction completes (commit/rollback).
- *
- *	  We don't use the XID to check if it's the same top-level transaction.
- *	  It's enough to know it was created in an in-progress transaction,
- *	  and we know it must be the current one because otherwise it wouldn't
- *	  see the sequence object.
- *
- *	  The XID may be valid even for non-transactional sequences - we simply
- *	  keep the XID logged to WAL, it's up to the reorderbuffer to decide if
- *	  the increment is transactional.
+ *	  FIXME
  *
  * -------------------------------------------------------------------------
  */
@@ -151,13 +119,6 @@ typedef struct ReorderBufferTXNByIdEnt
 	ReorderBufferTXN *txn;
 } ReorderBufferTXNByIdEnt;
 
-/* entry for hash table we use to track sequences created in running xacts */
-typedef struct ReorderBufferSequenceEnt
-{
-	RelFileNode		rnode;
-	TransactionId	xid;
-} ReorderBufferSequenceEnt;
-
 /* data structures for (relfilenode, ctid) => (cmin, cmax) mapping */
 typedef struct ReorderBufferTupleCidKey
 {
@@ -388,14 +349,6 @@ ReorderBufferAllocate(void)
 	buffer->by_txn = hash_create("ReorderBufferByXid", 1000, &hash_ctl,
 								 HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
 
-	/* hash table of sequences, mapping relfilenode to XID of transaction */
-	hash_ctl.keysize = sizeof(RelFileNode);
-	hash_ctl.entrysize = sizeof(ReorderBufferSequenceEnt);
-	hash_ctl.hcxt = buffer->context;
-
-	buffer->sequences = hash_create("ReorderBufferSequenceHash", 1000, &hash_ctl,
-								 HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
-
 	buffer->by_txn_last_xid = InvalidTransactionId;
 	buffer->by_txn_last_txn = NULL;
 
@@ -582,17 +535,11 @@ 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:
 		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
+		case REORDER_BUFFER_CHANGE_SEQUENCE:
 			break;
 	}
 
@@ -923,57 +870,6 @@ ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
 	}
 }
 
-/*
- * Treat the sequence increment as transactional?
- *
- * The hash table tracks all sequences created in in-progress transactions,
- * 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
-ReorderBufferSequenceIsTransactional(ReorderBuffer *rb,
-									 RelFileNode rnode, bool created)
-{
-	bool	found = false;
-
-	if (created)
-		return true;
-
-	hash_search(rb->sequences,
-				(void *) &rnode,
-				HASH_FIND,
-				&found);
-
-	return 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;
-
-		(void) hash_search(rb->sequences,
-					   (void *) &(ent->rnode),
-					   HASH_REMOVE, NULL);
-	}
-}
-
 /*
  * A transactional sequence increment is queued to be processed upon commit
  * and a non-transactional increment gets processed immediately.
@@ -985,166 +881,32 @@ ReorderBufferSequenceCleanup(ReorderBuffer *rb, TransactionId xid)
  */
 void
 ReorderBufferQueueSequence(ReorderBuffer *rb, TransactionId xid,
-						   Snapshot snapshot, XLogRecPtr lsn, RepOriginId origin_id,
-						   RelFileNode rnode, bool transactional, bool created,
-						   ReorderBufferTupleBuf *tuplebuf)
+						   XLogRecPtr lsn, RepOriginId origin_id,
+						   Oid reloid, RelFileNode rnode,
+						   int64 last, int64 log_cnt, bool is_called)
 {
-	/*
-	 * 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;
-
-		/* XXX Maybe check that we're still in the same top-level xact? */
-
-		/* OK, allocate and queue the change */
-		oldcontext = MemoryContextSwitchTo(rb->context);
-
-		change = ReorderBufferGetChange(rb);
-
-		change->action = REORDER_BUFFER_CHANGE_SEQUENCE;
-		change->origin_id = origin_id;
-
-		memcpy(&change->data.sequence.relnode, &rnode, sizeof(RelFileNode));
-
-		change->data.sequence.tuple = tuplebuf;
-
-		/* add it to the same subxact that created the sequence */
-		ReorderBufferQueueChange(rb, ent->xid, lsn, change, false);
-
-		MemoryContextSwitchTo(oldcontext);
-	}
-	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));
-
-			relation = RelationIdGetRelation(reloid);
-			tuple = &tuplebuf->tuple;
-			seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
+	MemoryContext oldcontext;
+	ReorderBufferChange *change;
 
-			rb->sequence(rb, txn, lsn, relation, transactional,
-						 seq->last_value, seq->log_cnt, seq->is_called);
+	/* OK, allocate and queue the change */
+	oldcontext = MemoryContextSwitchTo(rb->context);
 
-			RelationClose(relation);
+	change = ReorderBufferGetChange(rb);
 
-			TeardownHistoricSnapshot(false);
+	change->action = REORDER_BUFFER_CHANGE_SEQUENCE;
+	change->origin_id = origin_id;
 
-			AbortCurrentTransaction();
+	memcpy(&change->data.sequence.relnode, &rnode, sizeof(RelFileNode));
 
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
-		}
-		PG_CATCH();
-		{
-			TeardownHistoricSnapshot(true);
+	change->data.sequence.reloid = reloid;
+	change->data.sequence.last = last;
+	change->data.sequence.log_cnt = log_cnt;
+	change->data.sequence.is_called = is_called;
 
-			AbortCurrentTransaction();
-
-			if (using_subtxn)
-				RollbackAndReleaseCurrentSubTransaction();
+	/* add it to the same subxact that created the sequence */
+	ReorderBufferQueueChange(rb, xid, lsn, change, false);
 
-			PG_RE_THROW();
-		}
-		PG_END_TRY();
-	}
+	MemoryContextSwitchTo(oldcontext);
 }
 
 /*
@@ -1823,9 +1585,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);
@@ -2249,19 +2008,23 @@ ReorderBufferApplySequence(ReorderBuffer *rb, ReorderBufferTXN *txn,
 						   Relation relation, ReorderBufferChange *change,
 						   bool streaming)
 {
-	HeapTuple	tuple;
-	Form_pg_sequence_data seq;
-
-	tuple = &change->data.sequence.tuple->tuple;
-	seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
+	int64		last_value, log_cnt;
+	bool		is_called;
+/*
+ * FIXME is it possible that we write sequence state for T1 before T2, but
+ * then end up committing T2 first? If the sequence could be incremented
+ * in between, that might cause data corruption result.
+ */
+	last_value = change->data.sequence.last;
+	log_cnt = change->data.sequence.log_cnt;
+	is_called = change->data.sequence.is_called;
 
-	/* Only ever called from ReorderBufferApplySequence, so transational. */
 	if (streaming)
-		rb->stream_sequence(rb, txn, change->lsn, relation, true,
-							seq->last_value, seq->log_cnt, seq->is_called);
+		rb->stream_sequence(rb, txn, change->lsn, relation,
+							last_value, log_cnt, is_called);
 	else
-		rb->sequence(rb, txn, change->lsn, relation, true,
-					 seq->last_value, seq->log_cnt, seq->is_called);
+		rb->sequence(rb, txn, change->lsn, relation,
+					 last_value, log_cnt, is_called);
 }
 
 /*
@@ -2710,15 +2473,17 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				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",
+						elog(ERROR, "zz could not map filenode \"%s\" to relation OID",
 							 relpathperm(change->data.sequence.relnode,
 										 MAIN_FORKNUM));
 
-					relation = RelationIdGetRelation(reloid);
+					*/
+					relation = RelationIdGetRelation(change->data.sequence.reloid);
 
 					if (!RelationIsValid(relation))
 						elog(ERROR, "could not open relation with OID %u (for filenode \"%s\")",
@@ -4115,45 +3880,13 @@ 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:
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
 		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
 		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
+		case REORDER_BUFFER_CHANGE_SEQUENCE:
 			/* ReorderBufferChange contains everything important */
 			break;
 	}
@@ -4412,28 +4145,13 @@ 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:
 		case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT:
 		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
 		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
+		case REORDER_BUFFER_CHANGE_SEQUENCE:
 			/* ReorderBufferChange contains everything important */
 			break;
 	}
@@ -4729,34 +4447,11 @@ 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:
 		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
+		case REORDER_BUFFER_CHANGE_SEQUENCE:
 			break;
 	}
 
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 697fb23634c..d6adbd6e46d 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1220,8 +1220,7 @@ copy_sequence(Relation rel)
 
 	fetch_sequence_data(lrel.nspname, lrel.relname, &last_value, &log_cnt, &is_called);
 
-	/* tablesync sets the sequences in non-transactional way */
-	SetSequence(RelationGetRelid(rel), false, last_value, log_cnt, is_called);
+	SetSequence(RelationGetRelid(rel), last_value, log_cnt, is_called);
 
 	logicalrep_rel_close(relmapentry, NoLock);
 }
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f3868b3e1f8..fbc65f277ea 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1159,12 +1159,9 @@ apply_handle_sequence(StringInfo s)
 	logicalrep_read_sequence(s, &seq);
 
 	/*
-	 * Non-transactional sequence updates should not be part of a remote
-	 * transaction. There should not be any running transaction.
+	 * We should be in remote transaction.
 	 */
-	Assert((!seq.transactional) || in_remote_transaction);
-	Assert(!(!seq.transactional && in_remote_transaction));
-	Assert(!(!seq.transactional && IsTransactionState()));
+	Assert(in_remote_transaction);
 
 	/*
 	 * Make sure we're in a transaction (needed by SetSequence). For
@@ -1185,14 +1182,7 @@ apply_handle_sequence(StringInfo s)
 	LockRelationOid(relid, AccessExclusiveLock);
 
 	/* apply the sequence change */
-	SetSequence(relid, seq.transactional, seq.last_value, seq.log_cnt, seq.is_called);
-
-	/*
-	 * Commit the per-stream transaction (we only do this when not in
-	 * remote transaction, i.e. for non-transactional sequence updates.
-	 */
-	if (!in_remote_transaction)
-		CommitTransactionCommand();
+	SetSequence(relid, seq.last_value, seq.log_cnt, seq.is_called);
 }
 
 /*
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 9d33630464c..31be8064ebe 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -57,7 +57,7 @@ static void pgoutput_message(LogicalDecodingContext *ctx,
 							 Size sz, const char *message);
 static void pgoutput_sequence(LogicalDecodingContext *ctx,
 							  ReorderBufferTXN *txn, XLogRecPtr sequence_lsn,
-							  Relation relation, bool transactional,
+							  Relation relation,
 							  int64 last_value, int64 log_cnt, bool is_called);
 static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
 								   RepOriginId origin_id);
@@ -1712,12 +1712,13 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 static void
 pgoutput_sequence(LogicalDecodingContext *ctx,
 				  ReorderBufferTXN *txn, XLogRecPtr sequence_lsn,
-				  Relation relation, bool transactional,
+				  Relation relation,
 				  int64 last_value, int64 log_cnt, bool is_called)
 {
 	PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
 	TransactionId xid = InvalidTransactionId;
 	RelationSyncEntry *relentry;
+	PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
 
 	if (!data->sequences)
 		return;
@@ -1742,25 +1743,15 @@ pgoutput_sequence(LogicalDecodingContext *ctx,
 	if (!relentry->pubactions.pubsequence)
 		return;
 
-	/*
-	 * Output BEGIN if we haven't yet. Avoid for non-transactional
-	 * sequence changes.
-	 */
-	if (transactional)
-	{
-		PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
-
-		/* Send BEGIN if we haven't yet */
-		if (txndata && !txndata->sent_begin_txn)
-			pgoutput_send_begin(ctx, txn);
-	}
+	/* Send BEGIN if we haven't yet */
+	if (txndata && !txndata->sent_begin_txn)
+		pgoutput_send_begin(ctx, txn);
 
 	OutputPluginPrepareWrite(ctx, true);
 	logicalrep_write_sequence(ctx->out,
 							  relation,
 							  xid,
 							  sequence_lsn,
-							  transactional,
 							  last_value,
 							  log_cnt,
 							  is_called);
diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h
index cf8b6d48193..9a74721c97c 100644
--- a/src/include/access/rmgrlist.h
+++ b/src/include/access/rmgrlist.h
@@ -40,7 +40,7 @@ PG_RMGR(RM_BTREE_ID, "Btree", btree_redo, btree_desc, btree_identify, btree_xlog
 PG_RMGR(RM_HASH_ID, "Hash", hash_redo, hash_desc, hash_identify, NULL, NULL, hash_mask, NULL)
 PG_RMGR(RM_GIN_ID, "Gin", gin_redo, gin_desc, gin_identify, gin_xlog_startup, gin_xlog_cleanup, gin_mask, NULL)
 PG_RMGR(RM_GIST_ID, "Gist", gist_redo, gist_desc, gist_identify, gist_xlog_startup, gist_xlog_cleanup, gist_mask, NULL)
-PG_RMGR(RM_SEQ_ID, "Sequence", seq_redo, seq_desc, seq_identify, NULL, NULL, seq_mask, sequence_decode)
+PG_RMGR(RM_SEQ_ID, "Sequence", seq_redo, seq_desc, seq_identify, NULL, NULL, seq_mask, NULL)
 PG_RMGR(RM_SPGIST_ID, "SPGist", spg_redo, spg_desc, spg_identify, spg_xlog_startup, spg_xlog_cleanup, spg_mask, NULL)
 PG_RMGR(RM_BRIN_ID, "BRIN", brin_redo, brin_desc, brin_identify, NULL, NULL, brin_mask, NULL)
 PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_identify, NULL, NULL, NULL, NULL)
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 5bab90db8e0..0bc90c922ba 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -48,7 +48,6 @@ typedef FormData_pg_sequence_data *Form_pg_sequence_data;
 typedef struct xl_seq_rec
 {
 	RelFileNode node;
-	bool		created;	/* creates a new relfilenode (CREATE/ALTER) */
 	/* SEQUENCE TUPLE DATA FOLLOWS AT THE END */
 } xl_seq_rec;
 
@@ -60,9 +59,11 @@ extern ObjectAddress DefineSequence(ParseState *pstate, CreateSeqStmt *stmt);
 extern ObjectAddress AlterSequence(ParseState *pstate, AlterSeqStmt *stmt);
 extern void DeleteSequenceTuple(Oid relid);
 extern void ResetSequence(Oid seq_relid);
-extern void SetSequence(Oid seq_relid, bool transactional, int64 last_value, int64 log_cnt, bool is_called);
+extern void SetSequence(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called);
 extern void ResetSequenceCaches(void);
 
+extern void AtEOXact_Sequences(bool isCommit);
+
 extern void seq_redo(XLogReaderState *rptr);
 extern void seq_desc(StringInfo buf, XLogReaderState *rptr);
 extern const char *seq_identify(uint8 info);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 13ee10fdd4e..a466db394b8 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -125,7 +125,6 @@ typedef struct LogicalRepSequence
 	Oid			remoteid;		/* unique id of the remote sequence */
 	char	   *nspname;		/* schema name of remote sequence */
 	char	   *seqname;		/* name of the remote sequence */
-	bool		transactional;
 	int64		last_value;
 	int64		log_cnt;
 	bool		is_called;
@@ -245,7 +244,6 @@ extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecP
 									 bool transactional, const char *prefix, Size sz, const char *message);
 extern void logicalrep_write_sequence(StringInfo out, Relation rel,
 									  TransactionId xid, XLogRecPtr lsn,
-									  bool transactional,
 									  int64 last_value, int64 log_cnt,
 									  bool is_called);
 extern void logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata);
diff --git a/src/include/replication/message.h b/src/include/replication/message.h
index 7d7785292f1..a084acc06ab 100644
--- a/src/include/replication/message.h
+++ b/src/include/replication/message.h
@@ -27,13 +27,28 @@ typedef struct xl_logical_message
 	char		message[FLEXIBLE_ARRAY_MEMBER];
 } xl_logical_message;
 
+/*
+ * Generic logical decoding sequence wal record.
+ */
+typedef struct xl_logical_sequence
+{
+	RelFileNode	node;
+	Oid			reloid;
+	int64		last;			/* last value emitted for sequence */
+	int64		log_cnt;		/* last value cached for sequence */
+	bool		is_called;
+} xl_logical_sequence;
+
 #define SizeOfLogicalMessage	(offsetof(xl_logical_message, message))
+#define SizeOfLogicalSequence	(sizeof(xl_logical_sequence))
 
 extern XLogRecPtr LogLogicalMessage(const char *prefix, const char *message,
 									size_t size, bool transactional);
 
 /* RMGR API*/
 #define XLOG_LOGICAL_MESSAGE	0x00
+#define XLOG_LOGICAL_SEQUENCE	0x10
+
 void		logicalmsg_redo(XLogReaderState *record);
 void		logicalmsg_desc(StringInfo buf, XLogReaderState *record);
 const char *logicalmsg_identify(uint8 info);
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index fe85d49a030..abc05644516 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -95,7 +95,6 @@ typedef void (*LogicalDecodeSequenceCB) (struct LogicalDecodingContext *ctx,
 										 ReorderBufferTXN *txn,
 										 XLogRecPtr sequence_lsn,
 										 Relation rel,
-										 bool transactional,
 										 int64 last_value,
 										 int64 log_cnt,
 										 bool is_called);
@@ -219,7 +218,6 @@ typedef void (*LogicalDecodeStreamSequenceCB) (struct LogicalDecodingContext *ct
 											   ReorderBufferTXN *txn,
 											   XLogRecPtr sequence_lsn,
 											   Relation rel,
-											   bool transactional,
 											   int64 last_value,
 											   int64 log_cnt,
 											   bool is_called);
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 0bcc150b331..260630c2ba5 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -164,7 +164,10 @@ typedef struct ReorderBufferChange
 		struct
 		{
 			RelFileNode relnode;
-			ReorderBufferTupleBuf *tuple;
+			Oid			reloid;
+			int64		last;
+			int64		log_cnt;
+			bool		is_called;
 		}			sequence;
 	}			data;
 
@@ -443,7 +446,6 @@ typedef void (*ReorderBufferSequenceCB) (ReorderBuffer *rb,
 										 ReorderBufferTXN *txn,
 										 XLogRecPtr sequence_lsn,
 										 Relation rel,
-										 bool transactional,
 										 int64 last_value, int64 log_cnt,
 										 bool is_called);
 
@@ -518,7 +520,6 @@ typedef void (*ReorderBufferStreamSequenceCB) (ReorderBuffer *rb,
 											   ReorderBufferTXN *txn,
 											   XLogRecPtr sequence_lsn,
 											   Relation rel,
-											   bool transactional,
 											   int64 last_value, int64 log_cnt,
 											   bool is_called);
 
@@ -670,9 +671,9 @@ 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,
-									   ReorderBufferTupleBuf *tuplebuf);
+									   XLogRecPtr lsn, RepOriginId origin_id,
+									   Oid reloid, RelFileNode rnode,
+									   int64 last, int64 log_cnt, bool is_called);
 void		ReorderBufferCommit(ReorderBuffer *, TransactionId,
 								XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
 								TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_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