Add commit_command hook

Author: knizhnik

contrib/mmts/multimaster.c

@@ -111,6 +111,7 @@ static void MtmPrePrepareTransaction(MtmCurrentTrans* x);
 static void MtmPostPrepareTransaction(MtmCurrentTrans* x);
 static void MtmAbortPreparedTransaction(MtmCurrentTrans* x);
 static void MtmEndTransaction(MtmCurrentTrans* x, bool commit);
+static bool MtmTwoPhaseCommit(MtmCurrentTrans* x);
 static TransactionId MtmGetOldestXmin(Relation rel, bool ignoreVacuum);
 static bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 static TransactionId MtmAdjustOldestXid(TransactionId xid);
@@ -588,6 +589,11 @@ MtmXactCallback(XactEvent event, void *arg)
 	  case XACT_EVENT_ABORT: 
 		MtmEndTransaction(&MtmTx, false);
 		break;
+	  case XACT_EVENT_COMMIT_COMMAND:
+		if (!IsTransactionBlock()) { 
+			MtmTwoPhaseCommit(&MtmTx);
+		}
+		break;
 	  default:
         break;
 	}
@@ -1922,33 +1928,33 @@ MtmGenerateGid(char* gid)
 	sprintf(gid, "MTM-%d-%d-%d", MtmNodeId, MyProcPid, ++localCount);
 }
 
-static void MtmTwoPhaseCommit(char *completionTag)
+static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 {
-	MtmGenerateGid(MtmTx.gid);
-	if (!IsTransactionBlock()) { 
-		elog(WARNING, "Start transaction block for %d", MtmTx.xid);
-		BeginTransactionBlock();
-		CommitTransactionCommand();
-		StartTransactionCommand();
-	}
-	if (!PrepareTransactionBlock(MtmTx.gid))
-	{
-		elog(WARNING, "Failed to prepare transaction %s", MtmTx.gid);
-		/* report unsuccessful commit in completionTag */
-		if (completionTag) { 
-			strcpy(completionTag, "ROLLBACK");
+	if (x->isDistributed && x->containsDML) { 
+		MtmGenerateGid(x->gid);
+		if (!IsTransactionBlock()) { 
+			elog(WARNING, "Start transaction block for %d", x->xid);
+			BeginTransactionBlock();
+			CommitTransactionCommand();
+			StartTransactionCommand();
 		}
-		/* ??? Should we do explicit rollback */
-	} else { 
-		CommitTransactionCommand();
-		StartTransactionCommand();
-		if (MtmGetCurrentTransactionStatus() == TRANSACTION_STATUS_ABORTED) { 
-			FinishPreparedTransaction(MtmTx.gid, false);
-			elog(ERROR, "Transaction %s is aborted by DTM", MtmTx.gid);
-		} else {
-			FinishPreparedTransaction(MtmTx.gid, true);
+		if (!PrepareTransactionBlock(x->gid))
+		{
+			elog(WARNING, "Failed to prepare transaction %s", x->gid);
+			/* ??? Should we do explicit rollback */
+		} else { 
+			CommitTransactionCommand();
+			StartTransactionCommand();
+			if (MtmGetCurrentTransactionStatus() == TRANSACTION_STATUS_ABORTED) { 
+				FinishPreparedTransaction(x->gid, false);
+				elog(ERROR, "Transaction %s is aborted by DTM", x->gid);
+			} else {
+				FinishPreparedTransaction(x->gid, true);
+			}
 		}
+		return true;
 	}
+	return false;
 }
 
 static void MtmProcessUtility(Node *parsetree, const char *queryString,
@@ -1965,8 +1971,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 				switch (stmt->kind)
 				{					
 				case TRANS_STMT_COMMIT:
-					if (MtmTx.isDistributed && MtmTx.containsDML) {
-						MtmTwoPhaseCommit(completionTag);
+				  if (MtmTwoPhaseCommit(&MtmTx)) { 
 						return;
 					}
 					break;
@@ -2002,9 +2007,6 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		if (MtmProcessDDLCommand(queryString)) { 
 			return;
 		}
-		if (MtmTx.isDistributed && MtmTx.containsDML && !IsTransactionBlock()) { 
-			MtmTwoPhaseCommit(completionTag);
-		}
 	}
 	if (PreviousProcessUtilityHook != NULL)
 	{
@@ -2035,7 +2037,7 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 			}
         }
 		if (MtmTx.isDistributed && MtmTx.containsDML && !IsTransactionBlock()) { 
-			MtmTwoPhaseCommit(NULL);
+			MtmTwoPhaseCommit(&MtmTx);
 		}
     }
     if (PreviousExecutorFinishHook != NULL)

src/backend/access/transam/twophase.c

@@ -1248,6 +1248,9 @@ ParsePrepareRecord(uint8 info, char *xlrec, xl_xact_parsed_prepare *parsed)
 
 	hdr = (TwoPhaseFileHeader *) xlrec;
 	bufptr = xlrec + MAXALIGN(sizeof(TwoPhaseFileHeader));
+	bufptr += MAXALIGN(hdr->gidlen);
+	
+	strncpy(parsed->twophase_gid, bufptr, hdr->gidlen);
 
 	parsed->twophase_xid = hdr->xid;
 	parsed->dbId = hdr->database;
@@ -1267,7 +1270,7 @@ ParsePrepareRecord(uint8 info, char *xlrec, xl_xact_parsed_prepare *parsed)
 	parsed->msgs = (SharedInvalidationMessage *) bufptr;
 	bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
 
-	strncpy(parsed->twophase_gid, bufptr, hdr->gidlen);
+	// strncpy(parsed->twophase_gid, bufptr, hdr->gidlen);
 }
 
 

src/backend/access/transam/xact.c

@@ -2728,6 +2728,8 @@ CommitTransactionCommand(void)
 {
 	TransactionState s = CurrentTransactionState;
 
+	CallXactCallbacks(XACT_EVENT_COMMIT_COMMAND);
+
 	switch (s->blockState)
 	{
 			/*

src/include/access/xact.h

@@ -92,7 +92,8 @@ typedef enum
 	XACT_EVENT_PRE_PREPARE,
 	XACT_EVENT_POST_PREPARE,
 	XACT_EVENT_COMMIT_PREPARED,
-	XACT_EVENT_ABORT_PREPARED
+	XACT_EVENT_ABORT_PREPARED,
+	XACT_EVENT_COMMIT_COMMAND
 } XactEvent;
 
 typedef void (*XactCallback) (XactEvent event, void *arg);