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