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();