Fix handling of subtransactions working with temporary tables at replica

Author: knizhnik

src/backend/access/heap/heapam_visibility.c

@@ -464,7 +464,11 @@ HeapTupleSatisfiesUpdate(Relation relation, HeapTuple htup, CommandId curcid,
 	Assert(htup->t_tableOid != InvalidOid);
 
 	if (relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION && RecoveryInProgress())
+	{
+		AccessTempRelationAtReplica = true;
 		return TempTupleSatisfiesVisibility(htup, curcid, buffer) ? TM_Ok : TM_Invisible;
+	}
+	AccessTempRelationAtReplica = false;
 
 	if (!HeapTupleHeaderXminCommitted(tuple))
 	{
@@ -1747,7 +1751,11 @@ bool
 HeapTupleSatisfiesVisibility(Relation relation, HeapTuple tup, Snapshot snapshot, Buffer buffer)
 {
 	if (relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION && RecoveryInProgress())
+	{
+		AccessTempRelationAtReplica = true;
 		return TempTupleSatisfiesVisibility(tup, snapshot->curcid, buffer);
+	}
+	AccessTempRelationAtReplica = false;
 
 	switch (snapshot->snapshot_type)
 	{

src/backend/access/transam/transam.c

@@ -22,6 +22,7 @@
 #include "access/clog.h"
 #include "access/subtrans.h"
 #include "access/transam.h"
+#include "access/xact.h"
 #include "utils/snapmgr.h"
 
 /*
@@ -126,6 +127,9 @@ TransactionIdDidCommit(TransactionId transactionId)
 {
 	XidStatus	xidstatus;
 
+	if (AccessTempRelationAtReplica)
+		return !IsReplicaCurrentTransactionId(transactionId) && !IsReplicaTransactionAborted(transactionId);
+
 	xidstatus = TransactionLogFetch(transactionId);
 
 	/*
@@ -182,6 +186,9 @@ TransactionIdDidAbort(TransactionId transactionId)
 {
 	XidStatus	xidstatus;
 
+	if (AccessTempRelationAtReplica)
+		return IsReplicaTransactionAborted(transactionId);
+
 	xidstatus = TransactionLogFetch(transactionId);
 
 	/*

src/backend/access/transam/xact.c

@@ -287,10 +287,12 @@ typedef struct XactCallbackItem
 
 static XactCallbackItem *Xact_callbacks = NULL;
 
-static TransactionId replicaTransIdCount;
+static TransactionId replicaTransIdCount = FirstNormalTransactionId;
 static TransactionId replicaTopTransId;
 static Bitmapset*    replicaAbortedXids;
 
+bool AccessTempRelationAtReplica;
+
 /*
  * List of add-on start- and end-of-subxact callbacks
  */
@@ -902,6 +904,9 @@ TransactionIdIsCurrentTransactionId(TransactionId xid)
 {
 	TransactionState s;
 
+	if (AccessTempRelationAtReplica)
+		return IsReplicaCurrentTransactionId(xid);
+
 	/*
 	 * We always say that BootstrapTransactionId is "not my transaction ID"
 	 * even when it is (ie, during bootstrap).  Along with the fact that
@@ -2622,7 +2627,11 @@ AbortTransaction(void)
 
 	/* Mark transactions involved global temp table at replica as aborted */
 	if (TransactionIdIsValid(s->replicaTransactionId))
+	{
+		MemoryContext ctx = MemoryContextSwitchTo(TopMemoryContext);
 		replicaAbortedXids = bms_add_member(replicaAbortedXids, s->replicaTransactionId);
+		MemoryContextSwitchTo(ctx);
+	}
 
 	/* Make sure we have a valid memory context and resource owner */
 	AtAbort_Memory();
@@ -4915,7 +4924,11 @@ AbortSubTransaction(void)
 
 	/* Mark transactions involved global temp table at replica as aborted */
 	if (TransactionIdIsValid(s->replicaTransactionId))
+	{
+		MemoryContext ctx = MemoryContextSwitchTo(TopMemoryContext);
 		replicaAbortedXids = bms_add_member(replicaAbortedXids, s->replicaTransactionId);
+		MemoryContextSwitchTo(ctx);
+	}
 
 	/* Make sure we have a valid memory context and resource owner */
 	AtSubAbort_Memory();

src/backend/storage/ipc/procarray.c

@@ -994,6 +994,9 @@ TransactionIdIsInProgress(TransactionId xid)
 	int			i,
 				j;
 
+	if (AccessTempRelationAtReplica)
+		return IsReplicaCurrentTransactionId(xid) && !IsReplicaTransactionAborted(xid);
+
 	/*
 	 * Don't bother checking a transaction older than RecentXmin; it could not
 	 * possibly still be running.  (Note: in particular, this guarantees that

src/include/access/xact.h

@@ -41,6 +41,9 @@
 extern int	DefaultXactIsoLevel;
 extern PGDLLIMPORT int XactIsoLevel;
 
+extern bool AccessTempRelationAtReplica;
+
+
 /*
  * We implement three isolation levels internally.
  * The two stronger ones use one snapshot per database transaction;