Support monotonic sequences

Author: knizhnik

contrib/mmts/multimaster.c

@@ -47,6 +47,7 @@
 #include "utils/memutils.h"
 #include "commands/dbcommands.h"
 #include "commands/extension.h"
+#include "commands/sequence.h"
 #include "postmaster/autovacuum.h"
 #include "storage/pmsignal.h"
 #include "storage/proc.h"
@@ -266,11 +267,13 @@ static bool	 MtmBreakConnection;
 static bool	 MtmClusterLocked;
 static bool	 MtmInsideTransaction;
 static bool  MtmReferee;
+static bool  MtmMonotonicSequences;
 
 static ExecutorStart_hook_type PreviousExecutorStartHook;
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
 static ProcessUtility_hook_type PreviousProcessUtilityHook;
 static shmem_startup_hook_type PreviousShmemStartupHook;
+static seq_nextval_hook_t PreviousSeqNextvalHook;
 
 static nodemask_t lastKnownMatrix[MAX_NODES];
 
@@ -279,6 +282,7 @@ static void MtmExecutorFinish(QueryDesc *queryDesc);
 static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 ProcessUtilityContext context, ParamListInfo params,
 							 DestReceiver *dest, char *completionTag);
+static void MtmSeqNextvalHook(Oid seqid, int64 next);
 
 static bool MtmAtExitHookRegistered = false;
 
