Merge branch 'PGPROEE10_MULTIMASTER' of https://gitlab.postgrespro.ru/pgpro-dev/postgrespro into PGPROEE10_MULTIMASTER

Author: knizhnik

contrib/mmts/multimaster.c

@@ -290,7 +290,7 @@ 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 seq_nextval_hook_t PreviousSeqNextvalHook;
 
 static void MtmExecutorStart(QueryDesc *queryDesc, int eflags);
 static void MtmExecutorFinish(QueryDesc *queryDesc);
@@ -3390,8 +3390,8 @@ _PG_init(void)
 	PreviousProcessUtilityHook = ProcessUtility_hook;
 	ProcessUtility_hook = MtmProcessUtility;
 
-	// PreviousSeqNextvalHook = SeqNextvalHook;
-	// SeqNextvalHook = MtmSeqNextvalHook;
+	PreviousSeqNextvalHook = SeqNextvalHook;
+	SeqNextvalHook = MtmSeqNextvalHook;
 }
 
 /*
@@ -3403,7 +3403,7 @@ _PG_fini(void)
 	shmem_startup_hook = PreviousShmemStartupHook;
 	ExecutorFinish_hook = PreviousExecutorFinishHook;
 	ProcessUtility_hook = PreviousProcessUtilityHook;
-	// SeqNextvalHook = PreviousSeqNextvalHook;
+	SeqNextvalHook = PreviousSeqNextvalHook;
 }
 
 
@@ -5089,7 +5089,33 @@ static bool MtmFunctionProfileDependsOnTempTable(CreateFunctionStmt* func)
 	return false;
 }
 
+static void
+AdjustCreateSequence(List *options)
+{
+	bool has_increment = false, has_start = false;
+	ListCell   *option;
+
+	foreach(option, options)
+	{
+		DefElem    *defel = (DefElem *) lfirst(option);
+		if (strcmp(defel->defname, "increment") == 0)
+			has_increment = true;
+		else if (strcmp(defel->defname, "start") == 0)
+			has_start = true;
+	}
 
+	if (!has_increment)
+	{
+		DefElem *defel = makeDefElem("increment", (Node *) makeInteger(MtmMaxNodes), -1);
+		options = lappend(options, defel);
+	}
+
+	if (!has_start)
+	{
+		DefElem *defel = makeDefElem("start", (Node *) makeInteger(MtmNodeId), -1);
+		options = lappend(options, defel);
+	}
+}
 
 static void MtmProcessUtility(PlannedStmt *pstmt,
 										  const char *queryString, ProcessUtilityContext context,
@@ -5162,6 +5188,13 @@ static void MtmProcessUtility(PlannedStmt *pstmt,
 			elog(ERROR, "Multimaster doesn't support creating and dropping databases");
 			break;
 
+		case T_CreateSeqStmt:
+			{
+				CreateSeqStmt *stmt = (CreateSeqStmt *) parsetree;
+				AdjustCreateSequence(stmt->options);
+			}
+			break;
+
 		case T_CreateTableSpaceStmt:
 		case T_DropTableSpaceStmt:
 			{

contrib/mmts/pglogical_apply.c

@@ -1274,18 +1274,18 @@ void MtmExecutor(void* work, size_t size)
 				s.len = save_len;
 				break;
 			}
-			// case 'N':
-			// {
-			// 	int64 next;
-			// 	Oid relid;
-			// 	Assert(rel != NULL);
-			// 	relid = RelationGetRelid(rel);
-			// 	close_rel(rel);
-			// 	rel = NULL;
-			// 	next = pq_getmsgint64(&s);
-			// 	AdjustSequence(relid, next);
-			// 	break;
-			// }
+			case 'N':
+			{
+				int64 next;
+				Oid relid;
+				Assert(rel != NULL);
+				relid = RelationGetRelid(rel);
+				close_rel(rel);
+				rel = NULL;
+				next = pq_getmsgint64(&s);
+				AdjustSequence(relid, next);
+				break;
+			}
 		    case '0':
 			    Assert(rel != NULL);
 			    heap_truncate_one_rel(rel);

contrib/mmts/tests/reinit-mm.sh

@@ -77,6 +77,7 @@ do
         multimaster.arbiter_port = $arbiter_port
         multimaster.max_recovery_lag = 30GB
         multimaster.referee_connstring = 'dbname=$USER host=127.0.0.1 port=5440 sslmode=disable'
+        multimaster.monotonic_sequences = on
 SQL
 
     cat <<CONF >> tmp_check/node$i/pg_hba.conf

src/backend/commands/sequence.c

@@ -75,6 +75,8 @@ typedef struct SeqTableData
 
 typedef SeqTableData *SeqTable;
 
+seq_nextval_hook_t SeqNextvalHook;
+
 static HTAB *seqhashtab = NULL; /* hash table for SeqTable items */
 
 /*
@@ -614,6 +616,10 @@ nextval_internal(Oid relid, bool check_permissions)
 		elm->last += elm->increment;
 		relation_close(seqrel, NoLock);
 		last_used_seq = elm;
+
+		if (SeqNextvalHook)
+			SeqNextvalHook(relid, elm->last);
+
 		return elm->last;
 	}
 
@@ -810,6 +816,9 @@ nextval_internal(Oid relid, bool check_permissions)
 
 	relation_close(seqrel, NoLock);
 
+	if (SeqNextvalHook)
+		SeqNextvalHook(relid, result);
+
 	return result;
 }
 
@@ -879,6 +888,141 @@ lastval(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(result);
 }
 
+/*
+ * Bump last value to next iff next > value.
+ * Support routine for multimaster's monotonic sequences.
+ */
+void
+AdjustSequence(Oid relid, int64 next)
+{
+	SeqTable	elm;
+	Relation	seqrel;
+	Buffer		buf;
+	HeapTupleData seqdatatuple;
+	Form_pg_sequence_data seq;
+	HeapTuple	pgstuple;
+	Form_pg_sequence pgsform;
+	int64		maxv,
+				minv,
+				incby,
+				cache;
+	int64		last;
+
+	/* open and lock sequence */
+	init_sequence(relid, &elm, &seqrel);
+
+	if (pg_class_aclcheck(elm->relid, GetUserId(), ACL_UPDATE) != ACLCHECK_OK)
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+				 errmsg("permission denied for sequence %s",
+						RelationGetRelationName(seqrel))));
+
+	pgstuple = SearchSysCache1(SEQRELID, ObjectIdGetDatum(relid));
+	if (!HeapTupleIsValid(pgstuple))
+		elog(ERROR, "cache lookup failed for sequence %u", relid);
+	pgsform = (Form_pg_sequence) GETSTRUCT(pgstuple);
+	maxv = pgsform->seqmax;
+	minv = pgsform->seqmin;
+	incby = pgsform->seqincrement;
+	cache = pgsform->seqcache;
+	ReleaseSysCache(pgstuple);
+
+	/* cached number is greater than received */
+	if (elm->last != cache && elm->last + incby > next)
+	{
+		relation_close(seqrel, NoLock);
+		return;
+	}
+
+	/* read-only transactions may only modify temp sequences */
+	if (!seqrel->rd_islocaltemp)
+		PreventCommandIfReadOnly("setval()");
+
+	/*
+	 * Forbid this during parallel operation because, to make it work, the
+	 * cooperating backends would need to share the backend-local cached
+	 * sequence information.  Currently, we don't support that.
+	 */
+	PreventCommandIfParallelMode("setval()");
+
+	/* lock page' buffer and read tuple */
+	seq = read_seq_tuple(seqrel, &buf, &seqdatatuple);
+
+	if ((next < minv) || (next > maxv))
+	{
+		char		bufv[100],
+					bufm[100],
+					bufx[100];
+
+		snprintf(bufv, sizeof(bufv), INT64_FORMAT, next);
+		snprintf(bufm, sizeof(bufm), INT64_FORMAT, minv);
+		snprintf(bufx, sizeof(bufx), INT64_FORMAT, maxv);
+		ereport(ERROR,
+				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+				 errmsg("setval: value %s is out of bounds for sequence \"%s\" (%s..%s)",
+						bufv, RelationGetRelationName(seqrel),
+						bufm, bufx)));
+	}
+
+	last = seq->last_value;
+	if (seq->is_called)
+	{
+		last += incby;
+	}
+	if (last <= next)
+	{
+		next = last + incby*((next - last + incby)/incby);
+
+		/* 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();
+
+		/* ready to change the on-disk (or really, in-buffer) tuple */
+		ptrack_add_block(seqrel, BufferGetBlockNumber(buf));
+		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 *) seqdatatuple.t_data, seqdatatuple.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,4 +71,9 @@ extern void seq_desc(StringInfo buf, XLogReaderState *rptr);
 extern const char *seq_identify(uint8 info);
 extern void seq_mask(char *pagedata, BlockNumber blkno);
 
+typedef void (*seq_nextval_hook_t)(Oid seq_relid, int64 next);
+extern seq_nextval_hook_t SeqNextvalHook;
+
+extern void AdjustSequence(Oid relid, int64 next);
+
 #endif							/* SEQUENCE_H */