LCOV - code coverage report
Current view: top level - src/backend/commands - repack_worker.c (source / functions) Coverage Total Hit
Test: PostgreSQL 20devel Lines: 93.9 % 164 154
Test Date: 2026-10-16 08:15:56 Functions: 100.0 % 8 8
Legend: Lines:     hit not hit
Branches: + taken - not taken # not executed
Branches: 62.9 % 62 39

             Branch data     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
      59                 :          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)
      76         [ #  # ]:           0 :         ereport(ERROR,
      77                 :             :                 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      78                 :             :                 errmsg("could not map dynamic shared memory segment"));
      79                 :             : 
      80                 :          13 :     shared = (DecodingWorkerShared *) dsm_segment_address(seg);
      81                 :             : 
      82                 :             :     /* Arrange to signal the steering process if we exit. */
      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))
      91                 :           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                 :             :      */
      97                 :          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                 :             :      */
     110                 :          13 :     BackgroundWorkerInitializeConnectionByOid(shared->dbid, shared->roleid,
     111                 :             :                                               BGWORKER_BYPASS_ALLOWCONN |
     112                 :             :                                               BGWORKER_BYPASS_ROLELOGINCHECK);
     113                 :             : 
     114                 :             :     /* Adopt the steering backend's relevant timeouts. */
     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                 :             :      */
     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                 :             :     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();
     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
     185                 :          13 : RepackWorkerShutdown(int code, Datum arg)
     186                 :             : {
     187                 :             :     dsm_segment *seg;
     188                 :             :     DecodingWorkerShared *shared;
     189                 :             :     pid_t       pid;
     190                 :             :     ProcNumber  procno;
     191                 :             : 
     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);
     207                 :          13 : }
     208                 :             : 
     209                 :             : bool
     210                 :        2092 : AmRepackWorker(void)
     211                 :             : {
     212                 :        2092 :     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                 :             :     Assert(!TransactionIdIsValid(GetTopTransactionIdIfAny()));
     237                 :             : 
     238                 :             :     /* Make sure we can use logical decoding */
     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                 :             :      */
     249                 :          13 :     snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid);
     250                 :          13 :     ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true,
     251                 :             :                           false, false);
     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                 :             :      */
     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                 :             :     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                 :             :      */
     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                 :             :      */
     317                 :          13 :     oldcxt = MemoryContextSwitchTo(ctx->context);
     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);
     334                 :          13 :     ReplicationSlotDropAcquired(true);
     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);
     358                 :          13 :     pfree(snap_space);
     359                 :             : 
     360                 :             :     /* Increase the counter to tell the backend that the file is available. */
     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                 :        3004 :     {
     394                 :             :         XLogRecord *record;
     395                 :             :         XLogSegNo   segno_new;
     396                 :        3030 :         char       *errm = NULL;
     397                 :             :         XLogRecPtr  end_lsn;
     398                 :             : 
     399         [ -  + ]:        3030 :         CHECK_FOR_INTERRUPTS();
     400                 :             : 
     401                 :        3030 :         record = XLogReadRecord(ctx->reader, &errm);
     402         [ +  + ]:        3030 :         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                 :             :             {
     429                 :           0 :                 LogicalIncreaseRestartDecodingForSlot(end_lsn, end_lsn);
     430                 :           0 :                 LogicalConfirmReceivedLocation(end_lsn);
     431         [ #  # ]:           0 :                 elog(DEBUG1, "REPACK: confirmed receive location %X/%08X",
     432                 :             :                      LSN_FORMAT_ARGS(end_lsn));
     433                 :           0 :                 repack_current_segment = segno_new;
     434                 :             :             }
     435                 :             :         }
     436                 :             :         else
     437                 :             :         {
     438                 :             :             ReadLocalXLogPageNoWaitPrivate *priv;
     439                 :             : 
     440         [ -  + ]:          59 :             if (errm)
     441         [ #  # ]:           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                 :             :              */
     453                 :          59 :             priv = (ReadLocalXLogPageNoWaitPrivate *) ctx->reader->private_data;
     454         [ +  - ]:          59 :             if (priv->end_of_wal)
     455                 :             :                 /* Do not miss the end of WAL condition next time. */
     456                 :          59 :                 priv->end_of_wal = false;
     457                 :             :             else
     458         [ #  # ]:           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                 :             :          */
     467         [ +  + ]:        3030 :         if (!XLogRecPtrIsValid(lsn_upto))
     468                 :             :         {
     469                 :        2365 :             SpinLockAcquire(&shared->mutex);
     470                 :        2365 :             lsn_upto = shared->lsn_upto;
     471                 :             :             /* 'done' should be set at the same time as 'lsn_upto' */
     472                 :        2365 :             done = shared->done;
     473                 :        2365 :             SpinLockRelease(&shared->mutex);
     474                 :             :         }
     475         [ +  + ]:        3030 :         if (XLogRecPtrIsValid(lsn_upto) &&
     476         [ +  + ]:         691 :             ctx->reader->EndRecPtr >= lsn_upto)
     477                 :          26 :             break;
     478                 :             : 
     479         [ +  + ]:        3004 :         if (record == NULL)
     480                 :             :         {
     481                 :          48 :             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                 :             :              */
     499         [ +  - ]:          48 :             if (!XLogRecPtrIsValid(lsn_upto))
     500                 :          48 :                 timeout = 100;
     501                 :          48 :             res = WaitForLSN(WAIT_LSN_TYPE_PRIMARY_FLUSH,
     502                 :          48 :                              ctx->reader->EndRecPtr + 1,
     503                 :             :                              timeout);
     504   [ +  +  -  + ]:          48 :             if (res != WAIT_LSN_RESULT_SUCCESS &&
     505                 :             :                 res != WAIT_LSN_RESULT_TIMEOUT)
     506         [ #  # ]:           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                 :             :      */
     515                 :          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                 :     1408314 : change_useless_for_repack(XLogRecordBuffer *buf)
     532                 :             : {
     533                 :     1408314 :     XLogReaderState *r = buf->record;
     534                 :             :     RelFileLocator locator;
     535                 :             : 
     536                 :             :     /* TOAST locator should not be set unless the main is. */
     537                 :             :     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         [ +  + ]:     1408314 :     if (!OidIsValid(repacked_rel_locator.relNumber))
     545                 :     1407443 :         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))
     553                 :           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                 :             :      */
     559   [ +  +  +  -  :         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                 :             : }
        

Generated by: LCOV version 2.0-1