151 #define SNAPSHOT_EXPORT_DIR "pg_snapshots"
225 "cannot take query snapshot during a parallel operation");
280 "cannot update SecondarySnapshot during a parallel operation");
306 Snapshot OldestRegisteredSnapshot = NULL;
313 RegisteredLSN = OldestRegisteredSnapshot->
lsn;
324 return OldestRegisteredSnapshot;
459 int sourcepid,
PGPROC *sourceproc)
486 if (sourcesnap->
xcnt > 0)
511 if (sourceproc != NULL)
515 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
516 errmsg(
"could not import the requested snapshot"),
517 errdetail(
"The source transaction is not running anymore.")));
521 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
522 errmsg(
"could not import the requested snapshot"),
523 errdetail(
"The source process with PID %d is not running anymore.",
578 if (snapshot->
xcnt > 0)
581 memcpy(newsnap->
xip, snapshot->
xip,
713 elog(
ERROR,
"cannot modify commandid in active snapshot during a parallel operation");
819 if (snapshot == NULL)
832 if (snapshot == NULL)
1032 elog(
WARNING,
"registered snapshots seem to remain after cleanup");
1036 elog(
WARNING,
"snapshot %p still active", active);
1111 (
errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
1112 errmsg(
"cannot export a snapshot from a subtransaction")));
1176 for (
i = 0;
i < snapshot->
xcnt;
i++)
1194 for (
i = 0;
i < nchildren;
i++)
1205 snprintf(pathtmp,
sizeof(pathtmp),
"%s.tmp", path);
1209 errmsg(
"could not create file \"%s\": %m", pathtmp)));
1211 if (fwrite(
buf.data,
buf.len, 1, f) != 1)
1214 errmsg(
"could not write to file \"%s\": %m", pathtmp)));
1221 errmsg(
"could not write to file \"%s\": %m", pathtmp)));
1227 if (rename(pathtmp, path) < 0)
1230 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
1265 int prefixlen = strlen(prefix);
1268 if (strncmp(ptr, prefix, prefixlen) != 0)
1270 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1273 if (sscanf(ptr,
"%d", &
val) != 1)
1275 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1277 ptr = strchr(ptr,
'\n');
1280 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1290 int prefixlen = strlen(prefix);
1293 if (strncmp(ptr, prefix, prefixlen) != 0)
1295 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1298 if (sscanf(ptr,
"%u", &
val) != 1)
1300 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1302 ptr = strchr(ptr,
'\n');
1305 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1316 int prefixlen = strlen(prefix);
1318 if (strncmp(ptr, prefix, prefixlen) != 0)
1320 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1325 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1327 ptr = strchr(ptr,
'\n');
1330 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1346 struct stat stat_buf;
1367 (
errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
1368 errmsg(
"SET TRANSACTION SNAPSHOT must be called before any query")));
1376 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1377 errmsg(
"a snapshot-importing transaction must have isolation level SERIALIZABLE or REPEATABLE READ")));
1383 if (strspn(idstr,
"0123456789ABCDEF-") != strlen(idstr))
1385 (
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1386 errmsg(
"invalid snapshot identifier: \"%s\"", idstr)));
1398 if (errno == ENOENT)
1400 (
errcode(ERRCODE_UNDEFINED_OBJECT),
1401 errmsg(
"snapshot \"%s\" does not exist", idstr)));
1405 errmsg(
"could not open file \"%s\" for reading: %m",
1410 if (
fstat(fileno(f), &stat_buf))
1411 elog(
ERROR,
"could not stat file \"%s\": %m", path);
1415 if (fread(filebuf, stat_buf.
st_size, 1, f) != 1)
1416 elog(
ERROR,
"could not read file \"%s\": %m", path);
1418 filebuf[stat_buf.
st_size] =
'\0';
1425 memset(&snapshot, 0,
sizeof(snapshot));
1444 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1445 errmsg(
"invalid snapshot data in file \"%s\"", path)));
1448 for (
i = 0;
i < xcnt;
i++)
1460 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1461 errmsg(
"invalid snapshot data in file \"%s\"", path)));
1464 for (
i = 0;
i < xcnt;
i++)
1485 (
errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
1486 errmsg(
"invalid snapshot data in file \"%s\"", path)));
1498 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1499 errmsg(
"a serializable transaction cannot import a snapshot from a non-serializable transaction")));
1502 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1503 errmsg(
"a non-read-only serializable transaction cannot import a snapshot from a read-only transaction")));
1517 (
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1518 errmsg(
"cannot import a snapshot from a different database")));
1557 if (strcmp(s_de->
d_name,
".") == 0 ||
1558 strcmp(s_de->
d_name,
"..") == 0)
1563 if (unlink(
buf) != 0)
1566 errmsg(
"could not remove file \"%s\": %m",
buf)));
1626 Assert(historic_snapshot != NULL);
1698 serialized_snapshot.
xmin = snapshot->
xmin;
1699 serialized_snapshot.
xmax = snapshot->
xmax;
1700 serialized_snapshot.
xcnt = snapshot->
xcnt;
1706 serialized_snapshot.
lsn = snapshot->
lsn;
1714 serialized_snapshot.
subxcnt = 0;
1717 memcpy(start_address,
1721 if (snapshot->
xcnt > 0)
1732 if (serialized_snapshot.
subxcnt > 0)
1757 memcpy(&serialized_snapshot, start_address,
1770 snapshot->
xmin = serialized_snapshot.
xmin;
1771 snapshot->
xmax = serialized_snapshot.
xmax;
1772 snapshot->
xip = NULL;
1773 snapshot->
xcnt = serialized_snapshot.
xcnt;
1780 snapshot->
lsn = serialized_snapshot.
lsn;
1784 if (serialized_snapshot.
xcnt > 0)
1787 memcpy(snapshot->
xip, serialized_xids,
1792 if (serialized_snapshot.
subxcnt > 0)
1795 serialized_snapshot.
xcnt;
1796 memcpy(snapshot->
subxip, serialized_xids + serialized_snapshot.
xcnt,
#define OidIsValid(objectId)
elog(ERROR, "%s: %s", p2, msg)
int errcode_for_file_access(void)
int errdetail(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
FILE * AllocateFile(const char *name, const char *mode)
struct dirent * ReadDirExtended(DIR *dir, const char *dirname, int elevel)
DIR * AllocateDir(const char *dirname)
#define PG_RETURN_TEXT_P(x)
Assert(fmt[strlen(fmt) - 1] !='\n')
List * lappend(List *list, void *datum)
#define VirtualTransactionIdIsValid(vxid)
MemoryContext TopTransactionContext
char * pstrdup(const char *in)
void pfree(void *pointer)
void * MemoryContextAlloc(MemoryContext context, Size size)
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 MemoryContext MemoryContextSwitchTo(MemoryContext context)
static bool pg_lfind32(uint32 key, uint32 *base, uint32 nelem)
static int list_length(const List *l)
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)
ResourceOwner CurrentResourceOwner
void ResourceOwnerRememberSnapshot(ResourceOwner owner, Snapshot snapshot)
void ResourceOwnerForgetSnapshot(ResourceOwner owner, Snapshot snapshot)
void ResourceOwnerEnlargeSnapshots(ResourceOwner owner)
Size add_size(Size s1, Size s2)
Size mul_size(Size s1, Size s2)
static Snapshot HistoricSnapshot
static Snapshot FirstXactSnapshot
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
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)
static ActiveSnapshotElt * OldestActiveSnapshot
Snapshot RestoreSnapshot(char *start_address)
HTAB * HistoricSnapshotGetTupleCids(void)
void AtSubCommit_Snapshot(int level)
void UpdateActiveSnapshotCommandId(void)
static void SnapshotResetXmin(void)
static int parseIntFromText(const char *prefix, char **s, const char *filename)
static SnapshotData SecondarySnapshotData
char * ExportSnapshot(Snapshot snapshot)
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)
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
bool ThereAreNoPriorRegisteredSnapshots(void)
void RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc)
void SnapshotSetCommandId(CommandId curcid)
struct SerializedSnapshotData SerializedSnapshotData
void PopActiveSnapshot(void)
void PushCopiedSnapshot(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)
void InvalidateCatalogSnapshot(void)
void PushActiveSnapshotWithLevel(Snapshot snapshot, int snap_level)
Snapshot GetNonHistoricCatalogSnapshot(Oid relid)
struct ExportedSnapshot ExportedSnapshot
Snapshot GetOldestSnapshot(void)
static Snapshot CatalogSnapshot
Snapshot GetActiveSnapshot(void)
Datum pg_export_snapshot(PG_FUNCTION_ARGS)
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()
#define XLogRecPtrIsInvalid(r)
#define InvalidXLogRecPtr