Merge branch 'PGPROEE10_MULTIMASTER_stable' into PGPROEE10

Author: vbwagner

configure

@@ -2788,7 +2788,7 @@ fi
 
 
 
-PGPRO_VERSION="$PACKAGE_VERSION.2"
+PGPRO_VERSION="$PACKAGE_VERSION.3"
 PGPRO_PACKAGE_NAME="PostgresPro"
 PGPRO_EDITION="enterprise"
 PGPRO_EDN="ent"

configure.in

@@ -39,7 +39,7 @@ PGAC_ARG_REQ(with, extra-version, [STRING], [append STRING to version],
              [PG_VERSION="$PACKAGE_VERSION$withval"],
              [PG_VERSION="$PACKAGE_VERSION"])
 
-PGPRO_VERSION="$PACKAGE_VERSION.2"
+PGPRO_VERSION="$PACKAGE_VERSION.3"
 PGPRO_PACKAGE_NAME="PostgresPro"
 PGPRO_EDITION="enterprise"
 PGPRO_EDN="ent"

contrib/mmts/Cluster.pm

@@ -73,6 +73,7 @@ sub new
 	my $self = {
 		nodenum => $nodenum,
 		nodes => $nodes,
+		recv_timeout => 5,
 	};
 
 	bless $self, $class;

contrib/mmts/multimaster--1.0.sql

@@ -84,7 +84,7 @@ CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_make_table_local'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.broadcast_table(srcTable regclass, dstNodesMask bigint) RETURNS void
+CREATE FUNCTION mtm.broadcast_table(srcTable regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_broadcast_table'
 LANGUAGE C;
 

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);
@@ -602,14 +602,16 @@ csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, nodemask_
 		MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
 		if (ts != NULL) {
 			*participantsMask = ts->participantsMask;
-			/* If node is disables, then we are in a process of recovery of this node */
-			if (!ts->isLocal && BIT_CHECK(ts->participantsMask|Mtm->disabledNodeMask, nodeId-1)) {
+			if (!ts->isLocal)
 				snapshot = ts->snapshot;
-				Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession);
-			} else {
-				MTM_LOG1("Do not send transaction %s (%llu) to node %d participants mask %llx",
-						 ts->gid, (long64)ts->xid, nodeId, ts->participantsMask);
-			}
+			// /* If node is disables, then we are in a process of recovery of this node */
+			// if (!ts->isLocal && BIT_CHECK(ts->participantsMask|Mtm->disabledNodeMask, nodeId-1)) {
+			// 	snapshot = ts->snapshot;
+			// 	Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession);
+			// } else {
+			// 	MTM_LOG1("Do not send transaction %s (%llu) to node %d participants mask %llx",
+			// 			 ts->gid, (long64)ts->xid, nodeId, ts->participantsMask);
+			// }
 		}
 	}
 	MtmUnlock();
@@ -3389,8 +3391,8 @@ _PG_init(void)
 	PreviousProcessUtilityHook = ProcessUtility_hook;
 	ProcessUtility_hook = MtmProcessUtility;
 
-	// PreviousSeqNextvalHook = SeqNextvalHook;
-	// SeqNextvalHook = MtmSeqNextvalHook;
+	PreviousSeqNextvalHook = SeqNextvalHook;
+	SeqNextvalHook = MtmSeqNextvalHook;
 }
 
 /*
@@ -3402,7 +3404,7 @@ _PG_fini(void)
 	shmem_startup_hook = PreviousShmemStartupHook;
 	ExecutorFinish_hook = PreviousExecutorFinishHook;
 	ProcessUtility_hook = PreviousProcessUtilityHook;
-	// SeqNextvalHook = PreviousSeqNextvalHook;
+	SeqNextvalHook = PreviousSeqNextvalHook;
 }
 
 
@@ -4462,7 +4464,7 @@ Datum mtm_broadcast_table(PG_FUNCTION_ARGS)
 {
 	MtmCopyRequest copy;
 	copy.sourceTable = PG_GETARG_OID(0);
-	copy.targetNodes = PG_GETARG_INT64(1);
+	copy.targetNodes = ~Mtm->disabledNodeMask;
 	LogLogicalMessage("B", (char*)&copy, sizeof(copy), true);
 	MtmTx.containsDML = true;
 	PG_RETURN_VOID();
@@ -4724,7 +4726,7 @@ static void
 MtmGenerateGid(char* gid)
 {
 	static int localCount;
-	sprintf(gid, "MTM-%d-%d-%d-%ld", MtmNodeId, MyProcPid, ++localCount, (int64) GetCurrentTimestamp());
+	sprintf(gid, "MTM-%d-%d-%d-" INT64_FORMAT, MtmNodeId, MyProcPid, ++localCount, (int64) GetCurrentTimestamp());
 }
 
 /*
@@ -5088,7 +5090,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,
@@ -5161,6 +5189,14 @@ static void MtmProcessUtility(PlannedStmt *pstmt,
 			elog(ERROR, "Multimaster doesn't support creating and dropping databases");
 			break;
 
+		case T_CreateSeqStmt:
+			{
+				CreateSeqStmt *stmt = (CreateSeqStmt *) parsetree;
+				if (!MtmVolksWagenMode)
+					AdjustCreateSequence(stmt->options);
+			}
+			break;
+
 		case T_CreateTableSpaceStmt:
 		case T_DropTableSpaceStmt:
 			{
@@ -5453,11 +5489,9 @@ static void MtmProcessUtility(PlannedStmt *pstmt,
 		if (relid != InvalidOid) {
 			Oid constraint_oid;
 			Bitmapset* pk = get_primary_key_attnos(relid, true, &constraint_oid);
-			if (pk == NULL && !MtmVolksWagenMode) {
+			if (pk == NULL && !MtmVolksWagenMode && MtmIgnoreTablesWithoutPk) {
 				elog(WARNING,
-					 MtmIgnoreTablesWithoutPk
-					 ? "Table %s.%s without primary will not be replicated"
-					 : "Updates and deletes of table %s.%s without primary will not be replicated",
+					 "Table %s.%s without primary will not be replicated",
 					 create->relation->schemaname ? create->relation->schemaname : "public",
 					 create->relation->relname);
 			}