Reset GUC context at transaction begin

Author: knizhnik

multimaster.c

@@ -156,7 +156,7 @@ static void MtmShmemStartup(void);
 static BgwPool* MtmPoolConstructor(void);
 static bool MtmRunUtilityStmt(PGconn* conn, char const* sql, char **errmsg);
 static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError);
-static bool MtmProcessDDLCommand(char const* queryString, bool transactional);
+static void MtmProcessDDLCommand(char const* queryString, bool transactional);
 
 MtmState* Mtm;
 
@@ -175,7 +175,8 @@ static MtmConnectionInfo* MtmConnections;
 static MtmCurrentTrans MtmTx;
 static dlist_head MtmLsnMapping = DLIST_STATIC_INIT(MtmLsnMapping);
 
-static TransactionManager MtmTM = { 
+static TransactionManager MtmTM = 
+{ 
 	PgTransactionIdGetStatus, 
 	PgTransactionIdSetTreeStatus,
 	MtmGetSnapshot, 
@@ -4431,18 +4432,29 @@ char* MtmGucSerialize(void)
  * -------------------------------------------
  */
 
-static bool MtmProcessDDLCommand(char const* queryString, bool transactional)
+static void MtmProcessDDLCommand(char const* queryString, bool transactional)
 {
+	char* gucCtx = MtmGucSerialize();
+	if (*gucCtx) {
+		queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s; %s", gucCtx, queryString);
+	} else { 
+		queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
+	}
 	MTM_LOG3("Sending utility: %s", queryString);
 	if (transactional) {
 		/* DDL */
 		LogLogicalMessage("D", queryString, strlen(queryString) + 1, true);
 		MtmTx.containsDML = true;
-	} else {
+	} else {		
+		char* gucCtx = MtmGucSerialize();
+		return;
+		if (*gucCtx) {
+			queryString = psprintf("%s; %s", gucCtx, queryString);
+		}
 		/* CONCURRENT DDL */
 		XLogFlush(LogLogicalMessage("C", queryString, strlen(queryString) + 1, false));
 	}
-	return false;
+	return;
 }
 
 static void MtmFinishDDLCommand()
@@ -4530,16 +4542,18 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 
 		case T_VacuumStmt:
 		  skipCommand = true;		  
-		//   if (context == PROCESS_UTILITY_TOPLEVEL) {			  
-		// 	  MtmProcessDDLCommand(queryString, false, true);
-		// 	  MtmTx.isDistributed = false;
-		//   } else if (MtmApplyContext != NULL) {
-		// 	  MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
-		// 	  Assert(oldContext != MtmApplyContext);
-		// 	  MtmVacuumStmt = (VacuumStmt*)copyObject(parsetree);
-		// 	  MemoryContextSwitchTo(oldContext);
-		// 	  return;
-		//   }
+#if 0
+		  if (context == PROCESS_UTILITY_TOPLEVEL) {			  
+			  MtmProcessDDLCommand(queryString, false);
+			  MtmTx.isDistributed = false;
+		  } else if (MtmApplyContext != NULL) {
+			  MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
+			  Assert(oldContext != MtmApplyContext);
+			  MtmVacuumStmt = (VacuumStmt*)copyObject(parsetree);
+			  MemoryContextSwitchTo(oldContext);
+			  return;
+		  }
+#endif
 		  break;
 
 		case T_CreateDomainStmt:
@@ -4634,7 +4648,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 				if (indexStmt->concurrent) 
 				{
 					 if (context == PROCESS_UTILITY_TOPLEVEL) {
-						//  MtmProcessDDLCommand(queryString, false, true);
+						 MtmProcessDDLCommand(queryString, false);
 						 MtmTx.isDistributed = false;
 						 skipCommand = true;
 						 /* 
@@ -4661,7 +4675,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 				if (stmt->removeType == OBJECT_INDEX && stmt->concurrent)
 				{
 					if (context == PROCESS_UTILITY_TOPLEVEL) {
-						// MtmProcessDDLCommand(queryString, false, true);
+						MtmProcessDDLCommand(queryString, false);
 						MtmTx.isDistributed = false;
 						skipCommand = true;
 					} else if (MtmApplyContext != NULL) {

pglogical_apply.c

@@ -337,13 +337,11 @@ process_remote_begin(StringInfo s)
 {
 	GlobalTransactionId gtid;
 	csn_t snapshot;
-	char const* gucCtx;
 	int rc;
 
 	gtid.node = pq_getmsgint(s, 4); 
 	gtid.xid = pq_getmsgint(s, 4); 
 	snapshot = pq_getmsgint64(s);    
-	gucCtx = pq_getmsgstring(s);
 
 	Assert(gtid.node > 0);
 
@@ -359,19 +357,13 @@ process_remote_begin(StringInfo s)
 	StartTransactionCommand();
     MtmJoinTransaction(&gtid, snapshot);
 
-	if (*gucCtx || GucAltered) {
+	if (GucAltered) {
 		SPI_connect();
-		if (GucAltered) { 
-			GucAltered = *gucCtx != '\0';
-			gucCtx = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", gucCtx);
-		} else { 
-			GucAltered = true;
-		}
-		ActivePortal->sourceText = gucCtx;
-		rc = SPI_execute(gucCtx, false, 0);
+		GucAltered = false;
+		rc = SPI_execute("RESET SESSION AUTHORIZATION; reset all;", false, 0);
 		SPI_finish();
 		if (rc < 0) { 
-			elog(ERROR, "Failed to set GUC context %s: %d", gucCtx, rc);
+			elog(ERROR, "Failed to set reset context: %d", rc);
 		}
 	}
 
@@ -399,6 +391,7 @@ process_remote_message(StringInfo s)
 		case 'D':
 		{
 			int rc;
+			GucAltered = true;
 			MTM_LOG1("%d: Executing utility statement %s", MyProcPid, messageBody);
 			SPI_connect();
 			ActivePortal->sourceText = messageBody;

pglogical_proto.c

@@ -131,7 +131,6 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 		pq_sendint(out, MtmNodeId, 4);
 		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
 		pq_sendint64(out, csn);
-		pq_sendstring(out, MtmGucSerialize());
 
 		MtmTransactionRecords = 0;
 	}