125 #include <sys/stat.h>
327 bool need_full_snapshot,
328 bool in_slot_creation,
337 "snapshot builder context",
404 elog(
ERROR,
"cannot free a copied snapshot");
407 elog(
ERROR,
"cannot free an active snapshot");
418 return builder->
state;
483 elog(
ERROR,
"cannot free a copied snapshot");
545 memcpy(snapshot->
xip,
593 elog(
ERROR,
"cannot build an initial slot snapshot when snapshots exist");
597 elog(
ERROR,
"cannot build an initial slot snapshot before reaching a consistent state");
600 elog(
ERROR,
"cannot build an initial slot snapshot, not all transactions are monitored anymore");
604 elog(
ERROR,
"cannot build an initial slot snapshot when MyProc->xmin already is valid");
622 elog(
ERROR,
"cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
623 safeXid, snap->
xmin);
653 errmsg(
"initial slot snapshot too large")));
655 newxip[newxcnt++] = xid;
663 snap->
xcnt = newxcnt;
684 elog(
ERROR,
"cannot export a snapshot from within a transaction");
687 elog(
ERROR,
"can only export one snapshot at a time");
707 (
errmsg_plural(
"exported logical decoding snapshot: \"%s\" with %u transaction ID",
708 "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
710 snapname, snap->
xcnt)));
748 elog(
ERROR,
"clearing exported snapshot in wrong transaction state");
855 elog(
ERROR,
"xl_heap_new_cid record without a valid CommandId");
906 elog(
DEBUG2,
"adding a new snapshot to %u at %X/%X",
931 elog(
DEBUG1,
"increasing space for committed transactions to %u",
961 int surviving_xids = 0;
979 workspace[surviving_xids++] = builder->
committed.
xip[off];
986 elog(
DEBUG3,
"purged committed transactions from %u to %u, xmin: %u, xmax: %u",
1012 if (surviving_xids > 0)
1023 elog(
DEBUG3,
"purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
1039 bool needs_snapshot =
false;
1040 bool needs_timetravel =
false;
1041 bool sub_needs_timetravel =
false;
1071 needs_timetravel =
true;
1075 for (nxact = 0; nxact < nsubxacts; nxact++)
1085 sub_needs_timetravel =
true;
1086 needs_snapshot =
true;
1088 elog(
DEBUG1,
"found subtransaction %u:%u with catalog changes",
1103 else if (needs_timetravel)
1114 elog(
DEBUG2,
"found top level transaction %u, with catalog changes",
1116 needs_snapshot =
true;
1117 needs_timetravel =
true;
1120 else if (sub_needs_timetravel)
1123 elog(
DEBUG2,
"forced transaction %u to do timetravel due to one of its subtransactions",
1125 needs_timetravel =
true;
1128 else if (needs_timetravel)
1130 elog(
DEBUG2,
"forced transaction %u to do timetravel", xid);
1135 if (!needs_timetravel)
1141 Assert(!needs_snapshot || needs_timetravel);
1148 if (needs_timetravel &&
1152 builder->
xmax = xmax;
1275 elog(
DEBUG3,
"xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1311 else if (txn == NULL &&
1365 (
errmsg_internal(
"skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
1404 (
errmsg(
"logical decoding found consistent point at %X/%X",
1406 errdetail(
"There are no running transactions.")));
1453 (
errmsg(
"logical decoding found initial starting point at %X/%X",
1455 errdetail(
"Waiting for transactions (approximately %d) older than %u to end.",
1477 (
errmsg(
"logical decoding found initial consistent point at %X/%X",
1479 errdetail(
"Waiting for transactions (approximately %d) older than %u to end.",
1501 (
errmsg(
"logical decoding found consistent point at %X/%X",
1503 errdetail(
"There are no old transactions anymore.")));
1530 for (off = 0; off < running->
xcnt; off++)
1594 #define SnapBuildOnDiskConstantSize \
1595 offsetof(SnapBuildOnDisk, builder)
1596 #define SnapBuildOnDiskNotChecksummedSize \
1597 offsetof(SnapBuildOnDisk, version)
1599 #define SNAPBUILD_MAGIC 0x51A1E001
1600 #define SNAPBUILD_VERSION 6
1628 size_t catchange_xcnt;
1634 struct stat stat_buf;
1657 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
1665 ret =
stat(path, &stat_buf);
1667 if (ret != 0 && errno != ENOENT)
1670 errmsg(
"could not stat file \"%s\": %m", path)));
1697 elog(
DEBUG1,
"serializing snapshot to %s", path);
1700 sprintf(tmppath,
"pg_logical/snapshots/%X-%X.snap.%d.tmp",
1709 if (unlink(tmppath) != 0 && errno != ENOENT)
1712 errmsg(
"could not remove file \"%s\": %m", tmppath)));
1723 ondisk_c =
palloc0(needed_length);
1727 ondisk->
length = needed_length;
1758 if (catchange_xcnt > 0)
1761 memcpy(ondisk_c, catchange_xip, sz);
1770 O_CREAT | O_EXCL | O_WRONLY |
PG_BINARY);
1774 errmsg(
"could not open file \"%s\": %m", tmppath)));
1778 if ((
write(
fd, ondisk, needed_length)) != needed_length)
1780 int save_errno = errno;
1785 errno = save_errno ? save_errno : ENOSPC;
1788 errmsg(
"could not write to file \"%s\": %m", tmppath)));
1806 int save_errno = errno;
1812 errmsg(
"could not fsync file \"%s\": %m", tmppath)));
1819 errmsg(
"could not close file \"%s\": %m", tmppath)));
1827 if (rename(tmppath, path) != 0)
1831 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
1854 pfree(catchange_xip);
1874 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
1879 if (
fd < 0 && errno == ENOENT)
1884 errmsg(
"could not open file \"%s\": %m", path)));
1904 errmsg(
"snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1910 errmsg(
"snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1943 errmsg(
"could not close file \"%s\": %m", path)));
1951 errmsg(
"checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1952 path, checksum, ondisk.
checksum)));
1965 goto snapshot_not_interesting;
1972 goto snapshot_not_interesting;
2017 (
errmsg(
"logical decoding found consistent point at %X/%X",
2019 errdetail(
"Logical decoding will begin using saved snapshot.")));
2022 snapshot_not_interesting:
2041 if (readBytes !=
size)
2043 int save_errno = errno;
2052 errmsg(
"could not read file \"%s\": %m", path)));
2057 errmsg(
"could not read file \"%s\": read %d of %zu",
2058 path, readBytes,
size)));
2094 while ((snap_de =
ReadDir(snap_dir,
"pg_logical/snapshots")) != NULL)
2101 if (strcmp(snap_de->
d_name,
".") == 0 ||
2102 strcmp(snap_de->
d_name,
"..") == 0)
2105 snprintf(path,
sizeof(path),
"pg_logical/snapshots/%s", snap_de->
d_name);
2110 elog(
DEBUG1,
"only regular files expected: %s", path);
2123 if (sscanf(snap_de->
d_name,
"%X-%X.snap", &hi, &lo) != 2)
2126 (
errmsg(
"could not parse file name \"%s\"", path)));
2130 lsn = ((uint64) hi) << 32 | lo;
2135 elog(
DEBUG1,
"removing snapbuild snapshot %s", path);
2142 if (unlink(path) < 0)
2146 errmsg(
"could not remove file \"%s\": %m",
2163 struct stat stat_buf;
2165 sprintf(path,
"pg_logical/snapshots/%X-%X.snap",
2168 ret =
stat(path, &stat_buf);
2170 if (ret != 0 && errno != ENOENT)
2173 errmsg(
"could not stat file \"%s\": %m", path)));
#define Assert(condition)
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)
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
#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)
RT_SCOPE RT_RADIX_TREE *MemoryContext old_ctx
MemoryContextSwitchTo(old_ctx)
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
static pg_noinline void Size size
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)
bool SnapBuildSnapshotExists(XLogRecPtr lsn)
void CheckPointSnapBuild(void)
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
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)
SnapBuild * AllocateSnapshotBuilder(ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, bool in_slot_creation, XLogRecPtr two_phase_at)
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
struct SnapBuild::@20 catchange
XLogRecPtr start_decoding_at
TransactionId initial_xmin_horizon
bool building_full_snapshot
TransactionId next_phase_at
XLogRecPtr last_serialized_snapshot
bool includes_all_transactions
struct SnapBuild::@19 committed
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