trying to send utility_stmts on PrePrepare

Author: kelvich

bgwpool.c

@@ -25,7 +25,7 @@ static void BgwPoolMainLoop(Datum arg)
     void* work;
 
     BackgroundWorkerUnblockSignals();
-	BackgroundWorkerInitializeConnection(pool->dbname, NULL);
+	BackgroundWorkerInitializeConnection(pool->dbname, "stas");
 
     while(true) { 
         PGSemaphoreLock(&pool->available);
@@ -98,7 +98,7 @@ void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor)
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_main = BgwPoolMainLoop;
 	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
-    
+
     for (i = 0; i < nWorkers; i++) { 
         BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)malloc(sizeof(BgwPoolExecutorCtx));
         snprintf(worker.bgw_name, BGW_MAXLEN, "bgw_pool_worker_%d", i+1);

multimaster.c

@@ -197,6 +197,7 @@ char const* const MtmNodeStatusMnem[] =
 
 bool  MtmDoReplication;
 char* MtmDatabaseName;
+char* MtmUtilityStmt = NULL;
 
 int   MtmNodes;
 int   MtmNodeId;
@@ -213,8 +214,6 @@ int   MtmHeartbeatRecvTimeout;
 bool  MtmUseRaftable;
 bool  MtmUseDtm;
 
-// static int   reset_wrokers = 0;
-
 static char* MtmConnStrs;
 static int   MtmQueueSize;
 static int   MtmWorkers;
@@ -693,6 +692,10 @@ static const char* const isoLevelStr[] =
 static void 
 MtmBeginTransaction(MtmCurrentTrans* x)
 {
+	if (MtmUtilityStmt)
+		pfree(MtmUtilityStmt);
+	MtmUtilityStmt = NULL;
+
     if (x->snapshot == INVALID_CSN) { 
 		TransactionId xmin = (Mtm->gcCount >= MtmGcPeriod) ? PgGetOldestXmin(NULL, false) : InvalidTransactionId; /* Get oldest xmin outside critical section */
 
@@ -3087,7 +3090,14 @@ MtmGenerateGid(char* gid)
 
 static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 {
-	if (x->isDistributed && x->containsDML) { 
+	if (MtmUtilityStmt && !MyXactAccessedTempRel)
+	{
+		MtmProcessDDLCommand(MtmUtilityStmt);
+		pfree(MtmUtilityStmt);
+		MtmUtilityStmt = NULL;
+	}
+
+	if (!x->isReplicated && (x->isDistributed && x->containsDML)) { 
 		MtmGenerateGid(x->gid);
 		if (!x->isTransactionBlock) { 
 			BeginTransactionBlock();
@@ -3118,6 +3128,9 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 							 DestReceiver *dest, char *completionTag)
 {
 	bool skipCommand = false;
+
+	// skipCommand = MyXactAccessedTempRel;
+
 	MTM_LOG3("%d: Process utility statement %s", MyProcPid, queryString);
 	switch (nodeTag(parsetree))
 	{
@@ -3198,12 +3211,12 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			}
 			break;
 		case T_CreateTableAsStmt:
-			{
-				/* Do not replicate temp tables */
-				CreateTableAsStmt *stmt = (CreateTableAsStmt *) parsetree;
-				skipCommand = stmt->into->rel->relpersistence == RELPERSISTENCE_TEMP ||
-					(stmt->into->rel->schemaname && strcmp(stmt->into->rel->schemaname, "pg_temp") == 0);
-			}
+			// {
+			// 	/* Do not replicate temp tables */
+			// 	CreateTableAsStmt *stmt = (CreateTableAsStmt *) parsetree;
+			// 	skipCommand = stmt->into->rel->relpersistence == RELPERSISTENCE_TEMP ||
+			// 		(stmt->into->rel->schemaname && strcmp(stmt->into->rel->schemaname, "pg_temp") == 0);
+			// }
 			break;
 		case T_CreateStmt:
 			{
@@ -3306,11 +3319,26 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			skipCommand = false;
 			break;
 	}
-	if (!skipCommand && !MtmTx.isReplicated && context == PROCESS_UTILITY_TOPLEVEL) {
-		if (MtmProcessDDLCommand(queryString)) { 
-			return;
+	if (context == PROCESS_UTILITY_TOPLEVEL)
+	{
+		if (!skipCommand && !MtmTx.isReplicated) {
+			// if (MtmProcessDDLCommand(queryString)) { 
+			// 	return;
+			// }
+
+			MemoryContext oldcontext;
+
+			if (MtmUtilityStmt)
+				pfree(MtmUtilityStmt);
+
+			oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+			MtmUtilityStmt = palloc(strlen(queryString) + 1);
+			MemoryContextSwitchTo(oldcontext);
+
+			strncpy(MtmUtilityStmt, queryString, strlen(queryString) + 1);
 		}
 	}
+
 	if (PreviousProcessUtilityHook != NULL)
 	{
 		PreviousProcessUtilityHook(parsetree, queryString, context,

pglogical_receiver.c

@@ -468,14 +468,24 @@ pglogical_receiver_main(Datum main_arg)
 				{
 					stmt = copybuf + hdr_len;
 
-					if (buf.used >= MtmTransSpillThreshold*MB) { 
-						if (spill_file < 0) {
-							int file_id;
-							spill_file = MtmCreateSpillFile(nodeId, &file_id);
-							pq_sendbyte(&spill_info, 'F');
-							pq_sendint(&spill_info, nodeId, 4);
-							pq_sendint(&spill_info, file_id, 4);
-						}
+				if (buf.used >= MtmTransSpillThreshold*MB) { 
+					if (spill_file < 0) {
+						int file_id;
+						spill_file = MtmCreateSpillFile(nodeId, &file_id);
+						pq_sendbyte(&spill_info, 'F');
+						pq_sendint(&spill_info, nodeId, 4);
+						pq_sendint(&spill_info, file_id, 4);
+					}
+					ByteBufferAppend(&buf, ")", 1);
+					pq_sendbyte(&spill_info, '(');
+					pq_sendint(&spill_info, buf.used, 4);
+					MtmSpillToFile(spill_file, buf.data, buf.used);
+					ByteBufferReset(&buf);
+				}
+                ByteBufferAppend(&buf, stmt, rc - hdr_len);
+				if (stmt[0] == 'C') /* commit|prepare */
+                {
+					if (spill_file >= 0) { 
 						ByteBufferAppend(&buf, ")", 1);
 						pq_sendbyte(&spill_info, '(');
 						pq_sendint(&spill_info, buf.used, 4);