/*------------------------------------------------------------------------- * * snapbuild . c * * Infrastructure for building historic catalog snapshots based on contents * of the WAL , for the purpose of decoding heapam . c style values in the * WAL . * * NOTES : * * We build snapshots which can * only * be used to read catalog contents and we * do so by reading and interpreting the WAL stream . The aim is to build a * snapshot that behaves the same as a freshly taken MVCC snapshot would have * at the time the XLogRecord was generated . * * To build the snapshots we reuse the infrastructure built for Hot * Standby . The in - memory snapshots we build look different than HS ' because * we have different needs . To successfully decode data from the WAL we only * need to access catalog tables and ( sys | rel | cat ) cache , not the actual user * tables since the data we decode is wholly contained in the WAL * records . Also , our snapshots need to be different in comparison to normal * MVCC ones because in contrast to those we cannot fully rely on the clog and * pg_subtrans for information about committed transactions because they might * commit in the future from the POV of the WAL entry we ' re currently * decoding . This definition has the advantage that we only need to prevent * removal of catalog rows , while normal table ' s rows can still be * removed . This is achieved by using the replication slot mechanism . * * As the percentage of transactions modifying the catalog normally is fairly * small in comparisons to ones only manipulating user data , we keep track of * the committed catalog modifying ones inside [ xmin , xmax ) instead of keeping * track of all running transactions like it ' s done in a normal snapshot . Note * that we ' re generally only looking at transactions that have acquired an * xid . That is we keep a list of transactions between snapshot - > ( xmin , xmax ) * that we consider committed , everything else is considered aborted / in * progress . That also allows us not to care about subtransactions before they * have committed which means this module , in contrast to HS , doesn ' t have to * care about suboverflowed subtransactions and similar . * * One complexity of doing this is that to e . g . handle mixed DDL / DML * transactions we need Snapshots that see intermediate versions of the * catalog in a transaction . During normal operation this is achieved by using * CommandIds / cmin / cmax . The problem with that however is that for space * efficiency reasons , the cmin and cmax are not included in WAL records . We * cannot read the cmin / cmax from the tuple itself , either , because it is * reset on crash recovery . Even if we could , we could not decode combocids * which are only tracked in the original backend ' s memory . To work around * that , heapam writes an extra WAL record ( XLOG_HEAP2_NEW_CID ) every time a * catalog row is modified , which includes the cmin and cmax of the * tuple . During decoding , we insert the ctid - > ( cmin , cmax ) mappings into the * reorder buffer , and use them at visibility checks instead of the cmin / cmax * on the tuple itself . Check the reorderbuffer . c ' s comment above * ResolveCminCmaxDuringDecoding ( ) for details . * * To facilitate all this we need our own visibility routine , as the normal * ones are optimized for different usecases . * * To replace the normal catalog snapshots with decoding ones use the * SetupHistoricSnapshot ( ) and TeardownHistoricSnapshot ( ) functions . * * * * The snapbuild machinery is starting up in several stages , as illustrated * by the following graph describing the SnapBuild - > state transitions : * * + - - - - - - - - - - - - - - - - - - - - - - - - - + * + - - - - | START | - - - - - - - - - - - - - + * | + - - - - - - - - - - - - - - - - - - - - - - - - - + | * | | | * | | | * | running_xacts # 1 | * | | | * | | | * | v | * | + - - - - - - - - - - - - - - - - - - - - - - - - - + v * | | BUILDING_SNAPSHOT | - - - - - - - - - - - - > | * | + - - - - - - - - - - - - - - - - - - - - - - - - - + | * | | | * | | | * | running_xacts # 2 , xacts from # 1 finished | * | | | * | | | * | v | * | + - - - - - - - - - - - - - - - - - - - - - - - - - + v * | | FULL_SNAPSHOT | - - - - - - - - - - - - > | * | + - - - - - - - - - - - - - - - - - - - - - - - - - + | * | | | * running_xacts | saved snapshot * with zero xacts | at running_xacts ' s lsn * | | | * | running_xacts with xacts from # 2 finished | * | | | * | v | * | + - - - - - - - - - - - - - - - - - - - - - - - - - + | * + - - - > | SNAPBUILD_CONSISTENT | < - - - - - - - - - - - - + * + - - - - - - - - - - - - - - - - - - - - - - - - - + * * Initially the machinery is in the START stage . When an xl_running_xacts * record is read that is sufficiently new ( above the safe xmin horizon ) , * there ' s a state transition . If there were no running xacts when the * xl_running_xacts record was generated , we ' ll directly go into CONSISTENT * state , otherwise we ' ll switch to the BUILDING_SNAPSHOT state . Having a full * snapshot means that all transactions that start henceforth can be decoded * in their entirety , but transactions that started previously can ' t . In * FULL_SNAPSHOT we ' ll switch into CONSISTENT once all those previously * running transactions have committed or aborted . * * Only transactions that commit after CONSISTENT state has been reached will * be replayed , even though they might have started while still in * FULL_SNAPSHOT . That ensures that we ' ll reach a point where no previous * changes has been exported , but all the following ones will be . That point * is a convenient point to initialize replication from , which is why we * export a snapshot at that point , which * can * be used to read normal data . * * Copyright ( c ) 2012 - 2025 , PostgreSQL Global Development Group * * IDENTIFICATION * src / backend / replication / logical / snapbuild . c * * - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - #include "postgres.h" #include <sys/stat.h>#include <unistd.h>#include "access/heapam_xlog.h" #include "access/transam.h" #include "access/xact.h" #include "common/file_utils.h" #include "miscadmin.h" #include "pgstat.h" #include "replication/logical.h" #include "replication/reorderbuffer.h" #include "replication/snapbuild.h" #include "replication/snapbuild_internal.h" #include "storage/fd.h" #include "storage/lmgr.h" #include "storage/proc.h" #include "storage/procarray.h" #include "storage/standby.h" #include "utils/builtins.h" #include "utils/memutils.h" #include "utils/snapmgr.h" #include "utils/snapshot.h" /* * Starting a transaction - - which we need to do while exporting a snapshot - - * removes knowledge about the previously used resowner , so we save it here . static ResourceOwner SavedResourceOwnerDuringExport = NULL;static bool ExportInProgress = false ;/* ->committed and ->catchange manipulation */ static void SnapBuildPurgeOlderTxn(SnapBuild *builder);/* snapshot building/manipulation/distribution functions */ static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder);static void SnapBuildFreeSnapshot(Snapshot snap);static void SnapBuildSnapIncRefcount(Snapshot snap);static void SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid);static inline bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid,/* xlog reading helper functions for SnapBuildProcessRunningXacts */ static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running);static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff);/* serialization functions */ static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn);static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);static void SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path);/* * Allocate a new snapshot builder . * * xmin_horizon is the xid > = which we can be sure no catalog rows have been * removed , start_lsn is the LSN > = we want to replay commits . bool need_full_snapshot,bool in_slot_creation,/* allocate memory in own context, to have better accountability */ "snapshot builder context" ,sizeof (SnapBuild));/* Other struct members initialized by zeroing via palloc0 above */ 0 ;128 ; /* arbitrary number */ sizeof (TransactionId));true ;0 ;return builder;/* * Free a snapshot builder . void /* free snapshot explicitly, that contains some error checking */ if (builder->snapshot != NULL)/* other resources are deallocated via memory context reset */ /* * Free an unreferenced snapshot that has previously been built by us . static void /* make sure we don't get passed an external snapshot */ /* make sure nobody modified our snapshot */ 0 );/* slightly more likely, so it's checked even without c-asserts */ if (snap->copied)"cannot free a copied snapshot" );if (snap->active_count)"cannot free an active snapshot" );/* * In which state of snapshot building are we ? return builder->state;/* * Return the LSN at which the two - phase decoding was first enabled . return builder->two_phase_at;/* * Set the LSN at which two - phase decoding is enabled . void /* * Should the contents of transaction ending at ' ptr ' be decoded ? bool return ptr < builder->start_decoding_at;/* * Increase refcount of a snapshot . * * This is used when handing out a snapshot to some external resource or when * adding a Snapshot as builder - > snapshot . static void /* * Decrease refcount of a snapshot and free if the refcount reaches zero . * * Externally visible , so that external resources that have been handed an * IncRef ' ed Snapshot can adjust its refcount easily . void /* make sure we don't get passed an external snapshot */ /* make sure nobody modified our snapshot */ 0 );0 );/* slightly more likely, so it's checked even without casserts */ if (snap->copied)"cannot free a copied snapshot" );if (snap->active_count == 0 )/* * Build a new snapshot , based on currently committed catalog - modifying * transactions . * * In - progress transactions with catalog access are * not * allowed to modify * these snapshots ; they have to copy them and fill in appropriate - > curcid * and - > subxip / subxcnt values . static Snapshotsizeof (SnapshotData)sizeof (TransactionId) * builder->committed.xcntsizeof (TransactionId) * 1 /* toplevel xid */ ; /* * We misuse the original meaning of SnapshotData ' s xip and subxip fields * to make the more fitting for our needs . * * In the ' xip ' array we store transactions that have to be treated as * committed . Since we will only ever look at tuples from transactions * that have modified the catalog it ' s more efficient to store those few * that exist between xmin and xmax ( frequently there are none ) . * * Snapshots that are used in transactions that have modified the catalog * also use the ' subxip ' array to store their toplevel xid and all the * subtransaction xids so we can recognize when we need to treat rows as * visible that are not in xip but still need to be visible . Subxip only * gets filled when the transaction is copied into the context of a * catalog modifying transaction since we otherwise share a snapshot * between transactions . As long as a txn hasn ' t modified the catalog it * doesn ' t need to treat any uncommitted rows as visible , so there is no * need for those xids . * * Both arrays are qsort ' ed so that we can use bsearch ( ) on them . /* store all transactions to be treated as committed by this snapshot */ char *) snapshot + sizeof (SnapshotData));sizeof (TransactionId));/* sort so we can bsearch() */ sizeof (TransactionId), xidComparator);/* * Initially , subxip is empty , i . e . it ' s a snapshot to be used by * transactions that don ' t modify the catalog . Will be filled by * ReorderBufferCopySnap ( ) if necessary . 0 ;false ;false ;false ;0 ;0 ;0 ;return snapshot;/* * Build the initial slot snapshot and convert it to a normal snapshot that * is understood by HeapTupleSatisfiesMVCC . * * The snapshot will be usable directly in current transaction or exported * for loading in different transaction . int newxcnt = 0 ;/* don't allow older snapshots */ /* about to overwrite MyProc->xmin */ if (HaveRegisteredOrActiveSnapshot())"cannot build an initial slot snapshot when snapshots exist" );if (builder->state != SNAPBUILD_CONSISTENT)"cannot build an initial slot snapshot before reaching a consistent state" );if (!builder->committed.includes_all_transactions)"cannot build an initial slot snapshot, not all transactions are monitored anymore" );/* so we don't overwrite the existing value */ if (TransactionIdIsValid(MyProc->xmin))"cannot build an initial slot snapshot when MyProc->xmin already is valid" );/* * We know that snap - > xmin is alive , enforced by the logical xmin * mechanism . Due to that we can do this without locks , we ' re only * changing our own value . * * Building an initial snapshot is expensive and an unenforced xmin * horizon would have bad consequences , therefore always double - check that * the horizon is enforced . false );if (TransactionIdFollows(safeXid, snap->xmin))"cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u" ,/* allocate in transaction context */ sizeof (TransactionId) * GetMaxSnapshotXidCount());/* * snapbuild . c builds transactions in an " inverted " manner , which means it * stores committed transactions in - > xip , not ones in progress . Build a * classical snapshot by marking all non - committed transactions as * in - progress . This can be expensive . for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)void *test;/* * Check whether transaction committed using the decoding snapshot * meaning of - > xip . sizeof (TransactionId), xidComparator);if (test == NULL)if (newxcnt >= GetMaxSnapshotXidCount())"initial slot snapshot too large" )));/* adjust remaining snapshot fields as needed */ return snap;/* * Export a snapshot so it can be set in another session with SET TRANSACTION * SNAPSHOT . * * For that we need to start a transaction in the current backend as the * importing side checks whether the source transaction is still open to make * sure the xmin horizon hasn ' t advanced since then . const char *char *snapname;if (IsTransactionOrTransactionBlock())"cannot export a snapshot from within a transaction" );if (SavedResourceOwnerDuringExport)"can only export one snapshot at a time" );true ;/* There doesn't seem to a nice API to set these */ true ;/* * now that we ' ve built a plain snapshot , make it active and use the * normal mechanisms for exporting it "exported logical decoding snapshot: \" %s\" with %u transaction ID" ,"exported logical decoding snapshot: \" %s\" with %u transaction IDs" ,return snapname;/* * Ensure there is a snapshot and if not build one for current transaction . /* only build a new snapshot if we don't have a prebuilt one */ if (builder->snapshot == NULL)/* increase refcount for the snapshot builder */ return builder->snapshot;/* * Reset a previously SnapBuildExportSnapshot ( ) ' ed snapshot if there is * any . Aborts the previously started transaction and resets the resource * owner back to its original value . void void )/* nothing exported, that is the usual case */ if (!ExportInProgress)return ;if (!IsTransactionState())"clearing exported snapshot in wrong transaction state" );/* * AbortCurrentTransaction ( ) takes care of resetting the snapshot state , * so remember SavedResourceOwnerDuringExport . /* make sure nothing could have ever happened */ /* * Clear snapshot export state during transaction abort . void void )false ;/* * Handle the effects of a single heap change , appropriate to the current state * of the snapshot builder and returns whether changes made at ( xid , lsn ) can * be decoded . bool /* * We can ' t handle data in transactions if we haven ' t built a snapshot * yet , so don ' t store them . if (builder->state < SNAPBUILD_FULL_SNAPSHOT)return false ;/* * No point in keeping track of changes in transactions that we don ' t have * enough information about to decode . This means that they started before * we got into the SNAPBUILD_FULL_SNAPSHOT state . if (builder->state < SNAPBUILD_CONSISTENT &&return false ;/* * If the reorderbuffer doesn ' t yet have a snapshot , add one now , it will * be needed to decode the change we ' re currently processing . if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))/* only build a new snapshot if we don't have a prebuilt one */ if (builder->snapshot == NULL)/* increase refcount for the snapshot builder */ /* * Increase refcount for the transaction we ' re handing the snapshot * out to . return true ;/* * Do CommandId / combo CID handling after reading an xl_heap_new_cid record . * This implies that a transaction has done some form of write to system * catalogs . void /* * we only log new_cid ' s if a catalog tuple was modified , so mark the * transaction as containing catalog modifications /* figure out new command id */ if (xlrec->cmin != InvalidCommandId &&else if (xlrec->cmax != InvalidCommandId)else if (xlrec->cmin != InvalidCommandId)else /* silence compiler */ "xl_heap_new_cid record without a valid CommandId" );1 );/* * Add a new Snapshot and invalidation messages to all transactions we ' re * decoding that currently are in - progress so they can see new catalog contents * made by the transaction that just committed . This is necessary because those * in - progress transactions will use the new catalog ' s contents from here on * ( at the very least everything they do needs to be compatible with newer * catalog contents ) . static void d xid)/* * Iterate through all toplevel transactions . This can include * subtransactions which we just don ' t yet know to be that , but that ' s * fine , they will just get an unnecessary snapshot and invalidations * queued . /* * If we don ' t have a base snapshot yet , there are no changes in this * transaction which in turn implies we don ' t yet need a snapshot at * all . We ' ll add a snapshot when the first change gets queued . * * Similarly , we don ' t need to add invalidations to a transaction * whose base snapshot is not yet set . Once a base snapshot is built , * it will include the xids of committed transactions that have * modified the catalog , thus reflecting the new catalog contents . The * existing catalog cache will have already been invalidated after * processing the invalidations in the transaction that modified * catalogs , ensuring that a fresh cache is constructed during * decoding . * * NB : This works correctly even for subtransactions because * ReorderBufferAssignChild ( ) takes care to transfer the base snapshot * to the top - level transaction , and while iterating the changequeue * we ' ll get the change from the subtxn . if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))continue ;/* * We don ' t need to add snapshot or invalidations to prepared * transactions as they should not see the new catalog contents . if (rbtxn_is_prepared(txn))continue ;"adding a new snapshot and invalidations to %u at %X/%X" ,/* * increase the snapshot ' s refcount for the transaction we are handing * it out to /* * Add invalidation messages to the reorder buffer of in - progress * transactions except the current committed transaction , for which we * will execute invalidations at the end . * * It is required , otherwise , we will end up using the stale catcache * contents built by the current transaction even after its decoding , * which should have been invalidated due to concurrent catalog * changing transaction . * * Distribute only the invalidation messages generated by the current * committed transaction . Invalidation messages received from other * transactions would have already been propagated to the relevant * in - progress transactions . This transaction would have processed * those invalidations , ensuring that subsequent transactions observe * a consistent cache state . if (txn->xid != xid)if (ninvalidations > 0 )/* * Keep track of a new catalog changing transaction that has committed . static void if (builder->committed.xcnt == builder->committed.xcnt_space)2 + 1 ;"increasing space for committed transactions to %u" ,sizeof (TransactionId));/* * TODO : It might make sense to keep the array sorted here instead of * doing it every time we build a new snapshot . On the other hand this * gets called repeatedly when a transaction with subtransactions commits . /* * Remove knowledge about transactions we treat as committed or containing catalog * changes that are smaller than - > xmin . Those won ' t ever get checked via * the - > committed or - > catchange array , respectively . The committed xids will * get checked via the clog machinery . * * We can ideally remove the transaction from catchange array once it is * finished ( committed / aborted ) but that could be costly as we need to maintain * the xids order in the array . static void int off;int surviving_xids = 0 ;/* not ready yet */ if (!TransactionIdIsNormal(builder->xmin))return ;/* TODO: Neater algorithm than just copying and iterating? */ sizeof (TransactionId));/* copy xids that still are interesting to workspace */ for (off = 0 ; off < builder->committed.xcnt; off++)if (NormalTransactionIdPrecedes(builder->committed.xip[off],/* remove */ else /* copy workspace back to persistent state */ sizeof (TransactionId));"purged committed transactions from %u to %u, xmin: %u, xmax: %u" ,/* * Purge xids in - > catchange as well . The purged array must also be sorted * in xidComparator order . if (builder->catchange.xcnt > 0 )/* * Since catchange . xip is sorted , we find the lower bound of xids that * are still interesting . for (off = 0 ; off < builder->catchange.xcnt; off++)if (TransactionIdFollowsOrEquals(builder->catchange.xip[off],break ;if (surviving_xids > 0 )sizeof (TransactionId));else "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u" ,/* * Handle everything that needs to be done when a transaction commits void int nsubxacts, TransactionId *subxacts, uint32 xinfo)int nxact;bool needs_snapshot = false ;bool needs_timetravel = false ;bool sub_needs_timetravel = false ;/* * Transactions preceding BUILDING_SNAPSHOT will neither be decoded , nor * will they be part of a snapshot . So we don ' t need to record anything . if (builder->state == SNAPBUILD_START ||/* ensure that only commits after this are getting replayed */ if (builder->start_decoding_at <= lsn)1 ;return ;if (builder->state < SNAPBUILD_CONSISTENT)/* ensure that only commits after this are getting replayed */ if (builder->start_decoding_at <= lsn)1 ;/* * If building an exportable snapshot , force xid to be tracked , even * if the transaction didn ' t modify the catalog . if (builder->building_full_snapshot)true ;for (nxact = 0 ; nxact < nsubxacts; nxact++)/* * Add subtransaction to base snapshot if catalog modifying , we don ' t * distinguish to toplevel transactions there . if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))true ;true ;"found subtransaction %u:%u with catalog changes" ,if (NormalTransactionIdFollows(subxid, xmax))/* * If we ' re forcing timetravel we also need visibility information * about subtransaction , so keep track of subtransaction ' s state , even * if not catalog modifying . Don ' t need to distribute a snapshot in * that case . else if (needs_timetravel)if (NormalTransactionIdFollows(subxid, xmax))/* if top-level modified catalog, it'll need a snapshot */ if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))"found top level transaction %u, with catalog changes" ,true ;true ;else if (sub_needs_timetravel)/* track toplevel txn as well, subxact alone isn't meaningful */ "forced transaction %u to do timetravel due to one of its subtransactions" ,true ;else if (needs_timetravel)"forced transaction %u to do timetravel" , xid);if (!needs_timetravel)/* record that we cannot export a general snapshot anymore */ false ;/* * Adjust xmax of the snapshot builder , we only do that for committed , * catalog modifying , transactions , everything else isn ' t interesting for * us since we ' ll never look at the respective rows . if (needs_timetravel &&/* if there's any reason to build a historic snapshot, do so now */ if (needs_snapshot)/* * If we haven ' t built a complete snapshot yet there ' s no need to hand * it out , it wouldn ' t ( and couldn ' t ) be used anyway . if (builder->state < SNAPBUILD_FULL_SNAPSHOT)return ;/* * Decrease the snapshot builder ' s refcount of the old snapshot , note * that it still will be used if it has been handed out to the * reorderbuffer earlier . if (builder->snapshot)/* we might need to execute invalidations, add snapshot */ if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))/* refcount of the snapshot builder for the new snapshot */ /* * Add a new catalog snapshot and invalidations messages to all * currently running transactions . /* * Check the reorder buffer and the snapshot to see if the given transaction has * modified catalogs . static inline bool if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))return true ;/* * The transactions that have changed catalogs must have invalidation * info . if (!(xinfo & XACT_XINFO_HAS_INVALS))return false ;/* Check the catchange XID array */ return ((builder->catchange.xcnt > 0 ) &&sizeof (TransactionId), xidComparator) != NULL));/* ----------------------------------- * Snapshot building functions dealing with xlog records * - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - /* * Process a running xacts record , and use its information to first build a * historic snapshot and later to release resources that aren ' t needed * anymore . void nning)/* * If we ' re not consistent yet , inspect the record to see whether it * allows to get closer to being consistent . If we are consistent , dump * our snapshot so others or we , after a restart , can use it . if (builder->state < SNAPBUILD_CONSISTENT)/* returns false if there's no point in performing cleanup just yet */ if (!SnapBuildFindSnapshot(builder, lsn, running))return ;else /* * Update range of interesting xids based on the running xacts * information . We don ' t increase - > xmax using it , because once we are in * a consistent state we can do that ourselves and much more efficiently * so , because we only need to do it for catalog transactions since we * only ever look at those . * * NB : We only increase xmax when a catalog modifying transaction commits * ( see SnapBuildCommitTxn ) . Because of this , xmax can be lower than * xmin , which looks odd but is correct and actually more efficient , since * we hit fast paths in heapam_visibility . c . /* Remove transactions we don't need to keep track off anymore */ /* * Advance the xmin limit for the current replication slot , to allow * vacuum to clean up the tuples this slot has been protecting . * * The reorderbuffer might have an xmin among the currently running * snapshots ; use it if so . If not , we need only consider the snapshots * we ' ll produce later , which can ' t be less than the oldest running xid in * the record we ' re reading now . if (xmin == InvalidTransactionId)"xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u" ,/* * Also tell the slot where we can restart decoding from . We don ' t want to * do that after every commit because changing that implies an fsync of * the logical slot ' s state file , so we only do it every time we see a * running xacts record . * * Do so by looking for the oldest in progress transaction ( determined by * the first LSN of any of its relevant records ) . Every transaction * remembers the last location we stored the snapshot to disk before its * beginning . That point is where we can restart from . /* * Can ' t know about a serialized snapshot ' s location if we ' re not * consistent . if (builder->state < SNAPBUILD_CONSISTENT)return ;/* * oldest ongoing txn might have started when we didn ' t yet serialize * anything because we hadn ' t reached a consistent state yet . if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)/* * No in - progress transaction , can reuse the last serialized snapshot if * we have one . else if (txn == NULL &&/* * Build the start of a snapshot that ' s capable of decoding the catalog . * * Helper function for SnapBuildProcessRunningXacts ( ) while we ' re not yet * consistent . * * Returns true if there is a point in performing internal maintenance / cleanup * using the xl_running_xacts record . static bool /* --- * Build catalog decoding snapshot incrementally using information about * the currently running transactions . There are several ways to do that : * * a ) There were no running transactions when the xl_running_xacts record * was inserted , jump to CONSISTENT immediately . We might find such a * state while waiting on c ) ' s sub - states . * * b ) This ( in a previous run ) or another decoding slot serialized a * snapshot to disk that we can use . Can ' t use this method while finding * the start point for decoding changes as the restart LSN would be an * arbitrary LSN but we need to find the start point to extract changes * where we won ' t see the data for partial transactions . Also , we cannot * use this method when a slot needs a full snapshot for export or direct * use , as that snapshot will only contain catalog modifying transactions . * * c ) First incrementally build a snapshot for catalog tuples * ( BUILDING_SNAPSHOT ) , that requires all , already in - progress , * transactions to finish . Every transaction starting after that * ( FULL_SNAPSHOT state ) , has enough information to be decoded . But * for older running transactions no viable snapshot exists yet , so * CONSISTENT will only be reached once all of those have finished . * - - - /* * xl_running_xacts record is older than what we can use , we might not * have all necessary catalog rows anymore . if (TransactionIdIsNormal(builder->initial_xmin_horizon) &&"skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low" ,"initial xmin horizon of %u vs the snapshot's %u" ,return true ;/* * a ) No transaction were running , we can jump to consistent . * * This is not affected by races around xl_running_xacts , because we can * miss transaction commits , but currently not transactions starting . * * NB : We might have already started to incrementally assemble a snapshot , * so we need to be careful to deal with that . if (running->oldestRunningXid == running->nextXid)if (builder->start_decoding_at == InvalidXLogRecPtr ||/* can decode everything after this */ 1 ;/* As no transactions were running xmin/xmax can be trivially set. */ /* < are finished */ /* >= are running */ /* so we can safely use the faster comparisons */ "logical decoding found consistent point at %X/%X" ,"There are no running transactions." )));return false ;/* * b ) valid on disk state and while neither building full snapshot nor * creating a slot . else if (!builder->building_full_snapshot &&/* there won't be any state to cleanup */ return false ;/* * c ) transition from START to BUILDING_SNAPSHOT . * * In START state , and a xl_running_xacts record with running xacts is * encountered . In that case , switch to BUILDING_SNAPSHOT state , and * record xl_running_xacts - > nextXid . Once all running xacts have finished * ( i . e . they ' re all > = nextXid ) , we have a complete catalog snapshot . It * might look that we could use xl_running_xacts ' s - > xids information to * get there quicker , but that is problematic because transactions marked * as running , might already have inserted their commit record - it ' s * infeasible to change that with locking . else if (builder->state == SNAPBUILD_START)/* * Start with an xmin / xmax that ' s correct for future , when all the * currently running transactions have finished . We ' ll update both * while waiting for the pending transactions to finish . /* < are finished */ /* >= are running */ /* so we can safely use the faster comparisons */ "logical decoding found initial starting point at %X/%X" ,"Waiting for transactions (approximately %d) older than %u to end." ,/* * c ) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT . * * In BUILDING_SNAPSHOT state , and this xl_running_xacts ' oldestRunningXid * is > = than nextXid from when we switched to BUILDING_SNAPSHOT . This * means all transactions starting afterwards have enough information to * be decoded . Switch to FULL_SNAPSHOT . else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&"logical decoding found initial consistent point at %X/%X" ,"Waiting for transactions (approximately %d) older than %u to end." ,/* * c ) transition from FULL_SNAPSHOT to CONSISTENT . * * In FULL_SNAPSHOT state , and this xl_running_xacts ' oldestRunningXid is * > = than nextXid from when we switched to FULL_SNAPSHOT . This means all * transactions that are currently in progress have a catalog snapshot , * and all their changes have been collected . Switch to CONSISTENT . else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&"logical decoding found consistent point at %X/%X" ,"There are no old transactions anymore." )));/* * We already started to track running xacts and need to wait for all * in - progress ones to finish . We fall through to the normal processing of * records so incremental cleanup can be performed . return true ;/* --- * Iterate through xids in record , wait for all older than the cutoff to * finish . Then , if possible , log a new xl_running_xacts record . * * This isn ' t required for the correctness of decoding , but to : * a ) allow isolationtester to notice that we ' re currently waiting for * something . * b ) log a new xl_running_xacts record where it ' d be helpful , without having * to wait for bgwriter or checkpointer . * - - - static void int off;for (off = 0 ; off < running->xcnt; off++)/* * Upper layers should prevent that we ever need to wait on ourselves . * Check anyway , since failing to do so would either result in an * endless wait or an Assert ( ) failure . if (TransactionIdIsCurrentTransactionId(xid))"waiting for ourselves" );if (TransactionIdFollows(xid, cutoff))continue ;/* * All transactions we needed to finish finished - try to ensure there is * another xl_running_xacts record in a timely manner , without having to * wait for bgwriter or checkpointer to log one . During recovery we can ' t * enforce that , so we ' ll have to wait . if (!RecoveryInProgress())#define SnapBuildOnDiskConstantSize \#define SnapBuildOnDiskNotChecksummedSize \#define SNAPBUILD_MAGIC 0 x51A1E001#define SNAPBUILD_VERSION 6 /* * Store / Load a snapshot from disk , depending on the snapshot builder ' s state . * * Supposed to be used by external ( i . e . not snapbuild . c ) code that just read * a record that ' s a potential location for a serialized snapshot . void if (builder->state < SNAPBUILD_CONSISTENT)else /* * Serialize the snapshot ' builder ' at the location ' lsn ' if it hasn ' t already * been done by another decoding process . static void char *ondisk_c;int fd;char tmppath[MAXPGPATH];char path[MAXPGPATH];int ret;struct stat stat_buf;/* * no point in serializing if we cannot continue to work immediately after * restoring the snapshot if (builder->state < SNAPBUILD_CONSISTENT)return ;/* consistent snapshots have no next phase */ /* * We identify snapshots by the LSN they are valid for . We don ' t need to * include timelines in the name as each LSN maps to exactly one timeline * unless the user used pg_resetwal or similar . If a user did so , there ' s * no hope continuing to decode anyway . "%s/%X-%X.snap" ,/* * first check whether some other backend already has written the snapshot * for this LSN . It ' s perfectly fine if there ' s none , so we accept ENOENT * as a valid state . Everything else is an unexpected error . if (ret != 0 && errno != ENOENT)"could not stat file \" %s\": %m" , path)));else if (ret == 0 )/* * somebody else has already serialized to this point , don ' t overwrite * but remember location , so we don ' t need to read old data again . * * To be sure it has been synced to disk after the rename ( ) from the * tempfile filename to the real filename , we just repeat the fsync . * That ought to be cheap because in most scenarios it should already * be safely on disk . false );true );goto out;/* * there is an obvious race condition here between the time we stat ( 2 ) the * file and us writing the file . But we rename the file into place * atomically and all files created need to contain the same data anyway , * so this is perfectly fine , although a bit of a resource waste . Locking * seems like pointless complication . "serializing snapshot to %s" , path);/* to make sure only we will write to this tempfile, include pid */ "%s/%X-%X.snap.%d.tmp" ,/* * Unlink temporary file if it already exists , needs to have been before a * crash / error since we won ' t enter this function twice from within a * single decoding slot / backend and the temporary file contains the pid of * the current process . if (unlink(tmppath) != 0 && errno != ENOENT)"could not remove file \" %s\": %m" , tmppath)));/* Get the catalog modifying transactions that are yet not committed */ sizeof (SnapBuildOnDisk) +sizeof (TransactionId) * (builder->committed.xcnt + catchange_xcnt);char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,sizeof (SnapBuildOnDisk);sizeof (SnapBuild));/* NULL-ify memory-only data */ /* update catchange only on disk data */ sizeof (SnapBuild));/* copy committed xacts */ if (builder->committed.xcnt > 0 )sizeof (TransactionId) * builder->committed.xcnt;/* copy catalog modifying xacts */ if (catchange_xcnt > 0 )sizeof (TransactionId) * catchange_xcnt;/* we have valid data now, open tempfile and write it there */ if (fd < 0 )"could not open file \" %s\": %m" , tmppath)));0 ;if ((write(fd, ondisk, needed_length)) != needed_length)int save_errno = errno;/* if write didn't set errno, assume problem is no disk space */ "could not write to file \" %s\": %m" , tmppath)));/* * fsync the file before renaming so that even if we crash after this we * have either a fully valid file or nothing . * * It ' s safe to just ERROR on fsync ( ) here because we ' ll retry the whole * operation including the writes . * * TODO : Do the fsync ( ) via checkpoints / restartpoints , doing it here has * some noticeable overhead since it ' s performed synchronously during * decoding ? if (pg_fsync(fd) != 0 )int save_errno = errno;"could not fsync file \" %s\": %m" , tmppath)));if (CloseTransientFile(fd) != 0 )"could not close file \" %s\": %m" , tmppath)));true );/* * We may overwrite the work from some other backend , but that ' s ok , our * snapshot is valid as well , we ' ll just have done some superfluous work . if (rename(tmppath, path) != 0 )"could not rename file \" %s\" to \" %s\": %m" ,/* make sure we persist */ false );true );/* * Now there ' s no way we can lose the dumped state anymore , remember this * as a serialization point . /* be tidy */ if (ondisk)if (catchange_xip)/* * Restore the logical snapshot file contents to ' ondisk ' . * * ' context ' is the memory context where the catalog modifying / committed xid * will live . * If ' missing_ok ' is true , will not throw an error if the file is not found . bool bool missing_ok)int fd;char path[MAXPGPATH];"%s/%X-%X.snap" ,if (fd < 0 )if (missing_ok && errno == ENOENT)return false ;"could not open file \" %s\": %m" , path)));/* ---- * Make sure the snapshot had been stored safely to disk , that ' s normally * cheap . * Note that we do not need PANIC here , nobody will be able to use the * slot without fsyncing , and saving it won ' t succeed without an fsync ( ) * either . . . * - - - - false );true );/* read statically sized portion of snapshot */ if (ondisk->magic != SNAPBUILD_MAGIC)"snapbuild state file \" %s\" has wrong magic number: %u instead of %u" ,if (ondisk->version != SNAPBUILD_VERSION)"snapbuild state file \" %s\" has unsupported version: %u instead of %u" ,char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,/* read SnapBuild */ sizeof (SnapBuild), path);sizeof (SnapBuild));/* restore committed xacts information */ if (ondisk->builder.committed.xcnt > 0 )sizeof (TransactionId) * ondisk->builder.committed.xcnt;/* restore catalog modifying xacts information */ if (ondisk->builder.catchange.xcnt > 0 )sizeof (TransactionId) * ondisk->builder.catchange.xcnt;if (CloseTransientFile(fd) != 0 )"could not close file \" %s\": %m" , path)));/* verify checksum of what we've read */ if (!EQ_CRC32C(checksum, ondisk->checksum))"checksum mismatch for snapbuild state file \" %s\": is %u, should be %u" ,return true ;/* * Restore a snapshot into ' builder ' if previously one has been stored at the * location indicated by ' lsn ' . Returns true if successful , false otherwise . static bool /* no point in loading a snapshot if we're already there */ if (builder->state == SNAPBUILD_CONSISTENT)return false ;/* validate and restore the snapshot to 'ondisk' */ if (!SnapBuildRestoreSnapshot(&ondisk, lsn, builder->context, true ))return false ;/* * ok , we now have a sensible snapshot here , figure out if it has more * information than we have . /* * We are only interested in consistent snapshots for now , comparing * whether one incomplete snapshot is more " advanced " seems to be * unnecessarily complex . if (ondisk.builder.state < SNAPBUILD_CONSISTENT)goto snapshot_not_interesting;/* * Don ' t use a snapshot that requires an xmin that we cannot guarantee to * be available . if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))goto snapshot_not_interesting;/* * Consistent snapshots have no next phase . Reset next_phase_at as it is * possible that an old value may remain . /* ok, we think the snapshot is sensible, copy over everything important */ /* We only allocated/stored xcnt, not xcnt_space xids ! */ /* don't overwrite preallocated xip, if we don't have anything here */ if (builder->committed.xcnt > 0 )/* set catalog modifying transactions */ if (builder->catchange.xip)/* our snapshot is not interesting anymore, build a new one */ if (builder->snapshot != NULL)"logical decoding found consistent point at %X/%X" ,"Logical decoding will begin using saved snapshot." )));return true ;if (ondisk.builder.committed.xip != NULL)if (ondisk.builder.catchange.xip != NULL)return false ;/* * Read the contents of the serialized snapshot to ' dest ' . static void int fd, void *dest, Size size, const char *path)int readBytes;if (readBytes != size)int save_errno = errno;if (readBytes < 0 )"could not read file \" %s\": %m" , path)));else "could not read file \" %s\": read %d of %zu" ,/* * Remove all serialized snapshots that are not required anymore because no * slot can need them . This doesn ' t actually have to run during a checkpoint , * but it ' s a convenient point to schedule this . * * NB : We run this during checkpoints even if logical decoding is disabled so * we cleanup old slots at some point after it got disabled . void void )struct dirent *snap_de;char path[MAXPGPATH + sizeof (PG_LOGICAL_SNAPSHOTS_DIR)];/* * We start off with a minimum of the last redo pointer . No new * replication slot will start before that , so that ' s a safe upper bound * for removal . /* now check for the restart ptrs from existing slots */ /* don't start earlier than the restart lsn */ if (redo < cutoff)while ((snap_de = ReadDir(snap_dir, PG_LOGICAL_SNAPSHOTS_DIR)) != NULL)if (strcmp(snap_de->d_name, "." ) == 0 ||".." ) == 0 )continue ;sizeof (path), "%s/%s" , PG_LOGICAL_SNAPSHOTS_DIR, snap_de->d_name);false , DEBUG1);if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)"only regular files expected: %s" , path);continue ;/* * temporary filenames from SnapBuildSerialize ( ) include the LSN and * everything but are postfixed by . $ pid . tmp . We can just remove them * the same as other files because there can be none that are * currently being written that are older than cutoff . * * We just log a message if a file doesn ' t fit the pattern , it ' s * probably some editors lock / state file or similar . . . if (sscanf(snap_de->d_name, "%X-%X.snap" , &hi, &lo) != 2 )"could not parse file name \" %s\"" , path)));continue ;32 | lo;/* check whether we still need it */ if (lsn < cutoff || cutoff == InvalidXLogRecPtr)"removing snapbuild snapshot %s" , path);/* * It ' s not particularly harmful , though strange , if we can ' t * remove the file here . Don ' t prevent the checkpoint from * completing , that ' d be a cure worse than the disease . if (unlink(path) < 0 )"could not remove file \" %s\": %m" ,continue ;/* * Check if a logical snapshot at the specified point has been serialized . bool char path[MAXPGPATH];int ret;struct stat stat_buf;"%s/%X-%X.snap" ,if (ret != 0 && errno != ENOENT)"could not stat file \" %s\": %m" , path)));return ret == 0 ;
Messung V0.5 in Prozent C=93 H=90 G=91
¤ Diese beiden folgenden Angebotsgruppen bietet das Unternehmen0.70Angebot
(vorverarbeitet am 2026-08-08)
¤ *© Formatika GbR, Deutschland
