PostgreSQL Source Code  git master
snapbuild.h File Reference
#include "access/xlogdefs.h"
#include "utils/snapmgr.h"
Include dependency graph for snapbuild.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Typedefs

typedef struct SnapBuild SnapBuild
 

Enumerations

enum  SnapBuildState { SNAPBUILD_START = -1 , SNAPBUILD_BUILDING_SNAPSHOT = 0 , SNAPBUILD_FULL_SNAPSHOT = 1 , SNAPBUILD_CONSISTENT = 2 }
 

Functions

void CheckPointSnapBuild (void)
 
SnapBuildAllocateSnapshotBuilder (struct ReorderBuffer *reorder, TransactionId xmin_horizon, XLogRecPtr start_lsn, bool need_full_snapshot, XLogRecPtr two_phase_at)
 
void FreeSnapshotBuilder (SnapBuild *builder)
 
void SnapBuildSnapDecRefcount (Snapshot snap)
 
Snapshot SnapBuildInitialSnapshot (SnapBuild *builder)
 
const char * SnapBuildExportSnapshot (SnapBuild *builder)
 
void SnapBuildClearExportedSnapshot (void)
 
void SnapBuildResetExportedSnapshotState (void)
 
SnapBuildState SnapBuildCurrentState (SnapBuild *builder)
 
Snapshot SnapBuildGetOrBuildSnapshot (SnapBuild *builder)
 
bool SnapBuildXactNeedsSkip (SnapBuild *builder, XLogRecPtr ptr)
 
XLogRecPtr SnapBuildGetTwoPhaseAt (SnapBuild *builder)
 
void SnapBuildSetTwoPhaseAt (SnapBuild *builder, XLogRecPtr ptr)
 
void SnapBuildCommitTxn (SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts, uint32 xinfo)
 
