granicus.if.org Git - postgresql/blob - src/backend/access/transam/xlogutils.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * xlogutils.c
   4  *
   5  * PostgreSQL transaction log manager utility routines
   6  *
   7  * This file contains support routines that are used by XLOG replay functions.
   8  * None of this code is used during normal system operation.
   9  *
  10  *
  11  * Portions Copyright (c) 1996-2007, PostgreSQL Global Development Group
  12  * Portions Copyright (c) 1994, Regents of the University of California
  13  *
  14  * $PostgreSQL: pgsql/src/backend/access/transam/xlogutils.c,v 1.50 2007/05/02 23:18:03 tgl Exp $
  15  *
  16  *-------------------------------------------------------------------------
  17  */
  18 #include "postgres.h"
  19
  20 #include "access/xlogutils.h"
  21 #include "storage/bufpage.h"
  22 #include "storage/smgr.h"
  23 #include "utils/hsearch.h"
  24
  25
  26 /*
  27  * During XLOG replay, we may see XLOG records for incremental updates of
  28  * pages that no longer exist, because their relation was later dropped or
  29  * truncated.  (Note: this is only possible when full_page_writes = OFF,
  30  * since when it's ON, the first reference we see to a page should always
  31  * be a full-page rewrite not an incremental update.)  Rather than simply
  32  * ignoring such records, we make a note of the referenced page, and then
  33  * complain if we don't actually see a drop or truncate covering the page
  34  * later in replay.
  35  */
  36 typedef struct xl_invalid_page_key
  37 {
  38         RelFileNode node;                       /* the relation */
  39         BlockNumber blkno;                      /* the page */
  40 } xl_invalid_page_key;
  41
  42 typedef struct xl_invalid_page
  43 {
  44         xl_invalid_page_key key;        /* hash key ... must be first */
  45         bool            present;                /* page existed but contained zeroes */
  46 } xl_invalid_page;
  47
  48 static HTAB *invalid_page_tab = NULL;
  49
  50
  51 /* Log a reference to an invalid page */
  52 static void
  53 log_invalid_page(RelFileNode node, BlockNumber blkno, bool present)
  54 {
  55         xl_invalid_page_key key;
  56         xl_invalid_page *hentry;
  57         bool            found;
  58
  59         /*
  60          * Log references to invalid pages at DEBUG1 level.  This allows some
  61          * tracing of the cause (note the elog context mechanism will tell us
  62          * something about the XLOG record that generated the reference).
  63          */
  64         if (present)
  65                 elog(DEBUG1, "page %u of relation %u/%u/%u is uninitialized",
  66                          blkno, node.spcNode, node.dbNode, node.relNode);
  67         else
  68                 elog(DEBUG1, "page %u of relation %u/%u/%u does not exist",
  69                          blkno, node.spcNode, node.dbNode, node.relNode);
  70
  71         if (invalid_page_tab == NULL)
  72         {
  73                 /* create hash table when first needed */
  74                 HASHCTL         ctl;
  75
  76                 memset(&ctl, 0, sizeof(ctl));
  77                 ctl.keysize = sizeof(xl_invalid_page_key);
  78                 ctl.entrysize = sizeof(xl_invalid_page);
  79                 ctl.hash = tag_hash;
  80
  81                 invalid_page_tab = hash_create("XLOG invalid-page table",
  82                                                                            100,
  83                                                                            &ctl,
  84                                                                            HASH_ELEM | HASH_FUNCTION);
  85         }
  86
  87         /* we currently assume xl_invalid_page_key contains no padding */
  88         key.node = node;
  89         key.blkno = blkno;
  90         hentry = (xl_invalid_page *)
  91                 hash_search(invalid_page_tab, (void *) &key, HASH_ENTER, &found);
  92
  93         if (!found)
  94         {
  95                 /* hash_search already filled in the key */
  96                 hentry->present = present;
  97         }
  98         else
  99         {
 100                 /* repeat reference ... leave "present" as it was */
 101         }
 102 }
 103
 104 /* Forget any invalid pages >= minblkno, because they've been dropped */
 105 static void
 106 forget_invalid_pages(RelFileNode node, BlockNumber minblkno)
 107 {
 108         HASH_SEQ_STATUS status;
 109         xl_invalid_page *hentry;
 110
 111         if (invalid_page_tab == NULL)
 112                 return;                                 /* nothing to do */
 113
 114         hash_seq_init(&status, invalid_page_tab);
 115
 116         while ((hentry = (xl_invalid_page *) hash_seq_search(&status)) != NULL)
 117         {
 118                 if (RelFileNodeEquals(hentry->key.node, node) &&
 119                         hentry->key.blkno >= minblkno)
 120                 {
 121                         elog(DEBUG2, "page %u of relation %u/%u/%u has been dropped",
 122                                  hentry->key.blkno, hentry->key.node.spcNode,
 123                                  hentry->key.node.dbNode, hentry->key.node.relNode);
 124
 125                         if (hash_search(invalid_page_tab,
 126                                                         (void *) &hentry->key,
 127                                                         HASH_REMOVE, NULL) == NULL)
 128                                 elog(ERROR, "hash table corrupted");
 129                 }
 130         }
 131 }
 132
 133 /* Forget any invalid pages in a whole database */
 134 static void
 135 forget_invalid_pages_db(Oid dbid)
 136 {
 137         HASH_SEQ_STATUS status;
 138         xl_invalid_page *hentry;
 139
 140         if (invalid_page_tab == NULL)
 141                 return;                                 /* nothing to do */
 142
 143         hash_seq_init(&status, invalid_page_tab);
 144
 145         while ((hentry = (xl_invalid_page *) hash_seq_search(&status)) != NULL)
 146         {
 147                 if (hentry->key.node.dbNode == dbid)
 148                 {
 149                         elog(DEBUG2, "page %u of relation %u/%u/%u has been dropped",
 150                                  hentry->key.blkno, hentry->key.node.spcNode,
 151                                  hentry->key.node.dbNode, hentry->key.node.relNode);
 152
 153                         if (hash_search(invalid_page_tab,
 154                                                         (void *) &hentry->key,
 155                                                         HASH_REMOVE, NULL) == NULL)
 156                                 elog(ERROR, "hash table corrupted");
 157                 }
 158         }
 159 }
 160
 161 /* Complain about any remaining invalid-page entries */
 162 void
 163 XLogCheckInvalidPages(void)
 164 {
 165         HASH_SEQ_STATUS status;
 166         xl_invalid_page *hentry;
 167         bool            foundone = false;
 168
 169         if (invalid_page_tab == NULL)
 170                 return;                                 /* nothing to do */
 171
 172         hash_seq_init(&status, invalid_page_tab);
 173
 174         /*
 175          * Our strategy is to emit WARNING messages for all remaining entries and
 176          * only PANIC after we've dumped all the available info.
 177          */
 178         while ((hentry = (xl_invalid_page *) hash_seq_search(&status)) != NULL)
 179         {
 180                 if (hentry->present)
 181                         elog(WARNING, "page %u of relation %u/%u/%u was uninitialized",
 182                                  hentry->key.blkno, hentry->key.node.spcNode,
 183                                  hentry->key.node.dbNode, hentry->key.node.relNode);
 184                 else
 185                         elog(WARNING, "page %u of relation %u/%u/%u did not exist",
 186                                  hentry->key.blkno, hentry->key.node.spcNode,
 187                                  hentry->key.node.dbNode, hentry->key.node.relNode);
 188                 foundone = true;
 189         }
 190
 191         if (foundone)
 192                 elog(PANIC, "WAL contains references to invalid pages");
 193 }
 194
 195
 196 /*
 197  * XLogReadBuffer
 198  *              Read a page during XLOG replay
 199  *
 200  * This is functionally comparable to ReadBuffer followed by
 201  * LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE): you get back a pinned
 202  * and locked buffer.  (Getting the lock is not really necessary, since we
 203  * expect that this is only used during single-process XLOG replay, but
 204  * some subroutines such as MarkBufferDirty will complain if we don't.)
 205  *
 206  * If "init" is true then the caller intends to rewrite the page fully
 207  * using the info in the XLOG record.  In this case we will extend the
 208  * relation if needed to make the page exist, and we will not complain about
 209  * the page being "new" (all zeroes); in fact, we usually will supply a
 210  * zeroed buffer without reading the page at all, so as to avoid unnecessary
 211  * failure if the page is present on disk but has corrupt headers.
 212  *
 213  * If "init" is false then the caller needs the page to be valid already.
 214  * If the page doesn't exist or contains zeroes, we return InvalidBuffer.
 215  * In this case the caller should silently skip the update on this page.
 216  * (In this situation, we expect that the page was later dropped or truncated.
 217  * If we don't see evidence of that later in the WAL sequence, we'll complain
 218  * at the end of WAL replay.)
 219  */
 220 Buffer
 221 XLogReadBuffer(Relation reln, BlockNumber blkno, bool init)
 222 {
 223         BlockNumber lastblock = RelationGetNumberOfBlocks(reln);
 224         Buffer          buffer;
 225
 226         Assert(blkno != P_NEW);
 227
 228         if (blkno < lastblock)
 229         {
 230                 /* page exists in file */
 231                 if (init)
 232                         buffer = ReadOrZeroBuffer(reln, blkno);
 233                 else
 234                         buffer = ReadBuffer(reln, blkno);
 235         }
 236         else
 237         {
 238                 /* hm, page doesn't exist in file */
 239                 if (!init)
 240                 {
 241                         log_invalid_page(reln->rd_node, blkno, false);
 242                         return InvalidBuffer;
 243                 }
 244                 /* OK to extend the file */
 245                 /* we do this in recovery only - no rel-extension lock needed */
 246                 Assert(InRecovery);
 247                 buffer = InvalidBuffer;
 248                 while (blkno >= lastblock)
 249                 {
 250                         if (buffer != InvalidBuffer)
 251                                 ReleaseBuffer(buffer);
 252                         buffer = ReadBuffer(reln, P_NEW);
 253                         lastblock++;
 254                 }
 255                 Assert(BufferGetBlockNumber(buffer) == blkno);
 256         }
 257
 258         LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
 259
 260         if (!init)
 261         {
 262                 /* check that page has been initialized */
 263                 Page            page = (Page) BufferGetPage(buffer);
 264
 265                 if (PageIsNew((PageHeader) page))
 266                 {
 267                         UnlockReleaseBuffer(buffer);
 268                         log_invalid_page(reln->rd_node, blkno, true);
 269                         return InvalidBuffer;
 270                 }
 271         }
 272
 273         return buffer;
 274 }
 275
 276
 277 /*
 278  * Lightweight "Relation" cache --- this substitutes for the normal relcache
 279  * during XLOG replay.
 280  */
 281
 282 typedef struct XLogRelDesc
 283 {
 284         RelationData reldata;
 285         struct XLogRelDesc *lessRecently;
 286         struct XLogRelDesc *moreRecently;
 287 } XLogRelDesc;
 288
 289 typedef struct XLogRelCacheEntry
 290 {
 291         RelFileNode rnode;
 292         XLogRelDesc *rdesc;
 293 } XLogRelCacheEntry;
 294
 295 static HTAB *_xlrelcache;
 296 static XLogRelDesc *_xlrelarr = NULL;
 297 static Form_pg_class _xlpgcarr = NULL;
 298 static int      _xlast = 0;
 299 static int      _xlcnt = 0;
 300
 301 #define _XLOG_RELCACHESIZE      512
 302
 303 static void
 304 _xl_init_rel_cache(void)
 305 {
 306         HASHCTL         ctl;
 307
 308         _xlcnt = _XLOG_RELCACHESIZE;
 309         _xlast = 0;
 310         _xlrelarr = (XLogRelDesc *) malloc(sizeof(XLogRelDesc) * _xlcnt);
 311         memset(_xlrelarr, 0, sizeof(XLogRelDesc) * _xlcnt);
 312         _xlpgcarr = (Form_pg_class) malloc(sizeof(FormData_pg_class) * _xlcnt);
 313         memset(_xlpgcarr, 0, sizeof(FormData_pg_class) * _xlcnt);
 314
 315         _xlrelarr[0].moreRecently = &(_xlrelarr[0]);
 316         _xlrelarr[0].lessRecently = &(_xlrelarr[0]);
 317
 318         memset(&ctl, 0, sizeof(ctl));
 319         ctl.keysize = sizeof(RelFileNode);
 320         ctl.entrysize = sizeof(XLogRelCacheEntry);
 321         ctl.hash = tag_hash;
 322
 323         _xlrelcache = hash_create("XLOG relcache", _XLOG_RELCACHESIZE,
 324                                                           &ctl, HASH_ELEM | HASH_FUNCTION);
 325 }
 326
 327 static void
 328 _xl_remove_hash_entry(XLogRelDesc *rdesc)
 329 {
 330         Form_pg_class tpgc = rdesc->reldata.rd_rel;
 331         XLogRelCacheEntry *hentry;
 332
 333         rdesc->lessRecently->moreRecently = rdesc->moreRecently;
 334         rdesc->moreRecently->lessRecently = rdesc->lessRecently;
 335
 336         hentry = (XLogRelCacheEntry *) hash_search(_xlrelcache,
 337                                           (void *) &(rdesc->reldata.rd_node), HASH_REMOVE, NULL);
 338         if (hentry == NULL)
 339                 elog(PANIC, "_xl_remove_hash_entry: file was not found in cache");
 340
 341         RelationCloseSmgr(&(rdesc->reldata));
 342
 343         memset(rdesc, 0, sizeof(XLogRelDesc));
 344         memset(tpgc, 0, sizeof(FormData_pg_class));
 345         rdesc->reldata.rd_rel = tpgc;
 346 }
 347
 348 static XLogRelDesc *
 349 _xl_new_reldesc(void)
 350 {
 351         XLogRelDesc *res;
 352
 353         _xlast++;
 354         if (_xlast < _xlcnt)
 355         {
 356                 _xlrelarr[_xlast].reldata.rd_rel = &(_xlpgcarr[_xlast]);
 357                 return &(_xlrelarr[_xlast]);
 358         }
 359
 360         /* reuse */
 361         res = _xlrelarr[0].moreRecently;
 362
 363         _xl_remove_hash_entry(res);
 364
 365         _xlast--;
 366         return res;
 367 }
 368
 369
 370 void
 371 XLogInitRelationCache(void)
 372 {
 373         _xl_init_rel_cache();
 374         invalid_page_tab = NULL;
 375 }
 376
 377 void
 378 XLogCloseRelationCache(void)
 379 {
 380         HASH_SEQ_STATUS status;
 381         XLogRelCacheEntry *hentry;
 382
 383         if (!_xlrelarr)
 384                 return;
 385
 386         hash_seq_init(&status, _xlrelcache);
 387
 388         while ((hentry = (XLogRelCacheEntry *) hash_seq_search(&status)) != NULL)
 389                 _xl_remove_hash_entry(hentry->rdesc);
 390
 391         hash_destroy(_xlrelcache);
 392
 393         free(_xlrelarr);
 394         free(_xlpgcarr);
 395
 396         _xlrelarr = NULL;
 397 }
 398
 399 /*
 400  * Open a relation during XLOG replay
 401  *
 402  * Note: this once had an API that allowed NULL return on failure, but it
 403  * no longer does; any failure results in elog().
 404  */
 405 Relation
 406 XLogOpenRelation(RelFileNode rnode)
 407 {
 408         XLogRelDesc *res;
 409         XLogRelCacheEntry *hentry;
 410         bool            found;
 411
 412         hentry = (XLogRelCacheEntry *)
 413                 hash_search(_xlrelcache, (void *) &rnode, HASH_FIND, NULL);
 414
 415         if (hentry)
 416         {
 417                 res = hentry->rdesc;
 418
 419                 res->lessRecently->moreRecently = res->moreRecently;
 420                 res->moreRecently->lessRecently = res->lessRecently;
 421         }
 422         else
 423         {
 424                 res = _xl_new_reldesc();
 425
 426                 sprintf(RelationGetRelationName(&(res->reldata)), "%u", rnode.relNode);
 427
 428                 res->reldata.rd_node = rnode;
 429
 430                 /*
 431                  * We set up the lockRelId in case anything tries to lock the dummy
 432                  * relation.  Note that this is fairly bogus since relNode may be
 433                  * different from the relation's OID.  It shouldn't really matter
 434                  * though, since we are presumably running by ourselves and can't have
 435                  * any lock conflicts ...
 436                  */
 437                 res->reldata.rd_lockInfo.lockRelId.dbId = rnode.dbNode;
 438                 res->reldata.rd_lockInfo.lockRelId.relId = rnode.relNode;
 439
 440                 hentry = (XLogRelCacheEntry *)
 441                         hash_search(_xlrelcache, (void *) &rnode, HASH_ENTER, &found);
 442
 443                 if (found)
 444                         elog(PANIC, "xlog relation already present on insert into cache");
 445
 446                 hentry->rdesc = res;
 447
 448                 res->reldata.rd_targblock = InvalidBlockNumber;
 449                 res->reldata.rd_smgr = NULL;
 450                 RelationOpenSmgr(&(res->reldata));
 451
 452                 /*
 453                  * Create the target file if it doesn't already exist.  This lets us
 454                  * cope if the replay sequence contains writes to a relation that is
 455                  * later deleted.  (The original coding of this routine would instead
 456                  * return NULL, causing the writes to be suppressed. But that seems
 457                  * like it risks losing valuable data if the filesystem loses an inode
 458                  * during a crash.      Better to write the data until we are actually
 459                  * told to delete the file.)
 460                  */
 461                 smgrcreate(res->reldata.rd_smgr, res->reldata.rd_istemp, true);
 462         }
 463
 464         res->moreRecently = &(_xlrelarr[0]);
 465         res->lessRecently = _xlrelarr[0].lessRecently;
 466         _xlrelarr[0].lessRecently = res;
 467         res->lessRecently->moreRecently = res;
 468
 469         return &(res->reldata);
 470 }
 471
 472 /*
 473  * Drop a relation during XLOG replay
 474  *
 475  * This is called when the relation is about to be deleted; we need to ensure
 476  * that there is no dangling smgr reference in the xlog relation cache.
 477  *
 478  * Currently, we don't bother to physically remove the relation from the
 479  * cache, we just let it age out normally.
 480  *
 481  * This also takes care of removing any open "invalid-page" records for
 482  * the relation.
 483  */
 484 void
 485 XLogDropRelation(RelFileNode rnode)
 486 {
 487         XLogRelCacheEntry *hentry;
 488
 489         hentry = (XLogRelCacheEntry *)
 490                 hash_search(_xlrelcache, (void *) &rnode, HASH_FIND, NULL);
 491
 492         if (hentry)
 493         {
 494                 XLogRelDesc *rdesc = hentry->rdesc;
 495
 496                 RelationCloseSmgr(&(rdesc->reldata));
 497         }
 498
 499         forget_invalid_pages(rnode, 0);
 500 }
 501
 502 /*
 503  * Drop a whole database during XLOG replay
 504  *
 505  * As above, but for DROP DATABASE instead of dropping a single rel
 506  */
 507 void
 508 XLogDropDatabase(Oid dbid)
 509 {
 510         HASH_SEQ_STATUS status;
 511         XLogRelCacheEntry *hentry;
 512
 513         hash_seq_init(&status, _xlrelcache);
 514
 515         while ((hentry = (XLogRelCacheEntry *) hash_seq_search(&status)) != NULL)
 516         {
 517                 XLogRelDesc *rdesc = hentry->rdesc;
 518
 519                 if (hentry->rnode.dbNode == dbid)
 520                         RelationCloseSmgr(&(rdesc->reldata));
 521         }
 522
 523         forget_invalid_pages_db(dbid);
 524 }
 525
 526 /*
 527  * Truncate a relation during XLOG replay
 528  *
 529  * We don't need to do anything to the fake relcache, but we do need to
 530  * clean up any open "invalid-page" records for the dropped pages.
 531  */
 532 void
 533 XLogTruncateRelation(RelFileNode rnode, BlockNumber nblocks)
 534 {
 535         forget_invalid_pages(rnode, nblocks);
 536 }