143#define SNAPSHOT_EXPORT_DIR "pg_snapshots"
166 .
name =
"snapshot reference",
220 elog(
ERROR,
"cannot take query snapshot during logical decoding");
236 "cannot take query snapshot during a parallel operation");
291 "cannot update SecondarySnapshot during a parallel operation");
440 int sourcepid,
PGPROC *sourceproc)
467 if (sourcesnap->
xcnt > 0)
492 if (sourceproc != NULL)
496 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
497 errmsg(
"could not import the requested snapshot"),
498 errdetail(
"The source transaction is not running anymore.")));
502 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
503 errmsg(
"could not import the requested snapshot"),
504 errdetail(
"The source process with PID %d is not running anymore.",
559 if (snapshot->
xcnt > 0)
562 memcpy(newsnap->
xip, snapshot->
xip,
692 elog(
ERROR,
"cannot modify commandid in active snapshot during a parallel operation");
796 if (snapshot == NULL)
809 if (snapshot == NULL)
1005 elog(
WARNING,
"registered snapshots seem to remain after cleanup");
1009 elog(
WARNING,
"snapshot %p still active", active);
1083 (
errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
1084 errmsg(
"cannot export a snapshot from a subtransaction")));
1149 for (
i = 0;
i < snapshot->
xcnt;
i++)
1167 for (
i = 0;
i < nchildren;
i++)
1178 snprintf(pathtmp,
sizeof(pathtmp),
"%s.tmp", path);
1182 errmsg(
"could not create file \"%s\": %m", pathtmp)));
1184 if (fwrite(
buf.data,
buf.len, 1, f) != 1)
1187 errmsg(
"could not write to file \"%s\": %m", pathtmp)));
1194 errmsg(
"could not write to file \"%s\": %m", pathtmp)));
1200 if (rename(pathtmp, path) < 0)
1203 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
1238 int prefixlen = strlen(prefix);
1241 if (strncmp(ptr, prefix, prefixlen) != 0)
1243 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1246 if (sscanf(ptr,
"%d", &
val) != 1)
1248 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1250 ptr = strchr(ptr,
'\n');
1253 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1263 int prefixlen = strlen(prefix);
1266 if (strncmp(ptr, prefix, prefixlen) != 0)
1268 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1271 if (sscanf(ptr,
"%u", &
val) != 1)
1273 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1275 ptr = strchr(ptr,
'\n');
1278 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1289 int prefixlen = strlen(prefix);
1291 if (strncmp(ptr, prefix, prefixlen) != 0)
1293 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1298 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1300 ptr = strchr(ptr,
'\n');
1303 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1319 struct stat stat_buf;
1340 (
errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
1341 errmsg(
"SET TRANSACTION SNAPSHOT must be called before any query")));
1349 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1350 errmsg(
"a snapshot-importing transaction must have isolation level SERIALIZABLE or REPEATABLE READ")));
1356 if (strspn(idstr,
"0123456789ABCDEF-") != strlen(idstr))
1358 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1359 errmsg(
"invalid snapshot identifier: \"%s\"", idstr)));
1371 if (errno == ENOENT)
1373 (
errcode(ERRCODE_UNDEFINED_OBJECT),
1374 errmsg(
"snapshot \"%s\" does not exist", idstr)));
1378 errmsg(
"could not open file \"%s\" for reading: %m",
1383 if (
fstat(fileno(f), &stat_buf))
1384 elog(
ERROR,
"could not stat file \"%s\": %m", path);
1388 if (fread(filebuf, stat_buf.
st_size, 1, f) != 1)
1389 elog(
ERROR,
"could not read file \"%s\": %m", path);
1391 filebuf[stat_buf.
st_size] =
'\0';
1398 memset(&snapshot, 0,
sizeof(snapshot));
1417 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1418 errmsg(
"invalid snapshot data in file \"%s\"", path)));
1421 for (
i = 0;
i < xcnt;
i++)
1433 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1434 errmsg(
"invalid snapshot data in file \"%s\"", path)));
1437 for (
i = 0;
i < xcnt;
i++)
1458 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1459 errmsg(
"invalid snapshot data in file \"%s\"", path)));
1471 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1472 errmsg(
"a serializable transaction cannot import a snapshot from a non-serializable transaction")));
1475 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1476 errmsg(
"a non-read-only serializable transaction cannot import a snapshot from a read-only transaction")));
1490 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1491 errmsg(
"cannot import a snapshot from a different database")));
1530 if (strcmp(s_de->
d_name,
".") == 0 ||
1531 strcmp(s_de->
d_name,
"..") == 0)
1536 if (unlink(
buf) != 0)
1539 errmsg(
"could not remove file \"%s\": %m",
buf)));
1599 Assert(historic_snapshot != NULL);
1671 serialized_snapshot.
xmin = snapshot->
xmin;
1672 serialized_snapshot.
xmax = snapshot->
xmax;
1673 serialized_snapshot.
xcnt = snapshot->
xcnt;
1685 serialized_snapshot.
subxcnt = 0;
1688 memcpy(start_address,
1692 if (snapshot->
xcnt > 0)
1703 if (serialized_snapshot.
subxcnt > 0)
1728 memcpy(&serialized_snapshot, start_address,
1741 snapshot->
xmin = serialized_snapshot.
xmin;
1742 snapshot->
xmax = serialized_snapshot.
xmax;
1743 snapshot->
xip = NULL;
1744 snapshot->
xcnt = serialized_snapshot.
xcnt;
1753 if (serialized_snapshot.
xcnt > 0)
1756 memcpy(snapshot->
xip, serialized_xids,
1761 if (serialized_snapshot.
subxcnt > 0)
1764 serialized_snapshot.
xcnt;
1765 memcpy(snapshot->
subxip, serialized_xids + serialized_snapshot.
xcnt,
#define Assert(condition)
#define OidIsValid(objectId)
static void PGresult * res
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 * ReadDirExtended(DIR *dir, const char *dirname, int elevel)
DIR * AllocateDir(const char *dirname)
FILE * AllocateFile(const char *name, const char *mode)
#define PG_RETURN_TEXT_P(x)
List * lappend(List *list, void *datum)
#define VirtualTransactionIdIsValid(vxid)
void * MemoryContextAlloc(MemoryContext context, Size size)
MemoryContext TopTransactionContext
char * pstrdup(const char *in)
void pfree(void *pointer)
void pairingheap_remove(pairingheap *heap, pairingheap_node *node)
void pairingheap_add(pairingheap *heap, pairingheap_node *node)
pairingheap_node * pairingheap_first(pairingheap *heap)
#define pairingheap_is_empty(h)
#define pairingheap_is_singular(h)
#define pairingheap_container(type, membername, ptr)
#define pairingheap_const_container(type, membername, ptr)
#define pairingheap_reset(h)
static bool pg_lfind32(uint32 key, const uint32 *base, uint32 nelem)
static int list_length(const List *l)
static Datum PointerGetDatum(const void *X)
static Pointer DatumGetPointer(Datum X)
void SetSerializableTransactionSnapshot(Snapshot snapshot, VirtualTransactionId *sourcevxid, int sourcepid)
Snapshot GetSerializableTransactionSnapshot(Snapshot snapshot)
int GetMaxSnapshotSubxidCount(void)
Snapshot GetSnapshotData(Snapshot snapshot)
int GetMaxSnapshotXidCount(void)
bool ProcArrayInstallRestoredXmin(TransactionId xmin, PGPROC *proc)
bool ProcArrayInstallImportedXmin(TransactionId xmin, VirtualTransactionId *sourcevxid)
MemoryContextSwitchTo(old_ctx)
ResourceOwner CurrentResourceOwner
void ResourceOwnerForget(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
void ResourceOwnerRemember(ResourceOwner owner, Datum value, const ResourceOwnerDesc *kind)
void ResourceOwnerEnlarge(ResourceOwner owner)
#define RELEASE_PRIO_SNAPSHOT_REFS
@ RESOURCE_RELEASE_AFTER_LOCKS
Size add_size(Size s1, Size s2)
Size mul_size(Size s1, Size s2)
static pg_noinline void Size size
static Snapshot HistoricSnapshot
static Snapshot FirstXactSnapshot
static SnapshotData CatalogSnapshotData
void UnregisterSnapshotFromOwner(Snapshot snapshot, ResourceOwner owner)
static void SetTransactionSnapshot(Snapshot sourcesnap, VirtualTransactionId *sourcevxid, int sourcepid, PGPROC *sourceproc)
void AtSubAbort_Snapshot(int level)
void SerializeSnapshot(Snapshot snapshot, char *start_address)
SnapshotData SnapshotSelfData
void AtEOXact_Snapshot(bool isCommit, bool resetXmin)
struct ActiveSnapshotElt ActiveSnapshotElt
static Snapshot CurrentSnapshot
static Snapshot SecondarySnapshot
bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
static List * exportedSnapshots
static pairingheap RegisteredSnapshots
static void UnregisterSnapshotNoOwner(Snapshot snapshot)
Snapshot GetTransactionSnapshot(void)
Snapshot GetLatestSnapshot(void)
void TeardownHistoricSnapshot(bool is_error)
Snapshot GetCatalogSnapshot(Oid relid)
void UnregisterSnapshot(Snapshot snapshot)
void PushActiveSnapshot(Snapshot snapshot)
static Snapshot CopySnapshot(Snapshot snapshot)
Snapshot RestoreSnapshot(char *start_address)
void AtSubCommit_Snapshot(int level)
static void ResourceOwnerRememberSnapshot(ResourceOwner owner, Snapshot snap)
void UpdateActiveSnapshotCommandId(void)
static void SnapshotResetXmin(void)
static int parseIntFromText(const char *prefix, char **s, const char *filename)
static SnapshotData SecondarySnapshotData
static int xmin_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg)
TransactionId TransactionXmin
SnapshotData SnapshotAnyData
bool HistoricSnapshotActive(void)
void ImportSnapshot(const char *idstr)
bool ActiveSnapshotSet(void)
Snapshot RegisterSnapshot(Snapshot snapshot)
bool XactHasExportedSnapshots(void)
static const ResourceOwnerDesc snapshot_resowner_desc
void DeleteAllExportedSnapshotFiles(void)
static void parseVxidFromText(const char *prefix, char **s, const char *filename, VirtualTransactionId *vxid)
static void FreeSnapshot(Snapshot snapshot)
#define SNAPSHOT_EXPORT_DIR
bool HaveRegisteredOrActiveSnapshot(void)
void InvalidateCatalogSnapshotConditionally(void)
static SnapshotData CurrentSnapshotData
SnapshotData SnapshotToastData
static void ResourceOwnerForgetSnapshot(ResourceOwner owner, Snapshot snap)
bool ThereAreNoPriorRegisteredSnapshots(void)
void RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc)
void SnapshotSetCommandId(CommandId curcid)
struct SerializedSnapshotData SerializedSnapshotData
void PopActiveSnapshot(void)
void PushCopiedSnapshot(Snapshot snapshot)
char * ExportSnapshot(Snapshot snapshot)
Size EstimateSnapshotSpace(Snapshot snapshot)
static ActiveSnapshotElt * ActiveSnapshot
void SetupHistoricSnapshot(Snapshot historic_snapshot, HTAB *tuplecids)
static HTAB * tuplecid_data
Snapshot RegisterSnapshotOnOwner(Snapshot snapshot, ResourceOwner owner)
static TransactionId parseXidFromText(const char *prefix, char **s, const char *filename)
HTAB * HistoricSnapshotGetTupleCids(void)
void InvalidateCatalogSnapshot(void)
void PushActiveSnapshotWithLevel(Snapshot snapshot, int snap_level)
Snapshot GetNonHistoricCatalogSnapshot(Oid relid)
struct ExportedSnapshot ExportedSnapshot
static Snapshot CatalogSnapshot
Snapshot GetActiveSnapshot(void)
Datum pg_export_snapshot(PG_FUNCTION_ARGS)
static void ResOwnerReleaseSnapshot(Datum res)
struct SnapshotData * Snapshot
struct SnapshotData SnapshotData
void appendStringInfo(StringInfo str, const char *fmt,...)
void appendStringInfoString(StringInfo str, const char *s)
void initStringInfo(StringInfo str)
struct ActiveSnapshotElt * as_next
uint64 snapXactCompletionCount
SnapshotType snapshot_type
LocalTransactionId localTransactionId
TransactionId SubTransGetTopmostTransaction(TransactionId xid)
bool RelationHasSysCache(Oid relid)
bool RelationInvalidatesSnapshotsOnly(Oid relid)
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
#define InvalidTransactionId
#define FirstNormalTransactionId
#define TransactionIdIsValid(xid)
#define TransactionIdIsNormal(xid)
text * cstring_to_text(const char *s)
int GetCurrentTransactionNestLevel(void)
TransactionId GetTopTransactionIdIfAny(void)
bool IsSubTransaction(void)
bool IsInParallelMode(void)
int xactGetCommittedChildren(TransactionId **ptr)
CommandId GetCurrentCommandId(bool used)
#define XACT_SERIALIZABLE
#define IsolationUsesXactSnapshot()
#define IsolationIsSerializable()