fixes and crutches for scheduler

Author: kelvich

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

@@ -235,6 +235,7 @@ bool  MtmDoReplication;
 char* MtmDatabaseName;
 char* MtmDatabaseUser;
 Oid	  MtmDatabaseId;
+bool  MtmBackgroundWorker;
 
 int	  MtmNodes;
 int	  MtmNodeId;
@@ -898,7 +899,7 @@ MtmIsUserTransaction()
 		IsNormalProcessingMode() &&
 		MtmDoReplication &&
 		!am_walsender &&
-		!IsBackgroundWorker &&
+		!MtmBackgroundWorker &&
 		!IsAutoVacuumWorkerProcess();
 }
 
@@ -4865,7 +4866,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);
 }
@@ -4959,13 +4960,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);
@@ -4986,7 +4994,6 @@ char* MtmGucSerialize(void)
 			appendStringInfoString(serialized_gucs, cur_entry->value);
 		}
 		appendStringInfoString(serialized_gucs, "; ");
-		nvars++;
 	}
 
 	return serialized_gucs->data;
@@ -5007,7 +5014,7 @@ static void MtmProcessDDLCommand(char const* queryString, bool transactional)
 	{
 		char *gucCtx = MtmGucSerialize();
 		if (*gucCtx)
-			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s; %s", gucCtx, queryString);
+			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s %s", gucCtx, queryString);
 		else
 			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
 
@@ -5318,7 +5325,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);