src/bin/pg_rewind/parsexlog.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * parsexlog.c
   4  *        Functions for reading Write-Ahead-Log
   5  *
   6  * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *-------------------------------------------------------------------------
  10  */
  11
  12 #include "postgres_fe.h"
  13
  14 #include <unistd.h>
  15
  16 #include "access/rmgr.h"
  17 #include "access/xact.h"
  18 #include "access/xlog_internal.h"
  19 #include "access/xlogreader.h"
  20 #include "catalog/pg_control.h"
  21 #include "catalog/storage_xlog.h"
  22 #include "commands/dbcommands_xlog.h"
  23 #include "fe_utils/archive.h"
  24 #include "filemap.h"
  25 #include "pg_rewind.h"
  26
  27 /*
  28  * RmgrNames is an array of resource manager names, to make error messages
  29  * a bit nicer.
  30  */
  31 #define PG_RMGR(symname,name,redo,desc,identify,startup,cleanup,mask) \
  32   name,
  33
  34 static const char *RmgrNames[RM_MAX_ID + 1] = {
  35 #include "access/rmgrlist.h"
  36 };
  37
  38 static void extractPageInfo(XLogReaderState *record);
  39
  40 static int      xlogreadfd = -1;
  41 static XLogSegNo xlogreadsegno = -1;
  42 static char xlogfpath[MAXPGPATH];
  43
  44 typedef struct XLogPageReadPrivate
  45 {
  46         const char *restoreCommand;
  47         int                     tliIndex;
  48 } XLogPageReadPrivate;
  49
  50 static int      SimpleXLogPageRead(XLogReaderState *xlogreader,
  51                                                            XLogRecPtr targetPagePtr,
  52                                                            int reqLen, XLogRecPtr targetRecPtr, char *readBuf);
  53
  54 /*
  55  * Read WAL from the datadir/pg_wal, starting from 'startpoint' on timeline
  56  * index 'tliIndex' in target timeline history, until 'endpoint'. Make note of
  57  * the data blocks touched by the WAL records, and return them in a page map.
  58  *
  59  * 'endpoint' is the end of the last record to read. The record starting at
  60  * 'endpoint' is the first one that is not read.
  61  */
  62 void
  63 extractPageMap(const char *datadir, XLogRecPtr startpoint, int tliIndex,
  64                            XLogRecPtr endpoint, const char *restoreCommand)
  65 {
  66         XLogRecord *record;
  67         XLogReaderState *xlogreader;
  68         char       *errormsg;
  69         XLogPageReadPrivate private;
  70
  71         private.tliIndex = tliIndex;
  72         private.restoreCommand = restoreCommand;
  73         xlogreader = XLogReaderAllocate(WalSegSz, datadir,
  74                                                                         XL_ROUTINE(.page_read = &SimpleXLogPageRead),
  75                                                                         &private);
  76         if (xlogreader == NULL)
  77                 pg_fatal("out of memory while allocating a WAL reading processor");
  78
  79         XLogBeginRead(xlogreader, startpoint);
  80         do
  81         {
  82                 record = XLogReadRecord(xlogreader, &errormsg);
  83
  84                 if (record == NULL)
  85                 {
  86                         XLogRecPtr      errptr = xlogreader->EndRecPtr;
  87
  88                         if (errormsg)
  89                                 pg_fatal("could not read WAL record at %X/%X: %s",
  90                                                  LSN_FORMAT_ARGS(errptr),
  91                                                  errormsg);
  92                         else
  93                                 pg_fatal("could not read WAL record at %X/%X",
  94                                                  LSN_FORMAT_ARGS(errptr));
  95                 }
  96
  97                 extractPageInfo(xlogreader);
  98
  99         } while (xlogreader->EndRecPtr < endpoint);
 100
 101         /*
 102          * If 'endpoint' didn't point exactly at a record boundary, the caller
 103          * messed up.
 104          */
 105         Assert(xlogreader->EndRecPtr == endpoint);
 106
 107         XLogReaderFree(xlogreader);
 108         if (xlogreadfd != -1)
 109         {
 110                 close(xlogreadfd);
 111                 xlogreadfd = -1;
 112         }
 113 }
 114
 115 /*
 116  * Reads one WAL record. Returns the end position of the record, without
 117  * doing anything with the record itself.
 118  */
 119 XLogRecPtr
 120 readOneRecord(const char *datadir, XLogRecPtr ptr, int tliIndex,
 121                           const char *restoreCommand)
 122 {
 123         XLogRecord *record;
 124         XLogReaderState *xlogreader;
 125         char       *errormsg;
 126         XLogPageReadPrivate private;
 127         XLogRecPtr      endptr;
 128
 129         private.tliIndex = tliIndex;
 130         private.restoreCommand = restoreCommand;
 131         xlogreader = XLogReaderAllocate(WalSegSz, datadir,
 132                                                                         XL_ROUTINE(.page_read = &SimpleXLogPageRead),
 133                                                                         &private);
 134         if (xlogreader == NULL)
 135                 pg_fatal("out of memory while allocating a WAL reading processor");
 136
 137         XLogBeginRead(xlogreader, ptr);
 138         record = XLogReadRecord(xlogreader, &errormsg);
 139         if (record == NULL)
 140         {
 141                 if (errormsg)
 142                         pg_fatal("could not read WAL record at %X/%X: %s",
 143                                          LSN_FORMAT_ARGS(ptr), errormsg);
 144                 else
 145                         pg_fatal("could not read WAL record at %X/%X",
 146                                          LSN_FORMAT_ARGS(ptr));
 147         }
 148         endptr = xlogreader->EndRecPtr;
 149
 150         XLogReaderFree(xlogreader);
 151         if (xlogreadfd != -1)
 152         {
 153                 close(xlogreadfd);
 154                 xlogreadfd = -1;
 155         }
 156
 157         return endptr;
 158 }
 159
 160 /*
 161  * Find the previous checkpoint preceding given WAL location.
 162  */
 163 void
 164 findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex,
 165                                    XLogRecPtr *lastchkptrec, TimeLineID *lastchkpttli,
 166                                    XLogRecPtr *lastchkptredo, const char *restoreCommand)
 167 {
 168         /* Walk backwards, starting from the given record */
 169         XLogRecord *record;
 170         XLogRecPtr      searchptr;
 171         XLogReaderState *xlogreader;
 172         char       *errormsg;
 173         XLogPageReadPrivate private;
 174
 175         /*
 176          * The given fork pointer points to the end of the last common record,
 177          * which is not necessarily the beginning of the next record, if the
 178          * previous record happens to end at a page boundary. Skip over the page
 179          * header in that case to find the next record.
 180          */
 181         if (forkptr % XLOG_BLCKSZ == 0)
 182         {
 183                 if (XLogSegmentOffset(forkptr, WalSegSz) == 0)
 184                         forkptr += SizeOfXLogLongPHD;
 185                 else
 186                         forkptr += SizeOfXLogShortPHD;
 187         }
 188
 189         private.tliIndex = tliIndex;
 190         private.restoreCommand = restoreCommand;
 191         xlogreader = XLogReaderAllocate(WalSegSz, datadir,
 192                                                                         XL_ROUTINE(.page_read = &SimpleXLogPageRead),
 193                                                                         &private);
 194         if (xlogreader == NULL)
 195                 pg_fatal("out of memory while allocating a WAL reading processor");
 196
 197         searchptr = forkptr;
 198         for (;;)
 199         {
 200                 uint8           info;
 201
 202                 XLogBeginRead(xlogreader, searchptr);
 203                 record = XLogReadRecord(xlogreader, &errormsg);
 204
 205                 if (record == NULL)
 206                 {
 207                         if (errormsg)
 208                                 pg_fatal("could not find previous WAL record at %X/%X: %s",
 209                                                  LSN_FORMAT_ARGS(searchptr),
 210                                                  errormsg);
 211                         else
 212                                 pg_fatal("could not find previous WAL record at %X/%X",
 213                                                  LSN_FORMAT_ARGS(searchptr));
 214                 }
 215
 216                 /*
 217                  * Check if it is a checkpoint record. This checkpoint record needs to
 218                  * be the latest checkpoint before WAL forked and not the checkpoint
 219                  * where the primary has been stopped to be rewound.
 220                  */
 221                 info = XLogRecGetInfo(xlogreader) & ~XLR_INFO_MASK;
 222                 if (searchptr < forkptr &&
 223                         XLogRecGetRmid(xlogreader) == RM_XLOG_ID &&
 224                         (info == XLOG_CHECKPOINT_SHUTDOWN ||
 225                          info == XLOG_CHECKPOINT_ONLINE))
 226                 {
 227                         CheckPoint      checkPoint;
 228
 229                         memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));
 230                         *lastchkptrec = searchptr;
 231                         *lastchkpttli = checkPoint.ThisTimeLineID;
 232                         *lastchkptredo = checkPoint.redo;
 233                         break;
 234                 }
 235
 236                 /* Walk backwards to previous record. */
 237                 searchptr = record->xl_prev;
 238         }
 239
 240         XLogReaderFree(xlogreader);
 241         if (xlogreadfd != -1)
 242         {
 243                 close(xlogreadfd);
 244                 xlogreadfd = -1;
 245         }
 246 }
 247
 248 /* XLogReader callback function, to read a WAL page */
 249 static int
 250 SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 251                                    int reqLen, XLogRecPtr targetRecPtr, char *readBuf)
 252 {
 253         XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data;
 254         uint32          targetPageOff;
 255         XLogRecPtr      targetSegEnd;
 256         XLogSegNo       targetSegNo;
 257         int                     r;
 258
 259         XLByteToSeg(targetPagePtr, targetSegNo, WalSegSz);
 260         XLogSegNoOffsetToRecPtr(targetSegNo + 1, 0, WalSegSz, targetSegEnd);
 261         targetPageOff = XLogSegmentOffset(targetPagePtr, WalSegSz);
 262
 263         /*
 264          * See if we need to switch to a new segment because the requested record
 265          * is not in the currently open one.
 266          */
 267         if (xlogreadfd >= 0 &&
 268                 !XLByteInSeg(targetPagePtr, xlogreadsegno, WalSegSz))
 269         {
 270                 close(xlogreadfd);
 271                 xlogreadfd = -1;
 272         }
 273
 274         XLByteToSeg(targetPagePtr, xlogreadsegno, WalSegSz);
 275
 276         if (xlogreadfd < 0)
 277         {
 278                 char            xlogfname[MAXFNAMELEN];
 279
 280                 /*
 281                  * Since incomplete segments are copied into next timelines, switch to
 282                  * the timeline holding the required segment. Assuming this scan can
 283                  * be done both forward and backward, consider also switching timeline
 284                  * accordingly.
 285                  */
 286                 while (private->tliIndex < targetNentries - 1 &&
 287                            targetHistory[private->tliIndex].end < targetSegEnd)
 288                         private->tliIndex++;
 289                 while (private->tliIndex > 0 &&
 290                            targetHistory[private->tliIndex].begin >= targetSegEnd)
 291                         private->tliIndex--;
 292
 293                 XLogFileName(xlogfname, targetHistory[private->tliIndex].tli,
 294                                          xlogreadsegno, WalSegSz);
 295
 296                 snprintf(xlogfpath, MAXPGPATH, "%s/" XLOGDIR "/%s",
 297                                  xlogreader->segcxt.ws_dir, xlogfname);
 298
 299                 xlogreadfd = open(xlogfpath, O_RDONLY | PG_BINARY, 0);
 300
 301                 if (xlogreadfd < 0)
 302                 {
 303                         /*
 304                          * If we have no restore_command to execute, then exit.
 305                          */
 306                         if (private->restoreCommand == NULL)
 307                         {
 308                                 pg_log_error("could not open file \"%s\": %m", xlogfpath);
 309                                 return -1;
 310                         }
 311
 312                         /*
 313                          * Since we have restore_command, then try to retrieve missing WAL
 314                          * file from the archive.
 315                          */
 316                         xlogreadfd = RestoreArchivedFile(xlogreader->segcxt.ws_dir,
 317                                                                                          xlogfname,
 318                                                                                          WalSegSz,
 319                                                                                          private->restoreCommand);
 320
 321                         if (xlogreadfd < 0)
 322                                 return -1;
 323                         else
 324                                 pg_log_debug("using file \"%s\" restored from archive",
 325                                                          xlogfpath);
 326                 }
 327         }
 328
 329         /*
 330          * At this point, we have the right segment open.
 331          */
 332         Assert(xlogreadfd != -1);
 333
 334         /* Read the requested page */
 335         if (lseek(xlogreadfd, (off_t) targetPageOff, SEEK_SET) < 0)
 336         {
 337                 pg_log_error("could not seek in file \"%s\": %m", xlogfpath);
 338                 return -1;
 339         }
 340
 341
 342         r = read(xlogreadfd, readBuf, XLOG_BLCKSZ);
 343         if (r != XLOG_BLCKSZ)
 344         {
 345                 if (r < 0)
 346                         pg_log_error("could not read file \"%s\": %m", xlogfpath);
 347                 else
 348                         pg_log_error("could not read file \"%s\": read %d of %zu",
 349                                                  xlogfpath, r, (Size) XLOG_BLCKSZ);
 350
 351                 return -1;
 352         }
 353
 354         Assert(targetSegNo == xlogreadsegno);
 355
 356         xlogreader->seg.ws_tli = targetHistory[private->tliIndex].tli;
 357         return XLOG_BLCKSZ;
 358 }
 359
 360 /*
 361  * Extract information on which blocks the current record modifies.
 362  */
 363 static void
 364 extractPageInfo(XLogReaderState *record)
 365 {
 366         int                     block_id;
 367         RmgrId          rmid = XLogRecGetRmid(record);
 368         uint8           info = XLogRecGetInfo(record);
 369         uint8           rminfo = info & ~XLR_INFO_MASK;
 370
 371         /* Is this a special record type that I recognize? */
 372
 373         if (rmid == RM_DBASE_ID && rminfo == XLOG_DBASE_CREATE)
 374         {
 375                 /*
 376                  * New databases can be safely ignored. It won't be present in the
 377                  * source system, so it will be deleted. There's one corner-case,
 378                  * though: if a new, different, database is also created in the source
 379                  * system, we'll see that the files already exist and not copy them.
 380                  * That's OK, though; WAL replay of creating the new database, from
 381                  * the source systems's WAL, will re-copy the new database,
 382                  * overwriting the database created in the target system.
 383                  */
 384         }
 385         else if (rmid == RM_DBASE_ID && rminfo == XLOG_DBASE_DROP)
 386         {
 387                 /*
 388                  * An existing database was dropped. We'll see that the files don't
 389                  * exist in the target data dir, and copy them in toto from the source
 390                  * system. No need to do anything special here.
 391                  */
 392         }
 393         else if (rmid == RM_SMGR_ID && rminfo == XLOG_SMGR_CREATE)
 394         {
 395                 /*
 396                  * We can safely ignore these. The file will be removed from the
 397                  * target, if it doesn't exist in source system. If a file with same
 398                  * name is created in source system, too, there will be WAL records
 399                  * for all the blocks in it.
 400                  */
 401         }
 402         else if (rmid == RM_SMGR_ID && rminfo == XLOG_SMGR_TRUNCATE)
 403         {
 404                 /*
 405                  * We can safely ignore these. When we compare the sizes later on,
 406                  * we'll notice that they differ, and copy the missing tail from
 407                  * source system.
 408                  */
 409         }
 410         else if (rmid == RM_XACT_ID &&
 411                          ((rminfo & XLOG_XACT_OPMASK) == XLOG_XACT_COMMIT ||
 412                           (rminfo & XLOG_XACT_OPMASK) == XLOG_XACT_COMMIT_PREPARED ||
 413                           (rminfo & XLOG_XACT_OPMASK) == XLOG_XACT_ABORT ||
 414                           (rminfo & XLOG_XACT_OPMASK) == XLOG_XACT_ABORT_PREPARED))
 415         {
 416                 /*
 417                  * These records can include "dropped rels". We can safely ignore
 418                  * them, we will see that they are missing and copy them from the
 419                  * source.
 420                  */
 421         }
 422         else if (info & XLR_SPECIAL_REL_UPDATE)
 423         {
 424                 /*
 425                  * This record type modifies a relation file in some special way, but
 426                  * we don't recognize the type. That's bad - we don't know how to
 427                  * track that change.
 428                  */
 429                 pg_fatal("WAL record modifies a relation, but record type is not recognized: "
 430                                  "lsn: %X/%X, rmgr: %s, info: %02X",
 431                                  LSN_FORMAT_ARGS(record->ReadRecPtr),
 432                                  RmgrNames[rmid], info);
 433         }
 434
 435         for (block_id = 0; block_id <= record->max_block_id; block_id++)
 436         {
 437                 RelFileNode rnode;
 438                 ForkNumber      forknum;
 439                 BlockNumber blkno;
 440
 441                 if (!XLogRecGetBlockTag(record, block_id, &rnode, &forknum, &blkno))
 442                         continue;
 443
 444                 /* We only care about the main fork; others are copied in toto */
 445                 if (forknum != MAIN_FORKNUM)
 446                         continue;
 447
 448                 process_target_wal_block_change(forknum, rnode, blkno);
 449         }
 450 }