bool SnapBuildProcessChange (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
 
void SnapBuildProcessNewCid (SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, struct xl_heap_new_cid *xlrec)
 
void SnapBuildProcessRunningXacts (SnapBuild *builder, XLogRecPtr lsn, struct xl_running_xacts *running)
 
void SnapBuildSerializationPoint (SnapBuild *builder, XLogRecPtr lsn)
 
bool SnapBuildSnapshotExists (XLogRecPtr lsn)
 

Typedef Documentation

◆ SnapBuild

typedef struct SnapBuild SnapBuild

Definition at line 1 of file snapbuild.h.

Enumeration Type Documentation

◆ SnapBuildState

Enumerator
SNAPBUILD_START 
SNAPBUILD_BUILDING_SNAPSHOT 
SNAPBUILD_FULL_SNAPSHOT 
SNAPBUILD_CONSISTENT 

Definition at line 18 of file snapbuild.h.

19 {
20  /*
21  * Initial state, we can't do much yet.
22  */
23  SNAPBUILD_START = -1,
24 
25  /*
26  * Collecting committed transactions, to build the initial catalog
27  * snapshot.
28  */
30 
31  /*
32  * We have collected enough information to decode tuples in transactions
33  * that started after this.
34  *
35  * Once we reached this we start to collect changes. We cannot apply them
36  * yet, because they might be based on transactions that were still
37  * running when FULL_SNAPSHOT was reached.
38  */
40 
41  /*
42  * Found a point after SNAPBUILD_FULL_SNAPSHOT where all transactions that
43  * were running at that point finished. Till we reach that we hold off
44  * calling any commit callbacks.
45  */
SnapBuildState
Definition: snapbuild.h:19
@ SNAPBUILD_START
Definition: snapbuild.h:23
@ SNAPBUILD_BUILDING_SNAPSHOT
Definition: snapbuild.h:29
@ SNAPBUILD_FULL_SNAPSHOT
Definition: snapbuild.h:39
@ SNAPBUILD_CONSISTENT
Definition: snapbuild.h:46

Function Documentation

◆ AllocateSnapshotBuilder()

SnapBuild* AllocateSnapshotBuilder ( struct ReorderBuffer reorder,
TransactionId  xmin_horizon,
XLogRecPtr  start_lsn,
bool  need_full_snapshot,
XLogRecPtr  two_phase_at 
)

Definition at line 316 of file snapbuild.c.

321 {
323  MemoryContext oldcontext;
324  SnapBuild *builder;
325 
326  /* allocate memory in own context, to have better accountability */
328  "snapshot builder context",
330  oldcontext = MemoryContextSwitchTo(context);
331 
332  builder = palloc0(sizeof(SnapBuild));
333 
334  builder->state = SNAPBUILD_START;
335  builder->context = context;
336  builder->reorder = reorder;
337  /* Other struct members initialized by zeroing via palloc0 above */
338 
339  builder->committed.xcnt = 0;
340  builder->committed.xcnt_space = 128; /* arbitrary number */
341  builder->committed.xip =
342  palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
343  builder->committed.includes_all_transactions = true;
344 
345  builder->catchange.xcnt = 0;
346  builder->catchange.xip = NULL;
347 
348  builder->initial_xmin_horizon = xmin_horizon;
349  builder->start_decoding_at = start_lsn;
350  builder->building_full_snapshot = need_full_snapshot;
351  builder->two_phase_at = two_phase_at;
352 
353  MemoryContextSwitchTo(oldcontext);
354 
355  return builder;
356 }
uint32 TransactionId
Definition: c.h:652
void * palloc0(Size size)
Definition: mcxt.c:1346
MemoryContext CurrentMemoryContext
Definition: mcxt.c:143
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
tree context
Definition: radixtree.h:1833
MemoryContextSwitchTo(old_ctx)
struct SnapBuild::@20 catchange
XLogRecPtr start_decoding_at
Definition: snapbuild.c:170
SnapBuildState state
Definition: snapbuild.c:155
TransactionId initial_xmin_horizon
Definition: snapbuild.c:187
TransactionId * xip
Definition: snapbuild.c:247
XLogRecPtr two_phase_at
Definition: snapbuild.c:181
bool building_full_snapshot
Definition: snapbuild.c:190
size_t xcnt
Definition: snapbuild.c:221
size_t xcnt_space
Definition: snapbuild.c:224
bool includes_all_transactions
Definition: snapbuild.c:231
MemoryContext context
Definition: snapbuild.c:158
ReorderBuffer * reorder
Definition: snapbuild.c:205
struct SnapBuild::@19 committed

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, SnapBuild::building_full_snapshot, SnapBuild::catchange, SnapBuild::committed, SnapBuild::context, context, CurrentMemoryContext, SnapBuild::includes_all_transactions, SnapBuild::initial_xmin_horizon, MemoryContextSwitchTo(), palloc0(), SnapBuild::reorder, SNAPBUILD_START, SnapBuild::start_decoding_at, SnapBuild::state, SnapBuild::two_phase_at, SnapBuild::xcnt, SnapBuild::xcnt_space, and SnapBuild::xip.

Referenced by StartupDecodingContext().

◆ CheckPointSnapBuild()

void CheckPointSnapBuild ( void  )

Definition at line 2054 of file snapbuild.c.

2055 {
2056  XLogRecPtr cutoff;
2057  XLogRecPtr redo;
2058  DIR *snap_dir;
2059  struct dirent *snap_de;
2060  char path[MAXPGPATH + 21];
2061 
2062  /*
2063  * We start off with a minimum of the last redo pointer. No new
2064  * replication slot will start before that, so that's a safe upper bound
2065  * for removal.
2066  */
2067  redo = GetRedoRecPtr();
2068 
2069  /* now check for the restart ptrs from existing slots */
2071 
2072  /* don't start earlier than the restart lsn */
2073  if (redo < cutoff)
2074  cutoff = redo;
2075 
2076  snap_dir = AllocateDir("pg_logical/snapshots");
2077  while ((snap_de = ReadDir(snap_dir, "pg_logical/snapshots")) != NULL)
2078  {
2079  uint32 hi;
2080  uint32 lo;
2081  XLogRecPtr lsn;
2082  PGFileType de_type;
2083 
2084  if (strcmp(snap_de->d_name, ".") == 0 ||
2085  strcmp(snap_de->d_name, "..") == 0)
2086  continue;
2087 
2088  snprintf(path, sizeof(path), "pg_logical/snapshots/%s", snap_de->d_name);
2089  de_type = get_dirent_type(path, snap_de, false, DEBUG1);
2090 
2091  if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)
2092  {
2093  elog(DEBUG1, "only regular files expected: %s", path);
2094  continue;
2095  }
2096 
2097  /*
2098  * temporary filenames from SnapBuildSerialize() include the LSN and
2099  * everything but are postfixed by .$pid.tmp. We can just remove them
2100  * the same as other files because there can be none that are
2101  * currently being written that are older than cutoff.
2102  *
2103  * We just log a message if a file doesn't fit the pattern, it's
2104  * probably some editors lock/state file or similar...
2105  */
2106  if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
2107  {
2108  ereport(LOG,
2109  (errmsg("could not parse file name \"%s\"", path)));
2110  continue;
2111  }
2112 
2113  lsn = ((uint64) hi) << 32 | lo;
2114 
2115  /* check whether we still need it */
2116  if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
2117  {
2118  elog(DEBUG1, "removing snapbuild snapshot %s", path);
2119 
2120  /*
2121  * It's not particularly harmful, though strange, if we can't
2122  * remove the file here. Don't prevent the checkpoint from
2123  * completing, that'd be a cure worse than the disease.
2124  */
2125  if (unlink(path) < 0)
2126  {
2127  ereport(LOG,
2129  errmsg("could not remove file \"%s\": %m",
2130  path)));
2131  continue;
2132  }
2133  }
2134  }
2135  FreeDir(snap_dir);
2136 }
unsigned int uint32
Definition: c.h:506
int errcode_for_file_access(void)
Definition: elog.c:880
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define LOG
Definition: elog.h:31
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:224
#define ereport(elevel,...)
Definition: elog.h:149
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2909
int FreeDir(DIR *dir)
Definition: fd.c:2961
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2843
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:525
PGFileType
Definition: file_utils.h:19
@ PGFILETYPE_REG
Definition: file_utils.h:22
@ PGFILETYPE_ERROR
Definition: file_utils.h:20
#define MAXPGPATH
#define snprintf
Definition: port.h:238
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:1154
Definition: dirent.c:26
Definition: dirent.h:10
char d_name[MAX_PATH]
Definition: dirent.h:15
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6393
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References AllocateDir(), dirent::d_name, DEBUG1, elog, ereport, errcode_for_file_access(), errmsg(), FreeDir(), get_dirent_type(), GetRedoRecPtr(), InvalidXLogRecPtr, LOG, MAXPGPATH, PGFILETYPE_ERROR, PGFILETYPE_REG, ReadDir(), ReplicationSlotsComputeLogicalRestartLSN(), and snprintf.

