Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * repack_worker.c
4 : : * Implementation of the background worker for ad-hoc logical decoding
5 : : * during REPACK (CONCURRENTLY).
6 : : *
7 : : *
8 : : * Copyright (c) 2026, PostgreSQL Global Development Group
9 : : *
10 : : *
11 : : * IDENTIFICATION
12 : : * src/backend/commands/repack_worker.c
13 : : *
14 : : *-------------------------------------------------------------------------
15 : : */
16 : : #include "postgres.h"
17 : :
18 : : #include "access/table.h"
19 : : #include "access/xlog_internal.h"
20 : : #include "access/xlogutils.h"
21 : : #include "access/xlogwait.h"
22 : : #include "commands/progress.h"
23 : : #include "commands/repack.h"
24 : : #include "commands/repack_internal.h"
25 : : #include "libpq/libpq.h"
26 : : #include "libpq/pqmq.h"
27 : : #include "replication/snapbuild.h"
28 : : #include "storage/ipc.h"
29 : : #include "storage/proc.h"
30 : : #include "tcop/tcopprot.h"
31 : : #include "utils/guc.h"
32 : : #include "utils/memutils.h"
33 : :
34 : : #define PGREPACK_PLUGIN "pgrepack"
35 : :
36 : : static void RepackWorkerShutdown(int code, Datum arg);
37 : : static void set_repack_worker_setup_phase(DecodingWorkerShared *shared,
38 : : int setup_phase);
39 : : static LogicalDecodingContext *repack_setup_logical_decoding(Oid relid,
40 : : DecodingWorkerShared *shared);
41 : : static void repack_cleanup_logical_decoding(LogicalDecodingContext *ctx);
42 : : static void export_initial_snapshot(Snapshot snapshot,
43 : : DecodingWorkerShared *shared);
44 : : static bool decode_concurrent_changes(LogicalDecodingContext *ctx,
45 : : DecodingWorkerShared *shared);
46 : :
47 : : /* Is this process a REPACK worker? */
48 : : static bool am_repack_worker = false;
49 : :
50 : : /* The WAL segment being decoded. */
51 : : static XLogSegNo repack_current_segment = 0;
52 : :
53 : : /*
54 : : * Keep track of the table we're processing, to skip logical decoding of data
55 : : * from other relations.
56 : : */
57 : : static RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid};
58 : : static RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid};
59 : :
60 : :
61 : : /* REPACK decoding worker entry point */
62 : : void
213 alvherre@kurilemu.de 63 :CBC 13 : RepackWorkerMain(Datum main_arg)
64 : : {
65 : : dsm_segment *seg;
66 : : DecodingWorkerShared *shared;
67 : : shm_mq *mq;
68 : : shm_mq_handle *mqh;
69 : : LogicalDecodingContext *decoding_ctx;
70 : : SharedFileSet *sfs;
71 : : Snapshot snapshot;
72 : : char buf[32];
73 : :
74 : 13 : am_repack_worker = true;
75 : :
76 : 13 : BackgroundWorkerUnblockSignals();
77 : :
78 : 13 : seg = dsm_attach(DatumGetUInt32(main_arg));
79 [ - + ]: 13 : if (seg == NULL)
213 alvherre@kurilemu.de 80 [ # # ]:UBC 0 : ereport(ERROR,
81 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
82 : : errmsg("could not map dynamic shared memory segment"));
83 : :
213 alvherre@kurilemu.de 84 :CBC 13 : shared = (DecodingWorkerShared *) dsm_segment_address(seg);
85 : :
86 : : /* Arrange to signal the steering process if we exit. */
49 87 : 13 : before_shmem_exit(RepackWorkerShutdown, PointerGetDatum(seg));
88 : :
89 : : /*
90 : : * Join locking group - see the comments around the call of
91 : : * start_repack_decoding_worker().
92 : : */
93 [ - + ]: 13 : if (!BecomeLockGroupMember(GetPGProcByNumber(shared->backend_proc_number),
94 : : shared->backend_pid))
213 alvherre@kurilemu.de 95 :UBC 0 : return; /* The leader is not running anymore. */
96 : :
97 : : /*
98 : : * Setup a queue to send error messages to the backend that launched this
99 : : * worker.
100 : : */
213 alvherre@kurilemu.de 101 :CBC 13 : mq = (shm_mq *) (char *) BUFFERALIGN(shared->error_queue);
102 : 13 : shm_mq_set_sender(mq, MyProc);
103 : 13 : mqh = shm_mq_attach(mq, seg, NULL);
104 : 13 : pq_redirect_to_shm_mq(seg, mqh);
105 : 13 : pq_set_parallel_leader(shared->backend_pid,
106 : : shared->backend_proc_number);
107 : :
108 : : /*
109 : : * Connect to the database, skipping the connection authorization checks
110 : : * as parallel workers do. Note that we run as the owner of the table
111 : : * being repacked, who need not be able to log in or connect; the leader
112 : : * checked the invoking user's privileges before starting us.
113 : : */
199 114 : 13 : BackgroundWorkerInitializeConnectionByOid(shared->dbid, shared->roleid,
115 : : BGWORKER_BYPASS_ALLOWCONN |
116 : : BGWORKER_BYPASS_ROLELOGINCHECK);
117 : :
118 : : /* Adopt the steering backend's relevant timeouts. */
49 119 : 13 : snprintf(buf, sizeof(buf), "%d", shared->lock_timeout);
120 : 13 : SetConfigOption("lock_timeout", buf, PGC_SUSET, PGC_S_OVERRIDE);
121 : 13 : snprintf(buf, sizeof(buf), "%d", shared->transaction_timeout);
122 : 13 : SetConfigOption("transaction_timeout", buf, PGC_SUSET, PGC_S_OVERRIDE);
123 : :
124 : : /*
125 : : * Transaction is needed to open relation, and it also provides us with a
126 : : * resource owner.
127 : : */
213 128 : 13 : StartTransactionCommand();
129 : :
130 : 13 : shared = (DecodingWorkerShared *) dsm_segment_address(seg);
131 : :
132 : : /*
133 : : * Not sure the spinlock is needed here - the backend should not change
134 : : * anything in the shared memory until we have serialized the snapshot.
135 : : */
136 : 13 : SpinLockAcquire(&shared->mutex);
137 [ - + ]: 13 : Assert(!XLogRecPtrIsValid(shared->lsn_upto));
138 : 13 : sfs = &shared->sfs;
139 : 13 : SpinLockRelease(&shared->mutex);
140 : :
141 : 13 : SharedFileSetAttach(sfs, seg);
142 : :
143 : : /*
144 : : * Prepare to capture the concurrent data changes ourselves.
145 : : */
3 msawada@postgresql.o 146 :GNC 13 : decoding_ctx = repack_setup_logical_decoding(shared->relid, shared);
147 : :
148 : : /* Announce that we're ready. */
213 alvherre@kurilemu.de 149 :CBC 13 : SpinLockAcquire(&shared->mutex);
150 : 13 : shared->initialized = true;
151 : 13 : SpinLockRelease(&shared->mutex);
152 : 13 : ConditionVariableSignal(&shared->cv);
153 : :
154 : : /* There doesn't seem to a nice API to set these */
155 : 13 : XactIsoLevel = XACT_REPEATABLE_READ;
156 : 13 : XactReadOnly = true;
157 : :
158 : : /* Build the initial snapshot and export it. */
159 : 13 : snapshot = SnapBuildInitialSnapshot(decoding_ctx->snapshot_builder);
160 : 13 : export_initial_snapshot(snapshot, shared);
161 : :
162 : : /*
163 : : * Only historic snapshots should be used now. Do not let us restrict the
164 : : * progress of xmin horizon.
165 : : */
166 : 13 : InvalidateCatalogSnapshot();
167 : :
168 : : for (;;)
169 : 13 : {
170 : 26 : bool stop = decode_concurrent_changes(decoding_ctx, shared);
171 : :
172 [ + + ]: 26 : if (stop)
173 : 13 : break;
174 : : }
175 : :
176 : : /* Clean up and report termination to our steering process */
177 : 13 : repack_cleanup_logical_decoding(decoding_ctx);
178 : 13 : CommitTransactionCommand();
49 179 : 13 : pq_putmessage(PqRepackMsg_Terminate, NULL, 0);
180 : : }
181 : :
182 : : /*
183 : : * Make sure the repack steering process tries to read from our error queue one
184 : : * more time. This guards against the case where we exit uncleanly without
185 : : * sending an ErrorResponse to the leader, for example because some code calls
186 : : * proc_exit directly.
187 : : */
188 : : static void
213 189 : 13 : RepackWorkerShutdown(int code, Datum arg)
190 : : {
191 : : dsm_segment *seg;
192 : : DecodingWorkerShared *shared;
193 : : pid_t pid;
194 : : ProcNumber procno;
195 : :
49 196 : 13 : seg = (dsm_segment *) DatumGetPointer(arg);
197 : 13 : shared = (DecodingWorkerShared *) dsm_segment_address(seg);
198 : 13 : pid = shared->backend_pid;
199 : 13 : procno = shared->backend_proc_number;
200 : :
201 : : /*
202 : : * Detach from the shared memory segment before we signal the backend.
203 : : * Detaching also detaches the error message queue, and the backend learns
204 : : * that we are gone by reading that queue when it handles our signal. If
205 : : * we signaled first, the backend could read the queue while it still
206 : : * looks attached, and nothing would make it read again.
207 : : */
208 : 13 : dsm_detach(seg);
209 : :
210 : 13 : SendProcSignal(pid, PROCSIG_REPACK_MESSAGE, procno);
213 211 : 13 : }
212 : :
213 : : bool
214 : 2101 : AmRepackWorker(void)
215 : : {
216 : 2101 : return am_repack_worker;
217 : : }
218 : :
219 : : /*
220 : : * Tell the backend running REPACK which progress phase to report while we
221 : : * initialize the decoding.
222 : : */
223 : : static void
3 msawada@postgresql.o 224 :GNC 26 : set_repack_worker_setup_phase(DecodingWorkerShared *shared, int setup_phase)
225 : : {
226 : 26 : SpinLockAcquire(&shared->mutex);
227 : 26 : shared->setup_phase = setup_phase;
228 : 26 : SpinLockRelease(&shared->mutex);
229 : 26 : ConditionVariableSignal(&shared->cv);
230 : 26 : }
231 : :
232 : : /*
233 : : * This function is much like pg_create_logical_replication_slot() except that
234 : : * the new slot is neither released (if anyone else could read changes from
235 : : * our slot, we could miss changes other backends do while we copy the
236 : : * existing data into temporary table), nor persisted (it's easier to handle
237 : : * crash by restarting all the work from scratch).
238 : : */
239 : : static LogicalDecodingContext *
240 : 13 : repack_setup_logical_decoding(Oid relid, DecodingWorkerShared *shared)
241 : : {
242 : : Relation rel;
243 : : Oid toastrelid;
244 : : LogicalDecodingContext *ctx;
245 : : char slotname[NAMEDATALEN];
246 : : RepackDecodingState *dstate;
247 : : MemoryContext oldcxt;
248 : :
249 : : /*
250 : : * REPACK CONCURRENTLY is not allowed in a transaction block, so this
251 : : * should never fire.
252 : : */
213 alvherre@kurilemu.de 253 [ - + ]:CBC 13 : Assert(!TransactionIdIsValid(GetTopTransactionIdIfAny()));
254 : :
255 : : /* Make sure we can use logical decoding */
212 256 : 13 : CheckLogicalDecodingRequirements(true);
257 : :
258 : : /*
259 : : * Create the replication slot we'll use, and enable logical decoding in
260 : : * case it isn't already on.
261 : : *
262 : : * Make the slot RS_TEMPORARY so that it's removed on ERROR. A backend
263 : : * cannot execute multiple REPACK commands at a time, so the PID is enough
264 : : * to make the slot name unique.
265 : : */
149 266 : 13 : snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid);
267 : 13 : ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true,
268 : : false, false);
3 msawada@postgresql.o 269 :GNC 13 : set_repack_worker_setup_phase(shared,
270 : : PROGRESS_REPACK_PHASE_ENABLE_LOGICAL_DECODING);
213 alvherre@kurilemu.de 271 :CBC 13 : EnsureLogicalDecodingEnabled();
272 : :
273 : : /*
274 : : * Set up repacked_rel_locator and repacked_rel_toast_locator, which we
275 : : * use to skip decoding of unrelated relations. The steering backend
276 : : * already holds ShareUpdateExclusiveLock on both these tables, so we
277 : : * don't need any locks here, other than to avoid relation_open()'s
278 : : * assertion that we hold one.
279 : : */
280 : 13 : rel = table_open(relid, AccessShareLock);
281 : 13 : repacked_rel_locator = rel->rd_locator;
282 : 13 : toastrelid = rel->rd_rel->reltoastrelid;
283 [ + + ]: 13 : if (OidIsValid(toastrelid))
284 : : {
285 : : Relation toastrel;
286 : :
287 : : /* Avoid logical decoding of other TOAST relations. */
288 : 8 : toastrel = table_open(toastrelid, AccessShareLock);
289 : 8 : repacked_rel_toast_locator = toastrel->rd_locator;
290 : 8 : table_close(toastrel, AccessShareLock);
291 : : }
292 : 13 : table_close(rel, AccessShareLock);
293 : :
3 msawada@postgresql.o 294 :GNC 13 : set_repack_worker_setup_phase(shared,
295 : : PROGRESS_REPACK_PHASE_INIT_LOGICAL_DECODING);
296 : :
297 : : /*
298 : : * Set up our logical decoding context. We initially use the blocking
299 : : * read_local_xlog_page until we find the start point, and switch to the
300 : : * non-blocking interface afterwards.
301 : : */
149 alvherre@kurilemu.de 302 :CBC 13 : ctx = CreateInitDecodingContext(PGREPACK_PLUGIN,
303 : : NIL,
304 : : true,
305 : : true,
306 : : InvalidXLogRecPtr,
307 : 13 : XL_ROUTINE(.page_read = read_local_xlog_page,
308 : : .segment_open = wal_segment_open,
309 : : .segment_close = wal_segment_close),
310 : : NULL, NULL, NULL);
311 : :
312 : : /* Complete setup of output_writer_private */
313 : 13 : dstate = (RepackDecodingState *) ctx->output_writer_private;
314 : 13 : dstate->relid = relid;
315 : 13 : dstate->worker_cxt = CurrentMemoryContext;
316 : 13 : dstate->worker_resowner = CurrentResourceOwner;
317 : :
318 : : /* We don't have control on fast_forward, but verify it's sane */
319 [ - + ]: 13 : Assert(!ctx->fast_forward);
320 : :
321 : : /* Find our decoding starting point. */
322 : 13 : DecodingContextFindStartpoint(ctx);
323 : :
324 : : /* From this point on, we need non-blocking WAL reads */
325 : 13 : ctx->reader->routine.page_read = read_local_xlog_page_no_wait;
326 : :
327 : : /*
328 : : * Initialize repack_current_segment so that we can notice WAL segment
329 : : * boundaries.
330 : : */
213 331 : 13 : XLByteToSeg(ctx->reader->EndRecPtr, repack_current_segment,
332 : : wal_segment_size);
333 : :
334 : : /*
335 : : * Set up our reader private state to let the page-read callback notify
336 : : * when end-of-WAL has been reached. This lives in the same context as
337 : : * the logical decoding itself.
338 : : */
149 339 : 13 : oldcxt = MemoryContextSwitchTo(ctx->context);
213 340 : 13 : ctx->reader->private_data = palloc0_object(ReadLocalXLogPageNoWaitPrivate);
341 : 13 : MemoryContextSwitchTo(oldcxt);
342 : :
343 : 13 : return ctx;
344 : : }
345 : :
346 : : static void
347 : 13 : repack_cleanup_logical_decoding(LogicalDecodingContext *ctx)
348 : : {
349 : : RepackDecodingState *dstate;
350 : :
351 : 13 : dstate = (RepackDecodingState *) ctx->output_writer_private;
352 [ + + ]: 13 : if (dstate->slot)
353 : 4 : ExecDropSingleTupleTableSlot(dstate->slot);
354 : :
355 : 13 : FreeDecodingContext(ctx);
162 356 : 13 : ReplicationSlotDropAcquired(true);
213 357 : 13 : }
358 : :
359 : : /*
360 : : * Make snapshot available to the backend that launched the decoding worker.
361 : : */
362 : : static void
363 : 13 : export_initial_snapshot(Snapshot snapshot, DecodingWorkerShared *shared)
364 : : {
365 : : char fname[MAXPGPATH];
366 : : BufFile *file;
367 : : Size snap_size;
368 : : char *snap_space;
369 : :
370 : 13 : snap_size = EstimateSnapshotSpace(snapshot);
371 : 13 : snap_space = (char *) palloc(snap_size);
372 : 13 : SerializeSnapshot(snapshot, snap_space);
373 : :
374 : 13 : DecodingWorkerFileName(fname, shared->relid, shared->last_exported + 1);
375 : 13 : file = BufFileCreateFileSet(&shared->sfs.fs, fname);
376 : : /* To make restoration easier, write the snapshot size first. */
377 : 13 : BufFileWrite(file, &snap_size, sizeof(snap_size));
378 : 13 : BufFileWrite(file, snap_space, snap_size);
379 : 13 : BufFileClose(file);
212 380 : 13 : pfree(snap_space);
381 : :
382 : : /* Increase the counter to tell the backend that the file is available. */
213 383 : 13 : SpinLockAcquire(&shared->mutex);
384 : 13 : shared->last_exported++;
385 : 13 : SpinLockRelease(&shared->mutex);
386 : 13 : ConditionVariableSignal(&shared->cv);
387 : 13 : }
388 : :
389 : : /*
390 : : * Decode logical changes from the WAL sequence and store them to a file.
391 : : *
392 : : * If true is returned, there is no more work for the worker.
393 : : */
394 : : static bool
395 : 26 : decode_concurrent_changes(LogicalDecodingContext *ctx,
396 : : DecodingWorkerShared *shared)
397 : : {
398 : : RepackDecodingState *dstate;
399 : : XLogRecPtr lsn_upto;
400 : : bool done;
401 : : char fname[MAXPGPATH];
402 : :
403 : 26 : dstate = (RepackDecodingState *) ctx->output_writer_private;
404 : :
405 : : /* Open the output file. */
406 : 26 : DecodingWorkerFileName(fname, shared->relid, shared->last_exported + 1);
407 : 26 : dstate->file = BufFileCreateFileSet(&shared->sfs.fs, fname);
408 : :
409 : 26 : SpinLockAcquire(&shared->mutex);
410 : 26 : lsn_upto = shared->lsn_upto;
411 : 26 : done = shared->done;
412 : 26 : SpinLockRelease(&shared->mutex);
413 : :
414 : : while (true)
415 : 3889 : {
416 : : XLogRecord *record;
417 : : XLogSegNo segno_new;
418 : 3915 : char *errm = NULL;
419 : : XLogRecPtr end_lsn;
420 : :
421 [ - + ]: 3915 : CHECK_FOR_INTERRUPTS();
422 : :
423 : 3915 : record = XLogReadRecord(ctx->reader, &errm);
424 [ + + ]: 3915 : if (record)
425 : : {
426 : 2967 : LogicalDecodingProcessRecord(ctx, ctx->reader);
427 : :
428 : : /*
429 : : * We want to allow WAL to be recycled while REPACK is running.
430 : : *
431 : : * In normal usage of a replication slot, we need to be very
432 : : * careful not to advance the LSN until it's been confirmed as
433 : : * received by the remote. In REPACK's case, this is not needed:
434 : : * REPACK will never try to replay the same WAL after a crash, and
435 : : * if there _is_ a crash, the whole REPACK has to be started from
436 : : * scratch anyway.
437 : : *
438 : : * So here we disregard the careful LSN tracking and just move the
439 : : * LSN locations forward to what we've processed. Note that it
440 : : * would be bogus to move the xmin forward, though, so we don't
441 : : * touch that.
442 : : *
443 : : * This can be done on whatever schedule is convenient, but in
444 : : * order not to cause unnecessary load, we only do it as we cross
445 : : * each WAL segment boundary.
446 : : */
447 : 2967 : end_lsn = ctx->reader->EndRecPtr;
448 : 2967 : XLByteToSeg(end_lsn, segno_new, wal_segment_size);
449 [ - + ]: 2967 : if (segno_new != repack_current_segment)
450 : : {
159 alvherre@kurilemu.de 451 :UBC 0 : LogicalIncreaseRestartDecodingForSlot(end_lsn, end_lsn);
213 452 : 0 : LogicalConfirmReceivedLocation(end_lsn);
108 453 [ # # ]: 0 : elog(DEBUG1, "REPACK: confirmed receive location %X/%08X",
454 : : LSN_FORMAT_ARGS(end_lsn));
213 455 : 0 : repack_current_segment = segno_new;
456 : : }
457 : : }
458 : : else
459 : : {
460 : : ReadLocalXLogPageNoWaitPrivate *priv;
461 : :
213 alvherre@kurilemu.de 462 [ - + ]:CBC 948 : if (errm)
213 alvherre@kurilemu.de 463 [ # # ]:UBC 0 : ereport(ERROR,
464 : : errcode_for_file_access(),
465 : : errmsg("could not read WAL from timeline %u at %X/%08X: %s",
466 : : ctx->reader->currTLI,
467 : : LSN_FORMAT_ARGS(ctx->reader->EndRecPtr),
468 : : errm));
469 : :
470 : : /*
471 : : * In the decoding loop we do not want to get blocked when there
472 : : * is no more WAL available, otherwise the loop would become
473 : : * uninterruptible.
474 : : */
213 alvherre@kurilemu.de 475 :CBC 948 : priv = (ReadLocalXLogPageNoWaitPrivate *) ctx->reader->private_data;
476 [ + - ]: 948 : if (priv->end_of_wal)
477 : : /* Do not miss the end of WAL condition next time. */
478 : 948 : priv->end_of_wal = false;
479 : : else
213 alvherre@kurilemu.de 480 [ # # ]:UBC 0 : ereport(ERROR,
481 : : errcode(ERRCODE_DATA_CORRUPTED),
482 : : errmsg("could not read WAL record"));
483 : : }
484 : :
485 : : /*
486 : : * Whether we could read new record or not, keep checking if
487 : : * 'lsn_upto' was specified.
488 : : */
213 alvherre@kurilemu.de 489 [ + + ]:CBC 3915 : if (!XLogRecPtrIsValid(lsn_upto))
490 : : {
491 : 3650 : SpinLockAcquire(&shared->mutex);
492 : 3650 : lsn_upto = shared->lsn_upto;
493 : : /* 'done' should be set at the same time as 'lsn_upto' */
494 : 3650 : done = shared->done;
495 : 3650 : SpinLockRelease(&shared->mutex);
496 : : }
497 [ + + ]: 3915 : if (XLogRecPtrIsValid(lsn_upto) &&
498 [ + + ]: 290 : ctx->reader->EndRecPtr >= lsn_upto)
499 : 26 : break;
500 : :
501 [ + + ]: 3889 : if (record == NULL)
502 : : {
52 akorotkov@postgresql 503 : 934 : int timeout = 0;
504 : : WaitLSNResult res;
505 : :
506 : : /*
507 : : * Before we retry reading, wait until new WAL is flushed.
508 : : *
509 : : * There is a race condition such that the backend executing
510 : : * REPACK determines 'lsn_upto', but before it sets the shared
511 : : * variable, we reach the end of WAL. In that case we'd need to
512 : : * wait until the next WAL flush (unrelated to REPACK). Although
513 : : * that should not be a problem in a busy system, it might be
514 : : * noticeable in other cases, including regression tests (which
515 : : * are not necessarily executed in parallel). Therefore it makes
516 : : * sense to use timeout.
517 : : *
518 : : * If lsn_upto is valid, WAL records having LSN lower than that
519 : : * should already have been flushed to disk.
520 : : */
213 alvherre@kurilemu.de 521 [ + - ]: 934 : if (!XLogRecPtrIsValid(lsn_upto))
52 akorotkov@postgresql 522 : 934 : timeout = 100;
213 alvherre@kurilemu.de 523 : 934 : res = WaitForLSN(WAIT_LSN_TYPE_PRIMARY_FLUSH,
524 : 934 : ctx->reader->EndRecPtr + 1,
525 : : timeout);
526 [ + + - + ]: 934 : if (res != WAIT_LSN_RESULT_SUCCESS &&
527 : : res != WAIT_LSN_RESULT_TIMEOUT)
213 alvherre@kurilemu.de 528 [ # # ]:UBC 0 : ereport(ERROR,
529 : : errcode(ERRCODE_INTERNAL_ERROR),
530 : : errmsg("waiting for WAL failed"));
531 : : }
532 : : }
533 : :
534 : : /*
535 : : * Close the file so we can make it available to the backend.
536 : : */
213 alvherre@kurilemu.de 537 :CBC 26 : BufFileClose(dstate->file);
538 : 26 : dstate->file = NULL;
539 : 26 : SpinLockAcquire(&shared->mutex);
540 : 26 : shared->lsn_upto = InvalidXLogRecPtr;
541 : 26 : shared->last_exported++;
542 : 26 : SpinLockRelease(&shared->mutex);
543 : 26 : ConditionVariableSignal(&shared->cv);
544 : :
545 : 26 : return done;
546 : : }
547 : :
548 : : /*
549 : : * Does the WAL record contain a data change that this backend does not need
550 : : * to decode on behalf of REPACK (CONCURRENTLY)?
551 : : */
552 : : bool
553 : 1543910 : change_useless_for_repack(XLogRecordBuffer *buf)
554 : : {
555 : 1543910 : XLogReaderState *r = buf->record;
556 : : RelFileLocator locator;
557 : :
558 : : /* TOAST locator should not be set unless the main is. */
559 [ + + - + ]: 1543910 : Assert(!OidIsValid(repacked_rel_toast_locator.relNumber) ||
560 : : OidIsValid(repacked_rel_locator.relNumber));
561 : :
562 : : /*
563 : : * Backends not involved in REPACK (CONCURRENTLY) should not do the
564 : : * filtering.
565 : : */
566 [ + + ]: 1543910 : if (!OidIsValid(repacked_rel_locator.relNumber))
567 : 1543040 : return false;
568 : :
569 : : /*
570 : : * If the record does not contain the block 0, it's probably not INSERT /
571 : : * UPDATE / DELETE. In any case, we do not have enough information to
572 : : * filter the change out.
573 : : */
574 [ - + ]: 870 : if (!XLogRecGetBlockTagExtended(r, 0, &locator, NULL, NULL, NULL))
213 alvherre@kurilemu.de 575 :UBC 0 : return false;
576 : :
577 : : /*
578 : : * Decode the change if it belongs to the table we are repacking, or if it
579 : : * belongs to its TOAST relation.
580 : : */
213 alvherre@kurilemu.de 581 [ + + + - :CBC 870 : if (RelFileLocatorEquals(locator, repacked_rel_locator))
+ - ]
582 : 57 : return false;
583 [ + + ]: 813 : if (OidIsValid(repacked_rel_toast_locator.relNumber) &&
584 [ + + + - : 691 : RelFileLocatorEquals(locator, repacked_rel_toast_locator))
+ - ]
585 : 112 : return false;
586 : :
587 : : /* Filter out changes of other tables. */
588 : 701 : return true;
589 : : }
|
