Create fake portal

Author: knizhnik

bgwpool.c

@@ -20,12 +20,13 @@ static void BgwPoolMainLoop(BgwPool* pool)
 {
     int size;
     void* work;
+	static PortalData fakePortal;
 
 	MtmIsLogicalReceiver = true;
 
     BackgroundWorkerUnblockSignals();
 	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser);
-	ActivePortal = CreatePortal("", true, true);
+	ActivePortal = &fakePortal;
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";
 

multimaster.h

@@ -288,6 +288,7 @@ extern int   MtmReplicationNodeId;
 extern int   MtmNodes;
 extern int   MtmArbiterPort;
 extern char* MtmDatabaseName;
+extern char* MtmDatabaseUser;
 extern int   MtmConnectTimeout;
 extern int   MtmReconnectTimeout;
 extern int   MtmRaftPollDelay;

pglogical_apply.c

@@ -371,7 +371,8 @@ process_remote_transactional_message(StringInfo s)
 	}
 
 	MTM_LOG1("%d: Executing utility statement %s", MyProcPid, stmt);
-	SPI_connect();
+	SPI_connect();	
+	ActivePortal->sourceText = stmt;
 	rc = SPI_execute(stmt, false, 0);
 	SPI_finish();
 	if (rc < 0)
@@ -971,8 +972,8 @@ void MtmExecutor(void* work, size_t size)
     StringInfoData s;
     Relation rel = NULL;
 	int spill_file = -1;
-	int save_cursor;
-	int save_len;
+	int save_cursor = 0;
+	int save_len = 0;
     s.data = work;
     s.len = size;
     s.maxlen = -1;

pglogical_receiver.c

@@ -35,6 +35,8 @@
 #include "utils/memutils.h"
 #include "executor/spi.h"
 #include "replication/origin.h"
+#include "utils/portal.h"
+#include "tcop/pquery.h"
 
 #include "multimaster.h"
 #include "spill.h"
@@ -218,6 +220,8 @@ pglogical_receiver_main(Datum main_arg)
 	StringInfoData spill_info;
 	char *slotName;
 	char* connString = psprintf("replication=database %s", Mtm->nodes[nodeId-1].con.connStr);
+	static PortalData fakePortal;
+
 	slotName = psprintf(MULTIMASTER_SLOT_PATTERN, MtmNodeId);
 
 	MtmIsLogicalReceiver = true;
@@ -239,7 +243,10 @@ pglogical_receiver_main(Datum main_arg)
 	BackgroundWorkerUnblockSignals();
 
 	/* Connect to a database */
-	BackgroundWorkerInitializeConnection(MtmDatabaseName, NULL);
+	BackgroundWorkerInitializeConnection(MtmDatabaseName, MtmDatabaseUser);
+	ActivePortal = &fakePortal;
+	ActivePortal->status = PORTAL_ACTIVE;
+	ActivePortal->sourceText = "";
 
 	/* This is main loop of logical replication.
 	 * In case of errors we will try to reestablish connection.