Support rollback at replia

Author: knizhnik

contrib/pgstattuple/pgstattuple.c

@@ -349,7 +349,7 @@ pgstat_heap(Relation rel, FunctionCallInfo fcinfo)
 		/* must hold a buffer lock to call HeapTupleSatisfiesVisibility */
 		LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_SHARE);
 
-		if (HeapTupleSatisfiesVisibility(tuple, &SnapshotDirty, hscan->rs_cbuf))
+		if (HeapTupleSatisfiesVisibility(rel, tuple, &SnapshotDirty, hscan->rs_cbuf))
 		{
 			stat.tuple_len += tuple->t_len;
 			stat.tuple_count++;

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);
@@ -1612,7 +1613,7 @@ heap_hot_search_buffer(ItemPointer tid, Relation relation, Buffer buffer,
 			ItemPointerSet(&(heapTuple->t_self), BufferGetBlockNumber(buffer), offnum);
 
 			/* 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);
 			/* reset to original, non-redirected, tid */
@@ -1754,7 +1755,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;
@@ -1851,6 +1852,14 @@ ReleaseBulkInsertStatePin(BulkInsertState bistate)
 }
 
 
+static TransactionId
+GetTransactionId(Relation relation)
+{
+	TransactionId xid = relation->rd_rel->relpersistence == RELPERSISTENCE_SESSION
+		? GetReplicaTransactionId()
+		: GetCurrentTransactionId();
+}
+
 /*
  *	heap_insert		- insert tuple into a heap
  *
@@ -1873,7 +1882,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;
@@ -2110,7 +2119,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;
@@ -2449,7 +2458,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;
@@ -2514,7 +2523,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)
 	{
@@ -2633,7 +2642,7 @@ heap_delete(Relation relation, ItemPointer tid,
 	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;
 	}
 
@@ -2900,7 +2909,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;
@@ -3070,7 +3079,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);
@@ -3262,7 +3271,7 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 	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));
@@ -4018,7 +4027,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)
 	{
@@ -4193,7 +4202,7 @@ heap_lock_tuple(Relation relation, HeapTuple tuple,
 					TM_Result	res;
 
 					res = heap_lock_updated_tuple(relation, tuple, &t_ctid,
-												  GetCurrentTransactionId(),
+												  GetTransactionId(relation),
 												  mode);
 					if (res != TM_Ok)
 					{
@@ -4441,7 +4450,7 @@ heap_lock_tuple(Relation relation, HeapTuple tuple,
 				TM_Result	res;
 
 				res = heap_lock_updated_tuple(relation, tuple, &t_ctid,
-											  GetCurrentTransactionId(),
+											  GetTransactionId(relation),
 											  mode);
 				if (res != TM_Ok)
 				{
@@ -4550,7 +4559,7 @@ heap_lock_tuple(Relation relation, HeapTuple tuple,
 	 * 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();
@@ -5570,7 +5579,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;

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);
 
@@ -2161,7 +2162,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;
@@ -2481,7 +2482,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);
 	}
 }

src/backend/access/heap/heapam_visibility.c

@@ -1687,8 +1687,11 @@ 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)
+		return TempTupleSatisfiesMVCC(tup, snapshot, buffer);
+
 	switch (snapshot->snapshot_type)
 	{
 		case SNAPSHOT_MVCC:

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 replicXid;    /* pseudo XID for inserting data in global temp tables at replica */
 } TransactionStateData;
 
 typedef TransactionStateData *TransactionState;
@@ -286,6 +287,9 @@ typedef struct XactCallbackItem
 
 static XactCallbackItem *Xact_callbacks = NULL;
 
+static TransactionId replicaTransIdCount;
+static Bitmapset*    replicaAbortedXids;
+
 /*
  * List of add-on start- and end-of-subxact callbacks
  */
@@ -429,6 +433,22 @@ GetCurrentTransactionId(void)
 	return XidFromFullTransactionId(s->fullTransactionId);
 }
 
+
+TransactionId
+GetReplicaTransactionId(void)
+{
+	TransactionState s = CurrentTransactionState;
+	if (!TransactionIdIsValid(s->replicaTransactionId))
+		s->replicaTransactionId = ++replicaTransIdCount;
+	return s->replicaTransactionId;
+}
+
+bool
+IsReplicaTransactionAborted(TransactionId xid)
+{
+	return bms_is_member(replicaAbortedXids, xid);
+}
+
 /*
  *	GetCurrentTransactionIdIfAny
  *
@@ -1905,6 +1925,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 +2591,10 @@ 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))
+		replicaAbortedXids = bms_add_member(replicaAbortedXids, s->replicaTransactionId);
+
 	/* Make sure we have a valid memory context and resource owner */
 	AtAbort_Memory();
 	AtAbort_ResourceOwner();
@@ -4859,6 +4884,10 @@ 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))
+		replicaAbortedXids = bms_add_member(replicaAbortedXids, s->replicaTransactionId);
+
 	/* Make sure we have a valid memory context and resource owner */
 	AtSubAbort_Memory();
 	AtSubAbort_ResourceOwner();

src/include/access/heapam.h

@@ -195,7 +195,7 @@ extern void heap_vacuum_rel(Relation onerel,
 							struct VacuumParams *params, BufferAccessStrategy bstrategy);
 
 /* in heap/heapam_visibility.c */
-extern bool HeapTupleSatisfiesVisibility(HeapTuple stup, Snapshot snapshot,
+extern bool HeapTupleSatisfiesVisibility(Relation relation, HeapTuple stup, Snapshot snapshot,
 										 Buffer buffer);
 extern TM_Result HeapTupleSatisfiesUpdate(HeapTuple stup, CommandId curcid,
 										  Buffer buffer);

src/include/access/xact.h

@@ -440,4 +440,7 @@ extern void EnterParallelMode(void);
 extern void ExitParallelMode(void);
 extern bool IsInParallelMode(void);
 
+extern TransactionId GetReplicaTransactionId(void);
+extern bool IsReplicaTransactionAborted(TransactionId xid);
+
 #endif							/* XACT_H */