Referenced by CheckPointGuts().

◆ FreeSnapshotBuilder()

void FreeSnapshotBuilder ( SnapBuild builder)

Definition at line 362 of file snapbuild.c.

363 {
364  MemoryContext context = builder->context;
365 
366  /* free snapshot explicitly, that contains some error checking */
367  if (builder->snapshot != NULL)
368  {
370  builder->snapshot = NULL;
371  }
372 
373  /* other resources are deallocated via memory context reset */
375 }
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:454
void SnapBuildSnapDecRefcount(Snapshot snap)
Definition: snapbuild.c:457
Snapshot snapshot
Definition: snapbuild.c:195

References SnapBuild::context, context, MemoryContextDelete(), SnapBuildSnapDecRefcount(), and SnapBuild::snapshot.

Referenced by FreeDecodingContext().

◆ SnapBuildClearExportedSnapshot()

void SnapBuildClearExportedSnapshot ( void  )

Definition at line 729 of file snapbuild.c.

730 {
731  ResourceOwner tmpResOwner;
732 
733  /* nothing exported, that is the usual case */
734  if (!ExportInProgress)
735  return;
736 
737  if (!IsTransactionState())
738  elog(ERROR, "clearing exported snapshot in wrong transaction state");
739 
740  /*
741  * AbortCurrentTransaction() takes care of resetting the snapshot state,
742  * so remember SavedResourceOwnerDuringExport.
743  */
744  tmpResOwner = SavedResourceOwnerDuringExport;
745 
746  /* make sure nothing could have ever happened */
748 
749  CurrentResourceOwner = tmpResOwner;
750 }
#define ERROR
Definition: elog.h:39
ResourceOwner CurrentResourceOwner
Definition: resowner.c:165
static ResourceOwner SavedResourceOwnerDuringExport
Definition: snapbuild.c:282
static bool ExportInProgress
Definition: snapbuild.c:283
bool IsTransactionState(void)
Definition: xact.c:384
void AbortCurrentTransaction(void)
Definition: xact.c:3387

References AbortCurrentTransaction(), CurrentResourceOwner, elog, ERROR, ExportInProgress, IsTransactionState(), and SavedResourceOwnerDuringExport.

Referenced by exec_replication_command().

◆ SnapBuildCommitTxn()

void SnapBuildCommitTxn ( SnapBuild builder,
XLogRecPtr  lsn,
TransactionId  xid,
int  nsubxacts,
TransactionId subxacts,
uint32  xinfo 
)

Definition at line 1024 of file snapbuild.c.

