Close relation before processing of DDL messages

knizhnik · kelvich · commit b2430210b231 · 2017-11-13T02:17:39.000+03:00
diff --git a/multimaster.c b/multimaster.c
@@ -1276,7 +1276,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		MtmUnlock();
 		if (x->isTwoPhase) { 
 			if (x->status == TRANSACTION_STATUS_ABORTED) { 
-				MTM_ELOG(ERROR, "Prepare of user's 2PC transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);
+				MTM_ELOG(WARNING, "Prepare of user's 2PC transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);
 			}
 			MtmResetTransaction();
 		}
@@ -3412,7 +3412,7 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
 	MtmTx.isActive = true;
 	FinishPreparedTransaction(ts->gid, commit);
 	if (commit) { 
-		MTM_LOG1("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", ts->gid, (long64)ts->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());
+		MTM_LOG2("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", ts->gid, (long64)ts->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());
 	}
 	if (!insideTransaction) { 
 		CommitTransactionCommand();
@@ -4600,7 +4600,7 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 					MTM_ELOG(ERROR, "Transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);
 				} else {
 					FinishPreparedTransaction(x->gid, true);
-					MTM_LOG1("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", x->gid, (long64)x->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());
+					MTM_LOG2("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", x->gid, (long64)x->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());
 				}
 			}
 		}
@@ -4868,7 +4868,6 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 				case TRANS_STMT_ROLLBACK_PREPARED:
 				    Assert(!MtmTx.isTwoPhase);
   				    strcpy(MtmTx.gid, stmt->gid);
-					//MtmTx.isTwoPhase = true;
 					break;
 				default:
 					break;
diff --git a/pglogical_apply.c b/pglogical_apply.c
@@ -1150,6 +1150,8 @@ void MtmExecutor(void* work, size_t size)
 			}
 			case 'M':
 			{
+  			    close_rel(rel);
+				rel = NULL;
 				inside_transaction = !process_remote_message(&s);
 				break;
 			}
diff --git a/pglogical_proto.c b/pglogical_proto.c
@@ -171,6 +171,7 @@ static void
 pglogical_write_message(StringInfo out,
 						const char *prefix, Size sz, const char *message)
 {
+	MtmLastRelId = InvalidOid;
 	switch (*prefix) { 
 	  case 'L':
 		if (MtmIsRecoveredNode(MtmReplicationNodeId)) { 			

Original file line number	Diff line number	Diff line change
`@@ -1276,7 +1276,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)`
`1276`	`1276`	`MtmUnlock();`
`1277`	`1277`	`if (x->isTwoPhase) {`
`1278`	`1278`	`if (x->status == TRANSACTION_STATUS_ABORTED) {`
`1279`		`- MTM_ELOG(ERROR, "Prepare of user's 2PC transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);`
	`1279`	`+ MTM_ELOG(WARNING, "Prepare of user's 2PC transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);`
`1280`	`1280`	`}`
`1281`	`1281`	`MtmResetTransaction();`
`1282`	`1282`	`}`
`@@ -3412,7 +3412,7 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)`
`3412`	`3412`	`MtmTx.isActive = true;`
`3413`	`3413`	`FinishPreparedTransaction(ts->gid, commit);`
`3414`	`3414`	`if (commit) {`
`3415`		`- MTM_LOG1("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", ts->gid, (long64)ts->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());`
	`3415`	`+ MTM_LOG2("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", ts->gid, (long64)ts->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());`
`3416`	`3416`	`}`
`3417`	`3417`	`if (!insideTransaction) {`
`3418`	`3418`	`CommitTransactionCommand();`
`@@ -4600,7 +4600,7 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)`
`4600`	`4600`	`MTM_ELOG(ERROR, "Transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);`
`4601`	`4601`	`} else {`
`4602`	`4602`	`FinishPreparedTransaction(x->gid, true);`
`4603`		`- MTM_LOG1("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", x->gid, (long64)x->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());`
	`4603`	`+ MTM_LOG2("Distributed transaction %s (%lld) is committed at %lld with LSN=%lld", x->gid, (long64)x->xid, MtmGetCurrentTime(), (long64)GetXLogInsertRecPtr());`
`4604`	`4604`	`}`
`4605`	`4605`	`}`
`4606`	`4606`	`}`
`@@ -4868,7 +4868,6 @@ static void MtmProcessUtility(Node parsetree, const char queryString,`
`4868`	`4868`	`case TRANS_STMT_ROLLBACK_PREPARED:`
`4869`	`4869`	`Assert(!MtmTx.isTwoPhase);`
`4870`	`4870`	`strcpy(MtmTx.gid, stmt->gid);`
`4871`		`- //MtmTx.isTwoPhase = true;`
`4872`	`4871`	`break;`
`4873`	`4872`	`default:`
`4874`	`4873`	`break;`
Original file line number	Diff line number	Diff line change
`@@ -1150,6 +1150,8 @@ void MtmExecutor(void* work, size_t size)`
`1150`	`1150`	`}`
`1151`	`1151`	`case 'M':`
`1152`	`1152`	`{`
	`1153`	`+ close_rel(rel);`
	`1154`	`+ rel = NULL;`
`1153`	`1155`	`inside_transaction = !process_remote_message(&s);`
`1154`	`1156`	`break;`
`1155`	`1157`	`}`
Original file line number	Diff line number	Diff line change
`@@ -171,6 +171,7 @@ static void`
`171`	`171`	`pglogical_write_message(StringInfo out,`
`172`	`172`	`const char prefix, Size sz, const char message)`
`173`	`173`	`{`
	`174`	`+ MtmLastRelId = InvalidOid;`
`174`	`175`	`switch (*prefix) {`
`175`	`176`	`case 'L':`
`176`	`177`	`if (MtmIsRecoveredNode(MtmReplicationNodeId)) {`