4343#include "storage/pmsignal.h"
4444#include "storage/proc.h"
4545#include "utils/syscache.h"
46+ #include "replication/walsender.h"
4647#include "port/atomics.h"
4748
4849#include "sockhub/sockhub.h"
@@ -757,7 +758,7 @@ DtmXactCallback(XactEvent event, void *arg)
757758case XACT_EVENT_START :
758759//XTM_INFO("%d: normal=%d, initialized=%d, replication=%d, bgw=%d, vacuum=%d\n",
759760// getpid(), IsNormalProcessingMode(), dtm->initialized, MMDoReplication, IsBackgroundWorker, IsAutoVacuumWorkerProcess());
760- if (IsNormalProcessingMode ()&& dtm -> initialized && MMDoReplication && !IsBackgroundWorker && !IsAutoVacuumWorkerProcess ()) {
761+ if (IsNormalProcessingMode ()&& dtm -> initialized && MMDoReplication && !am_walsender && ! IsBackgroundWorker && !IsAutoVacuumWorkerProcess ()) {
761762MMBeginTransaction ();
762763 }
763764break ;
@@ -1190,7 +1191,7 @@ static void MMExecutor(int id, void* work, size_t size)
11901191{
11911192TransactionId xid = * (TransactionId * )work ;
11921193char * stmts = (char * )work + 4 ;
1193- int rc = SPI_ERROR_TRANSACTION ;
1194+ bool finished = false ;
11941195
11951196MMJoinTransaction (xid );
11961197
@@ -1201,9 +1202,10 @@ static void MMExecutor(int id, void* work, size_t size)
12011202
12021203PG_TRY ();
12031204 {
1204- rc = SPI_execute (stmts , false,0 );
1205+ int rc = SPI_execute (stmts , false,0 );
12051206SPI_finish ();
12061207PopActiveSnapshot ();
1208+ finished = true;
12071209if (rc != SPI_OK_INSERT && rc != SPI_OK_UPDATE && rc != SPI_OK_DELETE ) {
12081210ereport (LOG , (errmsg ("Executor %d: failed to apply transaction %u" ,
12091211id ,xid )));
@@ -1215,9 +1217,11 @@ static void MMExecutor(int id, void* work, size_t size)
12151217PG_CATCH ();
12161218 {
12171219FlushErrorState ();
1218- if (rc == SPI_ERROR_TRANSACTION ) {
1220+ if (! finished ) {
12191221SPI_finish ();
1220- PopActiveSnapshot ();
1222+ if (ActiveSnapshotSet ()) {
1223+ PopActiveSnapshot ();
1224+ }
12211225 }
12221226AbortCurrentTransaction ();
12231227 }