Add visibility functions for global temp table tuples

Author: knizhnik

contrib/pgrowlocks/pgrowlocks.c

@@ -158,7 +158,8 @@ pgrowlocks(PG_FUNCTION_ARGS)
 		/* must hold a buffer lock to call HeapTupleSatisfiesUpdate */
 		LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_SHARE);
 
-		htsu = HeapTupleSatisfiesUpdate(tuple,
+		htsu = HeapTupleSatisfiesUpdate(mydata->rel,
+										tuple,
 										GetCurrentCommandId(false),
 										hscan->rs_cbuf);
 		xmax = HeapTupleHeaderGetRawXmax(tuple->t_data);

src/backend/access/heap/heapam.c

@@ -1855,7 +1855,7 @@ ReleaseBulkInsertStatePin(BulkInsertState bistate)
 static TransactionId
 GetTransactionId(Relation relation)
 {
-	TransactionId xid = relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION
+	return relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION && RecoveryInProgress()
 		? GetReplicaTransactionId()
 		: GetCurrentTransactionId();
 }

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,14 +455,17 @@ 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;
 
 	Assert(ItemPointerIsValid(&htup->t_self));
 	Assert(htup->t_tableOid != InvalidOid);
 
+	if (relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION && RecoveryInProgress())
+		return TempTupleSatisfiesVisibility(htup, curcid, buffer) ? TM_Ok : TM_Invisible;
+
 	if (!HeapTupleHeaderXminCommitted(tuple))
 	{
 		if (HeapTupleHeaderXminInvalid(tuple))
@@ -1676,6 +1680,70 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
 		return true;
 }
 
+/*
+ * TempTupleSatisfiesVisibility
+ *		True iff global temp table tuple is visible for the given snapshot.
+ *
+ * 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;
+
+	if (tuple->t_infomask & HEAP_XMAX_IS_MULTI)
+	{
+		xmax = HeapTupleGetUpdateXid(tuple);
+
+		/* updating subtransaction must have aborted */
+		if (IsReplicaTransactionAborted(xmax))
+			return true;
+
+		return (HeapTupleHeaderGetCmax(tuple) >= curcid);
+	}
+	xmax = HeapTupleHeaderGetRawXmax(tuple);
+
+	/* deleting subtransaction must have aborted */
+	if (IsReplicaTransactionAborted(xmax))
+		return true;
+
+	if (IsReplicaCurrentTransactionId(xmax))
+		return (HeapTupleHeaderGetCmax(tuple) >= curcid); /* deleted after scan started */
+
+	/* xmax transaction committed */
+	return false;
+}
+
+
 /*
  * HeapTupleSatisfiesVisibility
  *		True iff heap tuple satisfies a time qual.
@@ -1689,8 +1757,8 @@ HeapTupleSatisfiesHistoricMVCC(HeapTuple htup, Snapshot snapshot,
 bool
 HeapTupleSatisfiesVisibility(Relation relation, HeapTuple tup, Snapshot snapshot, Buffer buffer)
 {
-	if (relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION)
-		return TempTupleSatisfiesMVCC(tup, snapshot, buffer);
+	if (relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION && RecoveryInProgress())
+		return TempTupleSatisfiesVisibility(tup, snapshot->curcid, buffer);
 
 	switch (snapshot->snapshot_type)
 	{

src/backend/access/transam/varsup.c

@@ -69,11 +69,6 @@ GetNewTransactionId(bool isSubXact)
 		return FullTransactionIdFromEpochAndXid(0, BootstrapTransactionId);
 	}
 
-	/* Make it possible to access global temporary tables at standby */
-	if (RecoveryInProgress())
-		return FullTransactionIdFromEpochAndXid(0, FrozenTransactionId);
-   	    /* elog(ERROR, "cannot assign TransactionIds during recovery"); */
-
 	LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
 
 	full_xid = ShmemVariableCache->nextFullXid;

src/backend/access/transam/xact.c

@@ -192,7 +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 replicXid;    /* pseudo XID for inserting data in global temp tables at replica */
+	TransactionId replicaTransactionId;    /* pseudo XID for inserting data in global temp tables at replica */
 } TransactionStateData;
 
 typedef TransactionStateData *TransactionState;
@@ -288,6 +288,7 @@ typedef struct XactCallbackItem
 static XactCallbackItem *Xact_callbacks = NULL;
 
 static TransactionId replicaTransIdCount;
+static TransactionId replicaTopTransId;
 static Bitmapset*    replicaAbortedXids;
 
 /*
@@ -433,7 +434,28 @@ GetCurrentTransactionId(void)
 	return XidFromFullTransactionId(s->fullTransactionId);
 }
 
+/*
+ *	GetCurrentTransactionIdIfAny
+ *
+ * This will return the XID of the current sub xact, if one is assigned.
+ * It will return InvalidTransactionId if we are not currently inside a
+ * transaction, or inside a transaction that hasn't been assigned an XID yet.
+ */
+TransactionId
+GetCurrentTransactionIdIfAny(void)
+{
+	return XidFromFullTransactionId(CurrentTransactionState->fullTransactionId);
+}
 
+/*
+ * 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)
 {
@@ -443,23 +465,28 @@ GetReplicaTransactionId(void)
 	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(replicaAbortedXids, xid);
+	return bms_is_member(xid, replicaAbortedXids);
 }
 
 /*
- *	GetCurrentTransactionIdIfAny
- *
- * This will return the XID of the current sub xact, if one is assigned.
- * It will return InvalidTransactionId if we are not currently inside a
- * transaction, or inside a transaction that hasn't been assigned an XID yet.
+ * 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.
  */
-TransactionId
-GetCurrentTransactionIdIfAny(void)
+bool
+IsReplicaCurrentTransactionId(TransactionId xid)
 {
-	return XidFromFullTransactionId(CurrentTransactionState->fullTransactionId);
+	return xid > replicaTopTransId;
 }
 
 /*
@@ -1912,6 +1939,8 @@ StartTransaction(void)
 	s = &TopTransactionStateData;
 	CurrentTransactionState = s;
 
+	replicaTopTransId = replicaTransIdCount;
+
 	Assert(!FullTransactionIdIsValid(XactTopFullTransactionId));
 
 	/* check the current transaction state */

src/include/access/heapam.h

@@ -197,7 +197,7 @@ extern void heap_vacuum_rel(Relation onerel,
 /* in heap/heapam_visibility.c */
 extern bool HeapTupleSatisfiesVisibility(Relation relation, HeapTuple stup, Snapshot snapshot,
 										 Buffer buffer);
-extern TM_Result HeapTupleSatisfiesUpdate(HeapTuple stup, CommandId curcid,
+extern TM_Result HeapTupleSatisfiesUpdate(Relation relation, HeapTuple stup, CommandId curcid,
 										  Buffer buffer);
 extern HTSV_Result HeapTupleSatisfiesVacuum(HeapTuple stup, TransactionId OldestXmin,
 											Buffer buffer);

src/include/access/xact.h

@@ -442,5 +442,6 @@ extern bool IsInParallelMode(void);
 
 extern TransactionId GetReplicaTransactionId(void);
 extern bool IsReplicaTransactionAborted(TransactionId xid);
+extern bool IsReplicaCurrentTransactionId(TransactionId xid);
 
 #endif							/* XACT_H */