Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * snapbuild.c
4 : : *
5 : : * Infrastructure for building historic catalog snapshots based on contents
6 : : * of the WAL, for the purpose of decoding heapam.c style values in the
7 : : * WAL.
8 : : *
9 : : * NOTES:
10 : : *
11 : : * We build snapshots which can *only* be used to read catalog contents and we
12 : : * do so by reading and interpreting the WAL stream. The aim is to build a
13 : : * snapshot that behaves the same as a freshly taken MVCC snapshot would have
14 : : * at the time the XLogRecord was generated.
15 : : *
16 : : * To build the snapshots we reuse the infrastructure built for Hot
17 : : * Standby. The in-memory snapshots we build look different than HS' because
18 : : * we have different needs. To successfully decode data from the WAL we only
19 : : * need to access catalog tables and (sys|rel|cat)cache, not the actual user
20 : : * tables since the data we decode is wholly contained in the WAL
21 : : * records. Also, our snapshots need to be different in comparison to normal
22 : : * MVCC ones because in contrast to those we cannot fully rely on the clog and
23 : : * pg_subtrans for information about committed transactions because they might
24 : : * commit in the future from the POV of the WAL entry we're currently
25 : : * decoding. This definition has the advantage that we only need to prevent
26 : : * removal of catalog rows, while normal table's rows can still be
27 : : * removed. This is achieved by using the replication slot mechanism.
28 : : *
29 : : * As the percentage of transactions modifying the catalog normally is fairly
30 : : * small in comparisons to ones only manipulating user data, we keep track of
31 : : * the committed catalog modifying ones inside [xmin, xmax) instead of keeping
32 : : * track of all running transactions like it's done in a normal snapshot. Note
33 : : * that we're generally only looking at transactions that have acquired an
34 : : * xid. That is we keep a list of transactions between snapshot->(xmin, xmax)
35 : : * that we consider committed, everything else is considered aborted/in
36 : : * progress. That also allows us not to care about subtransactions before they
37 : : * have committed which means this module, in contrast to HS, doesn't have to
38 : : * care about suboverflowed subtransactions and similar.
39 : : *
40 : : * One complexity of doing this is that to e.g. handle mixed DDL/DML
41 : : * transactions we need Snapshots that see intermediate versions of the
42 : : * catalog in a transaction. During normal operation this is achieved by using
43 : : * CommandIds/cmin/cmax. The problem with that however is that for space
44 : : * efficiency reasons, the cmin and cmax are not included in WAL records. We
45 : : * cannot read the cmin/cmax from the tuple itself, either, because it is
46 : : * reset on crash recovery. Even if we could, we could not decode combocids
47 : : * which are only tracked in the original backend's memory. To work around
48 : : * that, heapam writes an extra WAL record (XLOG_HEAP2_NEW_CID) every time a
49 : : * catalog row is modified, which includes the cmin and cmax of the
50 : : * tuple. During decoding, we insert the ctid->(cmin,cmax) mappings into the
51 : : * reorder buffer, and use them at visibility checks instead of the cmin/cmax
52 : : * on the tuple itself. Check the reorderbuffer.c's comment above
53 : : * ResolveCminCmaxDuringDecoding() for details.
54 : : *
55 : : * To facilitate all this we need our own visibility routine, as the normal
56 : : * ones are optimized for different usecases.
57 : : *
58 : : * To replace the normal catalog snapshots with decoding ones use the
59 : : * SetupHistoricSnapshot() and TeardownHistoricSnapshot() functions.
60 : : *
61 : : *
62 : : *
63 : : * The snapbuild machinery is starting up in several stages, as illustrated
64 : : * by the following graph describing the SnapBuild->state transitions:
65 : : *
66 : : * +-------------------------+
67 : : * +----| START |-------------+
68 : : * | +-------------------------+ |
69 : : * | | |
70 : : * | | |
71 : : * | running_xacts #1 |
72 : : * | | |
73 : : * | | |
74 : : * | v |
75 : : * | +-------------------------+ v
76 : : * | | BUILDING_SNAPSHOT |------------>|
77 : : * | +-------------------------+ |
78 : : * | | |
79 : : * | | |
80 : : * | running_xacts #2, xacts from #1 finished |
81 : : * | | |
82 : : * | | |
83 : : * | v |
84 : : * | +-------------------------+ v
85 : : * | | FULL_SNAPSHOT |------------>|
86 : : * | +-------------------------+ |
87 : : * | | |
88 : : * running_xacts | saved snapshot
89 : : * with zero xacts | at running_xacts's lsn
90 : : * | | |
91 : : * | running_xacts with xacts from #2 finished |
92 : : * | | |
93 : : * | v |
94 : : * | +-------------------------+ |
95 : : * +--->|SNAPBUILD_CONSISTENT |<------------+
96 : : * +-------------------------+
97 : : *
98 : : * Initially the machinery is in the START stage. When an xl_running_xacts
99 : : * record is read that is sufficiently new (above the safe xmin horizon),
100 : : * there's a state transition. If there were no running xacts when the
101 : : * xl_running_xacts record was generated, we'll directly go into CONSISTENT
102 : : * state, otherwise we'll switch to the BUILDING_SNAPSHOT state. Having a full
103 : : * snapshot means that all transactions that start henceforth can be decoded
104 : : * in their entirety, but transactions that started previously can't. In
105 : : * FULL_SNAPSHOT we'll switch into CONSISTENT once all those previously
106 : : * running transactions have committed or aborted.
107 : : *
108 : : * Only transactions that commit after CONSISTENT state has been reached will
109 : : * be replayed, even though they might have started while still in
110 : : * FULL_SNAPSHOT. That ensures that we'll reach a point where no previous
111 : : * changes has been exported, but all the following ones will be. That point
112 : : * is a convenient point to initialize replication from, which is why we
113 : : * export a snapshot at that point, which *can* be used to read normal data.
114 : : *
115 : : * Copyright (c) 2012-2026, PostgreSQL Global Development Group
116 : : *
117 : : * IDENTIFICATION
118 : : * src/backend/replication/logical/snapbuild.c
119 : : *
120 : : *-------------------------------------------------------------------------
121 : : */
122 : :
123 : : #include "postgres.h"
124 : :
125 : : #include <sys/stat.h>
126 : : #include <unistd.h>
127 : :
128 : : #include "access/heapam_xlog.h"
129 : : #include "access/transam.h"
130 : : #include "access/xact.h"
131 : : #include "common/file_utils.h"
132 : : #include "miscadmin.h"
133 : : #include "pgstat.h"
134 : : #include "replication/logical.h"
135 : : #include "replication/reorderbuffer.h"
136 : : #include "replication/snapbuild.h"
137 : : #include "replication/snapbuild_internal.h"
138 : : #include "storage/fd.h"
139 : : #include "storage/lmgr.h"
140 : : #include "storage/proc.h"
141 : : #include "storage/procarray.h"
142 : : #include "storage/standby.h"
143 : : #include "utils/builtins.h"
144 : : #include "utils/memutils.h"
145 : : #include "utils/snapmgr.h"
146 : : #include "utils/snapshot.h"
147 : : #include "utils/wait_event.h"
148 : :
149 : :
150 : : /*
151 : : * Starting a transaction -- which we need to do while exporting a snapshot --
152 : : * removes knowledge about the previously used resowner, so we save it here.
153 : : */
154 : : static ResourceOwner SavedResourceOwnerDuringExport = NULL;
155 : : static bool ExportInProgress = false;
156 : :
157 : : /* ->committed and ->catchange manipulation */
158 : : static void SnapBuildPurgeOlderTxn(SnapBuild *builder);
159 : :
160 : : /* snapshot building/manipulation/distribution functions */
161 : : static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder);
162 : :
163 : : static void SnapBuildFreeSnapshot(Snapshot snap);
164 : :
165 : : static void SnapBuildSnapIncRefcount(Snapshot snap);
166 : :
167 : : static void SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid);
168 : :
169 : : static inline bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid,
170 : : uint32 xinfo);
171 : :
172 : : /* xlog reading helper functions for SnapBuildProcessRunningXacts */
173 : : static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn,
174 : : xl_running_xacts *running);
175 : : static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff);
176 : :
177 : : /* serialization functions */
178 : : static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn);
179 : : static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);
180 : : static void SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path);
181 : :
182 : : /*
183 : : * Allocate a new snapshot builder.
184 : : *
185 : : * xmin_horizon is the xid >= which we can be sure no catalog rows have been
186 : : * removed, start_lsn is the LSN >= we want to replay commits.
187 : : */
188 : : SnapBuild *
4606 rhaas@postgresql.org 189 :CBC 1263 : AllocateSnapshotBuilder(ReorderBuffer *reorder,
190 : : TransactionId xmin_horizon,
191 : : XLogRecPtr start_lsn,
192 : : bool need_full_snapshot,
193 : : bool in_slot_creation,
194 : : XLogRecPtr two_phase_at)
195 : : {
196 : : MemoryContext context;
197 : : MemoryContext oldcontext;
198 : : SnapBuild *builder;
199 : :
200 : : /* allocate memory in own context, to have better accountability */
201 : 1263 : context = AllocSetContextCreate(CurrentMemoryContext,
202 : : "snapshot builder context",
203 : : ALLOCSET_DEFAULT_SIZES);
204 : 1263 : oldcontext = MemoryContextSwitchTo(context);
205 : :
306 michael@paquier.xyz 206 : 1263 : builder = palloc0_object(SnapBuild);
207 : :
4606 rhaas@postgresql.org 208 : 1263 : builder->state = SNAPBUILD_START;
209 : 1263 : builder->context = context;
210 : 1263 : builder->reorder = reorder;
211 : : /* Other struct members initialized by zeroing via palloc0 above */
212 : :
213 : 1263 : builder->committed.xcnt = 0;
3400 tgl@sss.pgh.pa.us 214 : 1263 : builder->committed.xcnt_space = 128; /* arbitrary number */
4606 rhaas@postgresql.org 215 : 1263 : builder->committed.xip =
220 msawada@postgresql.o 216 : 1263 : palloc0_array(TransactionId, builder->committed.xcnt_space);
4606 rhaas@postgresql.org 217 : 1263 : builder->committed.includes_all_transactions = true;
218 : :
1523 akapila@postgresql.o 219 : 1263 : builder->catchange.xcnt = 0;
220 : 1263 : builder->catchange.xip = NULL;
221 : :
4606 rhaas@postgresql.org 222 : 1263 : builder->initial_xmin_horizon = xmin_horizon;
4512 andres@anarazel.de 223 : 1263 : builder->start_decoding_at = start_lsn;
823 msawada@postgresql.o 224 : 1263 : builder->in_slot_creation = in_slot_creation;
3455 andres@anarazel.de 225 : 1263 : builder->building_full_snapshot = need_full_snapshot;
1916 akapila@postgresql.o 226 : 1263 : builder->two_phase_at = two_phase_at;
227 : :
4606 rhaas@postgresql.org 228 : 1263 : MemoryContextSwitchTo(oldcontext);
229 : :
230 : 1263 : return builder;
231 : : }
232 : :
233 : : /*
234 : : * Free a snapshot builder.
235 : : */
236 : : void
237 : 992 : FreeSnapshotBuilder(SnapBuild *builder)
238 : : {
239 : 992 : MemoryContext context = builder->context;
240 : :
241 : : /* free snapshot explicitly, that contains some error checking */
242 [ + + ]: 992 : if (builder->snapshot != NULL)
243 : : {
244 : 246 : SnapBuildSnapDecRefcount(builder->snapshot);
245 : 246 : builder->snapshot = NULL;
246 : : }
247 : :
248 : : /* other resources are deallocated via memory context reset */
249 : 992 : MemoryContextDelete(context);
250 : 992 : }
251 : :
252 : : /*
253 : : * Free an unreferenced snapshot that has previously been built by us.
254 : : */
255 : : static void
256 : 2090 : SnapBuildFreeSnapshot(Snapshot snap)
257 : : {
258 : : /* make sure we don't get passed an external snapshot */
2821 andres@anarazel.de 259 [ - + ]: 2090 : Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);
260 : :
261 : : /* make sure nobody modified our snapshot */
4606 rhaas@postgresql.org 262 [ - + ]: 2090 : Assert(snap->curcid == FirstCommandId);
263 [ - + ]: 2090 : Assert(!snap->suboverflowed);
264 [ - + ]: 2090 : Assert(!snap->takenDuringRecovery);
4197 heikki.linnakangas@i 265 [ - + ]: 2090 : Assert(snap->regd_count == 0);
266 : :
267 : : /* slightly more likely, so it's checked even without c-asserts */
4606 rhaas@postgresql.org 268 [ - + ]: 2090 : if (snap->copied)
4606 rhaas@postgresql.org 269 [ # # ]:UBC 0 : elog(ERROR, "cannot free a copied snapshot");
270 : :
4606 rhaas@postgresql.org 271 [ - + ]:CBC 2090 : if (snap->active_count)
4606 rhaas@postgresql.org 272 [ # # ]:UBC 0 : elog(ERROR, "cannot free an active snapshot");
273 : :
4606 rhaas@postgresql.org 274 :CBC 2090 : pfree(snap);
275 : 2090 : }
276 : :
277 : : /*
278 : : * In which state of snapshot building are we?
279 : : */
280 : : SnapBuildState
281 : 2097657 : SnapBuildCurrentState(SnapBuild *builder)
282 : : {
283 : 2097657 : return builder->state;
284 : : }
285 : :
286 : : /*
287 : : * Return the LSN at which the two-phase decoding was first enabled.
288 : : */
289 : : XLogRecPtr
1916 akapila@postgresql.o 290 : 37 : SnapBuildGetTwoPhaseAt(SnapBuild *builder)
291 : : {
292 : 37 : return builder->two_phase_at;
293 : : }
294 : :
295 : : /*
296 : : * Set the LSN at which two-phase decoding is enabled.
297 : : */
298 : : void
299 : 8 : SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
300 : : {
301 : 8 : builder->two_phase_at = ptr;
2051 302 : 8 : }
303 : :
304 : : /*
305 : : * Should the contents of transaction ending at 'ptr' be decoded?
306 : : */
307 : : bool
4606 rhaas@postgresql.org 308 : 374820 : SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
309 : : {
4512 andres@anarazel.de 310 : 374820 : return ptr < builder->start_decoding_at;
311 : : }
312 : :
313 : : /*
314 : : * Increase refcount of a snapshot.
315 : : *
316 : : * This is used when handing out a snapshot to some external resource or when
317 : : * adding a Snapshot as builder->snapshot.
318 : : */
319 : : static void
4606 rhaas@postgresql.org 320 : 8735 : SnapBuildSnapIncRefcount(Snapshot snap)
321 : : {
322 : 8735 : snap->active_count++;
323 : 8735 : }
324 : :
325 : : /*
326 : : * Decrease refcount of a snapshot and free if the refcount reaches zero.
327 : : *
328 : : * Externally visible, so that external resources that have been handed an
329 : : * IncRef'ed Snapshot can adjust its refcount easily.
330 : : */
331 : : void
332 : 8421 : SnapBuildSnapDecRefcount(Snapshot snap)
333 : : {
334 : : /* make sure we don't get passed an external snapshot */
2821 andres@anarazel.de 335 [ - + ]: 8421 : Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);
336 : :
337 : : /* make sure nobody modified our snapshot */
4606 rhaas@postgresql.org 338 [ - + ]: 8421 : Assert(snap->curcid == FirstCommandId);
339 [ - + ]: 8421 : Assert(!snap->suboverflowed);
340 [ - + ]: 8421 : Assert(!snap->takenDuringRecovery);
341 : :
4197 heikki.linnakangas@i 342 [ - + ]: 8421 : Assert(snap->regd_count == 0);
343 : :
344 [ - + ]: 8421 : Assert(snap->active_count > 0);
345 : :
346 : : /* slightly more likely, so it's checked even without casserts */
4606 rhaas@postgresql.org 347 [ - + ]: 8421 : if (snap->copied)
4606 rhaas@postgresql.org 348 [ # # ]:UBC 0 : elog(ERROR, "cannot free a copied snapshot");
349 : :
4606 rhaas@postgresql.org 350 :CBC 8421 : snap->active_count--;
4197 heikki.linnakangas@i 351 [ + + ]: 8421 : if (snap->active_count == 0)
4606 rhaas@postgresql.org 352 : 2090 : SnapBuildFreeSnapshot(snap);
353 : 8421 : }
354 : :
355 : : /*
356 : : * Build a new snapshot, based on currently committed catalog-modifying
357 : : * transactions.
358 : : *
359 : : * In-progress transactions with catalog access are *not* allowed to modify
360 : : * these snapshots; they have to copy them and fill in appropriate ->curcid
361 : : * and ->subxip/subxcnt values.
362 : : *
363 : : * Note that some of the transactions in the returned snapshot->xip might not
364 : : * have finished committing yet (we saw their commit WAL records, but we don't
365 : : * know if they've removed themselves from procarray). Callers that want to
366 : : * use the returned snapshot as an MVCC one must wait for this to happen; see
367 : : * SnapBuildInitialSnapshot().
368 : : */
369 : : static Snapshot
3407 andres@anarazel.de 370 : 2612 : SnapBuildBuildSnapshot(SnapBuild *builder)
371 : : {
372 : : Snapshot snapshot;
373 : : Size ssize;
374 : :
4606 rhaas@postgresql.org 375 [ - + ]: 2612 : Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);
376 : :
377 : 2612 : ssize = sizeof(SnapshotData)
378 : 2612 : + sizeof(TransactionId) * builder->committed.xcnt
379 : 2612 : + sizeof(TransactionId) * 1 /* toplevel xid */ ;
380 : :
381 : 2612 : snapshot = MemoryContextAllocZero(builder->context, ssize);
382 : :
2821 andres@anarazel.de 383 : 2612 : snapshot->snapshot_type = SNAPSHOT_HISTORIC_MVCC;
384 : :
385 : : /*
386 : : * We misuse the original meaning of SnapshotData's xip and subxip fields
387 : : * to make the more fitting for our needs.
388 : : *
389 : : * In the 'xip' array we store transactions that have to be treated as
390 : : * committed. Since we will only ever look at tuples from transactions
391 : : * that have modified the catalog it's more efficient to store those few
392 : : * that exist between xmin and xmax (frequently there are none).
393 : : *
394 : : * Snapshots that are used in transactions that have modified the catalog
395 : : * also use the 'subxip' array to store their toplevel xid and all the
396 : : * subtransaction xids so we can recognize when we need to treat rows as
397 : : * visible that are not in xip but still need to be visible. Subxip only
398 : : * gets filled when the transaction is copied into the context of a
399 : : * catalog modifying transaction since we otherwise share a snapshot
400 : : * between transactions. As long as a txn hasn't modified the catalog it
401 : : * doesn't need to treat any uncommitted rows as visible, so there is no
402 : : * need for those xids.
403 : : *
404 : : * Both arrays are qsort'ed so that we can use bsearch() on them.
405 : : */
4606 rhaas@postgresql.org 406 [ - + ]: 2612 : Assert(TransactionIdIsNormal(builder->xmin));
407 [ - + ]: 2612 : Assert(TransactionIdIsNormal(builder->xmax));
408 : :
409 : 2612 : snapshot->xmin = builder->xmin;
410 : 2612 : snapshot->xmax = builder->xmax;
411 : :
412 : : /* store all transactions to be treated as committed by this snapshot */
413 : 2612 : snapshot->xip =
414 : 2612 : (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
415 : 2612 : snapshot->xcnt = builder->committed.xcnt;
416 : 2612 : memcpy(snapshot->xip,
417 : 2612 : builder->committed.xip,
418 : 2612 : builder->committed.xcnt * sizeof(TransactionId));
419 : :
420 : : /* sort so we can bsearch() */
421 : 2612 : qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
422 : :
423 : : /*
424 : : * Initially, subxip is empty, i.e. it's a snapshot to be used by
425 : : * transactions that don't modify the catalog. Will be filled by
426 : : * ReorderBufferCopySnap() if necessary.
427 : : */
428 : 2612 : snapshot->subxcnt = 0;
429 : 2612 : snapshot->subxip = NULL;
430 : :
431 : 2612 : snapshot->suboverflowed = false;
432 : 2612 : snapshot->takenDuringRecovery = false;
433 : 2612 : snapshot->copied = false;
434 : 2612 : snapshot->curcid = FirstCommandId;
435 : 2612 : snapshot->active_count = 0;
4197 heikki.linnakangas@i 436 : 2612 : snapshot->regd_count = 0;
2247 andres@anarazel.de 437 : 2612 : snapshot->snapXactCompletionCount = 0;
438 : :
4606 rhaas@postgresql.org 439 : 2612 : return snapshot;
440 : : }
441 : :
442 : : /*
443 : : * Build the initial slot snapshot and convert it to a normal snapshot that
444 : : * is understood by HeapTupleSatisfiesMVCC.
445 : : *
446 : : * The snapshot will be usable directly in current transaction or exported
447 : : * for loading in different transaction.
448 : : */
449 : : Snapshot
3487 tgl@sss.pgh.pa.us 450 : 231 : SnapBuildInitialSnapshot(SnapBuild *builder)
451 : : {
452 : : Snapshot snap;
453 : : TransactionId xid;
454 : : TransactionId safeXid;
455 : : TransactionId *newxip;
4606 rhaas@postgresql.org 456 : 231 : int newxcnt = 0;
457 : :
3487 tgl@sss.pgh.pa.us 458 [ - + ]: 231 : Assert(XactIsoLevel == XACT_REPEATABLE_READ);
1421 akapila@postgresql.o 459 [ - + ]: 231 : Assert(builder->building_full_snapshot);
460 : :
461 : : /* don't allow older snapshots */
1242 tgl@sss.pgh.pa.us 462 : 231 : InvalidateCatalogSnapshot(); /* about to overwrite MyProc->xmin */
1421 akapila@postgresql.o 463 [ - + ]: 231 : if (HaveRegisteredOrActiveSnapshot())
1421 akapila@postgresql.o 464 [ # # ]:UBC 0 : elog(ERROR, "cannot build an initial slot snapshot when snapshots exist");
1421 akapila@postgresql.o 465 [ - + ]:CBC 231 : Assert(!HistoricSnapshotActive());
466 : :
4606 rhaas@postgresql.org 467 [ - + ]: 231 : if (builder->state != SNAPBUILD_CONSISTENT)
3490 peter_e@gmx.net 468 [ # # ]:UBC 0 : elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
469 : :
4606 rhaas@postgresql.org 470 [ - + ]:CBC 231 : if (!builder->committed.includes_all_transactions)
3490 peter_e@gmx.net 471 [ # # ]:UBC 0 : elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
472 : :
473 : : /* so we don't overwrite the existing value */
2251 andres@anarazel.de 474 [ - + ]:CBC 231 : if (TransactionIdIsValid(MyProc->xmin))
2251 andres@anarazel.de 475 [ # # ]:UBC 0 : elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
476 : :
3407 andres@anarazel.de 477 :CBC 231 : snap = SnapBuildBuildSnapshot(builder);
478 : :
479 : : /*
480 : : * Transactions appear in snap->xip as soon as their commit WAL records
481 : : * are decoded, but that doesn't imply they can be considered committed in
482 : : * a regular MVCC snapshot: that happens only when they remove themselves
483 : : * from procarray. Wait for this to happen; otherwise somebody using the
484 : : * snapshot might set hint bits incorrectly.
485 : : *
486 : : * This is not needed during recovery, because the decoded commit record
487 : : * has been replayed already.
488 : : */
18 alvherre@kurilemu.de 489 [ + - ]: 231 : if (!RecoveryInProgress())
490 : : {
491 : : RunningTransactions running;
492 : :
493 : 231 : running = GetRunningTransactionData();
494 : 231 : LWLockRelease(ProcArrayLock);
495 : 231 : LWLockRelease(XidGenLock);
496 : :
497 : : /*
498 : : * Note we walk the array up to running->xcnt only, omitting this wait
499 : : * for subtransactions: the subxacts are covered by their top-level
500 : : * transaction already, there's no need for an additional wait.
501 : : */
502 [ + + ]: 233 : for (int i = 0; i < running->xcnt; i++)
503 : : {
504 : 2 : TransactionId running_xid = running->xids[i];
505 : :
506 [ + - ]: 2 : if (bsearch(&running_xid, snap->xip, snap->xcnt,
507 : : sizeof(TransactionId), xidComparator) != NULL)
508 : 2 : XactLockTableWait(running_xid, NULL, NULL, XLTW_None);
509 : : }
510 : : }
511 : :
512 : : /*
513 : : * Building an initial snapshot is expensive and an unenforced xmin
514 : : * horizon would have bad consequences, therefore always double-check that
515 : : * the horizon is enforced.
516 : : */
1421 akapila@postgresql.o 517 : 231 : LWLockAcquire(ProcArrayLock, LW_SHARED);
518 : 231 : safeXid = GetOldestSafeDecodingTransactionId(false);
519 : 231 : LWLockRelease(ProcArrayLock);
520 : :
521 [ - + ]: 231 : if (TransactionIdFollows(safeXid, snap->xmin))
1421 akapila@postgresql.o 522 [ # # ]:UBC 0 : elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
523 : : safeXid, snap->xmin);
524 : :
525 : : /*
526 : : * We know that snap->xmin is alive, enforced by the logical xmin
527 : : * mechanism. Due to that we can do this without locks, we're only
528 : : * changing our own value.
529 : : */
2251 andres@anarazel.de 530 :CBC 231 : MyProc->xmin = snap->xmin;
531 : :
532 : : /* allocate in transaction context */
306 michael@paquier.xyz 533 : 231 : newxip = palloc_array(TransactionId, GetMaxSnapshotXidCount());
534 : :
535 : : /*
536 : : * snapbuild.c builds transactions in an "inverted" manner, which means it
537 : : * stores committed transactions in ->xip, not ones in progress. Build a
538 : : * classical snapshot by marking all non-committed transactions as
539 : : * in-progress. This can be expensive.
540 : : */
4606 rhaas@postgresql.org 541 [ + - - + : 233 : for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
+ + ]
542 : : {
543 : : void *test;
544 : :
545 : : /*
546 : : * Check whether transaction committed using the decoding snapshot
547 : : * meaning of ->xip.
548 : : */
549 : 2 : test = bsearch(&xid, snap->xip, snap->xcnt,
550 : : sizeof(TransactionId), xidComparator);
551 : :
552 [ - + ]: 2 : if (test == NULL)
553 : : {
4606 rhaas@postgresql.org 554 [ # # ]:UBC 0 : if (newxcnt >= GetMaxSnapshotXidCount())
3490 peter_e@gmx.net 555 [ # # ]: 0 : ereport(ERROR,
556 : : (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
557 : : errmsg("initial slot snapshot too large")));
558 : :
4606 rhaas@postgresql.org 559 : 0 : newxip[newxcnt++] = xid;
560 : : }
561 : :
4606 rhaas@postgresql.org 562 [ - + ]:CBC 2 : TransactionIdAdvance(xid);
563 : : }
564 : :
565 : : /* adjust remaining snapshot fields as needed */
2791 michael@paquier.xyz 566 : 231 : snap->snapshot_type = SNAPSHOT_MVCC;
4606 rhaas@postgresql.org 567 : 231 : snap->xcnt = newxcnt;
568 : 231 : snap->xip = newxip;
569 : :
3490 peter_e@gmx.net 570 : 231 : return snap;
571 : : }
572 : :
573 : : /*
574 : : * Export a snapshot so it can be set in another session with SET TRANSACTION
575 : : * SNAPSHOT.
576 : : *
577 : : * For that we need to start a transaction in the current backend as the
578 : : * importing side checks whether the source transaction is still open to make
579 : : * sure the xmin horizon hasn't advanced since then.
580 : : */
581 : : const char *
582 : 1 : SnapBuildExportSnapshot(SnapBuild *builder)
583 : : {
584 : : Snapshot snap;
585 : : char *snapname;
586 : :
587 [ - + ]: 1 : if (IsTransactionOrTransactionBlock())
3490 peter_e@gmx.net 588 [ # # ]:UBC 0 : elog(ERROR, "cannot export a snapshot from within a transaction");
589 : :
3490 peter_e@gmx.net 590 [ - + ]:CBC 1 : if (SavedResourceOwnerDuringExport)
3490 peter_e@gmx.net 591 [ # # ]:UBC 0 : elog(ERROR, "can only export one snapshot at a time");
592 : :
3490 peter_e@gmx.net 593 :CBC 1 : SavedResourceOwnerDuringExport = CurrentResourceOwner;
594 : 1 : ExportInProgress = true;
595 : :
596 : 1 : StartTransactionCommand();
597 : :
598 : : /* There doesn't seem to a nice API to set these */
599 : 1 : XactIsoLevel = XACT_REPEATABLE_READ;
600 : 1 : XactReadOnly = true;
601 : :
3487 tgl@sss.pgh.pa.us 602 : 1 : snap = SnapBuildInitialSnapshot(builder);
603 : :
604 : : /*
605 : : * now that we've built a plain snapshot, make it active and use the
606 : : * normal mechanisms for exporting it
607 : : */
4606 rhaas@postgresql.org 608 : 1 : snapname = ExportSnapshot(snap);
609 : :
610 [ + - ]: 1 : ereport(LOG,
611 : : (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
612 : : "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
613 : : snap->xcnt,
614 : : snapname, snap->xcnt)));
615 : 1 : return snapname;
616 : : }
617 : :
618 : : /*
619 : : * Ensure there is a snapshot and if not build one for current transaction.
620 : : */
621 : : Snapshot
1477 akapila@postgresql.o 622 : 8 : SnapBuildGetOrBuildSnapshot(SnapBuild *builder)
623 : : {
3841 simon@2ndQuadrant.co 624 [ - + ]: 8 : Assert(builder->state == SNAPBUILD_CONSISTENT);
625 : :
626 : : /* only build a new snapshot if we don't have a prebuilt one */
627 [ + + ]: 8 : if (builder->snapshot == NULL)
628 : : {
3407 andres@anarazel.de 629 : 1 : builder->snapshot = SnapBuildBuildSnapshot(builder);
630 : : /* increase refcount for the snapshot builder */
3841 simon@2ndQuadrant.co 631 : 1 : SnapBuildSnapIncRefcount(builder->snapshot);
632 : : }
633 : :
634 : 8 : return builder->snapshot;
635 : : }
636 : :
637 : : /*
638 : : * Reset a previously SnapBuildExportSnapshot()'ed snapshot if there is
639 : : * any. Aborts the previously started transaction and resets the resource
640 : : * owner back to its original value.
641 : : */
642 : : void
4076 andres@anarazel.de 643 : 6153 : SnapBuildClearExportedSnapshot(void)
644 : : {
645 : : ResourceOwner tmpResOwner;
646 : :
647 : : /* nothing exported, that is the usual case */
4606 rhaas@postgresql.org 648 [ + + ]: 6153 : if (!ExportInProgress)
649 : 6152 : return;
650 : :
651 [ - + ]: 1 : if (!IsTransactionState())
4606 rhaas@postgresql.org 652 [ # # ]:UBC 0 : elog(ERROR, "clearing exported snapshot in wrong transaction state");
653 : :
654 : : /*
655 : : * AbortCurrentTransaction() takes care of resetting the snapshot state,
656 : : * so remember SavedResourceOwnerDuringExport.
657 : : */
1820 michael@paquier.xyz 658 :CBC 1 : tmpResOwner = SavedResourceOwnerDuringExport;
659 : :
660 : : /* make sure nothing could have ever happened */
4606 rhaas@postgresql.org 661 : 1 : AbortCurrentTransaction();
662 : :
1820 michael@paquier.xyz 663 : 1 : CurrentResourceOwner = tmpResOwner;
664 : : }
665 : :
666 : : /*
667 : : * Clear snapshot export state during transaction abort.
668 : : */
669 : : void
670 : 35594 : SnapBuildResetExportedSnapshotState(void)
671 : : {
4606 rhaas@postgresql.org 672 : 35594 : SavedResourceOwnerDuringExport = NULL;
673 : 35594 : ExportInProgress = false;
674 : 35594 : }
675 : :
676 : : /*
677 : : * Handle the effects of a single heap change, appropriate to the current state
678 : : * of the snapshot builder and returns whether changes made at (xid, lsn) can
679 : : * be decoded.
680 : : */
681 : : bool
682 : 1523276 : SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
683 : : {
684 : : /*
685 : : * We can't handle data in transactions if we haven't built a snapshot
686 : : * yet, so don't store them.
687 : : */
688 [ - + ]: 1523276 : if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
4606 rhaas@postgresql.org 689 :UBC 0 : return false;
690 : :
691 : : /*
692 : : * No point in keeping track of changes in transactions that we don't have
693 : : * enough information about to decode. This means that they started before
694 : : * we got into the SNAPBUILD_FULL_SNAPSHOT state.
695 : : */
4606 rhaas@postgresql.org 696 [ + + - + ]:CBC 1523285 : if (builder->state < SNAPBUILD_CONSISTENT &&
2065 andres@anarazel.de 697 : 9 : TransactionIdPrecedes(xid, builder->next_phase_at))
4606 rhaas@postgresql.org 698 :UBC 0 : return false;
699 : :
700 : : /*
701 : : * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
702 : : * be needed to decode the change we're currently processing.
703 : : */
3873 andres@anarazel.de 704 [ + + ]:CBC 1523276 : if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
705 : : {
706 : : /* only build a new snapshot if we don't have a prebuilt one */
4606 rhaas@postgresql.org 707 [ + + ]: 4445 : if (builder->snapshot == NULL)
708 : : {
3407 andres@anarazel.de 709 : 472 : builder->snapshot = SnapBuildBuildSnapshot(builder);
710 : : /* increase refcount for the snapshot builder */
4606 rhaas@postgresql.org 711 : 472 : SnapBuildSnapIncRefcount(builder->snapshot);
712 : : }
713 : :
714 : : /*
715 : : * Increase refcount for the transaction we're handing the snapshot
716 : : * out to.
717 : : */
718 : 4445 : SnapBuildSnapIncRefcount(builder->snapshot);
719 : 4445 : ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
720 : : builder->snapshot);
721 : : }
722 : :
723 : 1523276 : return true;
724 : : }
725 : :
726 : : /*
727 : : * Do CommandId/combo CID handling after reading an xl_heap_new_cid record.
728 : : * This implies that a transaction has done some form of write to system
729 : : * catalogs.
730 : : */
731 : : void
732 : 30379 : SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
733 : : XLogRecPtr lsn, xl_heap_new_cid *xlrec)
734 : : {
735 : : CommandId cid;
736 : :
737 : : /*
738 : : * we only log new_cid's if a catalog tuple was modified, so mark the
739 : : * transaction as containing catalog modifications
740 : : */
4542 bruce@momjian.us 741 : 30379 : ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
742 : :
4606 rhaas@postgresql.org 743 : 30379 : ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
744 : : xlrec->target_locator, xlrec->target_tid,
745 : : xlrec->cmin, xlrec->cmax,
746 : : xlrec->combocid);
747 : :
748 : : /* figure out new command id */
749 [ + + ]: 30379 : if (xlrec->cmin != InvalidCommandId &&
750 [ + + ]: 24901 : xlrec->cmax != InvalidCommandId)
751 : 3517 : cid = Max(xlrec->cmin, xlrec->cmax);
752 [ + + ]: 26862 : else if (xlrec->cmax != InvalidCommandId)
753 : 5478 : cid = xlrec->cmax;
754 [ + - ]: 21384 : else if (xlrec->cmin != InvalidCommandId)
755 : 21384 : cid = xlrec->cmin;
756 : : else
757 : : {
4542 bruce@momjian.us 758 :UBC 0 : cid = InvalidCommandId; /* silence compiler */
4606 rhaas@postgresql.org 759 [ # # ]: 0 : elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
760 : : }
761 : :
4606 rhaas@postgresql.org 762 :CBC 30379 : ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
763 : 30379 : }
764 : :
765 : : /*
766 : : * Add a new Snapshot and invalidation messages to all transactions we're
767 : : * decoding that currently are in-progress so they can see new catalog contents
768 : : * made by the transaction that just committed. This is necessary because those
769 : : * in-progress transactions will use the new catalog's contents from here on
770 : : * (at the very least everything they do needs to be compatible with newer
771 : : * catalog contents).
772 : : */
773 : : static void
550 akapila@postgresql.o 774 : 1899 : SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid)
775 : : {
776 : : dlist_iter txn_i;
777 : : ReorderBufferTXN *txn;
778 : :
779 : : /*
780 : : * Iterate through all toplevel transactions. This can include
781 : : * subtransactions which we just don't yet know to be that, but that's
782 : : * fine, they will just get an unnecessary snapshot and invalidations
783 : : * queued.
784 : : */
4606 rhaas@postgresql.org 785 [ + - + + ]: 3833 : dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
786 : : {
787 : 1934 : txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
788 : :
789 [ - + ]: 1934 : Assert(TransactionIdIsValid(txn->xid));
790 : :
791 : : /*
792 : : * If we don't have a base snapshot yet, there are no changes in this
793 : : * transaction which in turn implies we don't yet need a snapshot at
794 : : * all. We'll add a snapshot when the first change gets queued.
795 : : *
796 : : * Similarly, we don't need to add invalidations to a transaction
797 : : * whose base snapshot is not yet set. Once a base snapshot is built,
798 : : * it will include the xids of committed transactions that have
799 : : * modified the catalog, thus reflecting the new catalog contents. The
800 : : * existing catalog cache will have already been invalidated after
801 : : * processing the invalidations in the transaction that modified
802 : : * catalogs, ensuring that a fresh cache is constructed during
803 : : * decoding.
804 : : *
805 : : * NB: This works correctly even for subtransactions because
806 : : * ReorderBufferAssignChild() takes care to transfer the base snapshot
807 : : * to the top-level transaction, and while iterating the changequeue
808 : : * we'll get the change from the subtxn.
809 : : */
810 [ + + ]: 1934 : if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
811 : 2 : continue;
812 : :
813 : : /*
814 : : * We don't need to add snapshot or invalidations to prepared
815 : : * transactions as they should not see the new catalog contents.
816 : : */
607 msawada@postgresql.o 817 [ + + ]: 1932 : if (rbtxn_is_prepared(txn))
2107 akapila@postgresql.o 818 : 31 : continue;
819 : :
462 alvherre@kurilemu.de 820 [ + + ]: 1901 : elog(DEBUG2, "adding a new snapshot and invalidations to %u at %X/%08X",
821 : : txn->xid, LSN_FORMAT_ARGS(lsn));
822 : :
823 : : /*
824 : : * increase the snapshot's refcount for the transaction we are handing
825 : : * it out to
826 : : */
4606 rhaas@postgresql.org 827 : 1901 : SnapBuildSnapIncRefcount(builder->snapshot);
828 : 1901 : ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
829 : : builder->snapshot);
830 : :
831 : : /*
832 : : * Add invalidation messages to the reorder buffer of in-progress
833 : : * transactions except the current committed transaction, for which we
834 : : * will execute invalidations at the end.
835 : : *
836 : : * It is required, otherwise, we will end up using the stale catcache
837 : : * contents built by the current transaction even after its decoding,
838 : : * which should have been invalidated due to concurrent catalog
839 : : * changing transaction.
840 : : *
841 : : * Distribute only the invalidation messages generated by the current
842 : : * committed transaction. Invalidation messages received from other
843 : : * transactions would have already been propagated to the relevant
844 : : * in-progress transactions. This transaction would have processed
845 : : * those invalidations, ensuring that subsequent transactions observe
846 : : * a consistent cache state.
847 : : */
550 akapila@postgresql.o 848 [ + + ]: 1901 : if (txn->xid != xid)
849 : : {
850 : : uint32 ninvalidations;
851 : 33 : SharedInvalidationMessage *msgs = NULL;
852 : :
853 : 33 : ninvalidations = ReorderBufferGetInvalidations(builder->reorder,
854 : : xid, &msgs);
855 : :
856 [ + + ]: 33 : if (ninvalidations > 0)
857 : : {
858 [ - + ]: 29 : Assert(msgs != NULL);
859 : :
483 msawada@postgresql.o 860 : 29 : ReorderBufferAddDistributedInvalidations(builder->reorder,
861 : : txn->xid, lsn,
862 : : ninvalidations, msgs);
863 : : }
864 : : }
865 : : }
4606 rhaas@postgresql.org 866 : 1899 : }
867 : :
868 : : /*
869 : : * Keep track of a new catalog changing transaction that has committed.
870 : : */
871 : : static void
872 : 1911 : SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
873 : : {
874 [ - + ]: 1911 : Assert(TransactionIdIsValid(xid));
875 : :
876 [ + + ]: 1911 : if (builder->committed.xcnt == builder->committed.xcnt_space)
877 : : {
4606 rhaas@postgresql.org 878 :GBC 1 : builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
879 : :
880 [ - + ]: 1 : elog(DEBUG1, "increasing space for committed transactions to %u",
881 : : (uint32) builder->committed.xcnt_space);
882 : :
220 msawada@postgresql.o 883 : 1 : builder->committed.xip = repalloc_array(builder->committed.xip,
884 : : TransactionId,
885 : : builder->committed.xcnt_space);
886 : : }
887 : :
888 : : /*
889 : : * TODO: It might make sense to keep the array sorted here instead of
890 : : * doing it every time we build a new snapshot. On the other hand this
891 : : * gets called repeatedly when a transaction with subtransactions commits.
892 : : */
4606 rhaas@postgresql.org 893 :CBC 1911 : builder->committed.xip[builder->committed.xcnt++] = xid;
894 : 1911 : }
895 : :
896 : : /*
897 : : * Remove knowledge about transactions we treat as committed or containing catalog
898 : : * changes that are smaller than ->xmin. Those won't ever get checked via
899 : : * the ->committed or ->catchange array, respectively. The committed xids will
900 : : * get checked via the clog machinery.
901 : : *
902 : : * We can ideally remove the transaction from catchange array once it is
903 : : * finished (committed/aborted) but that could be costly as we need to maintain
904 : : * the xids order in the array.
905 : : */
906 : : static void
1523 akapila@postgresql.o 907 : 569 : SnapBuildPurgeOlderTxn(SnapBuild *builder)
908 : : {
909 : : TransactionId *workspace;
4606 rhaas@postgresql.org 910 : 569 : int surviving_xids = 0;
911 : :
912 : : /* not ready yet */
913 [ - + ]: 569 : if (!TransactionIdIsNormal(builder->xmin))
4606 rhaas@postgresql.org 914 :UBC 0 : return;
915 : :
916 : : /* TODO: Neater algorithm than just copying and iterating? */
917 : : workspace =
4606 rhaas@postgresql.org 918 :CBC 569 : MemoryContextAlloc(builder->context,
919 : 569 : builder->committed.xcnt * sizeof(TransactionId));
920 : :
921 : : /* copy xids that still are interesting to workspace */
93 peter@eisentraut.org 922 [ + + ]:GNC 1040 : for (size_t off = 0; off < builder->committed.xcnt; off++)
923 : : {
4606 rhaas@postgresql.org 924 [ + - - + :CBC 471 : if (NormalTransactionIdPrecedes(builder->committed.xip[off],
+ + ]
925 : : builder->xmin))
926 : : ; /* remove */
927 : : else
928 : 3 : workspace[surviving_xids++] = builder->committed.xip[off];
929 : : }
930 : :
931 : : /* copy workspace back to persistent state */
932 : 569 : memcpy(builder->committed.xip, workspace,
933 : : surviving_xids * sizeof(TransactionId));
934 : :
935 [ - + ]: 569 : elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
936 : : (uint32) builder->committed.xcnt, (uint32) surviving_xids,
937 : : builder->xmin, builder->xmax);
938 : 569 : builder->committed.xcnt = surviving_xids;
939 : :
940 : 569 : pfree(workspace);
941 : :
942 : : /*
943 : : * Purge xids in ->catchange as well. The purged array must also be sorted
944 : : * in xidComparator order.
945 : : */
1505 akapila@postgresql.o 946 [ + + ]: 569 : if (builder->catchange.xcnt > 0)
947 : : {
948 : : size_t off;
949 : :
950 : : /*
951 : : * Since catchange.xip is sorted, we find the lower bound of xids that
952 : : * are still interesting.
953 : : */
954 [ + + ]: 9 : for (off = 0; off < builder->catchange.xcnt; off++)
955 : : {
956 [ + + ]: 6 : if (TransactionIdFollowsOrEquals(builder->catchange.xip[off],
957 : : builder->xmin))
958 : 1 : break;
959 : : }
960 : :
961 : 4 : surviving_xids = builder->catchange.xcnt - off;
962 : :
963 [ + + ]: 4 : if (surviving_xids > 0)
964 : : {
965 : 1 : memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
966 : : surviving_xids * sizeof(TransactionId));
967 : : }
968 : : else
969 : : {
970 : 3 : pfree(builder->catchange.xip);
971 : 3 : builder->catchange.xip = NULL;
972 : : }
973 : :
974 [ - + ]: 4 : elog(DEBUG3, "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
975 : : (uint32) builder->catchange.xcnt, (uint32) surviving_xids,
976 : : builder->xmin, builder->xmax);
977 : 4 : builder->catchange.xcnt = surviving_xids;
978 : : }
979 : : }
980 : :
981 : : /*
982 : : * Handle everything that needs to be done when a transaction commits
983 : : */
984 : : void
4606 rhaas@postgresql.org 985 : 4284 : SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
986 : : int nsubxacts, TransactionId *subxacts, uint32 xinfo)
987 : : {
988 : : int nxact;
989 : :
3439 andres@anarazel.de 990 : 4284 : bool needs_snapshot = false;
991 : 4284 : bool needs_timetravel = false;
4606 rhaas@postgresql.org 992 : 4284 : bool sub_needs_timetravel = false;
993 : :
994 : 4284 : TransactionId xmax = xid;
995 : :
996 : : /*
997 : : * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
998 : : * will they be part of a snapshot. So we don't need to record anything.
999 : : */
3439 andres@anarazel.de 1000 [ + - ]: 4284 : if (builder->state == SNAPBUILD_START ||
1001 [ - + - - ]: 4284 : (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
2065 andres@anarazel.de 1002 :UBC 0 : TransactionIdPrecedes(xid, builder->next_phase_at)))
1003 : : {
1004 : : /* ensure that only commits after this are getting replayed */
3439 1005 [ # # ]: 0 : if (builder->start_decoding_at <= lsn)
1006 : 0 : builder->start_decoding_at = lsn + 1;
1007 : 0 : return;
1008 : : }
1009 : :
4606 rhaas@postgresql.org 1010 [ + + ]:CBC 4284 : if (builder->state < SNAPBUILD_CONSISTENT)
1011 : : {
1012 : : /* ensure that only commits after this are getting replayed */
4512 andres@anarazel.de 1013 [ + + ]: 7 : if (builder->start_decoding_at <= lsn)
1014 : 4 : builder->start_decoding_at = lsn + 1;
1015 : :
1016 : : /*
1017 : : * If building an exportable snapshot, force xid to be tracked, even
1018 : : * if the transaction didn't modify the catalog.
1019 : : */
3439 1020 [ + + ]: 7 : if (builder->building_full_snapshot)
1021 : : {
1022 : 2 : needs_timetravel = true;
1023 : : }
1024 : : }
1025 : :
4606 rhaas@postgresql.org 1026 [ + + ]: 5609 : for (nxact = 0; nxact < nsubxacts; nxact++)
1027 : : {
1028 : 1325 : TransactionId subxid = subxacts[nxact];
1029 : :
1030 : : /*
1031 : : * Add subtransaction to base snapshot if catalog modifying, we don't
1032 : : * distinguish to toplevel transactions there.
1033 : : */
1523 akapila@postgresql.o 1034 [ + + ]: 1325 : if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))
1035 : : {
3439 andres@anarazel.de 1036 : 10 : sub_needs_timetravel = true;
1037 : 10 : needs_snapshot = true;
1038 : :
1039 [ - + ]: 10 : elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
1040 : : xid, subxid);
1041 : :
4606 rhaas@postgresql.org 1042 : 10 : SnapBuildAddCommittedTxn(builder, subxid);
1043 : :
1044 [ + - - + : 10 : if (NormalTransactionIdFollows(subxid, xmax))
+ - ]
1045 : 10 : xmax = subxid;
1046 : : }
1047 : :
1048 : : /*
1049 : : * If we're forcing timetravel we also need visibility information
1050 : : * about subtransaction, so keep track of subtransaction's state, even
1051 : : * if not catalog modifying. Don't need to distribute a snapshot in
1052 : : * that case.
1053 : : */
3439 andres@anarazel.de 1054 [ - + ]: 1315 : else if (needs_timetravel)
1055 : : {
4606 rhaas@postgresql.org 1056 :UBC 0 : SnapBuildAddCommittedTxn(builder, subxid);
1057 [ # # # # : 0 : if (NormalTransactionIdFollows(subxid, xmax))
# # ]
1058 : 0 : xmax = subxid;
1059 : : }
1060 : : }
1061 : :
1062 : : /* if top-level modified catalog, it'll need a snapshot */
1523 akapila@postgresql.o 1063 [ + + ]:CBC 4284 : if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))
1064 : : {
3439 andres@anarazel.de 1065 [ + + ]: 1898 : elog(DEBUG2, "found top level transaction %u, with catalog changes",
1066 : : xid);
1067 : 1898 : needs_snapshot = true;
1068 : 1898 : needs_timetravel = true;
4606 rhaas@postgresql.org 1069 : 1898 : SnapBuildAddCommittedTxn(builder, xid);
1070 : : }
3439 andres@anarazel.de 1071 [ + + ]: 2386 : else if (sub_needs_timetravel)
1072 : : {
1073 : : /* track toplevel txn as well, subxact alone isn't meaningful */
1453 akapila@postgresql.o 1074 [ - + ]: 1 : elog(DEBUG2, "forced transaction %u to do timetravel due to one of its subtransactions",
1075 : : xid);
1076 : 1 : needs_timetravel = true;
4606 rhaas@postgresql.org 1077 : 1 : SnapBuildAddCommittedTxn(builder, xid);
1078 : : }
3439 andres@anarazel.de 1079 [ + + ]: 2385 : else if (needs_timetravel)
1080 : : {
1081 [ - + ]: 2 : elog(DEBUG2, "forced transaction %u to do timetravel", xid);
1082 : :
4606 rhaas@postgresql.org 1083 : 2 : SnapBuildAddCommittedTxn(builder, xid);
1084 : : }
1085 : :
3439 andres@anarazel.de 1086 [ + + ]: 4284 : if (!needs_timetravel)
1087 : : {
1088 : : /* record that we cannot export a general snapshot anymore */
1089 : 2383 : builder->committed.includes_all_transactions = false;
1090 : : }
1091 : :
1092 [ + + - + ]: 4284 : Assert(!needs_snapshot || needs_timetravel);
1093 : :
1094 : : /*
1095 : : * Adjust xmax of the snapshot builder, we only do that for committed,
1096 : : * catalog modifying, transactions, everything else isn't interesting for
1097 : : * us since we'll never look at the respective rows.
1098 : : */
1099 [ + + ]: 4284 : if (needs_timetravel &&
1100 [ + - + + ]: 3802 : (!TransactionIdIsValid(builder->xmax) ||
1101 : 1901 : TransactionIdFollowsOrEquals(xmax, builder->xmax)))
1102 : : {
1103 : 1900 : builder->xmax = xmax;
1104 [ - + ]: 1900 : TransactionIdAdvance(builder->xmax);
1105 : : }
1106 : :
1107 : : /* if there's any reason to build a historic snapshot, do so now */
1108 [ + + ]: 4284 : if (needs_snapshot)
1109 : : {
1110 : : /*
1111 : : * If we haven't built a complete snapshot yet there's no need to hand
1112 : : * it out, it wouldn't (and couldn't) be used anyway.
1113 : : */
4606 rhaas@postgresql.org 1114 [ - + ]: 1899 : if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
4606 rhaas@postgresql.org 1115 :UBC 0 : return;
1116 : :
1117 : : /*
1118 : : * Decrease the snapshot builder's refcount of the old snapshot, note
1119 : : * that it still will be used if it has been handed out to the
1120 : : * reorderbuffer earlier.
1121 : : */
4606 rhaas@postgresql.org 1122 [ + - ]:CBC 1899 : if (builder->snapshot)
1123 : 1899 : SnapBuildSnapDecRefcount(builder->snapshot);
1124 : :
3407 andres@anarazel.de 1125 : 1899 : builder->snapshot = SnapBuildBuildSnapshot(builder);
1126 : :
1127 : : /* we might need to execute invalidations, add snapshot */
4606 rhaas@postgresql.org 1128 [ + + ]: 1899 : if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
1129 : : {
1130 : 8 : SnapBuildSnapIncRefcount(builder->snapshot);
1131 : 8 : ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
1132 : : builder->snapshot);
1133 : : }
1134 : :
1135 : : /* refcount of the snapshot builder for the new snapshot */
1136 : 1899 : SnapBuildSnapIncRefcount(builder->snapshot);
1137 : :
1138 : : /*
1139 : : * Add a new catalog snapshot and invalidations messages to all
1140 : : * currently running transactions.
1141 : : */
550 akapila@postgresql.o 1142 : 1899 : SnapBuildDistributeSnapshotAndInval(builder, lsn, xid);
1143 : : }
1144 : : }
1145 : :
1146 : : /*
1147 : : * Check the reorder buffer and the snapshot to see if the given transaction has
1148 : : * modified catalogs.
1149 : : */
1150 : : static inline bool
1523 1151 : 5609 : SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid,
1152 : : uint32 xinfo)
1153 : : {
1154 [ + + ]: 5609 : if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
1155 : 1904 : return true;
1156 : :
1157 : : /*
1158 : : * The transactions that have changed catalogs must have invalidation
1159 : : * info.
1160 : : */
1161 [ + + ]: 3705 : if (!(xinfo & XACT_XINFO_HAS_INVALS))
1162 : 3697 : return false;
1163 : :
1164 : : /* Check the catchange XID array */
1165 [ + + + - ]: 12 : return ((builder->catchange.xcnt > 0) &&
1166 : 4 : (bsearch(&xid, builder->catchange.xip, builder->catchange.xcnt,
1167 : : sizeof(TransactionId), xidComparator) != NULL));
1168 : : }
1169 : :
1170 : : /* -----------------------------------
1171 : : * Snapshot building functions dealing with xlog records
1172 : : * -----------------------------------
1173 : : */
1174 : :
1175 : : /*
1176 : : * Process a running xacts record, and use its information to first build a
1177 : : * historic snapshot and later to release resources that aren't needed
1178 : : * anymore.
1179 : : */
1180 : : void
142 alvherre@kurilemu.de 1181 : 1758 : SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
1182 : : {
1183 : : ReorderBufferTXN *txn;
1184 : : TransactionId xmin;
1185 : :
1186 : : /*
1187 : : * If we're not consistent yet, inspect the record to see whether it
1188 : : * allows to get closer to being consistent. If we are consistent, dump
1189 : : * our snapshot so others or we, after a restart, can use it.
1190 : : */
4606 rhaas@postgresql.org 1191 [ + + ]: 1758 : if (builder->state < SNAPBUILD_CONSISTENT)
1192 : : {
1193 : : /* returns false if there's no point in performing cleanup just yet */
1194 [ + + ]: 1219 : if (!SnapBuildFindSnapshot(builder, lsn, running))
1195 : 1187 : return;
1196 : : }
1197 : : else
1198 : 539 : SnapBuildSerialize(builder, lsn);
1199 : :
1200 : : /*
1201 : : * Update range of interesting xids based on the running xacts
1202 : : * information. We don't increase ->xmax using it, because once we are in
1203 : : * a consistent state we can do that ourselves and much more efficiently
1204 : : * so, because we only need to do it for catalog transactions since we
1205 : : * only ever look at those.
1206 : : *
1207 : : * NB: We only increase xmax when a catalog modifying transaction commits
1208 : : * (see SnapBuildCommitTxn). Because of this, xmax can be lower than
1209 : : * xmin, which looks odd but is correct and actually more efficient, since
1210 : : * we hit fast paths in heapam_visibility.c.
1211 : : */
1212 : 569 : builder->xmin = running->oldestRunningXid;
1213 : :
1214 : : /* Remove transactions we don't need to keep track off anymore */
1523 akapila@postgresql.o 1215 : 569 : SnapBuildPurgeOlderTxn(builder);
1216 : :
1217 : : /*
1218 : : * Advance the xmin limit for the current replication slot, to allow
1219 : : * vacuum to clean up the tuples this slot has been protecting.
1220 : : *
1221 : : * The reorderbuffer might have an xmin among the currently running
1222 : : * snapshots; use it if so. If not, we need only consider the snapshots
1223 : : * we'll produce later, which can't be less than the oldest running xid in
1224 : : * the record we're reading now.
1225 : : */
3030 alvherre@alvh.no-ip. 1226 : 569 : xmin = ReorderBufferGetOldestXmin(builder->reorder);
1227 [ + + ]: 569 : if (xmin == InvalidTransactionId)
1228 : 518 : xmin = running->oldestRunningXid;
1229 [ - + ]: 569 : elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
1230 : : builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
1231 : 569 : LogicalIncreaseXminForSlot(lsn, xmin);
1232 : :
1233 : : /*
1234 : : * Also tell the slot where we can restart decoding from. We don't want to
1235 : : * do that after every commit because changing that implies an fsync of
1236 : : * the logical slot's state file, so we only do it every time we see a
1237 : : * running xacts record.
1238 : : *
1239 : : * Do so by looking for the oldest in progress transaction (determined by
1240 : : * the first LSN of any of its relevant records). Every transaction
1241 : : * remembers the last location we stored the snapshot to disk before its
1242 : : * beginning. That point is where we can restart from.
1243 : : */
1244 : :
1245 : : /*
1246 : : * Can't know about a serialized snapshot's location if we're not
1247 : : * consistent.
1248 : : */
4606 rhaas@postgresql.org 1249 [ + + ]: 569 : if (builder->state < SNAPBUILD_CONSISTENT)
1250 : 23 : return;
1251 : :
1252 : 546 : txn = ReorderBufferGetOldestTXN(builder->reorder);
1253 : :
1254 : : /*
1255 : : * oldest ongoing txn might have started when we didn't yet serialize
1256 : : * anything because we hadn't reached a consistent state yet.
1257 : : */
340 alvherre@kurilemu.de 1258 [ + + + + ]: 546 : if (txn != NULL && XLogRecPtrIsValid(txn->restart_decoding_lsn))
4606 rhaas@postgresql.org 1259 : 28 : LogicalIncreaseRestartDecodingForSlot(lsn, txn->restart_decoding_lsn);
1260 : :
1261 : : /*
1262 : : * No in-progress transaction, can reuse the last serialized snapshot if
1263 : : * we have one.
1264 : : */
1265 [ + + ]: 518 : else if (txn == NULL &&
340 alvherre@kurilemu.de 1266 [ + + ]: 486 : XLogRecPtrIsValid(builder->reorder->current_restart_decoding_lsn) &&
1267 [ + - ]: 482 : XLogRecPtrIsValid(builder->last_serialized_snapshot))
4606 rhaas@postgresql.org 1268 : 482 : LogicalIncreaseRestartDecodingForSlot(lsn,
1269 : : builder->last_serialized_snapshot);
1270 : : }
1271 : :
1272 : :
1273 : : /*
1274 : : * Build the start of a snapshot that's capable of decoding the catalog.
1275 : : *
1276 : : * Helper function for SnapBuildProcessRunningXacts() while we're not yet
1277 : : * consistent.
1278 : : *
1279 : : * Returns true if there is a point in performing internal maintenance/cleanup
1280 : : * using the xl_running_xacts record.
1281 : : */
1282 : : static bool
1283 : 1219 : SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
1284 : : {
1285 : : /* ---
1286 : : * Build catalog decoding snapshot incrementally using information about
1287 : : * the currently running transactions. There are several ways to do that:
1288 : : *
1289 : : * a) There were no running transactions when the xl_running_xacts record
1290 : : * was inserted, jump to CONSISTENT immediately. We might find such a
1291 : : * state while waiting on c)'s sub-states.
1292 : : *
1293 : : * b) This (in a previous run) or another decoding slot serialized a
1294 : : * snapshot to disk that we can use. Can't use this method while finding
1295 : : * the start point for decoding changes as the restart LSN would be an
1296 : : * arbitrary LSN but we need to find the start point to extract changes
1297 : : * where we won't see the data for partial transactions. Also, we cannot
1298 : : * use this method when a slot needs a full snapshot for export or direct
1299 : : * use, as that snapshot will only contain catalog modifying transactions.
1300 : : *
1301 : : * c) First incrementally build a snapshot for catalog tuples
1302 : : * (BUILDING_SNAPSHOT), that requires all, already in-progress,
1303 : : * transactions to finish. Every transaction starting after that
1304 : : * (FULL_SNAPSHOT state), has enough information to be decoded. But
1305 : : * for older running transactions no viable snapshot exists yet, so
1306 : : * CONSISTENT will only be reached once all of those have finished.
1307 : : * ---
1308 : : */
1309 : :
1310 : : /*
1311 : : * xl_running_xacts record is older than what we can use, we might not
1312 : : * have all necessary catalog rows anymore.
1313 : : */
1314 [ + + ]: 1219 : if (TransactionIdIsNormal(builder->initial_xmin_horizon) &&
1315 [ + - - + : 543 : NormalTransactionIdPrecedes(running->oldestRunningXid,
- + ]
1316 : : builder->initial_xmin_horizon))
1317 : : {
4606 rhaas@postgresql.org 1318 [ # # ]:UBC 0 : ereport(DEBUG1,
1319 : : errmsg_internal("skipping snapshot at %X/%08X while building logical decoding snapshot, xmin horizon too low",
1320 : : LSN_FORMAT_ARGS(lsn)),
1321 : : errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
1322 : : builder->initial_xmin_horizon, running->oldestRunningXid));
1323 : :
1324 : :
3439 andres@anarazel.de 1325 : 0 : SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);
1326 : :
4606 rhaas@postgresql.org 1327 : 0 : return true;
1328 : : }
1329 : :
1330 : : /*
1331 : : * a) No transaction were running, we can jump to consistent.
1332 : : *
1333 : : * This is not affected by races around xl_running_xacts, because we can
1334 : : * miss transaction commits, but currently not transactions starting.
1335 : : *
1336 : : * NB: We might have already started to incrementally assemble a snapshot,
1337 : : * so we need to be careful to deal with that.
1338 : : */
3439 andres@anarazel.de 1339 [ + + ]:CBC 1219 : if (running->oldestRunningXid == running->nextXid)
1340 : : {
340 alvherre@kurilemu.de 1341 [ + + ]: 1178 : if (!XLogRecPtrIsValid(builder->start_decoding_at) ||
4512 andres@anarazel.de 1342 [ + + ]: 649 : builder->start_decoding_at <= lsn)
1343 : : /* can decode everything after this */
1344 : 531 : builder->start_decoding_at = lsn + 1;
1345 : :
1346 : : /* As no transactions were running xmin/xmax can be trivially set. */
3400 tgl@sss.pgh.pa.us 1347 : 1178 : builder->xmin = running->nextXid; /* < are finished */
1348 : 1178 : builder->xmax = running->nextXid; /* >= are running */
1349 : :
1350 : : /* so we can safely use the faster comparisons */
4606 rhaas@postgresql.org 1351 [ - + ]: 1178 : Assert(TransactionIdIsNormal(builder->xmin));
1352 [ - + ]: 1178 : Assert(TransactionIdIsNormal(builder->xmax));
1353 : :
1354 : 1178 : builder->state = SNAPBUILD_CONSISTENT;
2065 andres@anarazel.de 1355 : 1178 : builder->next_phase_at = InvalidTransactionId;
1356 : :
185 fujii@postgresql.org 1357 [ + + + + ]: 1178 : ereport(LogicalDecodingLogLevel(),
1358 : : errmsg("logical decoding found consistent point at %X/%08X",
1359 : : LSN_FORMAT_ARGS(lsn)),
1360 : : errdetail("There are no running transactions."));
1361 : :
4606 rhaas@postgresql.org 1362 : 1178 : return false;
1363 : : }
1364 : :
1365 : : /*
1366 : : * b) valid on disk state and while neither building full snapshot nor
1367 : : * creating a slot.
1368 : : */
3455 andres@anarazel.de 1369 [ + + ]: 41 : else if (!builder->building_full_snapshot &&
823 msawada@postgresql.o 1370 [ + + + + ]: 52 : !builder->in_slot_creation &&
3455 andres@anarazel.de 1371 : 20 : SnapBuildRestore(builder, lsn))
1372 : : {
1373 : : /* there won't be any state to cleanup */
4606 rhaas@postgresql.org 1374 : 9 : return false;
1375 : : }
1376 : :
1377 : : /*
1378 : : * c) transition from START to BUILDING_SNAPSHOT.
1379 : : *
1380 : : * In START state, and a xl_running_xacts record with running xacts is
1381 : : * encountered. In that case, switch to BUILDING_SNAPSHOT state, and
1382 : : * record xl_running_xacts->nextXid. Once all running xacts have finished
1383 : : * (i.e. they're all >= nextXid), we have a complete catalog snapshot. It
1384 : : * might look that we could use xl_running_xacts's ->xids information to
1385 : : * get there quicker, but that is problematic because transactions marked
1386 : : * as running, might already have inserted their commit record - it's
1387 : : * infeasible to change that with locking.
1388 : : */
3439 andres@anarazel.de 1389 [ + + ]: 32 : else if (builder->state == SNAPBUILD_START)
1390 : : {
1391 : 17 : builder->state = SNAPBUILD_BUILDING_SNAPSHOT;
2065 1392 : 17 : builder->next_phase_at = running->nextXid;
1393 : :
1394 : : /*
1395 : : * Start with an xmin/xmax that's correct for future, when all the
1396 : : * currently running transactions have finished. We'll update both
1397 : : * while waiting for the pending transactions to finish.
1398 : : */
3400 tgl@sss.pgh.pa.us 1399 : 17 : builder->xmin = running->nextXid; /* < are finished */
1400 : 17 : builder->xmax = running->nextXid; /* >= are running */
1401 : :
1402 : : /* so we can safely use the faster comparisons */
4606 rhaas@postgresql.org 1403 [ - + ]: 17 : Assert(TransactionIdIsNormal(builder->xmin));
1404 [ - + ]: 17 : Assert(TransactionIdIsNormal(builder->xmax));
1405 : :
1406 [ + - ]: 17 : ereport(LOG,
1407 : : errmsg("logical decoding found initial starting point at %X/%08X",
1408 : : LSN_FORMAT_ARGS(lsn)),
1409 : : errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1410 : : running->xcnt, running->nextXid));
1411 : :
3439 andres@anarazel.de 1412 : 17 : SnapBuildWaitSnapshot(running, running->nextXid);
1413 : : }
1414 : :
1415 : : /*
1416 : : * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
1417 : : *
1418 : : * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
1419 : : * is >= than nextXid from when we switched to BUILDING_SNAPSHOT. This
1420 : : * means all transactions starting afterwards have enough information to
1421 : : * be decoded. Switch to FULL_SNAPSHOT.
1422 : : */
1423 [ + + + + ]: 23 : else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
2065 1424 : 8 : TransactionIdPrecedesOrEquals(builder->next_phase_at,
1425 : : running->oldestRunningXid))
1426 : : {
3439 1427 : 7 : builder->state = SNAPBUILD_FULL_SNAPSHOT;
2065 1428 : 7 : builder->next_phase_at = running->nextXid;
1429 : :
3439 1430 [ + - ]: 7 : ereport(LOG,
1431 : : errmsg("logical decoding found initial consistent point at %X/%08X",
1432 : : LSN_FORMAT_ARGS(lsn)),
1433 : : errdetail("Waiting for transactions (approximately %d) older than %u to end.",
1434 : : running->xcnt, running->nextXid));
1435 : :
1436 : 7 : SnapBuildWaitSnapshot(running, running->nextXid);
1437 : : }
1438 : :
1439 : : /*
1440 : : * c) transition from FULL_SNAPSHOT to CONSISTENT.
1441 : : *
1442 : : * In FULL_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid is
1443 : : * >= than nextXid from when we switched to FULL_SNAPSHOT. This means all
1444 : : * transactions that are currently in progress have a catalog snapshot,
1445 : : * and all their changes have been collected. Switch to CONSISTENT.
1446 : : */
1447 [ + + + - ]: 15 : else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
2065 1448 : 7 : TransactionIdPrecedesOrEquals(builder->next_phase_at,
1449 : : running->oldestRunningXid))
1450 : : {
3439 1451 : 7 : builder->state = SNAPBUILD_CONSISTENT;
2065 1452 : 7 : builder->next_phase_at = InvalidTransactionId;
1453 : :
185 fujii@postgresql.org 1454 [ + + + + ]: 7 : ereport(LogicalDecodingLogLevel(),
1455 : : errmsg("logical decoding found consistent point at %X/%08X",
1456 : : LSN_FORMAT_ARGS(lsn)),
1457 : : errdetail("There are no old transactions anymore."));
1458 : : }
1459 : :
1460 : : /*
1461 : : * We already started to track running xacts and need to wait for all
1462 : : * in-progress ones to finish. We fall through to the normal processing of
1463 : : * records so incremental cleanup can be performed.
1464 : : */
4606 rhaas@postgresql.org 1465 : 30 : return true;
1466 : : }
1467 : :
1468 : : /* ---
1469 : : * Iterate through xids in record, wait for all older than the cutoff to
1470 : : * finish. Then, if possible, log a new xl_running_xacts record.
1471 : : *
1472 : : * This isn't required for the correctness of decoding, but to:
1473 : : * a) allow isolationtester to notice that we're currently waiting for
1474 : : * something.
1475 : : * b) log a new xl_running_xacts record where it'd be helpful, without having
1476 : : * to wait for bgwriter or checkpointer.
1477 : : * ---
1478 : : */
1479 : : static void
3439 andres@anarazel.de 1480 : 24 : SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
1481 : : {
1482 : : int off;
1483 : :
1484 [ + + ]: 46 : for (off = 0; off < running->xcnt; off++)
1485 : : {
1486 : 24 : TransactionId xid = running->xids[off];
1487 : :
1488 : : /*
1489 : : * Upper layers should prevent that we ever need to wait on ourselves.
1490 : : * Check anyway, since failing to do so would either result in an
1491 : : * endless wait or an Assert() failure.
1492 : : */
1493 [ - + ]: 24 : if (TransactionIdIsCurrentTransactionId(xid))
3439 andres@anarazel.de 1494 [ # # ]:UBC 0 : elog(ERROR, "waiting for ourselves");
1495 : :
3439 andres@anarazel.de 1496 [ - + ]:CBC 24 : if (TransactionIdFollows(xid, cutoff))
3439 andres@anarazel.de 1497 :UBC 0 : continue;
1498 : :
3439 andres@anarazel.de 1499 :CBC 24 : XactLockTableWait(xid, NULL, NULL, XLTW_None);
1500 : : }
1501 : :
1502 : : /*
1503 : : * All transactions we needed to finish finished - try to ensure there is
1504 : : * another xl_running_xacts record in a timely manner, without having to
1505 : : * wait for bgwriter or checkpointer to log one. During recovery we can't
1506 : : * enforce that, so we'll have to wait.
1507 : : */
1508 [ + - ]: 22 : if (!RecoveryInProgress())
1509 : : {
142 alvherre@kurilemu.de 1510 : 22 : LogStandbySnapshot();
1511 : : }
3439 andres@anarazel.de 1512 : 22 : }
1513 : :
1514 : : #define SnapBuildOnDiskConstantSize \
1515 : : offsetof(SnapBuildOnDisk, builder)
1516 : : #define SnapBuildOnDiskNotChecksummedSize \
1517 : : offsetof(SnapBuildOnDisk, version)
1518 : :
1519 : : #define SNAPBUILD_MAGIC 0x51A1E001
1520 : : #define SNAPBUILD_VERSION 6
1521 : :
1522 : : /*
1523 : : * Store/Load a snapshot from disk, depending on the snapshot builder's state.
1524 : : *
1525 : : * Supposed to be used by external (i.e. not snapbuild.c) code that just read
1526 : : * a record that's a potential location for a serialized snapshot.
1527 : : */
1528 : : void
4606 rhaas@postgresql.org 1529 : 106 : SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
1530 : : {
1531 [ - + ]: 106 : if (builder->state < SNAPBUILD_CONSISTENT)
4606 rhaas@postgresql.org 1532 :UBC 0 : SnapBuildRestore(builder, lsn);
1533 : : else
4606 rhaas@postgresql.org 1534 :CBC 106 : SnapBuildSerialize(builder, lsn);
1535 : 106 : }
1536 : :
1537 : : /*
1538 : : * Serialize the snapshot 'builder' at the location 'lsn' if it hasn't already
1539 : : * been done by another decoding process.
1540 : : */
1541 : : static void
1542 : 645 : SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
1543 : : {
1544 : : Size needed_length;
2098 akapila@postgresql.o 1545 : 645 : SnapBuildOnDisk *ondisk = NULL;
1523 1546 : 645 : TransactionId *catchange_xip = NULL;
1547 : : MemoryContext old_ctx;
1548 : : size_t catchange_xcnt;
1549 : : char *ondisk_c;
1550 : : int fd;
1551 : : char tmppath[MAXPGPATH];
1552 : : char path[MAXPGPATH];
1553 : : int ret;
1554 : : struct stat stat_buf;
1555 : : Size sz;
1556 : :
340 alvherre@kurilemu.de 1557 [ - + ]: 645 : Assert(XLogRecPtrIsValid(lsn));
1558 [ + + - + ]: 645 : Assert(!XLogRecPtrIsValid(builder->last_serialized_snapshot) ||
1559 : : builder->last_serialized_snapshot <= lsn);
1560 : :
1561 : : /*
1562 : : * no point in serializing if we cannot continue to work immediately after
1563 : : * restoring the snapshot
1564 : : */
4606 rhaas@postgresql.org 1565 [ - + ]: 645 : if (builder->state < SNAPBUILD_CONSISTENT)
4606 rhaas@postgresql.org 1566 :UBC 0 : return;
1567 : :
1568 : : /* consistent snapshots have no next phase */
2065 andres@anarazel.de 1569 [ - + ]:CBC 645 : Assert(builder->next_phase_at == InvalidTransactionId);
1570 : :
1571 : : /*
1572 : : * We identify snapshots by the LSN they are valid for. We don't need to
1573 : : * include timelines in the name as each LSN maps to exactly one timeline
1574 : : * unless the user used pg_resetwal or similar. If a user did so, there's
1575 : : * no hope continuing to decode anyway.
1576 : : */
773 michael@paquier.xyz 1577 : 645 : sprintf(path, "%s/%X-%X.snap",
1578 : : PG_LOGICAL_SNAPSHOTS_DIR,
2057 peter@eisentraut.org 1579 : 645 : LSN_FORMAT_ARGS(lsn));
1580 : :
1581 : : /*
1582 : : * first check whether some other backend already has written the snapshot
1583 : : * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
1584 : : * as a valid state. Everything else is an unexpected error.
1585 : : */
4606 rhaas@postgresql.org 1586 : 645 : ret = stat(path, &stat_buf);
1587 : :
1588 [ + + - + ]: 645 : if (ret != 0 && errno != ENOENT)
4606 rhaas@postgresql.org 1589 [ # # ]:UBC 0 : ereport(ERROR,
1590 : : (errcode_for_file_access(),
1591 : : errmsg("could not stat file \"%s\": %m", path)));
1592 : :
4606 rhaas@postgresql.org 1593 [ + + ]:CBC 645 : else if (ret == 0)
1594 : : {
1595 : : /*
1596 : : * somebody else has already serialized to this point, don't overwrite
1597 : : * but remember location, so we don't need to read old data again.
1598 : : *
1599 : : * To be sure it has been synced to disk after the rename() from the
1600 : : * tempfile filename to the real filename, we just repeat the fsync.
1601 : : * That ought to be cheap because in most scenarios it should already
1602 : : * be safely on disk.
1603 : : */
1604 : 299 : fsync_fname(path, false);
773 michael@paquier.xyz 1605 : 299 : fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
1606 : :
4606 rhaas@postgresql.org 1607 : 299 : builder->last_serialized_snapshot = lsn;
1608 : 299 : goto out;
1609 : : }
1610 : :
1611 : : /*
1612 : : * there is an obvious race condition here between the time we stat(2) the
1613 : : * file and us writing the file. But we rename the file into place
1614 : : * atomically and all files created need to contain the same data anyway,
1615 : : * so this is perfectly fine, although a bit of a resource waste. Locking
1616 : : * seems like pointless complication.
1617 : : */
1618 [ + + ]: 346 : elog(DEBUG1, "serializing snapshot to %s", path);
1619 : :
1620 : : /* to make sure only we will write to this tempfile, include pid */
773 michael@paquier.xyz 1621 : 346 : sprintf(tmppath, "%s/%X-%X.snap.%d.tmp",
1622 : : PG_LOGICAL_SNAPSHOTS_DIR,
2057 peter@eisentraut.org 1623 : 346 : LSN_FORMAT_ARGS(lsn), MyProcPid);
1624 : :
1625 : : /*
1626 : : * Unlink temporary file if it already exists, needs to have been before a
1627 : : * crash/error since we won't enter this function twice from within a
1628 : : * single decoding slot/backend and the temporary file contains the pid of
1629 : : * the current process.
1630 : : */
4606 rhaas@postgresql.org 1631 [ + - - + ]: 346 : if (unlink(tmppath) != 0 && errno != ENOENT)
4606 rhaas@postgresql.org 1632 [ # # ]:UBC 0 : ereport(ERROR,
1633 : : (errcode_for_file_access(),
1634 : : errmsg("could not remove file \"%s\": %m", tmppath)));
1635 : :
1523 akapila@postgresql.o 1636 :CBC 346 : old_ctx = MemoryContextSwitchTo(builder->context);
1637 : :
1638 : : /* Get the catalog modifying transactions that are yet not committed */
1639 : 346 : catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
1440 drowley@postgresql.o 1640 : 346 : catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);
1641 : :
4606 rhaas@postgresql.org 1642 : 346 : needed_length = sizeof(SnapBuildOnDisk) +
1523 akapila@postgresql.o 1643 : 346 : sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);
1644 : :
1645 : 346 : ondisk_c = palloc0(needed_length);
4606 rhaas@postgresql.org 1646 : 346 : ondisk = (SnapBuildOnDisk *) ondisk_c;
1647 : 346 : ondisk->magic = SNAPBUILD_MAGIC;
1648 : 346 : ondisk->version = SNAPBUILD_VERSION;
1649 : 346 : ondisk->length = needed_length;
4360 heikki.linnakangas@i 1650 : 346 : INIT_CRC32C(ondisk->checksum);
1651 : 346 : COMP_CRC32C(ondisk->checksum,
1652 : : ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1653 : : SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
4606 rhaas@postgresql.org 1654 : 346 : ondisk_c += sizeof(SnapBuildOnDisk);
1655 : :
1656 : 346 : memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
1657 : : /* NULL-ify memory-only data */
1658 : 346 : ondisk->builder.context = NULL;
1659 : 346 : ondisk->builder.snapshot = NULL;
1660 : 346 : ondisk->builder.reorder = NULL;
1661 : 346 : ondisk->builder.committed.xip = NULL;
1523 akapila@postgresql.o 1662 : 346 : ondisk->builder.catchange.xip = NULL;
1663 : : /* update catchange only on disk data */
1664 : 346 : ondisk->builder.catchange.xcnt = catchange_xcnt;
1665 : :
4360 heikki.linnakangas@i 1666 : 346 : COMP_CRC32C(ondisk->checksum,
1667 : : &ondisk->builder,
1668 : : sizeof(SnapBuild));
1669 : :
1670 : : /* copy committed xacts */
1523 akapila@postgresql.o 1671 [ + + ]: 346 : if (builder->committed.xcnt > 0)
1672 : : {
1673 : 69 : sz = sizeof(TransactionId) * builder->committed.xcnt;
1674 : 69 : memcpy(ondisk_c, builder->committed.xip, sz);
1675 : 69 : COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1676 : 69 : ondisk_c += sz;
1677 : : }
1678 : :
1679 : : /* copy catalog modifying xacts */
1680 [ + + ]: 346 : if (catchange_xcnt > 0)
1681 : : {
1682 : 11 : sz = sizeof(TransactionId) * catchange_xcnt;
1683 : 11 : memcpy(ondisk_c, catchange_xip, sz);
1684 : 11 : COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
1685 : 11 : ondisk_c += sz;
1686 : : }
1687 : :
4352 andres@anarazel.de 1688 : 346 : FIN_CRC32C(ondisk->checksum);
1689 : :
1690 : : /* we have valid data now, open tempfile and write it there */
4606 rhaas@postgresql.org 1691 : 346 : fd = OpenTransientFile(tmppath,
1692 : : O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1693 [ - + ]: 346 : if (fd < 0)
4606 rhaas@postgresql.org 1694 [ # # ]:UBC 0 : ereport(ERROR,
1695 : : (errcode_for_file_access(),
1696 : : errmsg("could not open file \"%s\": %m", tmppath)));
1697 : :
2990 michael@paquier.xyz 1698 :CBC 346 : errno = 0;
3495 rhaas@postgresql.org 1699 : 346 : pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
4606 1700 [ - + ]: 346 : if ((write(fd, ondisk, needed_length)) != needed_length)
1701 : : {
3031 michael@paquier.xyz 1702 :UBC 0 : int save_errno = errno;
1703 : :
4606 rhaas@postgresql.org 1704 : 0 : CloseTransientFile(fd);
1705 : :
1706 : : /* if write didn't set errno, assume problem is no disk space */
3031 michael@paquier.xyz 1707 [ # # ]: 0 : errno = save_errno ? save_errno : ENOSPC;
4606 rhaas@postgresql.org 1708 [ # # ]: 0 : ereport(ERROR,
1709 : : (errcode_for_file_access(),
1710 : : errmsg("could not write to file \"%s\": %m", tmppath)));
1711 : : }
3495 rhaas@postgresql.org 1712 :CBC 346 : pgstat_report_wait_end();
1713 : :
1714 : : /*
1715 : : * fsync the file before renaming so that even if we crash after this we
1716 : : * have either a fully valid file or nothing.
1717 : : *
1718 : : * It's safe to just ERROR on fsync() here because we'll retry the whole
1719 : : * operation including the writes.
1720 : : *
1721 : : * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
1722 : : * some noticeable overhead since it's performed synchronously during
1723 : : * decoding?
1724 : : */
1725 : 346 : pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_SYNC);
4606 1726 [ - + ]: 346 : if (pg_fsync(fd) != 0)
1727 : : {
3031 michael@paquier.xyz 1728 :UBC 0 : int save_errno = errno;
1729 : :
4606 rhaas@postgresql.org 1730 : 0 : CloseTransientFile(fd);
3031 michael@paquier.xyz 1731 : 0 : errno = save_errno;
4606 rhaas@postgresql.org 1732 [ # # ]: 0 : ereport(ERROR,
1733 : : (errcode_for_file_access(),
1734 : : errmsg("could not fsync file \"%s\": %m", tmppath)));
1735 : : }
3495 rhaas@postgresql.org 1736 :CBC 346 : pgstat_report_wait_end();
1737 : :
2655 peter@eisentraut.org 1738 [ - + ]: 346 : if (CloseTransientFile(fd) != 0)
2774 michael@paquier.xyz 1739 [ # # ]:UBC 0 : ereport(ERROR,
1740 : : (errcode_for_file_access(),
1741 : : errmsg("could not close file \"%s\": %m", tmppath)));
1742 : :
773 michael@paquier.xyz 1743 :CBC 346 : fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
1744 : :
1745 : : /*
1746 : : * We may overwrite the work from some other backend, but that's ok, our
1747 : : * snapshot is valid as well, we'll just have done some superfluous work.
1748 : : */
4606 rhaas@postgresql.org 1749 [ - + ]: 346 : if (rename(tmppath, path) != 0)
1750 : : {
4606 rhaas@postgresql.org 1751 [ # # ]:UBC 0 : ereport(ERROR,
1752 : : (errcode_for_file_access(),
1753 : : errmsg("could not rename file \"%s\" to \"%s\": %m",
1754 : : tmppath, path)));
1755 : : }
1756 : :
1757 : : /* make sure we persist */
4606 rhaas@postgresql.org 1758 :CBC 346 : fsync_fname(path, false);
773 michael@paquier.xyz 1759 : 346 : fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
1760 : :
1761 : : /*
1762 : : * Now there's no way we can lose the dumped state anymore, remember this
1763 : : * as a serialization point.
1764 : : */
4606 rhaas@postgresql.org 1765 : 346 : builder->last_serialized_snapshot = lsn;
1766 : :
1523 akapila@postgresql.o 1767 : 346 : MemoryContextSwitchTo(old_ctx);
1768 : :
4606 rhaas@postgresql.org 1769 : 645 : out:
1770 : 645 : ReorderBufferSetRestartPoint(builder->reorder,
1771 : : builder->last_serialized_snapshot);
1772 : : /* be tidy */
2098 akapila@postgresql.o 1773 [ + + ]: 645 : if (ondisk)
1774 : 346 : pfree(ondisk);
1523 1775 [ + + ]: 645 : if (catchange_xip)
1776 : 11 : pfree(catchange_xip);
1777 : : }
1778 : :
1779 : : /*
1780 : : * Restore the logical snapshot file contents to 'ondisk'.
1781 : : *
1782 : : * 'context' is the memory context where the catalog modifying/committed xid
1783 : : * will live.
1784 : : * If 'missing_ok' is true, will not throw an error if the file is not found.
1785 : : */
1786 : : bool
580 msawada@postgresql.o 1787 : 22 : SnapBuildRestoreSnapshot(SnapBuildOnDisk *ondisk, XLogRecPtr lsn,
1788 : : MemoryContext context, bool missing_ok)
1789 : : {
1790 : : int fd;
1791 : : pg_crc32c checksum;
1792 : : Size sz;
1793 : : char path[MAXPGPATH];
1794 : :
1795 : 22 : sprintf(path, "%s/%X-%X.snap",
1796 : : PG_LOGICAL_SNAPSHOTS_DIR,
1797 : 22 : LSN_FORMAT_ARGS(lsn));
1798 : :
3306 peter_e@gmx.net 1799 : 22 : fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
1800 : :
728 msawada@postgresql.o 1801 [ + + ]: 22 : if (fd < 0)
1802 : : {
1803 [ + - + - ]: 11 : if (missing_ok && errno == ENOENT)
1804 : 11 : return false;
1805 : :
4606 rhaas@postgresql.org 1806 [ # # ]:UBC 0 : ereport(ERROR,
1807 : : (errcode_for_file_access(),
1808 : : errmsg("could not open file \"%s\": %m", path)));
1809 : : }
1810 : :
1811 : : /* ----
1812 : : * Make sure the snapshot had been stored safely to disk, that's normally
1813 : : * cheap.
1814 : : * Note that we do not need PANIC here, nobody will be able to use the
1815 : : * slot without fsyncing, and saving it won't succeed without an fsync()
1816 : : * either...
1817 : : * ----
1818 : : */
4606 rhaas@postgresql.org 1819 :CBC 11 : fsync_fname(path, false);
773 michael@paquier.xyz 1820 : 11 : fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);
1821 : :
1822 : : /* read statically sized portion of snapshot */
748 peter@eisentraut.org 1823 : 11 : SnapBuildRestoreContents(fd, ondisk, SnapBuildOnDiskConstantSize, path);
1824 : :
728 msawada@postgresql.o 1825 [ - + ]: 11 : if (ondisk->magic != SNAPBUILD_MAGIC)
4606 rhaas@postgresql.org 1826 [ # # ]:UBC 0 : ereport(ERROR,
1827 : : (errcode(ERRCODE_DATA_CORRUPTED),
1828 : : errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
1829 : : path, ondisk->magic, SNAPBUILD_MAGIC)));
1830 : :
728 msawada@postgresql.o 1831 [ - + ]:CBC 11 : if (ondisk->version != SNAPBUILD_VERSION)
4606 rhaas@postgresql.org 1832 [ # # ]:UBC 0 : ereport(ERROR,
1833 : : (errcode(ERRCODE_DATA_CORRUPTED),
1834 : : errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
1835 : : path, ondisk->version, SNAPBUILD_VERSION)));
1836 : :
4360 heikki.linnakangas@i 1837 :CBC 11 : INIT_CRC32C(checksum);
1838 : 11 : COMP_CRC32C(checksum,
1839 : : ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
1840 : : SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
1841 : :
1842 : : /* read SnapBuild */
748 peter@eisentraut.org 1843 : 11 : SnapBuildRestoreContents(fd, &ondisk->builder, sizeof(SnapBuild), path);
728 msawada@postgresql.o 1844 : 11 : COMP_CRC32C(checksum, &ondisk->builder, sizeof(SnapBuild));
1845 : :
1846 : : /* restore committed xacts information */
1847 [ + + ]: 11 : if (ondisk->builder.committed.xcnt > 0)
1848 : : {
1849 : 5 : sz = sizeof(TransactionId) * ondisk->builder.committed.xcnt;
1850 : 5 : ondisk->builder.committed.xip = MemoryContextAllocZero(context, sz);
748 peter@eisentraut.org 1851 : 5 : SnapBuildRestoreContents(fd, ondisk->builder.committed.xip, sz, path);
728 msawada@postgresql.o 1852 : 5 : COMP_CRC32C(checksum, ondisk->builder.committed.xip, sz);
1853 : : }
1854 : :
1855 : : /* restore catalog modifying xacts information */
1856 [ + + ]: 11 : if (ondisk->builder.catchange.xcnt > 0)
1857 : : {
1858 : 5 : sz = sizeof(TransactionId) * ondisk->builder.catchange.xcnt;
1859 : 5 : ondisk->builder.catchange.xip = MemoryContextAllocZero(context, sz);
748 peter@eisentraut.org 1860 : 5 : SnapBuildRestoreContents(fd, ondisk->builder.catchange.xip, sz, path);
728 msawada@postgresql.o 1861 : 5 : COMP_CRC32C(checksum, ondisk->builder.catchange.xip, sz);
1862 : : }
1863 : :
2655 peter@eisentraut.org 1864 [ - + ]: 11 : if (CloseTransientFile(fd) != 0)
2774 michael@paquier.xyz 1865 [ # # ]:UBC 0 : ereport(ERROR,
1866 : : (errcode_for_file_access(),
1867 : : errmsg("could not close file \"%s\": %m", path)));
1868 : :
4352 andres@anarazel.de 1869 :CBC 11 : FIN_CRC32C(checksum);
1870 : :
1871 : : /* verify checksum of what we've read */
728 msawada@postgresql.o 1872 [ - + ]: 11 : if (!EQ_CRC32C(checksum, ondisk->checksum))
4606 rhaas@postgresql.org 1873 [ # # ]:UBC 0 : ereport(ERROR,
1874 : : (errcode(ERRCODE_DATA_CORRUPTED),
1875 : : errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
1876 : : path, checksum, ondisk->checksum)));
1877 : :
728 msawada@postgresql.o 1878 :CBC 11 : return true;
1879 : : }
1880 : :
1881 : : /*
1882 : : * Restore a snapshot into 'builder' if previously one has been stored at the
1883 : : * location indicated by 'lsn'. Returns true if successful, false otherwise.
1884 : : */
1885 : : static bool
1886 : 20 : SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
1887 : : {
1888 : : SnapBuildOnDisk ondisk;
1889 : :
1890 : : /* no point in loading a snapshot if we're already there */
1891 [ - + ]: 20 : if (builder->state == SNAPBUILD_CONSISTENT)
728 msawada@postgresql.o 1892 :UBC 0 : return false;
1893 : :
1894 : : /* validate and restore the snapshot to 'ondisk' */
580 msawada@postgresql.o 1895 [ + + ]:CBC 20 : if (!SnapBuildRestoreSnapshot(&ondisk, lsn, builder->context, true))
728 1896 : 11 : return false;
1897 : :
1898 : : /*
1899 : : * ok, we now have a sensible snapshot here, figure out if it has more
1900 : : * information than we have.
1901 : : */
1902 : :
1903 : : /*
1904 : : * We are only interested in consistent snapshots for now, comparing
1905 : : * whether one incomplete snapshot is more "advanced" seems to be
1906 : : * unnecessarily complex.
1907 : : */
4606 rhaas@postgresql.org 1908 [ - + ]: 9 : if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
4606 rhaas@postgresql.org 1909 :UBC 0 : goto snapshot_not_interesting;
1910 : :
1911 : : /*
1912 : : * Don't use a snapshot that requires an xmin that we cannot guarantee to
1913 : : * be available.
1914 : : */
4606 rhaas@postgresql.org 1915 [ - + ]:CBC 9 : if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
4606 rhaas@postgresql.org 1916 :UBC 0 : goto snapshot_not_interesting;
1917 : :
1918 : : /*
1919 : : * Consistent snapshots have no next phase. Reset next_phase_at as it is
1920 : : * possible that an old value may remain.
1921 : : */
2065 andres@anarazel.de 1922 [ - + ]:CBC 9 : Assert(ondisk.builder.next_phase_at == InvalidTransactionId);
1196 dgustafsson@postgres 1923 : 9 : builder->next_phase_at = InvalidTransactionId;
1924 : :
1925 : : /* ok, we think the snapshot is sensible, copy over everything important */
4606 rhaas@postgresql.org 1926 : 9 : builder->xmin = ondisk.builder.xmin;
1927 : 9 : builder->xmax = ondisk.builder.xmax;
1928 : 9 : builder->state = ondisk.builder.state;
1929 : :
1930 : 9 : builder->committed.xcnt = ondisk.builder.committed.xcnt;
1931 : : /* We only allocated/stored xcnt, not xcnt_space xids ! */
1932 : : /* don't overwrite preallocated xip, if we don't have anything here */
1933 [ + + ]: 9 : if (builder->committed.xcnt > 0)
1934 : : {
1935 : 3 : pfree(builder->committed.xip);
1936 : 3 : builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
1937 : 3 : builder->committed.xip = ondisk.builder.committed.xip;
1938 : : }
1939 : 9 : ondisk.builder.committed.xip = NULL;
1940 : :
1941 : : /* set catalog modifying transactions */
1523 akapila@postgresql.o 1942 [ - + ]: 9 : if (builder->catchange.xip)
1523 akapila@postgresql.o 1943 :UBC 0 : pfree(builder->catchange.xip);
1523 akapila@postgresql.o 1944 :CBC 9 : builder->catchange.xcnt = ondisk.builder.catchange.xcnt;
1945 : 9 : builder->catchange.xip = ondisk.builder.catchange.xip;
1946 : 9 : ondisk.builder.catchange.xip = NULL;
1947 : :
1948 : : /* our snapshot is not interesting anymore, build a new one */
4606 rhaas@postgresql.org 1949 [ - + ]: 9 : if (builder->snapshot != NULL)
1950 : : {
4606 rhaas@postgresql.org 1951 :UBC 0 : SnapBuildSnapDecRefcount(builder->snapshot);
1952 : : }
3407 andres@anarazel.de 1953 :CBC 9 : builder->snapshot = SnapBuildBuildSnapshot(builder);
4606 rhaas@postgresql.org 1954 : 9 : SnapBuildSnapIncRefcount(builder->snapshot);
1955 : :
1956 : 9 : ReorderBufferSetRestartPoint(builder->reorder, lsn);
1957 : :
1958 [ - + ]: 9 : Assert(builder->state == SNAPBUILD_CONSISTENT);
1959 : :
185 fujii@postgresql.org 1960 [ + + + + ]: 9 : ereport(LogicalDecodingLogLevel(),
1961 : : errmsg("logical decoding found consistent point at %X/%08X",
1962 : : LSN_FORMAT_ARGS(lsn)),
1963 : : errdetail("Logical decoding will begin using saved snapshot."));
4606 rhaas@postgresql.org 1964 : 9 : return true;
1965 : :
4606 rhaas@postgresql.org 1966 :UBC 0 : snapshot_not_interesting:
1967 [ # # ]: 0 : if (ondisk.builder.committed.xip != NULL)
1968 : 0 : pfree(ondisk.builder.committed.xip);
1523 akapila@postgresql.o 1969 [ # # ]: 0 : if (ondisk.builder.catchange.xip != NULL)
1970 : 0 : pfree(ondisk.builder.catchange.xip);
4606 rhaas@postgresql.org 1971 : 0 : return false;
1972 : : }
1973 : :
1974 : : /*
1975 : : * Read the contents of the serialized snapshot to 'dest'.
1976 : : */
1977 : : static void
748 peter@eisentraut.org 1978 :CBC 32 : SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path)
1979 : : {
1980 : : ssize_t readBytes;
1981 : :
1523 akapila@postgresql.o 1982 : 32 : pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
1983 : 32 : readBytes = read(fd, dest, size);
1984 : 32 : pgstat_report_wait_end();
1985 [ - + ]: 32 : if (readBytes != size)
1986 : : {
1523 akapila@postgresql.o 1987 :UBC 0 : int save_errno = errno;
1988 : :
1989 : 0 : CloseTransientFile(fd);
1990 : :
1991 [ # # ]: 0 : if (readBytes < 0)
1992 : : {
1993 : 0 : errno = save_errno;
1994 [ # # ]: 0 : ereport(ERROR,
1995 : : (errcode_for_file_access(),
1996 : : errmsg("could not read file \"%s\": %m", path)));
1997 : : }
1998 : : else
1999 [ # # ]: 0 : ereport(ERROR,
2000 : : (errcode(ERRCODE_DATA_CORRUPTED),
2001 : : errmsg("could not read file \"%s\": read %zd of %zu",
2002 : : path, readBytes, size)));
2003 : : }
1523 akapila@postgresql.o 2004 :CBC 32 : }
2005 : :
2006 : : /*
2007 : : * Remove all serialized snapshots that are not required anymore because no
2008 : : * slot can need them. This doesn't actually have to run during a checkpoint,
2009 : : * but it's a convenient point to schedule this.
2010 : : *
2011 : : * NB: We run this during checkpoints even if logical decoding is disabled so
2012 : : * we cleanup old slots at some point after it got disabled.
2013 : : */
2014 : : void
4606 rhaas@postgresql.org 2015 : 1998 : CheckPointSnapBuild(void)
2016 : : {
2017 : : XLogRecPtr cutoff;
2018 : : XLogRecPtr redo;
2019 : : DIR *snap_dir;
2020 : : struct dirent *snap_de;
2021 : : char path[MAXPGPATH + sizeof(PG_LOGICAL_SNAPSHOTS_DIR)];
2022 : :
2023 : : /*
2024 : : * We start off with a minimum of the last redo pointer. No new
2025 : : * replication slot will start before that, so that's a safe upper bound
2026 : : * for removal.
2027 : : */
2028 : 1998 : redo = GetRedoRecPtr();
2029 : :
2030 : : /* now check for the restart ptrs from existing slots */
2031 : 1998 : cutoff = ReplicationSlotsComputeLogicalRestartLSN();
2032 : :
2033 : : /* don't start earlier than the restart lsn */
2034 [ + + ]: 1998 : if (redo < cutoff)
2035 : 1 : cutoff = redo;
2036 : :
773 michael@paquier.xyz 2037 : 1998 : snap_dir = AllocateDir(PG_LOGICAL_SNAPSHOTS_DIR);
2038 [ + + ]: 6303 : while ((snap_de = ReadDir(snap_dir, PG_LOGICAL_SNAPSHOTS_DIR)) != NULL)
2039 : : {
2040 : : uint32 hi;
2041 : : uint32 lo;
2042 : : XLogRecPtr lsn;
2043 : : PGFileType de_type;
2044 : :
4606 rhaas@postgresql.org 2045 [ + + ]: 4305 : if (strcmp(snap_de->d_name, ".") == 0 ||
2046 [ + + ]: 2307 : strcmp(snap_de->d_name, "..") == 0)
2047 : 3996 : continue;
2048 : :
773 michael@paquier.xyz 2049 : 309 : snprintf(path, sizeof(path), "%s/%s", PG_LOGICAL_SNAPSHOTS_DIR, snap_de->d_name);
1501 2050 : 309 : de_type = get_dirent_type(path, snap_de, false, DEBUG1);
2051 : :
2052 [ + - - + ]: 309 : if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)
2053 : : {
4606 rhaas@postgresql.org 2054 [ # # ]:UBC 0 : elog(DEBUG1, "only regular files expected: %s", path);
2055 : 0 : continue;
2056 : : }
2057 : :
2058 : : /*
2059 : : * temporary filenames from SnapBuildSerialize() include the LSN and
2060 : : * everything but are postfixed by .$pid.tmp. We can just remove them
2061 : : * the same as other files because there can be none that are
2062 : : * currently being written that are older than cutoff.
2063 : : *
2064 : : * We just log a message if a file doesn't fit the pattern, it's
2065 : : * probably some editors lock/state file or similar...
2066 : : */
4606 rhaas@postgresql.org 2067 [ - + ]:CBC 309 : if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
2068 : : {
4606 rhaas@postgresql.org 2069 [ # # ]:UBC 0 : ereport(LOG,
2070 : : (errmsg("could not parse file name \"%s\"", path)));
2071 : 0 : continue;
2072 : : }
2073 : :
4606 rhaas@postgresql.org 2074 :CBC 309 : lsn = ((uint64) hi) << 32 | lo;
2075 : :
2076 : : /* check whether we still need it */
340 alvherre@kurilemu.de 2077 [ + + + + ]: 309 : if (lsn < cutoff || !XLogRecPtrIsValid(cutoff))
2078 : : {
4606 rhaas@postgresql.org 2079 [ + + ]: 202 : elog(DEBUG1, "removing snapbuild snapshot %s", path);
2080 : :
2081 : : /*
2082 : : * It's not particularly harmful, though strange, if we can't
2083 : : * remove the file here. Don't prevent the checkpoint from
2084 : : * completing, that'd be a cure worse than the disease.
2085 : : */
2086 [ - + ]: 202 : if (unlink(path) < 0)
2087 : : {
4606 rhaas@postgresql.org 2088 [ # # ]:UBC 0 : ereport(LOG,
2089 : : (errcode_for_file_access(),
2090 : : errmsg("could not remove file \"%s\": %m",
2091 : : path)));
2092 : 0 : continue;
2093 : : }
2094 : : }
2095 : : }
4606 rhaas@postgresql.org 2096 :CBC 1998 : FreeDir(snap_dir);
2097 : 1998 : }
2098 : :
2099 : : /*
2100 : : * Check if a logical snapshot at the specified point has been serialized.
2101 : : */
2102 : : bool
922 akapila@postgresql.o 2103 : 15 : SnapBuildSnapshotExists(XLogRecPtr lsn)
2104 : : {
2105 : : char path[MAXPGPATH];
2106 : : int ret;
2107 : : struct stat stat_buf;
2108 : :
424 fujii@postgresql.org 2109 : 15 : sprintf(path, "%s/%X-%X.snap",
2110 : : PG_LOGICAL_SNAPSHOTS_DIR,
922 akapila@postgresql.o 2111 : 15 : LSN_FORMAT_ARGS(lsn));
2112 : :
2113 : 15 : ret = stat(path, &stat_buf);
2114 : :
2115 [ + + - + ]: 15 : if (ret != 0 && errno != ENOENT)
922 akapila@postgresql.o 2116 [ # # ]:UBC 0 : ereport(ERROR,
2117 : : (errcode_for_file_access(),
2118 : : errmsg("could not stat file \"%s\": %m", path)));
2119 : :
922 akapila@postgresql.o 2120 :CBC 15 : return ret == 0;
2121 : : }
|
