Rewrite origin LSN calculation

Author: knizhnik

multimaster.c

@@ -1133,6 +1133,16 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 	}
 }
 
+static void 
+MtmLogAbortLogicalMessage(int nodeId, char const* gid)
+{
+	MtmAbortLogicalMessage msg;
+	strcpy(msg.gid, gid);
+	msg.origin_node = nodeId;
+	msg.origin_lsn = replorigin_session_origin_lsn;
+	XLogFlush(LogLogicalMessage("A", (char*)&msg, sizeof msg, false)); 
+}
+
 static void 
 MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 {
@@ -1154,7 +1164,8 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		}
 		if (ts != NULL) { 
 			if (*ts->gid)  
-				MTM_LOG2("TRANSLOG: %s transaction %s status %d", (commit ? "commit" : "rollback"), ts->gid, ts->status);
+				MTM_LOG1("TRANSLOG: %s transaction git=%s xid=%d node=%d dxid=%d status %d", 
+						 (commit ? "commit" : "rollback"), ts->gid, ts->xid, ts->gtid.node, ts->gtid.xid, ts->status);
 			if (commit) {
 				if (!(ts->status == TRANSACTION_STATUS_UNKNOWN 
 					  || (ts->status == TRANSACTION_STATUS_IN_PROGRESS && Mtm->status == MTM_RECOVERY)))  
@@ -1213,7 +1224,8 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 				}
 				MtmTransactionListAppend(ts);
 				if (*x->gid) { 
-					LogLogicalMessage("A", x->gid, strlen(x->gid) + 1, false);
+					replorigin_session_origin_lsn = InvalidXLogRecPtr;
+					MtmLogAbortLogicalMessage(MtmNodeId, x->gid);
 				}
 			}
 			MtmSend2PCMessage(ts, MSG_ABORTED); /* send notification to coordinator */
@@ -2826,20 +2838,23 @@ void MtmReleaseRecoverySlot(int nodeId)
 	if (Mtm->recoverySlot == nodeId) { 
 		Mtm->recoverySlot = 0;
 	}
-}
+}		
 
-void MtmRollbackPreparedTransaction(char const* gid)
+void MtmRollbackPreparedTransaction(int nodeId, char const* gid)
 {
-	MTM_LOG1("Abort prepared transaction %s", gid);
-	if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
+	XidStatus status = MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED);
+	MTM_LOG1("Abort prepared transaction %s status %d", gid, status);
+	if (status == TRANSACTION_STATUS_UNKNOWN) { 
 		MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
 		MtmResetTransaction();
 		StartTransactionCommand();
-		MtmBeginSession(MtmReplicationNodeId);
+		MtmBeginSession(nodeId);
 		MtmSetCurrentTransactionGID(gid);
 		FinishPreparedTransaction(gid, false);
 		CommitTransactionCommand();
-		MtmEndSession(MtmReplicationNodeId, true);
+		MtmEndSession(nodeId, true);
+	} else if (status == TRANSACTION_STATUS_IN_PROGRESS) {
+		MtmLogAbortLogicalMessage(nodeId, gid);
 	}
 }	
 
@@ -3159,19 +3174,11 @@ bool MtmFilterTransaction(char* record, int size)
 	  default:
 		break;
 	}
-	duplicate = Mtm->status == MTM_RECOVERY && origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLSN;
+	//duplicate = Mtm->status == MTM_RECOVERY && origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLSN;
+	duplicate = origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLSN;
 
-	MTM_LOG1("%s transaction %s from node %d lsn %lx, origin node %d, original lsn=%lx, current lsn=%lx", 
-			 duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, origin_node, origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
-	if (Mtm->status == MTM_RECOVERY) { 
-		if (Mtm->nodes[origin_node-1].restartLSN < origin_lsn) { 
-			Mtm->nodes[origin_node-1].restartLSN = origin_lsn;
-		}
-	} else { 
-		if (Mtm->nodes[replication_node-1].restartLSN < end_lsn) { 
-			Mtm->nodes[replication_node-1].restartLSN = end_lsn;
-		}
-	}
+	MTM_LOG1("%s transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
+			 duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, flags, origin_node, origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
 	return duplicate;
 }
 

multimaster.h

@@ -149,6 +149,13 @@ typedef struct
 	pgid_t         gid;    /* Global transaction identifier */
 } MtmArbiterMessage;
 
+typedef struct MtmAbortLogicalMessage
+{
+	pgid_t         gid;
+	int            origin_node;
+	XLogRecPtr     origin_lsn;
+} MtmAbortLogicalMessage;
+
 typedef struct MtmMessageQueue
 {
 	MtmArbiterMessage msg;
@@ -364,7 +371,7 @@ extern PGconn *PQconnectdb_safe(const char *conninfo);
 extern void MtmBeginSession(int nodeId);
 extern void MtmEndSession(int nodeId, bool unlock);
 extern void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit);