1026 {
1027  int nxact;
1028 
1029  bool needs_snapshot = false;
1030  bool needs_timetravel = false;
1031  bool sub_needs_timetravel = false;
1032 
1033  TransactionId xmax = xid;
1034 
1035  /*
1036  * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
1037  * will they be part of a snapshot. So we don't need to record anything.
1038  */
1039  if (builder->state == SNAPBUILD_START ||
1040  (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
1041  TransactionIdPrecedes(xid, builder->next_phase_at)))
1042  {
1043  /* ensure that only commits after this are getting replayed */
1044  if (builder->start_decoding_at <= lsn)
1045  builder->start_decoding_at = lsn + 1;
1046  return;
1047  }
1048 
1049  if (builder->state < SNAPBUILD_CONSISTENT)
1050  {
1051  /* ensure that only commits after this are getting replayed */
1052  if (builder->start_decoding_at <= lsn)
1053  builder->start_decoding_at = lsn + 1;
1054 
1055  /*
1056  * If building an exportable snapshot, force xid to be tracked, even
1057  * if the transaction didn't modify the catalog.
1058  */
1059  if (builder->building_full_snapshot)
1060  {
1061  needs_timetravel = true;
1062  }
1063  }
1064 
1065  for (nxact = 0; nxact < nsubxacts; nxact++)
1066  {
1067  TransactionId subxid = subxacts[nxact];
1068 
1069  /*
1070  * Add subtransaction to base snapshot if catalog modifying, we don't
1071  * distinguish to toplevel transactions there.
1072  */
1073  if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))
1074  {
1075  sub_needs_timetravel = true;
1076  needs_snapshot = true;
1077 
1078  elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
1079  xid, subxid);
1080 
1081  SnapBuildAddCommittedTxn(builder, subxid);
1082 
1083  if (NormalTransactionIdFollows(subxid, xmax))
1084  xmax = subxid;
1085  }
1086 
1087  /*
1088  * If we're forcing timetravel we also need visibility information
1089  * about subtransaction, so keep track of subtransaction's state, even
1090  * if not catalog modifying. Don't need to distribute a snapshot in
1091  * that case.
1092  */
1093  else if (needs_timetravel)
1094  {
1095  SnapBuildAddCommittedTxn(builder, subxid);
1096  if (NormalTransactionIdFollows(subxid, xmax))
1097  xmax = subxid;
1098  }
1099  }
1100 
1101  /* if top-level modified catalog, it'll need a snapshot */
1102  if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))
1103  {
1104  elog(DEBUG2, "found top level transaction %u, with catalog changes",
1105  xid);
1106  needs_snapshot = true;
1107  needs_timetravel = true;
1108  SnapBuildAddCommittedTxn(builder, xid);
1109  }
1110  else if (sub_needs_timetravel)
1111  {
1112  /* track toplevel txn as well, subxact alone isn't meaningful */
1113  elog(DEBUG2, "forced transaction %u to do timetravel due to one of its subtransactions",
1114  xid);
1115  needs_timetravel = true;
1116  SnapBuildAddCommittedTxn(builder, xid);
1117  }
1118  else if (needs_timetravel)
1119  {
1120  elog(DEBUG2, "forced transaction %u to do timetravel", xid);
1121 
1122  SnapBuildAddCommittedTxn(builder, xid);
1123  }
1124 
1125  if (!needs_timetravel)
1126  {
1127  /* record that we cannot export a general snapshot anymore */
1128  builder->committed.includes_all_transactions = false;
1129  }
1130 
1131  Assert(!needs_snapshot || needs_timetravel);
1132 
1133  /*
1134  * Adjust xmax of the snapshot builder, we only do that for committed,
1135  * catalog modifying, transactions, everything else isn't interesting for
1136  * us since we'll never look at the respective rows.
1137  */
1138  if (needs_timetravel &&
1139  (!TransactionIdIsValid(builder->xmax) ||
1140  TransactionIdFollowsOrEquals(xmax, builder->xmax)))
1141  {
1142  builder->xmax = xmax;
1143  TransactionIdAdvance(builder->xmax);
1144  }
1145 
1146  /* if there's any reason to build a historic snapshot, do so now */
1147  if (needs_snapshot)
1148  {
1149  /*
1150  * If we haven't built a complete snapshot yet there's no need to hand
1151  * it out, it wouldn't (and couldn't) be used anyway.
1152  */
1153  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
1154  return;
1155 
1156  /*
1157  * Decrease the snapshot builder's refcount of the old snapshot, note
1158  * that it still will be used if it has been handed out to the
1159  * reorderbuffer earlier.
1160  */
1161  if (builder->snapshot)
1163 
1164  builder->snapshot = SnapBuildBuildSnapshot(builder);
1165 
1166  /* we might need to execute invalidations, add snapshot */
1167  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
1168  {
1170  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
1171  builder->snapshot);
1172  }
1173 
1174  /* refcount of the snapshot builder for the new snapshot */
1176 
1177  /* add a new catalog snapshot to all currently running transactions */
1179  }
1180 }
#define Assert(condition)
Definition: c.h:858
#define DEBUG2
Definition: elog.h:29
void ReorderBufferSetBaseSnapshot(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *rb, TransactionId xid)
static void SnapBuildSnapIncRefcount(Snapshot snap)
Definition: snapbuild.c:445
static void SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:913
static bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid, uint32 xinfo)
Definition: snapbuild.c:1187
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder)
Definition: snapbuild.c:489
static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:860
TransactionId xmax
Definition: snapbuild.c:164
TransactionId next_phase_at
Definition: snapbuild.c:212
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:280
bool TransactionIdFollowsOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:329
#define NormalTransactionIdFollows(id1, id2)
Definition: transam.h:152
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define TransactionIdAdvance(dest)
Definition: transam.h:91

References Assert, SnapBuild::building_full_snapshot, SnapBuild::committed, DEBUG1, DEBUG2, elog, SnapBuild::includes_all_transactions, SnapBuild::next_phase_at, NormalTransactionIdFollows, SnapBuild::reorder, ReorderBufferSetBaseSnapshot(), ReorderBufferXidHasBaseSnapshot(), SNAPBUILD_BUILDING_SNAPSHOT, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SNAPBUILD_START, SnapBuildAddCommittedTxn(), SnapBuildBuildSnapshot(), SnapBuildDistributeNewCatalogSnapshot(), SnapBuildSnapDecRefcount(), SnapBuildSnapIncRefcount(), SnapBuildXidHasCatalogChanges(), SnapBuild::snapshot, SnapBuild::start_decoding_at, SnapBuild::state, TransactionIdAdvance, TransactionIdFollowsOrEquals(), TransactionIdIsValid, TransactionIdPrecedes(), and SnapBuild::xmax.

Referenced by DecodeCommit().

◆ SnapBuildCurrentState()

SnapBuildState SnapBuildCurrentState ( SnapBuild builder)

Definition at line 406 of file snapbuild.c.

407 {
408  return builder->state;
409 }

References SnapBuild::state.

Referenced by DecodePrepare(), DecodingContextReady(), heap2_decode(), heap_decode(), logicalmsg_decode(), ReorderBufferCanStartStreaming(), and xact_decode().

◆ SnapBuildExportSnapshot()

const char* SnapBuildExportSnapshot ( SnapBuild builder)

Definition at line 668 of file snapbuild.c.

