125 #include <sys/stat.h>
320 bool need_full_snapshot,
329 "snapshot builder context",
395 elog(
ERROR,
"cannot free a copied snapshot");
398 elog(
ERROR,
"cannot free an active snapshot");
409 return builder->
state;
474 elog(
ERROR,
"cannot free a copied snapshot");
536 memcpy(snapshot->
xip,
584 elog(
ERROR,
"cannot build an initial slot snapshot when snapshots exist");
588 elog(
ERROR,
"cannot build an initial slot snapshot before reaching a consistent state");
591 elog(
ERROR,
"cannot build an initial slot snapshot, not all transactions are monitored anymore");
595 elog(
ERROR,
"cannot build an initial slot snapshot when MyProc->xmin already is valid");
613 elog(
ERROR,
"cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
614 safeXid, snap->
xmin);
644 errmsg(
"initial slot snapshot too large")));
646 newxip[newxcnt++] = xid;
654 snap->
xcnt = newxcnt;
675 elog(
ERROR,
"cannot export a snapshot from within a transaction");
678 elog(
ERROR,
"can only export one snapshot at a time");
698 (
errmsg_plural(
"exported logical decoding snapshot: \"%s\" with %u transaction ID",
699 "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
701 snapname, snap->
xcnt)));
739 elog(
ERROR,
"clearing exported snapshot in wrong transaction state");
846 elog(
ERROR,
"xl_heap_new_cid record without a valid CommandId");
897 elog(
DEBUG2,
"adding a new snapshot to %u at %X/%X",
922 elog(
DEBUG1,
"increasing space for committed transactions to %u",
952 int surviving_xids = 0;
970 workspace[surviving_xids++] = builder->
committed.
xip[off];
977 elog(
DEBUG3,
"purged committed transactions from %u to %u, xmin: %u, xmax: %u",
1003 if (surviving_xids > 0)
1014 elog(
DEBUG3,
"purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
1030 bool needs_snapshot =
false;
1031 bool needs_timetravel =
false;
1032 bool sub_needs_timetravel =
false;
1062 needs_timetravel =
true;
1066 for (nxact = 0; nxact < nsubxacts; nxact++)
1076 sub_needs_timetravel =
true;
1077 needs_snapshot =
true;
1079 elog(
DEBUG1,
"found subtransaction %u:%u with catalog changes",
1094 else if (needs_timetravel)
1105 elog(
DEBUG2,
"found top level transaction %u, with catalog changes",
1107 needs_snapshot =
true;
1108 needs_timetravel =
true;
1111 else if (sub_needs_timetravel)
1114 elog(
DEBUG2,
"forced transaction %u to do timetravel due to one of its subtransactions",
1116 needs_timetravel =
true;
1119 else if (needs_timetravel)
1121 elog(
DEBUG2,
"forced transaction %u to do timetravel", xid);
1126 if (!needs_timetravel)
1132 Assert(!needs_snapshot || needs_timetravel);
1139 if (needs_timetravel &&
1143 builder->
xmax = xmax;
1266 elog(
DEBUG3,
"xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1302 else if (txn == NULL &&
1354 (
errmsg_internal(
"skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1393 (
errmsg(
"logical decoding found consistent point at %X/%X",
1395 errdetail(
"There are no running transactions.")));
1437 (
errmsg(
"logical decoding found initial starting point at %X/%X",
1439 errdetail(
"Waiting for transactions (approximately %d) older than %u to end.",
1461 (
errmsg(
"logical decoding found initial consistent point at %X/%X",
1463 errdetail(
"Waiting for transactions (approximately %d) older than %u to end.",
1485 (
errmsg(
"logical decoding found consistent point at %X/%X",
1487 errdetail(
"There are no old transactions anymore.")));
1514 for (off = 0; off < running->
xcnt; off++)
1578 #define SnapBuildOnDiskConstantSize \
1579 offsetof(SnapBuildOnDisk, builder)
1580 #define SnapBuildOnDiskNotChecksummedSize \
1581 offsetof(SnapBuildOnDisk, version)
1583 #define SNAPBUILD_MAGIC 0x51A1E001
1584 #define SNAPBUILD_VERSION 5
1612 size_t catchange_xcnt;
1618 struct stat stat_buf;
1641 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
1649 ret =
stat(path, &stat_buf);
1651 if (ret != 0 && errno != ENOENT)
1654 errmsg(
"could not stat file \"%s\": %m", path)));
1681 elog(
DEBUG1,
"serializing snapshot to %s", path);
1684 sprintf(tmppath,
"pg_logical/snapshots/%X-%X.snap.%d.tmp",
1693 if (unlink(tmppath) != 0 && errno != ENOENT)
1696 errmsg(
"could not remove file \"%s\": %m", tmppath)));
1707 ondisk_c =
palloc0(needed_length);
1711 ondisk->
length = needed_length;
1742 if (catchange_xcnt > 0)
1745 memcpy(ondisk_c, catchange_xip, sz);
1754 O_CREAT | O_EXCL | O_WRONLY |
PG_BINARY);
1758 errmsg(
"could not open file \"%s\": %m", tmppath)));
1762 if ((
write(
fd, ondisk, needed_length)) != needed_length)
1764 int save_errno = errno;
1769 errno = save_errno ? save_errno : ENOSPC;
1772 errmsg(
"could not write to file \"%s\": %m", tmppath)));
1790 int save_errno = errno;
1796 errmsg(
"could not fsync file \"%s\": %m", tmppath)));
1803 errmsg(
"could not close file \"%s\": %m", tmppath)));
1811 if (rename(tmppath, path) != 0)
1815 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
1838 pfree(catchange_xip);
1858 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
1863 if (
fd < 0 && errno == ENOENT)
1868 errmsg(
"could not open file \"%s\": %m", path)));
1888 errmsg(
"snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1894 errmsg(
"snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1927 errmsg(
"could not close file \"%s\": %m", path)));
1935 errmsg(
"checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1936 path, checksum, ondisk.
checksum)));
1949 goto snapshot_not_interesting;
1956 goto snapshot_not_interesting;
2001 (
errmsg(
"logical decoding found consistent point at %X/%X",
2003 errdetail(
"Logical decoding will begin using saved snapshot.")));
2006 snapshot_not_interesting:
2025 if (readBytes != size)
2027 int save_errno = errno;
2036 errmsg(
"could not read file \"%s\": %m", path)));
2041 errmsg(
"could not read file \"%s\": read %d of %zu",
2042 path, readBytes, size)));
2078 while ((snap_de =
ReadDir(snap_dir,
"pg_logical/snapshots")) != NULL)
2085 if (strcmp(snap_de->
d_name,
".") == 0 ||
2086 strcmp(snap_de->
d_name,
"..") == 0)
2089 snprintf(path,
sizeof(path),
"pg_logical/snapshots/%s", snap_de->
d_name);
2094 elog(
DEBUG1,
"only regular files expected: %s", path);
2107 if (sscanf(snap_de->
d_name,
"%X-%X.snap", &hi, &lo) != 2)
2110 (
errmsg(
"could not parse file name \"%s\"", path)));
2114 lsn = ((uint64) hi) << 32 | lo;
2119 elog(
DEBUG1,
"removing snapbuild snapshot %s", path);
2126 if (unlink(path) < 0)
2130 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)
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