@@ -3136,6 +3140,19 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomBoolVariable(
+		"multimaster.monotonic_sequences",
+		"Enforce monotinic behaviour of sequence values obtained from different nodes",
+		NULL,
+		&MtmMonotonicSequences,
+		false,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomBoolVariable(
 		"multimaster.ignore_tables_without_pk",
 		"Do not replicate tables without primary key",
@@ -3390,6 +3407,9 @@ _PG_init(void)
 
 	PreviousProcessUtilityHook = ProcessUtility_hook;
 	ProcessUtility_hook = MtmProcessUtility;
+
+	PreviousSeqNextvalHook = SeqNextvalHook; 
+	SeqNextvalHook = MtmSeqNextvalHook;
 }
 
 /*
@@ -3401,6 +3421,7 @@ _PG_fini(void)
 	shmem_startup_hook = PreviousShmemStartupHook;
 	ExecutorFinish_hook = PreviousExecutorFinishHook;
 	ProcessUtility_hook = PreviousProcessUtilityHook;
+	SeqNextvalHook = PreviousSeqNextvalHook;
 }
 
 
@@ -5338,6 +5359,17 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 	}
 }
 
+static void MtmSeqNextvalHook(Oid seqid, int64 next)
+{
+	if (MtmMonotonicSequences)
+	{
+		MtmSeqPosition pos;
+		pos.seqid = seqid;
+		pos.next = next;
+		LogLogicalMessage("N", (char*)&pos, sizeof(pos), true);		
+	}
+}
+
 /*
  * -------------------------------------------
  * Executor pool interface

contrib/mmts/multimaster.h

@@ -337,6 +337,12 @@ typedef struct MtmFlushPosition
 } MtmFlushPosition;
 
 
+typedef struct MtmSeqPosition
+{
+	Oid   seqid;
+	int64 next;
+} MtmSeqPosition;
+ 
 #define MtmIsCoordinator(ts) (ts->gtid.node == MtmNodeId)
 
 extern char const* const MtmNodeStatusMnem[];

contrib/mmts/pglogical_apply.c

@@ -22,6 +22,7 @@
 #include "commands/vacuum.h"
 #include "commands/tablespace.h"
 #include "commands/defrem.h"
+#include "commands/sequence.h"
 #include "parser/parse_utilcmd.h"
 
 #include "libpq/pqformat.h"
@@ -1157,7 +1158,18 @@ void MtmExecutor(void* work, size_t size)
 				s.len = save_len;
 				break;
 			}
+ 		    case 'N':
+			{
+				int64 next;
+			    Assert(rel != NULL);
+				next = pq_getmsgint64(&s); 
+				AdjustSequence(RelationGetRelid(rel), next);
+  			    close_rel(rel);
+				rel = NULL;
+				break;
+			}			   
 		    case '0':
+			    Assert(rel != NULL);
 			    heap_truncate_one_rel(rel);
 				break;
 			case 'M':

contrib/mmts/pglogical_proto.c

@@ -168,6 +168,17 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 	}
 }
 
+
+static void pglogical_seq_nextval(StringInfo out, LogicalDecodingContext *ctx, MtmSeqPosition* pos)
+{
+	Relation rel = heap_open(pos->seqid, NoLock);
+	pglogical_write_rel(out, ctx->output_plugin_private, rel);
+	heap_close(rel, NoLock);
+	pq_sendbyte(out, 'N');
+	pq_sendint64(out, pos->next);
+}
+	
+
 static void pglogical_broadcast_table(StringInfo out, LogicalDecodingContext *ctx, MtmCopyRequest* copy)
 {
 	if (BIT_CHECK(copy->targetNodes, MtmReplicationNodeId-1)) { 
@@ -229,6 +240,9 @@ pglogical_write_message(StringInfo out, LogicalDecodingContext *ctx,
 	  case 'B':
 		pglogical_broadcast_table(out, ctx, (MtmCopyRequest*)message);
 		return;
+	  case 'N':
+		pglogical_seq_nextval(out, ctx, (MtmSeqPosition*)message);
+		return;		
 	}
 	pq_sendbyte(out, 'M');
 	pq_sendbyte(out, *prefix);

src/backend/commands/sequence.c

@@ -80,6 +80,8 @@ typedef struct SeqTableData
 
 typedef SeqTableData *SeqTable;
 
+seq_nextval_hook_t SeqNextvalHook;
+
 static HTAB *seqhashtab = NULL; /* hash table for SeqTable items */
 
 /*
@@ -585,6 +587,10 @@ nextval_internal(Oid relid)
 		elm->last += elm->increment;
 		relation_close(seqrel, NoLock);
 		last_used_seq = elm;
+
+		if (SeqNextvalHook)
+			SeqNextvalHook(relid, elm->last);
+
 		return elm->last;
 	}
 
@@ -771,6 +777,9 @@ nextval_internal(Oid relid)
 
 	relation_close(seqrel, NoLock);
 
+	if (SeqNextvalHook)
+		SeqNextvalHook(relid, result);
+
 	return result;
 }
 
@@ -840,6 +849,87 @@ lastval(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(result);
 }
 
+void AdjustSequence(Oid relid, int64 next)
+{
+	SeqTable	elm;
+	Relation	seqrel;
+	Buffer		buf;
+	int64       last;
+	HeapTupleData seqtuple;
+	Form_pg_sequence seq;
+
+	/* open and AccessShareLock sequence */
+	init_sequence(relid, &elm, &seqrel);
+
+	/* lock page' buffer and read tuple */
+	seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
+
+	if (elm->last != elm->cached && elm->last + elm->increment > next) /* cached number is greater than received */
+	{
+		relation_close(seqrel, NoLock);
+		return;
+	}
+
+	/* lock page' buffer and read tuple */
+	seq = read_seq_tuple(elm, seqrel, &buf, &seqtuple);
+
+	last = seq->last_value;
+	if (seq->is_called) 
+	{
+		last += elm->increment;
+	}
+	if (last <= next) 
+	{
+		Assert(next >= seq->min_value && next <= seq->max_value);
+	
+		/* Set the currval() state only if iscalled = true */
+		if (seq->is_called)
+		{
+			elm->last = next;		/* last returned number */
+			elm->last_valid = true;
+		}
+		
+		/* In any case, forget any future cached numbers */
+		elm->cached = elm->last;
+		
+		/* check the comment above nextval_internal()'s equivalent call. */
+		if (RelationNeedsWAL(seqrel))
+			GetTopTransactionId();
+		
+		START_CRIT_SECTION();
+
+		seq->last_value = next;		/* last fetched number */
+		seq->log_cnt = 0;
+		
+		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;
+			XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
+			XLogRegisterData((char *) seqtuple.t_data, seqtuple.t_len);
+			
+			recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
+			
+			PageSetLSN(page, recptr);
+		}
+		END_CRIT_SECTION();
+	}
+
+	UnlockReleaseBuffer(buf);
+
+	relation_close(seqrel, NoLock);
+}	
+		
+
 /*
  * Main internal procedure that handles 2 & 3 arg forms of SETVAL.
  *

src/include/commands/sequence.h

@@ -71,6 +71,8 @@ extern Datum setval_oid(PG_FUNCTION_ARGS);
 extern Datum setval3_oid(PG_FUNCTION_ARGS);
 extern Datum lastval(PG_FUNCTION_ARGS);
 
+extern void  AdjustSequence(Oid seqid, int64 next);
+
 extern Datum pg_sequence_parameters(PG_FUNCTION_ARGS);
 
 extern ObjectAddress DefineSequence(CreateSeqStmt *stmt);
@@ -82,4 +84,9 @@ extern void seq_redo(XLogReaderState *rptr);
 extern void seq_desc(StringInfo buf, XLogReaderState *rptr);
 extern const char *seq_identify(uint8 info);
 
+typedef void (*seq_nextval_hook_t)(Oid seq_relid, int64 next);
+
+extern seq_nextval_hook_t SeqNextvalHook;
+
+
 #endif   /* SEQUENCE_H */