replica-gtt-xact.diff

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index e954482..2f5f5ef 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -444,7 +444,7 @@ heapgetpage(TableScanDesc sscan, BlockNumber page)
 			if (all_visible)
 				valid = true;
 			else
-				valid = HeapTupleSatisfiesVisibility(&loctup, snapshot, buffer);
+				valid = HeapTupleSatisfiesVisibility(scan->rs_base.rs_rd, &loctup, snapshot, buffer);
 
 			CheckForSerializableConflictOut(valid, scan->rs_base.rs_rd,
 											&loctup, buffer, snapshot);
@@ -664,7 +664,8 @@ heapgettup(HeapScanDesc scan,
 				/*
 				 * if current tuple qualifies, return it.
 				 */
-				valid = HeapTupleSatisfiesVisibility(tuple,
+				valid = HeapTupleSatisfiesVisibility(scan->rs_base.rs_rd,
+													 tuple,
 													 snapshot,
 													 scan->rs_cbuf);
 
@@ -1474,7 +1475,7 @@ heap_fetch(Relation relation,
 	/*
 	 * check tuple visibility, then release lock
 	 */
-	valid = HeapTupleSatisfiesVisibility(tuple, snapshot, buffer);
+	valid = HeapTupleSatisfiesVisibility(relation, tuple, snapshot, buffer);
 
 	if (valid)
 		PredicateLockTuple(relation, tuple, snapshot);
@@ -1609,7 +1610,7 @@ heap_hot_search_buffer(ItemPointer tid, Relation relation, Buffer buffer,
 		if (!skip)
 		{
 			/* If it's visible per the snapshot, we must return it */
-			valid = HeapTupleSatisfiesVisibility(heapTuple, snapshot, buffer);
+			valid = HeapTupleSatisfiesVisibility(relation, heapTuple, snapshot, buffer);
 			CheckForSerializableConflictOut(valid, relation, heapTuple,
 											buffer, snapshot);
 
@@ -1749,7 +1750,7 @@ heap_get_latest_tid(TableScanDesc sscan,
 		 * Check tuple visibility; if visible, set it as the new result
 		 * candidate.
 		 */
-		valid = HeapTupleSatisfiesVisibility(&tp, snapshot, buffer);
+		valid = HeapTupleSatisfiesVisibility(relation, &tp, snapshot, buffer);
 		CheckForSerializableConflictOut(valid, relation, &tp, buffer, snapshot);
 		if (valid)
 			*tid = ctid;
@@ -1846,6 +1847,14 @@ ReleaseBulkInsertStatePin(BulkInsertState bistate)
 }
 
 
+static TransactionId
+GetTransactionId(Relation relation)
+{
+	return relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION && RecoveryInProgress()
+		? GetReplicaTransactionId()
+		: GetCurrentTransactionId();
+}
+
 /*
  *	heap_insert		- insert tuple into a heap
  *
@@ -1868,7 +1877,7 @@ void
 heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 			int options, BulkInsertState bistate)
 {
-	TransactionId xid = GetCurrentTransactionId();
+	TransactionId xid = GetTransactionId(relation);
 	HeapTuple	heaptup;
 	Buffer		buffer;
 	Buffer		vmbuffer = InvalidBuffer;
@@ -2105,7 +2114,7 @@ void
 heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
 				  CommandId cid, int options, BulkInsertState bistate)
 {
-	TransactionId xid = GetCurrentTransactionId();
+	TransactionId xid = GetTransactionId(relation);
 	HeapTuple  *heaptuples;
 	int			i;
 	int			ndone;
@@ -2444,7 +2453,7 @@ heap_delete(Relation relation, ItemPointer tid,
 			TM_FailureData *tmfd, bool changingPart)
 {
 	TM_Result	result;
-	TransactionId xid = GetCurrentTransactionId();
+	TransactionId xid = GetTransactionId(relation);
 	ItemId		lp;
 	HeapTupleData tp;
 	Page		page;
@@ -2509,7 +2518,7 @@ heap_delete(Relation relation, ItemPointer tid,
 	tp.t_self = *tid;
 
 l1:
-	result = HeapTupleSatisfiesUpdate(&tp, cid, buffer);
+	result = HeapTupleSatisfiesUpdate(relation, &tp, cid, buffer);
 
 	if (result == TM_Invisible)
 	{
@@ -2628,7 +2637,7 @@ l1:
 	if (crosscheck != InvalidSnapshot && result == TM_Ok)
 	{
 		/* Perform additional check for transaction-snapshot mode RI updates */
-		if (!HeapTupleSatisfiesVisibility(&tp, crosscheck, buffer))
+		if (!HeapTupleSatisfiesVisibility(relation, &tp, crosscheck, buffer))
 			result = TM_Updated;
 	}
 
@@ -2895,7 +2904,7 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 			TM_FailureData *tmfd, LockTupleMode *lockmode)
 {
 	TM_Result	result;
-	TransactionId xid = GetCurrentTransactionId();
+	TransactionId xid = GetTransactionId(relation);
 	Bitmapset  *hot_attrs;
 	Bitmapset  *key_attrs;
 	Bitmapset  *id_attrs;
@@ -3065,7 +3074,7 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 l2:
 	checked_lockers = false;
 	locker_remains = false;
-	result = HeapTupleSatisfiesUpdate(&oldtup, cid, buffer);
+	result = HeapTupleSatisfiesUpdate(relation, &oldtup, cid, buffer);
 
 	/* see below about the "no wait" case */
 	Assert(result != TM_BeingModified || wait);
@@ -3258,7 +3267,7 @@ l2:
 	if (crosscheck != InvalidSnapshot && result == TM_Ok)
 	{
 		/* Perform additional check for transaction-snapshot mode RI updates */
-		if (!HeapTupleSatisfiesVisibility(&oldtup, crosscheck, buffer))
+		if (!HeapTupleSatisfiesVisibility(relation, &oldtup, crosscheck, buffer))
 		{
 			result = TM_Updated;
 			Assert(!ItemPointerEquals(&oldtup.t_self, &oldtup.t_data->t_ctid));
@@ -4014,7 +4023,7 @@ heap_lock_tuple(Relation relation, HeapTuple tuple,
 	tuple->t_tableOid = RelationGetRelid(relation);
 
 l3:
-	result = HeapTupleSatisfiesUpdate(tuple, cid, *buffer);
+	result = HeapTupleSatisfiesUpdate(relation, tuple, cid, *buffer);
 
 	if (result == TM_Invisible)
 	{
@@ -4189,7 +4198,7 @@ l3:
 					TM_Result	res;
 
 					res = heap_lock_updated_tuple(relation, tuple, &t_ctid,
-												  GetCurrentTransactionId(),
+												  GetTransactionId(relation),
 												  mode);
 					if (res != TM_Ok)
 					{
@@ -4437,7 +4446,7 @@ l3:
 				TM_Result	res;
 
 				res = heap_lock_updated_tuple(relation, tuple, &t_ctid,
-											  GetCurrentTransactionId(),
+											  GetTransactionId(relation),
 											  mode);
 				if (res != TM_Ok)
 				{
@@ -4546,7 +4555,7 @@ failed:
 	 * state if multixact.c elogs.
 	 */
 	compute_new_xmax_infomask(xmax, old_infomask, tuple->t_data->t_infomask2,
-							  GetCurrentTransactionId(), mode, false,
+							  GetTransactionId(relation), mode, false,
 							  &xid, &new_infomask, &new_infomask2);
 
 	START_CRIT_SECTION();
@@ -5566,7 +5575,7 @@ heap_finish_speculative(Relation relation, ItemPointer tid)
 void
 heap_abort_speculative(Relation relation, ItemPointer tid)
 {
-	TransactionId xid = GetCurrentTransactionId();
+	TransactionId xid = GetTransactionId(relation);
 	ItemId		lp;
 	HeapTupleData tp;
 	Page		page;
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 2dd8821..67a553a 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -226,7 +226,8 @@ heapam_tuple_satisfies_snapshot(Relation rel, TupleTableSlot *slot,
 	 * Caller should be holding pin, but not lock.
 	 */
 	LockBuffer(bslot->buffer, BUFFER_LOCK_SHARE);
-	res = HeapTupleSatisfiesVisibility(bslot->base.tuple, snapshot,
+
+	res = HeapTupleSatisfiesVisibility(rel, bslot->base.tuple, snapshot,
 									   bslot->buffer);
 	LockBuffer(bslot->buffer, BUFFER_LOCK_UNLOCK);
 
@@ -673,6 +674,7 @@ heapam_relation_copy_data(Relation rel, const RelFileNode *newrnode)
 			 * init fork of an unlogged relation.
 			 */
 			if (rel->rd_rel->relpersistence == RELPERSISTENCE_PERMANENT ||
+				rel->rd_rel->relpersistence == RELPERSISTENCE_SESSION ||
 				(rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED &&
 				 forkNum == INIT_FORKNUM))
 				log_smgrcreate(newrnode, forkNum);
@@ -2162,7 +2164,7 @@ heapam_scan_bitmap_next_block(TableScanDesc scan,
 			loctup.t_len = ItemIdGetLength(lp);
 			loctup.t_tableOid = scan->rs_rd->rd_id;
 			ItemPointerSet(&loctup.t_self, page, offnum);
-			valid = HeapTupleSatisfiesVisibility(&loctup, snapshot, buffer);
+			valid = HeapTupleSatisfiesVisibility(scan->rs_rd, &loctup, snapshot, buffer);
 			if (valid)
 			{
 				hscan->rs_vistuples[ntup++] = offnum;
@@ -2482,7 +2484,7 @@ SampleHeapTupleVisible(TableScanDesc scan, Buffer buffer,
 	else
 	{
 		/* Otherwise, we have to check the tuple individually. */
-		return HeapTupleSatisfiesVisibility(tuple, scan->rs_snapshot,
+		return HeapTupleSatisfiesVisibility(scan->rs_rd, tuple, scan->rs_snapshot,
 											buffer);
 	}
 }
diff --git a/src/backend/access/heap/heapam_visibility.c b/src/backend/access/heap/heapam_visibility.c
index 537e681..3076f6a 100644
--- a/src/backend/access/heap/heapam_visibility.c
+++ b/src/backend/access/heap/heapam_visibility.c
@@ -77,6 +77,7 @@
 #include "utils/combocid.h"
 #include "utils/snapmgr.h"
 
+static bool TempTupleSatisfiesVisibility(HeapTuple htup, CommandId curcid, Buffer buffer);
 
 /*
  * SetHintBits()
@@ -454,7 +455,7 @@ HeapTupleSatisfiesToast(HeapTuple htup, Snapshot snapshot,
  *	test for it themselves.)
  */
 TM_Result
-HeapTupleSatisfiesUpdate(HeapTuple htup, CommandId curcid,
+HeapTupleSatisfiesUpdate(Relation relation, HeapTuple htup, CommandId curcid,
 						 Buffer buffer)
 {
 	HeapTupleHeader tuple = htup->t_data;
@@ -462,6 +463,13 @@ HeapTupleSatisfiesUpdate(HeapTuple htup, CommandId curcid,
 	Assert(ItemPointerIsValid(&htup->t_self));
 	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))
 	{
 		if (HeapTupleHeaderXminInvalid(tuple))
@@ -1677,6 +1685,59 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
 }
 
 /*
+ * TempTupleSatisfiesVisibility
+ *		True iff global temp table tuple is visible for the current transaction.
+ *
+ * Temporary tables are visible only for current backend, so there is no need to
+ * handle cases with tuples committed by other backends. We only need to exclude
+ * modifications done by aborted transactions or after start of table scan.
+ *
+ */
+static bool
+TempTupleSatisfiesVisibility(HeapTuple htup, CommandId curcid, Buffer buffer)
+{
+	HeapTupleHeader tuple = htup->t_data;
+	TransactionId xmin;
+	TransactionId xmax;
+
+	Assert(ItemPointerIsValid(&htup->t_self));
+	Assert(htup->t_tableOid != InvalidOid);
+
+	if (HeapTupleHeaderXminInvalid(tuple))
+		return false;
+
+	xmin = HeapTupleHeaderGetRawXmin(tuple);
+
+	if (IsReplicaTransactionAborted(xmin))
+		return false;
+
+	if (IsReplicaCurrentTransactionId(xmin)
+		&& HeapTupleHeaderGetCmin(tuple) >= curcid)
+	{
+		return false;	/* inserted after scan started */
+	}
+
+	if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid */
+		return true;
+
+	if (HEAP_XMAX_IS_LOCKED_ONLY(tuple->t_infomask))	/* not deleter */
+		return true;
+
+	xmax = (tuple->t_infomask & HEAP_XMAX_IS_MULTI)
+	    ? HeapTupleGetUpdateXid(tuple)
+		: HeapTupleHeaderGetRawXmax(tuple);
+
+	if (IsReplicaTransactionAborted(xmax))
+		return true; /* updating subtransaction aborted */
+
+	if (!IsReplicaCurrentTransactionId(xmax))
+		return false; /* updating transaction committed */
+
+	return (HeapTupleHeaderGetCmax(tuple) >= curcid);	/* updated after scan started */
+}
+
+
+/*
  * HeapTupleSatisfiesVisibility
  *		True iff heap tuple satisfies a time qual.
  *
@@ -1687,8 +1748,15 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
  *	if so, the indicated buffer is marked dirty.
  */
 bool
-HeapTupleSatisfiesVisibility(HeapTuple tup, Snapshot snapshot, Buffer buffer)
+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)
 	{
 		case SNAPSHOT_MVCC:
diff --git a/src/backend/access/transam/transam.c b/src/backend/access/transam/transam.c
index 365ddfb..bce9c4a 100644
--- a/src/backend/access/transam/transam.c
+++ b/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);
 
 	/*
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 9162286..cf6bf4e 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -192,6 +192,7 @@ typedef struct TransactionStateData
 	int			parallelModeLevel;	/* Enter/ExitParallelMode counter */
 	bool		chain;			/* start a new block after this one */
 	struct TransactionStateData *parent;	/* back link to parent */
+	TransactionId replicaTransactionId;    /* pseudo XID for inserting data in global temp tables at replica */
 } TransactionStateData;
 
 typedef TransactionStateData *TransactionState;
@@ -286,6 +287,12 @@ typedef struct XactCallbackItem
 
 static XactCallbackItem *Xact_callbacks = NULL;
 
+static TransactionId replicaTransIdCount = FirstNormalTransactionId;
+static TransactionId replicaTopTransId;
+static Bitmapset*    replicaAbortedXids;
+
+bool AccessTempRelationAtReplica;
+
 /*
  * List of add-on start- and end-of-subxact callbacks
  */
@@ -443,6 +450,48 @@ GetCurrentTransactionIdIfAny(void)
 }
 
 /*
+ * Transactions at replica can update only global temporary tables.
+ * Them are assigned backend-local XIDs which are independent from normal XIDs received from primary node.
+ * So tuples of temporary tables at replica requires special visibility rules.
+ *
+ * XIDs for such transactions at replica are created on demand (when tuple of temp table is updated).
+ * XID wrap-around and adjusting XID horizon is not supported. So number of such transactions at replica is
+ * limited by 2^32 and require up to 2^29 in-memory bitmap for marking aborted transactions.
+  */
+TransactionId
+GetReplicaTransactionId(void)
+{
+	TransactionState s = CurrentTransactionState;
+	if (!TransactionIdIsValid(s->replicaTransactionId))
+		s->replicaTransactionId = ++replicaTransIdCount;
+	return s->replicaTransactionId;
+}
+
+/*
+ * At replica transaction can update only temporary tables
+ * and them are assigned special XIDs (not related with normal XIDs received from primary node).
+ * As far as we see only own transaction it is not necessary to mark committed transactions.
+ * Only marking aborted ones is enough. All transactions which are not marked as aborted are treated as
+ * committed or self in-progress transactions.
+ */
+bool
+IsReplicaTransactionAborted(TransactionId xid)
+{
+	return bms_is_member(xid, replicaAbortedXids);
+}
+
+/*
+ * As far as XIDs for transactions at replica are generated individually for each backends,
+ * we can check that XID belongs to the current transaction or any of its subtransactions by
+ * just comparing it with XID of top transaction.
+ */
+bool
+IsReplicaCurrentTransactionId(TransactionId xid)
+{
+	return xid > replicaTopTransId;
+}
+
+/*
  *	GetTopFullTransactionId
  *
  * This will return the FullTransactionId of the main transaction, assigning
@@ -855,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
@@ -1206,7 +1258,7 @@ static TransactionId
 RecordTransactionCommit(void)
 {
 	TransactionId xid = GetTopTransactionIdIfAny();
-	bool		markXidCommitted = TransactionIdIsValid(xid);
+	bool		markXidCommitted = TransactionIdIsNormal(xid);
 	TransactionId latestXid = InvalidTransactionId;
 	int			nrels;
 	RelFileNode *rels;
@@ -1624,7 +1676,7 @@ RecordTransactionAbort(bool isSubXact)
 	 * rels to delete (note that this routine is not responsible for actually
 	 * deleting 'em).  We cannot have any child XIDs, either.
 	 */
-	if (!TransactionIdIsValid(xid))
+	if (!TransactionIdIsNormal(xid))
 	{
 		/* Reset XactLastRecEnd until the next transaction writes something */
 		if (!isSubXact)
@@ -1892,6 +1944,8 @@ StartTransaction(void)
 	s = &TopTransactionStateData;
 	CurrentTransactionState = s;
 
+	replicaTopTransId = replicaTransIdCount;
+
 	Assert(!FullTransactionIdIsValid(XactTopFullTransactionId));
 
 	/* check the current transaction state */
@@ -1905,6 +1959,7 @@ StartTransaction(void)
 	 */
 	s->state = TRANS_START;
 	s->fullTransactionId = InvalidFullTransactionId;	/* until assigned */
+	s->replicaTransactionId = InvalidTransactionId;	/* until assigned */
 
 	/* Determine if statements are logged in this transaction */
 	xact_is_sampled = log_xact_sample_rate != 0 &&
@@ -2570,6 +2625,14 @@ AbortTransaction(void)
 	/* Prevent cancel/die interrupt while cleaning up */
 	HOLD_INTERRUPTS();
 
+	/* 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();
 	AtAbort_ResourceOwner();
@@ -2991,6 +3054,9 @@ CommitTransactionCommand(void)
 			 * and then clean up.
 			 */
 		case TBLOCK_ABORT_PENDING:
+			if (GetCurrentTransactionIdIfAny() == FrozenTransactionId)
+				elog(FATAL, "Transaction is aborted at standby");
+
 			AbortTransaction();
 			CleanupTransaction();
 			s->blockState = TBLOCK_DEFAULT;
@@ -4880,6 +4946,14 @@ AbortSubTransaction(void)
 	/* Prevent cancel/die interrupt while cleaning up */
 	HOLD_INTERRUPTS();
 
+	/* 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();
 	AtSubAbort_ResourceOwner();