669 {
670  Snapshot snap;
671  char *snapname;
672 
674  elog(ERROR, "cannot export a snapshot from within a transaction");
675 
677  elog(ERROR, "can only export one snapshot at a time");
678 
680  ExportInProgress = true;
681 
683 
684  /* There doesn't seem to a nice API to set these */
686  XactReadOnly = true;
687 
688  snap = SnapBuildInitialSnapshot(builder);
689 
690  /*
691  * now that we've built a plain snapshot, make it active and use the
692  * normal mechanisms for exporting it
693  */
694  snapname = ExportSnapshot(snap);
695 
696  ereport(LOG,
697  (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
698  "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
699  snap->xcnt,
700  snapname, snap->xcnt)));
701  return snapname;
702 }
int errmsg_plural(const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: elog.c:1180
Snapshot SnapBuildInitialSnapshot(SnapBuild *builder)
Definition: snapbuild.c:569
char * ExportSnapshot(Snapshot snapshot)
Definition: snapmgr.c:1095
uint32 xcnt
Definition: snapshot.h:169
bool IsTransactionOrTransactionBlock(void)
Definition: xact.c:4933
bool XactReadOnly
Definition: xact.c:80
void StartTransactionCommand(void)
Definition: xact.c:2995
int XactIsoLevel
Definition: xact.c:77
#define XACT_REPEATABLE_READ
Definition: xact.h:38

References CurrentResourceOwner, elog, ereport, errmsg_plural(), ERROR, ExportInProgress, ExportSnapshot(), IsTransactionOrTransactionBlock(), LOG, SavedResourceOwnerDuringExport, SnapBuildInitialSnapshot(), StartTransactionCommand(), XACT_REPEATABLE_READ, XactIsoLevel, XactReadOnly, and SnapshotData::xcnt.

Referenced by CreateReplicationSlot().

◆ SnapBuildGetOrBuildSnapshot()

Snapshot SnapBuildGetOrBuildSnapshot ( SnapBuild builder)

Definition at line 708 of file snapbuild.c.

709 {
710  Assert(builder->state == SNAPBUILD_CONSISTENT);
711 
712  /* only build a new snapshot if we don't have a prebuilt one */
713  if (builder->snapshot == NULL)
714  {
715  builder->snapshot = SnapBuildBuildSnapshot(builder);
716  /* increase refcount for the snapshot builder */
718  }
719 
720  return builder->snapshot;
721 }

References Assert, SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, and SnapBuild::state.

Referenced by logicalmsg_decode().

◆ SnapBuildGetTwoPhaseAt()

XLogRecPtr SnapBuildGetTwoPhaseAt ( SnapBuild builder)

Definition at line 415 of file snapbuild.c.

416 {
417  return builder->two_phase_at;
418 }

References SnapBuild::two_phase_at.

Referenced by DecodeCommit().

◆ SnapBuildInitialSnapshot()

Snapshot SnapBuildInitialSnapshot ( SnapBuild builder)

Definition at line 569 of file snapbuild.c.

570 {
571  Snapshot snap;
572  TransactionId xid;
573  TransactionId safeXid;
574  TransactionId *newxip;
575  int newxcnt = 0;
576 
578  Assert(builder->building_full_snapshot);
579 
580  /* don't allow older snapshots */
581  InvalidateCatalogSnapshot(); /* about to overwrite MyProc->xmin */
583  elog(ERROR, "cannot build an initial slot snapshot when snapshots exist");
585 
586  if (builder->state != SNAPBUILD_CONSISTENT)
587  elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
588 
589  if (!builder->committed.includes_all_transactions)
590  elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
591 
592  /* so we don't overwrite the existing value */
594  elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
595 
596  snap = SnapBuildBuildSnapshot(builder);
597 
598  /*
599  * We know that snap->xmin is alive, enforced by the logical xmin
600  * mechanism. Due to that we can do this without locks, we're only
601  * changing our own value.
602  *
603  * Building an initial snapshot is expensive and an unenforced xmin
604  * horizon would have bad consequences, therefore always double-check that
605  * the horizon is enforced.
606  */
607  LWLockAcquire(ProcArrayLock, LW_SHARED);
608  safeXid = GetOldestSafeDecodingTransactionId(false);
609  LWLockRelease(ProcArrayLock);
610 
611  if (TransactionIdFollows(safeXid, snap->xmin))
612  elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
613  safeXid, snap->xmin);
614 
615  MyProc->xmin = snap->xmin;
616 
617  /* allocate in transaction context */
618  newxip = (TransactionId *)
620 
621  /*
622  * snapbuild.c builds transactions in an "inverted" manner, which means it
623  * stores committed transactions in ->xip, not ones in progress. Build a
624  * classical snapshot by marking all non-committed transactions as
625  * in-progress. This can be expensive.
626  */
627  for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
628  {
629  void *test;
630 
631  /*
632  * Check whether transaction committed using the decoding snapshot
633  * meaning of ->xip.
634  */
635  test = bsearch(&xid, snap->xip, snap->xcnt,
636  sizeof(TransactionId), xidComparator);
637 
638  if (test == NULL)
639  {
640  if (newxcnt >= GetMaxSnapshotXidCount())
641  ereport(ERROR,
643  errmsg("initial slot snapshot too large")));
644 
645  newxip[newxcnt++] = xid;
646  }
647 
649  }
650 
651  /* adjust remaining snapshot fields as needed */
653  snap->xcnt = newxcnt;
654  snap->xip = newxip;
655 
656  return snap;
657 }
int errcode(int sqlerrcode)
Definition: elog.c:857
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1170
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1783
@ LW_SHARED
Definition: lwlock.h:115
void * palloc(Size size)
Definition: mcxt.c:1316
#define ERRCODE_T_R_SERIALIZATION_FAILURE
Definition: pgbench.c:76
static void test(void)
TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly)
Definition: procarray.c:2932
int GetMaxSnapshotXidCount(void)
Definition: procarray.c:2057
bool HistoricSnapshotActive(void)
Definition: snapmgr.c:1672
bool HaveRegisteredOrActiveSnapshot(void)
Definition: snapmgr.c:1624
void InvalidateCatalogSnapshot(void)
Definition: snapmgr.c:422
@ SNAPSHOT_MVCC
Definition: snapshot.h:50
PGPROC * MyProc
Definition: proc.c:66
TransactionId xmin
Definition: proc.h:173
TransactionId xmin
Definition: snapshot.h:157
TransactionId xmax
Definition: snapshot.h:158
SnapshotType snapshot_type
Definition: snapshot.h:144
TransactionId * xip
Definition: snapshot.h:168
bool TransactionIdFollows(TransactionId id1, TransactionId id2)
Definition: transam.c:314
#define NormalTransactionIdPrecedes(id1, id2)
Definition: transam.h:147
int xidComparator(const void *arg1, const void *arg2)
Definition: xid.c:139

