merge guc serializatuion fix

Author: knizhnik

contrib/mmts/arbiter.c

@@ -715,6 +715,8 @@ static void MtmSender(Datum arg)
 	int nNodes = MtmMaxNodes;
 	int i;
 
+	MtmBackgroundWorker = true;
+
 	MtmBuffer* txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
 	MTM_ELOG(LOG, "Start arbiter sender %d", MyProcPid);
 	InitializeTimeouts();
@@ -802,6 +804,8 @@ static void MtmMonitor(Datum arg)
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
 
+	MtmBackgroundWorker = true;
+
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();
 
@@ -838,7 +842,9 @@ static void MtmReceiver(Datum arg)
 	pqsignal(SIGINT, SetStop);
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
-	
+
+	MtmBackgroundWorker = true;
+
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();
 

contrib/mmts/bgwpool.c

@@ -36,6 +36,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
 
 	MTM_ELOG(LOG, "Start background worker %d, shutdown=%d", MyProcPid, pool->shutdown);
 
+	MtmBackgroundWorker = true;
 	MtmIsLogicalReceiver = true;
 	MtmPool = pool;
 

contrib/mmts/multimaster.c

@@ -238,6 +238,7 @@ bool  MtmDoReplication;
 char* MtmDatabaseName;
 char* MtmDatabaseUser;
 Oid	  MtmDatabaseId;
+bool  MtmBackgroundWorker;
 
 int	  MtmNodes;
 int	  MtmNodeId;
@@ -902,7 +903,7 @@ MtmIsUserTransaction()
 		IsNormalProcessingMode() &&
 		MtmDoReplication &&
 		!am_walsender &&
-		!IsBackgroundWorker &&
+		!MtmBackgroundWorker &&
 		!IsAutoVacuumWorkerProcess();
 }
 
@@ -4924,7 +4925,7 @@ static void MtmGucInit(void)
 	 */
 	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
 	current_role = GetConfigOptionByName("session_authorization", NULL, false);
-	if (strcmp(MtmDatabaseUser, current_role) != 0)
+	if (current_role && *current_role && strcmp(MtmDatabaseUser, current_role) != 0)
 		MtmGucUpdate("session_authorization", current_role);
 	MemoryContextSwitchTo(oldcontext);
 }
@@ -5018,13 +5019,20 @@ char* MtmGucSerialize(void)
 {
 	StringInfo serialized_gucs;
 	dlist_iter iter;
-	int nvars = 0;
+	const char *search_path;
 
 	if (!MtmGucHash)
 		MtmGucInit();
 
 	serialized_gucs = makeStringInfo();
 
+	/*
+	 * Crutch for scheduler. It sets search_path through SetConfigOption()
+	 * so our callback do not react on that.
+	 */
+	search_path = GetConfigOption("search_path", false, true);
+	appendStringInfo(serialized_gucs, "SET search_path TO %s; ", search_path);
+
 	dlist_foreach(iter, &MtmGucList)
 	{
 		MtmGucEntry *cur_entry = dlist_container(MtmGucEntry, list_node, iter.cur);
@@ -5045,7 +5053,6 @@ char* MtmGucSerialize(void)
 			appendStringInfoString(serialized_gucs, cur_entry->value);
 		}
 		appendStringInfoString(serialized_gucs, "; ");
-		nvars++;
 	}
 
 	return serialized_gucs->data;
@@ -5163,6 +5170,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		case T_CheckPointStmt:
 		case T_ReindexStmt:
 		case T_ExplainStmt:
+		case T_AlterSystemStmt:
 			skipCommand = true;
 			break;
 
@@ -5374,7 +5382,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 	if (!skipCommand && !MtmTx.isReplicated && (context == PROCESS_UTILITY_TOPLEVEL || MtmUtilityProcessedInXid != GetCurrentTransactionId()))
 	{
 		MtmUtilityProcessedInXid = GetCurrentTransactionId();
-		if (context == PROCESS_UTILITY_TOPLEVEL) {
+		if (context == PROCESS_UTILITY_TOPLEVEL || !ActivePortal) {
 			MtmProcessDDLCommand(queryString, true);
 		} else {
 			MtmProcessDDLCommand(ActivePortal->sourceText, true);

contrib/mmts/multimaster.h

@@ -378,6 +378,7 @@ extern MemoryContext MtmApplyContext;
 extern lsn_t MtmSenderWalEnd;
 extern timestamp_t MtmRefreshClusterStatusSchedule;
 extern MtmConnectionInfo* MtmConnections;
+extern bool MtmBackgroundWorker;
 
 
 extern void  MtmArbiterInitialize(void);

contrib/mmts/pglogical_receiver.c

@@ -224,6 +224,8 @@ pglogical_receiver_main(Datum main_arg)
 	char* connString = psprintf("replication=database %s", Mtm->nodes[nodeId-1].con.connStr);
 	static PortalData fakePortal;
 
+	MtmBackgroundWorker = true;
+
 	ByteBufferAlloc(&buf);
 
 	slotName = psprintf(MULTIMASTER_SLOT_PATTERN, MtmNodeId);