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-09-26 04:15:43 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                 :          11 : 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                 :          11 :     am_repack_worker = true;
      71                 :             : 
      72                 :          11 :     BackgroundWorkerUnblockSignals();
      73                 :             : 
      74                 :          11 :     seg = dsm_attach(DatumGetUInt32(main_arg));
      75         [ -  + ]:          11 :     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                 :          11 :     shared = (DecodingWorkerShared *) dsm_segment_address(seg);
      81                 :             : 
      82                 :             :     /* Arrange to signal the steering process if we exit. */
      83                 :          11 :     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         [ -  + ]:          11 :     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                 :          11 :     mq = (shm_mq *) (char *) BUFFERALIGN(shared->error_queue);
      98                 :          11 :     shm_mq_set_sender(mq, MyProc);
      99                 :          11 :     mqh = shm_mq_attach(mq, seg, NULL);
     100                 :          11 :     pq_redirect_to_shm_mq(seg, mqh);
     101                 :          11 :     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                 :          11 :     BackgroundWorkerInitializeConnectionByOid(shared->dbid, shared->roleid,
     111                 :             :                                               BGWORKER_BYPASS_ALLOWCONN |
     112                 :             :                                               BGWORKER_BYPASS_ROLELOGINCHECK);
     113                 :             : 
     114                 :             :     /* Adopt the steering backend's relevant timeouts. */
     115                 :          11 :     snprintf(buf, sizeof(buf), "%d", shared->lock_timeout);
     116                 :          11 :     SetConfigOption("lock_timeout", buf, PGC_SUSET, PGC_S_OVERRIDE);
     117                 :          11 :     snprintf(buf, sizeof(buf), "%d", shared->transaction_timeout);
     118                 :          11 :     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                 :          11 :     StartTransactionCommand();
     125                 :             : 
     126                 :          11 :     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                 :          11 :     SpinLockAcquire(&shared->mutex);
     133                 :             :     Assert(!XLogRecPtrIsValid(shared->lsn_upto));
     134                 :          11 :     sfs = &shared->sfs;
     135                 :          11 :     SpinLockRelease(&shared->mutex);
     136                 :             : 
     137                 :          11 :     SharedFileSetAttach(sfs, seg);
     138                 :             : 
     139                 :             :     /*
     140                 :             :      * Prepare to capture the concurrent data changes ourselves.
     141                 :             :      */
     142                 :          11 :     decoding_ctx = repack_setup_logical_decoding(shared->relid);
     143                 :             : 
     144                 :             :     /* Announce that we're ready. */
     145                 :          11 :     SpinLockAcquire(&shared->mutex);
     146                 :          11 :     shared->initialized = true;
     147                 :          11 :     SpinLockRelease(&shared->mutex);
     148                 :          11 :     ConditionVariableSignal(&shared->cv);
     149                 :             : 
     150                 :             :     /* There doesn't seem to a nice API to set these */
     151                 :          11 :     XactIsoLevel = XACT_REPEATABLE_READ;
     152                 :          11 :     XactReadOnly = true;
     153                 :             : 
     154                 :             :     /* Build the initial snapshot and export it. */
     155                 :          11 :     snapshot = SnapBuildInitialSnapshot(decoding_ctx->snapshot_builder);
     156                 :          11 :     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                 :          11 :     InvalidateCatalogSnapshot();
     163                 :             : 
     164                 :             :     for (;;)
     165                 :          11 :     {
     166                 :          22 :         bool        stop = decode_concurrent_changes(decoding_ctx, shared);
     167                 :             : 
     168         [ +  + ]:          22 :         if (stop)
     169                 :          11 :             break;
     170                 :             :     }
     171                 :             : 
     172                 :             :     /* Clean up and report termination to our steering process */
     173                 :          11 :     repack_cleanup_logical_decoding(decoding_ctx);
     174                 :          11 :     CommitTransactionCommand();
     175                 :          11 :     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                 :          11 : RepackWorkerShutdown(int code, Datum arg)
     186                 :             : {
     187                 :             :     dsm_segment *seg;
     188                 :             :     DecodingWorkerShared *shared;
     189                 :             :     pid_t       pid;
     190                 :             :     ProcNumber  procno;
     191                 :             : 
     192                 :          11 :     seg = (dsm_segment *) DatumGetPointer(arg);
     193                 :          11 :     shared = (DecodingWorkerShared *) dsm_segment_address(seg);
     194                 :          11 :     pid = shared->backend_pid;
     195                 :          11 :     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                 :          11 :     dsm_detach(seg);
     205                 :             : 
     206                 :          11 :     SendProcSignal(pid, PROCSIG_REPACK_MESSAGE, procno);
     207                 :          11 : }
     208                 :             : 
     209                 :             : bool
     210                 :        2039 : AmRepackWorker(void)
     211                 :             : {
     212                 :        2039 :     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                 :          11 : 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                 :          11 :     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                 :          11 :     snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid);
     250                 :          11 :     ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true,
     251                 :             :                           false, false);
     252                 :          11 :     EnsureLogicalDecodingEnabled();
     253                 :             : 
     254                 :             :     /*
     255                 :             :      * Set up repacked_rel_locator and repacked_rel_toast_locator, which we
     256                 :             :      * use to skip decoding of unrelated relations.
     257                 :             :      */
     258                 :          11 :     rel = table_open(relid, AccessShareLock);
     259                 :          11 :     repacked_rel_locator = rel->rd_locator;
     260                 :          11 :     toastrelid = rel->rd_rel->reltoastrelid;
     261         [ +  + ]:          11 :     if (OidIsValid(toastrelid))
     262                 :             :     {
     263                 :             :         Relation    toastrel;
     264                 :             : 
     265                 :             :         /* Avoid logical decoding of other TOAST relations. */
     266                 :           6 :         toastrel = table_open(toastrelid, AccessShareLock);
     267                 :           6 :         repacked_rel_toast_locator = toastrel->rd_locator;
     268                 :           6 :         table_close(toastrel, AccessShareLock);
     269                 :             :     }
     270                 :          11 :     table_close(rel, AccessShareLock);
     271                 :             : 
     272                 :             :     /*
     273                 :             :      * Set up our logical decoding context.  We initially use the blocking
     274                 :             :      * read_local_xlog_page until we find the start point, and switch to the
     275                 :             :      * non-blocking interface afterwards.
     276                 :             :      */
     277                 :          11 :     ctx = CreateInitDecodingContext(PGREPACK_PLUGIN,
     278                 :             :                                     NIL,
     279                 :             :                                     true,
     280                 :             :                                     true,
     281                 :             :                                     InvalidXLogRecPtr,
     282                 :          11 :                                     XL_ROUTINE(.page_read = read_local_xlog_page,
     283                 :             :                                                .segment_open = wal_segment_open,
     284                 :             :                                                .segment_close = wal_segment_close),
     285                 :             :                                     NULL, NULL, NULL);
     286                 :             : 
     287                 :             :     /* Complete setup of output_writer_private */
     288                 :          11 :     dstate = (RepackDecodingState *) ctx->output_writer_private;
     289                 :          11 :     dstate->relid = relid;
     290                 :          11 :     dstate->worker_cxt = CurrentMemoryContext;
     291                 :          11 :     dstate->worker_resowner = CurrentResourceOwner;
     292                 :             : 
     293                 :             :     /* We don't have control on fast_forward, but verify it's sane */
     294                 :             :     Assert(!ctx->fast_forward);
     295                 :             : 
     296                 :             :     /* Find our decoding starting point. */
     297                 :          11 :     DecodingContextFindStartpoint(ctx);
     298                 :             : 
     299                 :             :     /* From this point on, we need non-blocking WAL reads */
     300                 :          11 :     ctx->reader->routine.page_read = read_local_xlog_page_no_wait;
     301                 :             : 
     302                 :             :     /*
     303                 :             :      * Initialize repack_current_segment so that we can notice WAL segment
     304                 :             :      * boundaries.
     305                 :             :      */
     306                 :          11 :     XLByteToSeg(ctx->reader->EndRecPtr, repack_current_segment,
     307                 :             :                 wal_segment_size);
     308                 :             : 
     309                 :             :     /*
     310                 :             :      * Set up our reader private state to let the page-read callback notify
     311                 :             :      * when end-of-WAL has been reached.  This lives in the same context as
     312                 :             :      * the logical decoding itself.
     313                 :             :      */
     314                 :          11 :     oldcxt = MemoryContextSwitchTo(ctx->context);
     315                 :          11 :     ctx->reader->private_data = palloc0_object(ReadLocalXLogPageNoWaitPrivate);
     316                 :          11 :     MemoryContextSwitchTo(oldcxt);
     317                 :             : 
     318                 :          11 :     return ctx;
     319                 :             : }
     320                 :             : 
     321                 :             : static void
     322                 :          11 : repack_cleanup_logical_decoding(LogicalDecodingContext *ctx)
     323                 :             : {
     324                 :             :     RepackDecodingState *dstate;
     325                 :             : 
     326                 :          11 :     dstate = (RepackDecodingState *) ctx->output_writer_private;
     327         [ +  + ]:          11 :     if (dstate->slot)
     328                 :           2 :         ExecDropSingleTupleTableSlot(dstate->slot);
     329                 :             : 
     330                 :          11 :     FreeDecodingContext(ctx);
     331                 :          11 :     ReplicationSlotDropAcquired(true);
     332                 :          11 : }
     333                 :             : 
     334                 :             : /*
     335                 :             :  * Make snapshot available to the backend that launched the decoding worker.
     336                 :             :  */
     337                 :             : static void
     338                 :          11 : export_initial_snapshot(Snapshot snapshot, DecodingWorkerShared *shared)
     339                 :             : {
     340                 :             :     char        fname[MAXPGPATH];
     341                 :             :     BufFile    *file;
     342                 :             :     Size        snap_size;
     343                 :             :     char       *snap_space;
     344                 :             : 
     345                 :          11 :     snap_size = EstimateSnapshotSpace(snapshot);
     346                 :          11 :     snap_space = (char *) palloc(snap_size);
     347                 :          11 :     SerializeSnapshot(snapshot, snap_space);
     348                 :             : 
     349                 :          11 :     DecodingWorkerFileName(fname, shared->relid, shared->last_exported + 1);
     350                 :          11 :     file = BufFileCreateFileSet(&shared->sfs.fs, fname);
     351                 :             :     /* To make restoration easier, write the snapshot size first. */
     352                 :          11 :     BufFileWrite(file, &snap_size, sizeof(snap_size));
     353                 :          11 :     BufFileWrite(file, snap_space, snap_size);
     354                 :          11 :     BufFileClose(file);
     355                 :          11 :     pfree(snap_space);
     356                 :             : 
     357                 :             :     /* Increase the counter to tell the backend that the file is available. */
     358                 :          11 :     SpinLockAcquire(&shared->mutex);
     359                 :          11 :     shared->last_exported++;
     360                 :          11 :     SpinLockRelease(&shared->mutex);
     361                 :          11 :     ConditionVariableSignal(&shared->cv);
     362                 :          11 : }
     363                 :             : 
     364                 :             : /*
     365                 :             :  * Decode logical changes from the WAL sequence and store them to a file.
     366                 :             :  *
     367                 :             :  * If true is returned, there is no more work for the worker.
     368                 :             :  */
     369                 :             : static bool
     370                 :          22 : decode_concurrent_changes(LogicalDecodingContext *ctx,
     371                 :             :                           DecodingWorkerShared *shared)
     372                 :             : {
     373                 :             :     RepackDecodingState *dstate;
     374                 :             :     XLogRecPtr  lsn_upto;
     375                 :             :     bool        done;
     376                 :             :     char        fname[MAXPGPATH];
     377                 :             : 
     378                 :          22 :     dstate = (RepackDecodingState *) ctx->output_writer_private;
     379                 :             : 
     380                 :             :     /* Open the output file. */
     381                 :          22 :     DecodingWorkerFileName(fname, shared->relid, shared->last_exported + 1);
     382                 :          22 :     dstate->file = BufFileCreateFileSet(&shared->sfs.fs, fname);
     383                 :             : 
     384                 :          22 :     SpinLockAcquire(&shared->mutex);
     385                 :          22 :     lsn_upto = shared->lsn_upto;
     386                 :          22 :     done = shared->done;
     387                 :          22 :     SpinLockRelease(&shared->mutex);
     388                 :             : 
     389                 :             :     while (true)
     390                 :        9574 :     {
     391                 :             :         XLogRecord *record;
     392                 :             :         XLogSegNo   segno_new;
     393                 :        9596 :         char       *errm = NULL;
     394                 :             :         XLogRecPtr  end_lsn;
     395                 :             : 
     396         [ -  + ]:        9596 :         CHECK_FOR_INTERRUPTS();
     397                 :             : 
     398                 :        9596 :         record = XLogReadRecord(ctx->reader, &errm);
     399         [ +  + ]:        9596 :         if (record)
     400                 :             :         {
     401                 :        2186 :             LogicalDecodingProcessRecord(ctx, ctx->reader);
     402                 :             : 
     403                 :             :             /*
     404                 :             :              * We want to allow WAL to be recycled while REPACK is running.
     405                 :             :              *
     406                 :             :              * In normal usage of a replication slot, we need to be very
     407                 :             :              * careful not to advance the LSN until it's been confirmed as
     408                 :             :              * received by the remote.  In REPACK's case, this is not needed:
     409                 :             :              * REPACK will never try to replay the same WAL after a crash, and
     410                 :             :              * if there _is_ a crash, the whole REPACK has to be started from
     411                 :             :              * scratch anyway.
     412                 :             :              *
     413                 :             :              * So here we disregard the careful LSN tracking and just move the
     414                 :             :              * LSN locations forward to what we've processed.  Note that it
     415                 :             :              * would be bogus to move the xmin forward, though, so we don't
     416                 :             :              * touch that.
     417                 :             :              *
     418                 :             :              * This can be done on whatever schedule is convenient, but in
     419                 :             :              * order not to cause unnecessary load, we only do it as we cross
     420                 :             :              * each WAL segment boundary.
     421                 :             :              */
     422                 :        2186 :             end_lsn = ctx->reader->EndRecPtr;
     423                 :        2186 :             XLByteToSeg(end_lsn, segno_new, wal_segment_size);
     424         [ -  + ]:        2186 :             if (segno_new != repack_current_segment)
     425                 :             :             {
     426                 :           0 :                 LogicalIncreaseRestartDecodingForSlot(end_lsn, end_lsn);
     427                 :           0 :                 LogicalConfirmReceivedLocation(end_lsn);
     428         [ #  # ]:           0 :                 elog(DEBUG1, "REPACK: confirmed receive location %X/%08X",
     429                 :             :                      LSN_FORMAT_ARGS(end_lsn));
     430                 :           0 :                 repack_current_segment = segno_new;
     431                 :             :             }
     432                 :             :         }
     433                 :             :         else
     434                 :             :         {
     435                 :             :             ReadLocalXLogPageNoWaitPrivate *priv;
     436                 :             : 
     437         [ -  + ]:        7410 :             if (errm)
     438         [ #  # ]:           0 :                 ereport(ERROR,
     439                 :             :                         errcode_for_file_access(),
     440                 :             :                         errmsg("could not read WAL from timeline %u at %X/%08X: %s",
     441                 :             :                                ctx->reader->currTLI,
     442                 :             :                                LSN_FORMAT_ARGS(ctx->reader->EndRecPtr),
     443                 :             :                                errm));
     444                 :             : 
     445                 :             :             /*
     446                 :             :              * In the decoding loop we do not want to get blocked when there
     447                 :             :              * is no more WAL available, otherwise the loop would become
     448                 :             :              * uninterruptible.
     449                 :             :              */
     450                 :        7410 :             priv = (ReadLocalXLogPageNoWaitPrivate *) ctx->reader->private_data;
     451         [ +  - ]:        7410 :             if (priv->end_of_wal)
     452                 :             :                 /* Do not miss the end of WAL condition next time. */
     453                 :        7410 :                 priv->end_of_wal = false;
     454                 :             :             else
     455         [ #  # ]:           0 :                 ereport(ERROR,
     456                 :             :                         errcode(ERRCODE_DATA_CORRUPTED),
     457                 :             :                         errmsg("could not read WAL record"));
     458                 :             :         }
     459                 :             : 
     460                 :             :         /*
     461                 :             :          * Whether we could read new record or not, keep checking if
     462                 :             :          * 'lsn_upto' was specified.
     463                 :             :          */
     464         [ +  + ]:        9596 :         if (!XLogRecPtrIsValid(lsn_upto))
     465                 :             :         {
     466                 :        9131 :             SpinLockAcquire(&shared->mutex);
     467                 :        9131 :             lsn_upto = shared->lsn_upto;
     468                 :             :             /* 'done' should be set at the same time as 'lsn_upto' */
     469                 :        9131 :             done = shared->done;
     470                 :        9131 :             SpinLockRelease(&shared->mutex);
     471                 :             :         }
     472         [ +  + ]:        9596 :         if (XLogRecPtrIsValid(lsn_upto) &&
     473         [ +  + ]:         487 :             ctx->reader->EndRecPtr >= lsn_upto)
     474                 :          22 :             break;
     475                 :             : 
     476         [ +  + ]:        9574 :         if (record == NULL)
     477                 :             :         {
     478                 :        7402 :             int         timeout = 0;
     479                 :             :             WaitLSNResult res;
     480                 :             : 
     481                 :             :             /*
     482                 :             :              * Before we retry reading, wait until new WAL is flushed.
     483                 :             :              *
     484                 :             :              * There is a race condition such that the backend executing
     485                 :             :              * REPACK determines 'lsn_upto', but before it sets the shared
     486                 :             :              * variable, we reach the end of WAL. In that case we'd need to
     487                 :             :              * wait until the next WAL flush (unrelated to REPACK). Although
     488                 :             :              * that should not be a problem in a busy system, it might be
     489                 :             :              * noticeable in other cases, including regression tests (which
     490                 :             :              * are not necessarily executed in parallel). Therefore it makes
     491                 :             :              * sense to use timeout.
     492                 :             :              *
     493                 :             :              * If lsn_upto is valid, WAL records having LSN lower than that
     494                 :             :              * should already have been flushed to disk.
     495                 :             :              */
     496         [ +  - ]:        7402 :             if (!XLogRecPtrIsValid(lsn_upto))
     497                 :        7402 :                 timeout = 100;
     498                 :        7402 :             res = WaitForLSN(WAIT_LSN_TYPE_PRIMARY_FLUSH,
     499                 :        7402 :                              ctx->reader->EndRecPtr + 1,
     500                 :             :                              timeout);
     501   [ +  +  -  + ]:        7402 :             if (res != WAIT_LSN_RESULT_SUCCESS &&
     502                 :             :                 res != WAIT_LSN_RESULT_TIMEOUT)
     503         [ #  # ]:           0 :                 ereport(ERROR,
     504                 :             :                         errcode(ERRCODE_INTERNAL_ERROR),
     505                 :             :                         errmsg("waiting for WAL failed"));
     506                 :             :         }
     507                 :             :     }
     508                 :             : 
     509                 :             :     /*
     510                 :             :      * Close the file so we can make it available to the backend.
     511                 :             :      */
     512                 :          22 :     BufFileClose(dstate->file);
     513                 :          22 :     dstate->file = NULL;
     514                 :          22 :     SpinLockAcquire(&shared->mutex);
     515                 :          22 :     shared->lsn_upto = InvalidXLogRecPtr;
     516                 :          22 :     shared->last_exported++;
     517                 :          22 :     SpinLockRelease(&shared->mutex);
     518                 :          22 :     ConditionVariableSignal(&shared->cv);
     519                 :             : 
     520                 :          22 :     return done;
     521                 :             : }
     522                 :             : 
     523                 :             : /*
     524                 :             :  * Does the WAL record contain a data change that this backend does not need
     525                 :             :  * to decode on behalf of REPACK (CONCURRENTLY)?
     526                 :             :  */
     527                 :             : bool
     528                 :     1485364 : change_useless_for_repack(XLogRecordBuffer *buf)
     529                 :             : {
     530                 :     1485364 :     XLogReaderState *r = buf->record;
     531                 :             :     RelFileLocator locator;
     532                 :             : 
     533                 :             :     /* TOAST locator should not be set unless the main is. */
     534                 :             :     Assert(!OidIsValid(repacked_rel_toast_locator.relNumber) ||
     535                 :             :            OidIsValid(repacked_rel_locator.relNumber));
     536                 :             : 
     537                 :             :     /*
     538                 :             :      * Backends not involved in REPACK (CONCURRENTLY) should not do the
     539                 :             :      * filtering.
     540                 :             :      */
     541         [ +  + ]:     1485364 :     if (!OidIsValid(repacked_rel_locator.relNumber))
     542                 :     1484798 :         return false;
     543                 :             : 
     544                 :             :     /*
     545                 :             :      * If the record does not contain the block 0, it's probably not INSERT /
     546                 :             :      * UPDATE / DELETE. In any case, we do not have enough information to
     547                 :             :      * filter the change out.
     548                 :             :      */
     549         [ -  + ]:         566 :     if (!XLogRecGetBlockTagExtended(r, 0, &locator, NULL, NULL, NULL))
     550                 :           0 :         return false;
     551                 :             : 
     552                 :             :     /*
     553                 :             :      * Decode the change if it belongs to the table we are repacking, or if it
     554                 :             :      * belongs to its TOAST relation.
     555                 :             :      */
     556   [ +  +  +  -  :         566 :     if (RelFileLocatorEquals(locator, repacked_rel_locator))
                   +  - ]
     557                 :          40 :         return false;
     558         [ +  + ]:         526 :     if (OidIsValid(repacked_rel_toast_locator.relNumber) &&
     559   [ +  +  +  -  :         403 :         RelFileLocatorEquals(locator, repacked_rel_toast_locator))
                   +  - ]
     560                 :          50 :         return false;
     561                 :             : 
     562                 :             :     /* Filter out changes of other tables. */
     563                 :         476 :     return true;
     564                 :             : }
        

Generated by: LCOV version 2.0-1