References Assert, SnapBuild::building_full_snapshot, SnapBuild::committed, elog, ereport, errcode(), ERRCODE_T_R_SERIALIZATION_FAILURE, errmsg(), ERROR, GetMaxSnapshotXidCount(), GetOldestSafeDecodingTransactionId(), HaveRegisteredOrActiveSnapshot(), HistoricSnapshotActive(), SnapBuild::includes_all_transactions, InvalidateCatalogSnapshot(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyProc, NormalTransactionIdPrecedes, palloc(), SNAPBUILD_CONSISTENT, SnapBuildBuildSnapshot(), SNAPSHOT_MVCC, SnapshotData::snapshot_type, SnapBuild::state, test(), TransactionIdAdvance, TransactionIdFollows(), TransactionIdIsValid, XACT_REPEATABLE_READ, XactIsoLevel, SnapshotData::xcnt, xidComparator(), SnapshotData::xip, SnapshotData::xmax, PGPROC::xmin, and SnapshotData::xmin.

Referenced by CreateReplicationSlot(), and SnapBuildExportSnapshot().

◆ SnapBuildProcessChange()

bool SnapBuildProcessChange ( SnapBuild builder,
TransactionId  xid,
XLogRecPtr  lsn 
)

Definition at line 768 of file snapbuild.c.

769 {
770  /*
771  * We can't handle data in transactions if we haven't built a snapshot
772  * yet, so don't store them.
773  */
774  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
775  return false;
776 
777  /*
778  * No point in keeping track of changes in transactions that we don't have
779  * enough information about to decode. This means that they started before
780  * we got into the SNAPBUILD_FULL_SNAPSHOT state.
781  */
782  if (builder->state < SNAPBUILD_CONSISTENT &&
783  TransactionIdPrecedes(xid, builder->next_phase_at))
784  return false;
785 
786  /*
787  * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
788  * be needed to decode the change we're currently processing.
789  */
790  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
791  {
792  /* only build a new snapshot if we don't have a prebuilt one */
793  if (builder->snapshot == NULL)
794  {
795  builder->snapshot = SnapBuildBuildSnapshot(builder);
796  /* increase refcount for the snapshot builder */
798  }
799 
800  /*
801  * Increase refcount for the transaction we're handing the snapshot
802  * out to.
803  */
805  ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
806  builder->snapshot);
807  }
808 
809  return true;
810 }

References SnapBuild::next_phase_at, SnapBuild::reorder, ReorderBufferSetBaseSnapshot(), ReorderBufferXidHasBaseSnapshot(), SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SnapBuildBuildSnapshot(), SnapBuildSnapIncRefcount(), SnapBuild::snapshot, SnapBuild::state, and TransactionIdPrecedes().

Referenced by heap2_decode(), heap_decode(), and logicalmsg_decode().

◆ SnapBuildProcessNewCid()

void SnapBuildProcessNewCid ( SnapBuild builder,
TransactionId  xid,
XLogRecPtr  lsn,
struct xl_heap_new_cid xlrec 
)

Definition at line 818 of file snapbuild.c.

820 {
821  CommandId cid;
822 
823  /*
824  * we only log new_cid's if a catalog tuple was modified, so mark the
825  * transaction as containing catalog modifications
826  */
827  ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
828 
829  ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
830  xlrec->target_locator, xlrec->target_tid,
831  xlrec->cmin, xlrec->cmax,
832  xlrec->combocid);
833 
834  /* figure out new command id */
835  if (xlrec->cmin != InvalidCommandId &&
836  xlrec->cmax != InvalidCommandId)
837  cid = Max(xlrec->cmin, xlrec->cmax);
838  else if (xlrec->cmax != InvalidCommandId)
839  cid = xlrec->cmax;
840  else if (xlrec->cmin != InvalidCommandId)
841  cid = xlrec->cmin;
842  else
843  {
844  cid = InvalidCommandId; /* silence compiler */
845  elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
846  }
847 
848  ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
849 }
#define InvalidCommandId
Definition: c.h:669
#define Max(x, y)
Definition: c.h:998
uint32 CommandId
Definition: c.h:666
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)
CommandId cmin
Definition: heapam_xlog.h:454
CommandId combocid
Definition: heapam_xlog.h:456
ItemPointerData target_tid
Definition: heapam_xlog.h:462
TransactionId top_xid
Definition: heapam_xlog.h:453
CommandId cmax
Definition: heapam_xlog.h:455
RelFileLocator target_locator
Definition: heapam_xlog.h:461

