Ressurect support of concurrent DDL statements

Author: knizhnik

multimaster.c

@@ -133,7 +133,7 @@ static void MtmBeginTransaction(MtmCurrentTrans* x);
 static void MtmPrePrepareTransaction(MtmCurrentTrans* x);
 static void MtmPostPrepareTransaction(MtmCurrentTrans* x);
 static void MtmAbortPreparedTransaction(MtmCurrentTrans* x);
-static void MtmCommitPreparedTransaction(MtmCurrentTrans* x);
+static void MtmPreCommitPreparedTransaction(MtmCurrentTrans* x);
 static void MtmEndTransaction(MtmCurrentTrans* x, bool commit);
 static bool MtmTwoPhaseCommit(MtmCurrentTrans* x);
 static TransactionId MtmGetOldestXmin(Relation rel, bool ignoreVacuum);
@@ -705,8 +705,8 @@ MtmXactCallback(XactEvent event, void *arg)
 	  case XACT_EVENT_ABORT_PREPARED:
 		MtmAbortPreparedTransaction(&MtmTx);
 		break;
-	  case XACT_EVENT_COMMIT_PREPARED:
-		MtmCommitPreparedTransaction(&MtmTx);
+	  case XACT_EVENT_PRE_COMMIT_PREPARED:
+		MtmPreCommitPreparedTransaction(&MtmTx);
 		break;
 	  case XACT_EVENT_COMMIT:
 		MtmEndTransaction(&MtmTx, true);
@@ -1149,7 +1149,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 }
 
 static void 
-MtmCommitPreparedTransaction(MtmCurrentTrans* x)
+MtmPreCommitPreparedTransaction(MtmCurrentTrans* x)
 {
     MtmTransMap* tm;
 	MtmTransState* ts;
@@ -1181,6 +1181,7 @@ MtmCommitPreparedTransaction(MtmCurrentTrans* x)
 		Mtm2PCVoting(x, ts);
 
 		x->xid = ts->xid;
+		x->csn = ts->csn;
 		x->isPrepared = true;
 	}
 	MtmUnlock();
@@ -3529,6 +3530,7 @@ bool MtmFilterTransaction(char* record, int size)
 	}
 	restart_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
     if (Mtm->nodes[origin_node-1].restartLSN < restart_lsn) {
+		Assert(restart_lsn != InvalidXLogRecPtr);
 		MTM_LOG2("[restartlsn] node %d: %lx -> %lx (MtmFilterTransaction)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, restart_lsn);
 		Mtm->nodes[origin_node-1].restartLSN = restart_lsn;
     } else {
@@ -4442,19 +4444,14 @@ static void MtmProcessDDLCommand(char const* queryString, bool transactional)
 	}
 	MTM_LOG3("Sending utility: %s", queryString);
 	if (transactional) {
-		/* DDL */
+		/* Transactional DDL */
 		LogLogicalMessage("D", queryString, strlen(queryString) + 1, true);
 		MtmTx.containsDML = true;
-	} else {		
-		char* gucCtx = MtmGucSerialize();
-		return;
-		if (*gucCtx) {
-			queryString = psprintf("%s; %s", gucCtx, queryString);
-		}
-		/* CONCURRENT DDL */
+	} else {	
+		MTM_LOG1("Execute concurrent DDL: %s", queryString);
+		/* Concurrent DDL */
 		XLogFlush(LogLogicalMessage("C", queryString, strlen(queryString) + 1, false));
 	}
-	return;
 }
 
 static void MtmFinishDDLCommand()
@@ -4542,7 +4539,6 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 
 		case T_VacuumStmt:
 		  skipCommand = true;		  
-#if 0
 		  if (context == PROCESS_UTILITY_TOPLEVEL) {			  
 			  MtmProcessDDLCommand(queryString, false);
 			  MtmTx.isDistributed = false;
@@ -4553,7 +4549,6 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			  MemoryContextSwitchTo(oldContext);
 			  return;
 		  }
-#endif
 		  break;
 
 		case T_CreateDomainStmt:

pglogical_apply.c

@@ -384,6 +384,7 @@ process_remote_message(StringInfo s)
 		{
 			MTM_LOG1("%d: Executing non-tx utility statement %s", MyProcPid, messageBody);
 			SetCurrentStatementStartTimestamp();
+			MtmResetTransaction();
 			StartTransactionCommand();
 			standalone = true;
 			/* intentional falldown to the next case */
@@ -413,7 +414,6 @@ process_remote_message(StringInfo s)
 														 false, false,
 														 NULL,
 														 NULL);
-					
 					/* Run parse analysis ... */
 					MtmIndexStmt = transformIndexStmt(relid, MtmIndexStmt, messageBody);
 

pglogical_receiver.c

@@ -543,9 +543,13 @@ pglogical_receiver_main(Datum main_arg)
 						MtmSpillToFile(spill_file, buf.data, buf.used);
 						ByteBufferReset(&buf);
 					}
-					if (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A')) {
+					if (stmt[0] == 'M' && (stmt[1] == 'L' || stmt[1] == 'A' || stmt[1] == 'C')) {
 						MTM_LOG3("Process '%c' message from %d", stmt[1], nodeId);
-						MtmExecutor(stmt, rc - hdr_len);
+						if ( stmt[1] == 'C') { /* concurrent DDL */
+							MtmExecute(stmt, rc - hdr_len);
+						} else {
+							MtmExecutor(stmt, rc - hdr_len);
+						}
 					} else { 
 						ByteBufferAppend(&buf, stmt, rc - hdr_len);
 						if (stmt[0] == 'C') /* commit */