-extern void MtmRollbackPreparedTransaction(char const* gid);
+extern void MtmRollbackPreparedTransaction(int nodeId, char const* gid);
 extern bool MtmFilterTransaction(char* record, int size);
 
 #endif

pglogical_apply.c

@@ -427,7 +427,14 @@ process_remote_message(StringInfo s)
 		}
 	    case 'A':
 		{
-		    MtmRollbackPreparedTransaction(messageBody);
+			MtmAbortLogicalMessage* msg = (MtmAbortLogicalMessage*)messageBody;
+			int origin_node = msg->origin_node;
+			Assert(messageSize == sizeof(MtmAbortLogicalMessage));
+			if (Mtm->nodes[origin_node-1].restartLSN < msg->origin_lsn) { 
+				Mtm->nodes[origin_node-1].restartLSN = msg->origin_lsn;
+			}
+			replorigin_session_origin_lsn = msg->origin_lsn; 
+			MtmRollbackPreparedTransaction(origin_node, msg->gid);
 			standalone = true;
 			break;
 		}
@@ -597,17 +604,16 @@ process_remote_commit(StringInfo in)
 
 	/* read fields */
 	pq_getmsgint64(in); /* commit_lsn */
-	replorigin_session_origin_lsn = end_lsn = pq_getmsgint64(in); /* end_lsn */
+	end_lsn = pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
 
 	origin_node = pq_getmsgbyte(in);
 	origin_lsn = pq_getmsgint64(in);
 
-	if (origin_node != MtmReplicationNodeId) {
-		replorigin_advance(Mtm->nodes[origin_node-1].originId, origin_lsn, GetXLogInsertRecPtr(),
-						   false /* backward */ , false /* WAL */ );
+	replorigin_session_origin_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
+	if (Mtm->nodes[origin_node-1].restartLSN < replorigin_session_origin_lsn) { 
+		Mtm->nodes[origin_node-1].restartLSN = replorigin_session_origin_lsn;
 	}
-
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 
 	switch(PGLOGICAL_XACT_EVENT(flags))
@@ -617,9 +623,9 @@ process_remote_commit(StringInfo in)
 			MTM_LOG2("%d: PGLOGICAL_COMMIT commit", MyProcPid);
 			if (IsTransactionState()) {
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
-				MtmBeginSession(MtmReplicationNodeId);
+				MtmBeginSession(origin_node);
 				CommitTransactionCommand();
-				MtmEndSession(MtmReplicationNodeId, true);
+				MtmEndSession(origin_node, true);
 			}
 			break;
 		}
@@ -632,12 +638,12 @@ process_remote_commit(StringInfo in)
 				AbortCurrentTransaction();
 			} else { 				
 				/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
-				MTM_LOG2("PGLOGICAL_PREPARE commit: gid=%s", gid);
+				MTM_LOG1("PGLOGICAL_PREPARE commit: gid=%s", gid);
 				BeginTransactionBlock();
 				CommitTransactionCommand();
 				StartTransactionCommand();
 
-				MtmBeginSession(MtmReplicationNodeId);
+				MtmBeginSession(origin_node);
 				/* PREPARE itself */
 				MtmSetCurrentTransactionGID(gid);
 				PrepareTransactionBlock(gid);
@@ -650,7 +656,7 @@ process_remote_commit(StringInfo in)
 					FinishPreparedTransaction(gid, false);
 					CommitTransactionCommand();					
 				}	
-				MtmEndSession(MtmReplicationNodeId, true);
+				MtmEndSession(origin_node, true);
 			}
 			break;
 		}
@@ -659,22 +665,22 @@ process_remote_commit(StringInfo in)
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			csn = pq_getmsgint64(in); 
 			gid = pq_getmsgstring(in);
-			MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%ld", csn, gid, end_lsn);
+			MTM_LOG1("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%lx", csn, gid, end_lsn);
 			MtmResetTransaction();
 			StartTransactionCommand();
-			MtmBeginSession(MtmReplicationNodeId);
+			MtmBeginSession(origin_node);
 			MtmSetCurrentTransactionCSN(csn);
 			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, true);
 			CommitTransactionCommand();
-			MtmEndSession(MtmReplicationNodeId, true);
+			MtmEndSession(origin_node, true);
 			break;
 		}
 		case PGLOGICAL_ABORT_PREPARED:
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			MtmRollbackPreparedTransaction(gid);
+			MtmRollbackPreparedTransaction(origin_node, gid);
 			break;
 		}
 		default:

pglogical_receiver.c

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