References xl_heap_new_cid::cmax, xl_heap_new_cid::cmin, xl_heap_new_cid::combocid, elog, ERROR, InvalidCommandId, Max, SnapBuild::reorder, ReorderBufferAddNewCommandId(), ReorderBufferAddNewTupleCids(), ReorderBufferXidSetCatalogChanges(), xl_heap_new_cid::target_locator, xl_heap_new_cid::target_tid, and xl_heap_new_cid::top_xid.

Referenced by heap2_decode().

◆ SnapBuildProcessRunningXacts()

void SnapBuildProcessRunningXacts ( SnapBuild builder,
XLogRecPtr  lsn,
struct xl_running_xacts running 
)

Definition at line 1217 of file snapbuild.c.

1218 {
1219  ReorderBufferTXN *txn;
1220  TransactionId xmin;
1221 
1222  /*
1223  * If we're not consistent yet, inspect the record to see whether it
1224  * allows to get closer to being consistent. If we are consistent, dump
1225  * our snapshot so others or we, after a restart, can use it.
1226  */
1227  if (builder->state < SNAPBUILD_CONSISTENT)
1228  {
1229  /* returns false if there's no point in performing cleanup just yet */
1230  if (!SnapBuildFindSnapshot(builder, lsn, running))
1231  return;
1232  }
1233  else
1234  SnapBuildSerialize(builder, lsn);
1235 
1236  /*
1237  * Update range of interesting xids based on the running xacts
1238  * information. We don't increase ->xmax using it, because once we are in
1239  * a consistent state we can do that ourselves and much more efficiently
1240  * so, because we only need to do it for catalog transactions since we
1241  * only ever look at those.
1242  *
1243  * NB: We only increase xmax when a catalog modifying transaction commits
1244  * (see SnapBuildCommitTxn). Because of this, xmax can be lower than
1245  * xmin, which looks odd but is correct and actually more efficient, since
1246  * we hit fast paths in heapam_visibility.c.
1247  */
1248  builder->xmin = running->oldestRunningXid;
1249 
1250  /* Remove transactions we don't need to keep track off anymore */
1251  SnapBuildPurgeOlderTxn(builder);
1252 
1253  /*
1254  * Advance the xmin limit for the current replication slot, to allow
1255  * vacuum to clean up the tuples this slot has been protecting.
1256  *
1257  * The reorderbuffer might have an xmin among the currently running
1258  * snapshots; use it if so. If not, we need only consider the snapshots
1259  * we'll produce later, which can't be less than the oldest running xid in
1260  * the record we're reading now.
1261  */
1262  xmin = ReorderBufferGetOldestXmin(builder->reorder);
1263  if (xmin == InvalidTransactionId)
1264  xmin = running->oldestRunningXid;
1265  elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1266  builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
1267  LogicalIncreaseXminForSlot(lsn, xmin);
1268 
1269  /*
1270  * Also tell the slot where we can restart decoding from. We don't want to
1271  * do that after every commit because changing that implies an fsync of
1272  * the logical slot's state file, so we only do it every time we see a
1273  * running xacts record.
1274  *
1275  * Do so by looking for the oldest in progress transaction (determined by
1276  * the first LSN of any of its relevant records). Every transaction
1277  * remembers the last location we stored the snapshot to disk before its
1278  * beginning. That point is where we can restart from.
1279  */
1280 
1281  /*
1282  * Can't know about a serialized snapshot's location if we're not
1283  * consistent.
1284  */
1285  if (builder->state < SNAPBUILD_CONSISTENT)
1286  return;
1287 
1288  txn = ReorderBufferGetOldestTXN(builder->reorder);
1289 
1290  /*
1291  * oldest ongoing txn might have started when we didn't yet serialize
1292  * anything because we hadn't reached a consistent state yet.
1293  */
1294  if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)
1296 
1297  /*
1298  * No in-progress transaction, can reuse the last serialized snapshot if
1299  * we have one.
1300  */
1301  else if (txn == NULL &&
1305  builder->last_serialized_snapshot);
1306 }
#define DEBUG3
Definition: elog.h:28
void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
Definition: logical.c:1760
void LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
Definition: logical.c:1692
TransactionId ReorderBufferGetOldestXmin(ReorderBuffer *rb)
ReorderBufferTXN * ReorderBufferGetOldestTXN(ReorderBuffer *rb)
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1605
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1319
static void SnapBuildPurgeOlderTxn(SnapBuild *builder)
Definition: snapbuild.c:947
XLogRecPtr restart_decoding_lsn
XLogRecPtr current_restart_decoding_lsn
TransactionId xmin
Definition: snapbuild.c:161
XLogRecPtr last_serialized_snapshot
Definition: snapbuild.c:200
TransactionId oldestRunningXid
Definition: standbydefs.h:53
#define InvalidTransactionId
Definition: transam.h:31

