120 #include <sys/stat.h>
315 bool need_full_snapshot,
324 "snapshot builder context",
390 elog(
ERROR,
"cannot free a copied snapshot");
393 elog(
ERROR,
"cannot free an active snapshot");
404 return builder->
state;
469 elog(
ERROR,
"cannot free a copied snapshot");
531 memcpy(snapshot->
xip,
579 elog(
ERROR,
"cannot build an initial slot snapshot when snapshots exist");
583 elog(
ERROR,
"cannot build an initial slot snapshot before reaching a consistent state");
586 elog(
ERROR,
"cannot build an initial slot snapshot, not all transactions are monitored anymore");
590 elog(
ERROR,
"cannot build an initial slot snapshot when MyProc->xmin already is valid");
608 elog(
ERROR,
"cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
609 safeXid, snap->
xmin);
639 errmsg(
"initial slot snapshot too large")));
641 newxip[newxcnt++] = xid;
649 snap->
xcnt = newxcnt;
670 elog(
ERROR,
"cannot export a snapshot from within a transaction");
673 elog(
ERROR,
"can only export one snapshot at a time");
693 (
errmsg_plural(
"exported logical decoding snapshot: \"%s\" with %u transaction ID",
694 "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
696 snapname, snap->
xcnt)));
734 elog(
ERROR,
"clearing exported snapshot in wrong transaction state");
841 elog(
ERROR,
"xl_heap_new_cid record without a valid CommandId");
892 elog(
DEBUG2,
"adding a new snapshot to %u at %X/%X",
917 elog(
DEBUG1,
"increasing space for committed transactions to %u",
947 int surviving_xids = 0;
965 workspace[surviving_xids++] = builder->
committed.
xip[off];
972 elog(
DEBUG3,
"purged committed transactions from %u to %u, xmin: %u, xmax: %u",
998 if (surviving_xids > 0)
1009 elog(
DEBUG3,
"purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
1025 bool needs_snapshot =
false;
1026 bool needs_timetravel =
false;
1027 bool sub_needs_timetravel =
false;
1057 needs_timetravel =
true;
1061 for (nxact = 0; nxact < nsubxacts; nxact++)
1071 sub_needs_timetravel =
true;
1072 needs_snapshot =
true;
1074 elog(
DEBUG1,
"found subtransaction %u:%u with catalog changes",
1089 else if (needs_timetravel)
1100 elog(
DEBUG2,
"found top level transaction %u, with catalog changes",
1102 needs_snapshot =
true;
1103 needs_timetravel =
true;
1106 else if (sub_needs_timetravel)
1109 elog(
DEBUG2,
"forced transaction %u to do timetravel due to one of its subtransactions",
1111 needs_timetravel =
true;
1114 else if (needs_timetravel)
1116 elog(
DEBUG2,
"forced transaction %u to do timetravel", xid);
1121 if (!needs_timetravel)
1127 Assert(!needs_snapshot || needs_timetravel);
1134 if (needs_timetravel &&
1138 builder->
xmax = xmax;
1261 elog(
DEBUG3,
"xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1297 else if (txn == NULL &&
1349 (
errmsg_internal(
"skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1388 (
errmsg(
"logical decoding found consistent point at %X/%X",
1390 errdetail(
"There are no running transactions.")));
1432 (
errmsg(
"logical decoding found initial starting point at %X/%X",
1434 errdetail(
"Waiting for transactions (approximately %d) older than %u to end.",
1456 (
errmsg(
"logical decoding found initial consistent point at %X/%X",
1458 errdetail(
"Waiting for transactions (approximately %d) older than %u to end.",
1480 (
errmsg(
"logical decoding found consistent point at %X/%X",
1482 errdetail(
"There are no old transactions anymore.")));
1509 for (off = 0; off < running->
xcnt; off++)
1573 #define SnapBuildOnDiskConstantSize \
1574 offsetof(SnapBuildOnDisk, builder)
1575 #define SnapBuildOnDiskNotChecksummedSize \
1576 offsetof(SnapBuildOnDisk, version)
1578 #define SNAPBUILD_MAGIC 0x51A1E001
1579 #define SNAPBUILD_VERSION 5
1607 size_t catchange_xcnt;
1613 struct stat stat_buf;
1636 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
1644 ret =
stat(path, &stat_buf);
1646 if (ret != 0 && errno != ENOENT)
1649 errmsg(
"could not stat file \"%s\": %m", path)));
1676 elog(
DEBUG1,
"serializing snapshot to %s", path);
1679 sprintf(tmppath,
"pg_logical/snapshots/%X-%X.snap.%d.tmp",
1688 if (unlink(tmppath) != 0 && errno != ENOENT)
1691 errmsg(
"could not remove file \"%s\": %m", tmppath)));
1702 ondisk_c =
palloc0(needed_length);
1706 ondisk->
length = needed_length;
1737 if (catchange_xcnt > 0)
1740 memcpy(ondisk_c, catchange_xip, sz);
1749 O_CREAT | O_EXCL | O_WRONLY |
PG_BINARY);
1753 errmsg(
"could not open file \"%s\": %m", tmppath)));
1757 if ((
write(
fd, ondisk, needed_length)) != needed_length)
1759 int save_errno = errno;
1764 errno = save_errno ? save_errno : ENOSPC;
1767 errmsg(
"could not write to file \"%s\": %m", tmppath)));
1785 int save_errno = errno;
1791 errmsg(
"could not fsync file \"%s\": %m", tmppath)));
1798 errmsg(
"could not close file \"%s\": %m", tmppath)));
1806 if (rename(tmppath, path) != 0)
1810 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
1833 pfree(catchange_xip);
1853 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
1858 if (
fd < 0 && errno == ENOENT)
1863 errmsg(
"could not open file \"%s\": %m", path)));
1883 errmsg(
"snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1889 errmsg(
"snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1922 errmsg(
"could not close file \"%s\": %m", path)));
1930 errmsg(
"checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1931 path, checksum, ondisk.
checksum)));
1944 goto snapshot_not_interesting;
1951 goto snapshot_not_interesting;
1992 (
errmsg(
"logical decoding found consistent point at %X/%X",
1994 errdetail(
"Logical decoding will begin using saved snapshot.")));
1997 snapshot_not_interesting:
2016 if (readBytes != size)
2018 int save_errno = errno;
2027 errmsg(
"could not read file \"%s\": %m", path)));
2032 errmsg(
"could not read file \"%s\": read %d of %zu",
2069 while ((snap_de =
ReadDir(snap_dir,
"pg_logical/snapshots")) != NULL)
2076 if (strcmp(snap_de->
d_name,
".") == 0 ||
2077 strcmp(snap_de->
d_name,
"..") == 0)
2080 snprintf(path,
sizeof(path),
"pg_logical/snapshots/%s", snap_de->
d_name);
2085 elog(
DEBUG1,
"only regular files expected: %s", path);
2098 if (sscanf(snap_de->
d_name,
"%X-%X.snap", &hi, &lo) != 2)
2101 (
errmsg(
"could not parse file name \"%s\"", path)));
2105 lsn = ((uint64) hi) << 32 | lo;
2110 elog(
DEBUG1,
"removing snapbuild snapshot %s", path);
2117 if (unlink(path) < 0)
2121 errmsg(
"could not remove file \"%s\": %m",
elog(ERROR, "%s: %s", p2, msg)
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
int errmsg_internal(const char *fmt,...)
int errdetail_internal(const char *fmt,...)
int errcode_for_file_access(void)
int errdetail(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
struct dirent * ReadDir(DIR *dir, const char *dirname)
int CloseTransientFile(int fd)
void fsync_fname(const char *fname, bool isdir)
int OpenTransientFile(const char *fileName, int fileFlags)
DIR * AllocateDir(const char *dirname)
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
#define dlist_foreach(iter, lhead)
static uint32 dclist_count(const dclist_head *head)
#define dlist_container(type, membername, ptr)
Assert(fmt[strlen(fmt) - 1] !='\n')
void XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid, XLTW_Oper oper)
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
void LWLockRelease(LWLock *lock)
void pfree(void *pointer)
void * palloc0(Size size)
void * MemoryContextAllocZero(MemoryContext context, Size size)
MemoryContext CurrentMemoryContext
void * repalloc(void *pointer, Size size)
void * MemoryContextAlloc(MemoryContext context, Size size)
void MemoryContextDelete(MemoryContext context)
#define AllocSetContextCreate
#define ALLOCSET_DEFAULT_SIZES
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
#define ERRCODE_DATA_CORRUPTED
#define COMP_CRC32C(crc, data, len)
#define EQ_CRC32C(c1, c2)
#define ERRCODE_T_R_SERIALIZATION_FAILURE
#define qsort(a, b, c, d)
static int fd(const char *x, int i)
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
int GetMaxSnapshotXidCount(void)
TransactionId * ReorderBufferGetCatalogChangesXacts(ReorderBuffer *rb)
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
void ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, CommandId cid)
void ReorderBufferAddNewTupleCids(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, RelFileLocator locator, ItemPointerData tid, CommandId cmin, CommandId cmax, CommandId combocid)
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *rb, TransactionId xid)
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
void ReorderBufferAddSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
void ReorderBufferSetRestartPoint(ReorderBuffer *rb, XLogRecPtr ptr)
#define rbtxn_prepared(txn)
#define rbtxn_skip_prepared(txn)
ResourceOwner CurrentResourceOwner
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
void SnapBuildSnapDecRefcount(Snapshot snap)
#define SNAPBUILD_VERSION
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
void SnapBuildResetExportedSnapshotState(void)
void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
static void SnapBuildSnapIncRefcount(Snapshot snap)
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
const char * SnapBuildExportSnapshot(SnapBuild *builder)
XLogRecPtr SnapBuildGetTwoPhaseAt(SnapBuild *builder)
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
#define SnapBuildOnDiskNotChecksummedSize
void FreeSnapshotBuilder(SnapBuild *builder)
void CheckPointSnapBuild(void)
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr two_phase_at)
static bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid, uint32 xinfo)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder)
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
static ResourceOwner SavedResourceOwnerDuringExport
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts, uint32 xinfo)
static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
void SnapBuildClearExportedSnapshot(void)
static void SnapBuildFreeSnapshot(Snapshot snap)
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
static bool ExportInProgress
struct SnapBuildOnDisk SnapBuildOnDisk
static void SnapBuildPurgeOlderTxn(SnapBuild *builder)
#define SnapBuildOnDiskConstantSize
static void SnapBuildRestoreContents(int fd, char *dest, Size size, const char *path)
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
@ SNAPBUILD_BUILDING_SNAPSHOT
@ SNAPBUILD_FULL_SNAPSHOT
char * ExportSnapshot(Snapshot snapshot)
bool HistoricSnapshotActive(void)
bool HaveRegisteredOrActiveSnapshot(void)
void InvalidateCatalogSnapshot(void)
struct SnapshotData SnapshotData
XLogRecPtr LogStandbySnapshot(void)
XLogRecPtr restart_decoding_lsn
dclist_head catchange_txns
dlist_head toplevel_by_lsn
XLogRecPtr current_restart_decoding_lsn
XLogRecPtr start_decoding_at
TransactionId initial_xmin_horizon
struct SnapBuild::@16 committed
bool building_full_snapshot
TransactionId next_phase_at
struct SnapBuild::@17 catchange
XLogRecPtr last_serialized_snapshot
bool includes_all_transactions
uint64 snapXactCompletionCount
SnapshotType snapshot_type
ItemPointerData target_tid
RelFileLocator target_locator
TransactionId oldestRunningXid
TransactionId xids[FLEXIBLE_ARRAY_MEMBER]
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
#define InvalidTransactionId
#define NormalTransactionIdPrecedes(id1, id2)
#define NormalTransactionIdFollows(id1, id2)
#define TransactionIdIsValid(xid)
#define TransactionIdIsNormal(xid)
#define TransactionIdAdvance(dest)
@ WAIT_EVENT_SNAPBUILD_SYNC
@ WAIT_EVENT_SNAPBUILD_WRITE
@ WAIT_EVENT_SNAPBUILD_READ
static void pgstat_report_wait_start(uint32 wait_event_info)
static void pgstat_report_wait_end(void)
bool IsTransactionOrTransactionBlock(void)
bool IsTransactionState(void)
void StartTransactionCommand(void)
bool TransactionIdIsCurrentTransactionId(TransactionId xid)
void AbortCurrentTransaction(void)
#define XACT_REPEATABLE_READ
#define XACT_XINFO_HAS_INVALS
int xidComparator(const void *arg1, const void *arg2)
bool RecoveryInProgress(void)
XLogRecPtr GetRedoRecPtr(void)
#define LSN_FORMAT_ARGS(lsn)
#define InvalidXLogRecPtr