Support monotonic sequences

knizhnik · kelvich · commit 76741a11cf49 · 2017-11-13T02:21:36.000+03:00
diff --git a/multimaster.c b/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"
@@ -267,11 +268,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];
 
@@ -280,6 +283,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;
 
@@ -3137,6 +3141,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",
@@ -3404,6 +3421,9 @@ _PG_init(void)
 
 	PreviousProcessUtilityHook = ProcessUtility_hook;
 	ProcessUtility_hook = MtmProcessUtility;
+
+	PreviousSeqNextvalHook = SeqNextvalHook; 
+	SeqNextvalHook = MtmSeqNextvalHook;
 }
 
 /*
@@ -3415,6 +3435,7 @@ _PG_fini(void)
 	shmem_startup_hook = PreviousShmemStartupHook;
 	ExecutorFinish_hook = PreviousExecutorFinishHook;
 	ProcessUtility_hook = PreviousProcessUtilityHook;
+	SeqNextvalHook = PreviousSeqNextvalHook;
 }
 
 
@@ -5352,6 +5373,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
diff --git a/multimaster.h b/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[];
diff --git a/pglogical_apply.c b/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':
diff --git a/pglogical_proto.c b/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);
