diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 2874dc0612..c919d0fa77 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -405,6 +405,26 @@ DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) { xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r); + /* + * We rely on HEAP2_NEW_CID records and XACT_INVALIDATIONS to know + * if the transaction has changed the catalog, and that information + * is not serialized to SnapBuilder. Therefore, if the logical + * decoding decodes the commit record of the transaction that actually + * has done catalog changes without these records, we miss to add + * the xid to the snapshot so up creating the wrong snapshot. To + * avoid such a problem, if the COMMIT record of the xid listed in + * the RUNNING_XACTS record read at the start of logical decoding + * has XACT_XINFO_HAS_INVALS flag, we mark both the top transaction + * and its substransactions as containing catalog changes (see also + * ReorderBufferSetLastRunningXactsCatalogChanges()). Since we cannot + * know which transactions actually have done catalog changes only + * by reading the COMMIT record we do that for both. So we might + * mark an xid that actually has not done that but it’s not a + * problem since we use historic snapshot only for reading system + * catalogs. + */ + ReorderBufferProcessLastRunningXacts(ctx->reorder, running); + SnapBuildProcessRunningXacts(builder, buf->origptr, running); /* @@ -689,6 +709,15 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, commit_time = parsed->origin_timestamp; } + /* + * Set the last running xacts as containing catalog change if necessary. + * This must be done before SnapBuildCommitTxn() so that we include catalog + * change transactions to the historic snapshot. + */ + ReorderBufferSetLastRunningXactsCatalogChanges(ctx->reorder, xid, parsed->xinfo, + parsed->nsubxacts, parsed->subxacts, + buf->origptr); + SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid, parsed->nsubxacts, parsed->subxacts); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 46e66608cf..e2b688e107 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -346,6 +346,9 @@ ReorderBufferAllocate(void) buffer->outbufsize = 0; buffer->size = 0; + buffer->last_running_xacts = NULL; + buffer->n_last_running_xacts = -1; + buffer->spillTxns = 0; buffer->spillCount = 0; buffer->spillBytes = 0; @@ -5155,3 +5158,91 @@ restart: *cmax = ent->cmax; return true; } + +void +ReorderBufferProcessLastRunningXacts(ReorderBuffer *rb, xl_running_xacts *running) +{ + /* Quick exit if there is no longer last running xacts */ + if (likely(rb->n_last_running_xacts == 0)) + return; + + /* First call, build the last running xact list */ + if (rb->n_last_running_xacts == -1) + { + int nxacts = running->subxcnt + running->xcnt; + Size sz = sizeof(TransactionId) * nxacts;; + + rb->last_running_xacts = MemoryContextAlloc(rb->context, sz); + memcpy(rb->last_running_xacts, running->xids, sz); + qsort(rb->last_running_xacts, nxacts, sizeof(TransactionId), xidComparator); + + rb->n_last_running_xacts = nxacts; + + return; + } + + /* + * Purge xids in the last running xacts list if we can do that for at least + * one xid. + */ + if (NormalTransactionIdPrecedes(rb->last_running_xacts[0], + running->oldestRunningXid)) + { + TransactionId *workspace; + int nxids = 0; + + workspace = MemoryContextAlloc(rb->context, rb->n_last_running_xacts); + for (int i = 0; i < rb->n_last_running_xacts; i++) + { + if (NormalTransactionIdPrecedes(rb->last_running_xacts[i], + running->oldestRunningXid)) + ; /* remove */ + else + workspace[nxids++] = rb->last_running_xacts[i]; + } + + if (nxids > 0) + memcpy(rb->last_running_xacts, workspace, sizeof(TransactionId) * nxids); + else + { + pfree(rb->last_running_xacts); + rb->last_running_xacts = NULL; + } + + rb->n_last_running_xacts = nxids; + } +} + +void +ReorderBufferSetLastRunningXactsCatalogChanges(ReorderBuffer *rb, TransactionId xid, + uint32 xinfo, int subxcnt, + TransactionId *subxacts, XLogRecPtr lsn) +{ + void *test; + + /* + * Skip if there is no longer last running xacts information or the COMMIT record + * doesn't have invalidation message, which is a common case. + */ + if (likely(rb->n_last_running_xacts == 0 || !(xinfo & XACT_XINFO_HAS_INVALS))) + return; + + test = bsearch(&xid, rb->last_running_xacts, rb->n_last_running_xacts, + sizeof(TransactionId), xidComparator); + + /* + * If this committed transaction is the one that was running at the time when + * decoding the first RUNNING_XACTS record and have done catalog changes, we + * can mark the top transaction and its subtransactions as catalog-changes. + */ + if (test != NULL) + { + ReorderBufferXidSetCatalogChanges(rb, xid, lsn); + + for (int i = 0; i < subxcnt; i++) + { + ReorderBufferAssignChild(rb, xid, subxacts[i], lsn); + ReorderBufferXidSetCatalogChanges(rb, subxacts[i], lsn); + } + } +} diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 5b40ff75f7..234d7c0c61 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -12,6 +12,7 @@ #include "access/htup_details.h" #include "lib/ilist.h" #include "storage/sinval.h" +#include "storage/standby.h" #include "utils/hsearch.h" #include "utils/relcache.h" #include "utils/snapshot.h" @@ -593,6 +594,9 @@ struct ReorderBuffer /* memory accounting */ Size size; + TransactionId *last_running_xacts; + int n_last_running_xacts; /* -1 for initial value */ + /* * Statistics about transactions spilled to disk. * @@ -682,4 +686,9 @@ void ReorderBufferSetRestartPoint(ReorderBuffer *, XLogRecPtr ptr); void StartupReorderBuffer(void); +void ReorderBufferProcessLastRunningXacts(ReorderBuffer *rb, xl_running_xacts *running); +void ReorderBufferSetLastRunningXactsCatalogChanges(ReorderBuffer *rb, TransactionId xid, + uint32 xinfo, int subxcnt, + TransactionId *subxacts, XLogRecPtr lsn); + #endif