References ReorderBuffer::current_restart_decoding_lsn, DEBUG3, elog, InvalidTransactionId, InvalidXLogRecPtr, SnapBuild::last_serialized_snapshot, LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), xl_running_xacts::oldestRunningXid, SnapBuild::reorder, ReorderBufferGetOldestTXN(), ReorderBufferGetOldestXmin(), ReorderBufferTXN::restart_decoding_lsn, SNAPBUILD_CONSISTENT, SnapBuildFindSnapshot(), SnapBuildPurgeOlderTxn(), SnapBuildSerialize(), SnapBuild::state, SnapBuild::xmax, and SnapBuild::xmin.

Referenced by standby_decode().

◆ SnapBuildResetExportedSnapshotState()

void SnapBuildResetExportedSnapshotState ( void  )

Definition at line 756 of file snapbuild.c.

757 {
759  ExportInProgress = false;
760 }

References ExportInProgress, and SavedResourceOwnerDuringExport.

Referenced by AbortTransaction().

◆ SnapBuildSerializationPoint()

void SnapBuildSerializationPoint ( SnapBuild builder,
XLogRecPtr  lsn 
)

Definition at line 1592 of file snapbuild.c.

1593 {
1594  if (builder->state < SNAPBUILD_CONSISTENT)
1595  SnapBuildRestore(builder, lsn);
1596  else
1597  SnapBuildSerialize(builder, lsn);
1598 }
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1845

References SNAPBUILD_CONSISTENT, SnapBuildRestore(), SnapBuildSerialize(), and SnapBuild::state.

Referenced by xlog_decode().

◆ SnapBuildSetTwoPhaseAt()

void SnapBuildSetTwoPhaseAt ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 424 of file snapbuild.c.

425 {
426  builder->two_phase_at = ptr;
427 }

References SnapBuild::two_phase_at.

Referenced by CreateDecodingContext().

◆ SnapBuildSnapDecRefcount()

void SnapBuildSnapDecRefcount ( Snapshot  snap)

Definition at line 457 of file snapbuild.c.

458 {
459  /* make sure we don't get passed an external snapshot */
461 
462  /* make sure nobody modified our snapshot */
463  Assert(snap->curcid == FirstCommandId);
464  Assert(!snap->suboverflowed);
465  Assert(!snap->takenDuringRecovery);
466 
467  Assert(snap->regd_count == 0);
468 
469  Assert(snap->active_count > 0);
470 
471  /* slightly more likely, so it's checked even without casserts */
472  if (snap->copied)
473  elog(ERROR, "cannot free a copied snapshot");
474 
475  snap->active_count--;
476  if (snap->active_count == 0)
477  SnapBuildFreeSnapshot(snap);
478 }
#define FirstCommandId
Definition: c.h:668
static void SnapBuildFreeSnapshot(Snapshot snap)
Definition: snapbuild.c:381
@ SNAPSHOT_HISTORIC_MVCC
Definition: snapshot.h:109
bool copied
Definition: snapshot.h:185
uint32 regd_count
Definition: snapshot.h:205
uint32 active_count
Definition: snapshot.h:204
CommandId curcid
Definition: snapshot.h:187
bool suboverflowed
Definition: snapshot.h:182
bool takenDuringRecovery
Definition: snapshot.h:184

References SnapshotData::active_count, Assert, SnapshotData::copied, SnapshotData::curcid, elog, ERROR, FirstCommandId, SnapshotData::regd_count, SnapBuildFreeSnapshot(), SNAPSHOT_HISTORIC_MVCC, SnapshotData::snapshot_type, SnapshotData::suboverflowed, and SnapshotData::takenDuringRecovery.

Referenced by FreeSnapshotBuilder(), ReorderBufferCleanupTXN(), ReorderBufferFreeSnap(), ReorderBufferTransferSnapToParent(), SnapBuildCommitTxn(), and SnapBuildRestore().

◆ SnapBuildSnapshotExists()

bool SnapBuildSnapshotExists ( XLogRecPtr  lsn)

Definition at line 2142 of file snapbuild.c.

2143 {
2144  char path[MAXPGPATH];
2145  int ret;
2146  struct stat stat_buf;
2147 
2148  sprintf(path, "pg_logical/snapshots/%X-%X.snap",
2149  LSN_FORMAT_ARGS(lsn));
2150 
2151  ret = stat(path, &stat_buf);
2152 
2153  if (ret != 0 && errno != ENOENT)
2154  ereport(ERROR,
2156  errmsg("could not stat file \"%s\": %m", path)));
2157 
2158  return ret == 0;
2159 }
#define sprintf
Definition: port.h:240
#define stat
Definition: win32_port.h:284
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43

References ereport, errcode_for_file_access(), errmsg(), ERROR, LSN_FORMAT_ARGS, MAXPGPATH, sprintf, and stat.

Referenced by update_local_synced_slot().

◆ SnapBuildXactNeedsSkip()

bool SnapBuildXactNeedsSkip ( SnapBuild builder,
XLogRecPtr  ptr 
)

Definition at line 433 of file snapbuild.c.

434 {
435  return ptr < builder->start_decoding_at;
436 }

References SnapBuild::start_decoding_at.

Referenced by AssertTXNLsnOrder(), DecodeTXNNeedSkip(), logicalmsg_decode(), and ReorderBufferCanStartStreaming().