Fix memory context usage for concurrent DDL statements

Author: knizhnik

multimaster.c

@@ -3941,18 +3941,18 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			break;
 
 		case T_VacuumStmt:
-		  if (context == PROCESS_UTILITY_TOPLEVEL) {
+		  skipCommand = true;		  
+		  if (context == PROCESS_UTILITY_TOPLEVEL) {			  
 			  MtmProcessDDLCommand(queryString, false, true);
 			  MtmTx.isDistributed = false;
-			  skipCommand = true;		  
-			  break;
-		  } else {
+		  } else if (MtmApplyContext != NULL) {
 			  MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
 			  Assert(oldContext != MtmApplyContext);
 			  MtmVacuumStmt = (VacuumStmt*)copyObject(parsetree);
 			  MemoryContextSwitchTo(oldContext);
 			  return;
 		  }
+		  break;
 
 		case T_CreateDomainStmt:
 			/* Detect temp tables access */
@@ -4049,10 +4049,10 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 						 MtmProcessDDLCommand(queryString, false, true);
 						 MtmTx.isDistributed = false;
 						 skipCommand = true;
-					 } else { 
+					 } else if (MtmApplyContext != NULL) { 
 						 MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
 						 Assert(oldContext != MtmApplyContext);
-						 MtmIndexStmt = indexStmt;
+						 MtmIndexStmt = (IndexStmt*)copyObject(indexStmt);
 						 MemoryContextSwitchTo(oldContext);
 						 return;
 					 }
@@ -4069,10 +4069,10 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 						MtmProcessDDLCommand(queryString, false, true);
 						MtmTx.isDistributed = false;
 						skipCommand = true;
-					} else {
+					} else if (MtmApplyContext != NULL) {
 						 MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
 						 Assert(oldContext != MtmApplyContext);
-						 MtmDropStmt = stmt;
+						 MtmDropStmt = (DropStmt*)copyObject(stmt);
 						 MemoryContextSwitchTo(oldContext);
 						 return;
 					}

pglogical_apply.c

@@ -390,19 +390,21 @@ process_remote_message(StringInfo s)
 			if (rc < 0) { 
 				elog(ERROR, "Failed to execute utility statement %s", messageBody);
 			} else { 
+				PushActiveSnapshot(GetTransactionSnapshot());
+
 				if (MtmVacuumStmt != NULL) { 
 					ExecVacuum(MtmVacuumStmt, 1);
 				} else if (MtmIndexStmt != NULL) {
+					MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
 					Oid relid =	RangeVarGetRelidExtended(MtmIndexStmt->relation, ShareUpdateExclusiveLock,
 														 false, false,
 														 NULL,
 														 NULL);
 
 					/* Run parse analysis ... */
-					//MtmIndexStmt = transformIndexStmt(relid, MtmIndexStmt, messageBody);
-
-					PushActiveSnapshot(GetTransactionSnapshot());
+					MtmIndexStmt = transformIndexStmt(relid, MtmIndexStmt, messageBody);
 
+					MemoryContextSwitchTo(oldContext);
 
 					DefineIndex(relid,		/* OID of heap relation */
 								MtmIndexStmt,
@@ -412,12 +414,11 @@ process_remote_message(StringInfo s)
 								false,		/* skip_build */
 								false);		/* quiet */
 
-					if (ActiveSnapshotSet())
-						PopActiveSnapshot();
-
 				} else if (MtmDropStmt != NULL) { 
 					RemoveObjects(MtmDropStmt);
 				}
+				if (ActiveSnapshotSet())
+					PopActiveSnapshot();
 			}
 			if (standalone) { 
 				CommitTransactionCommand();