granicus.if.org Git - postgresql/blob - src/backend/storage/page/bufpage.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * bufpage.c
   4  *        POSTGRES standard buffer page code.
   5  *
   6  * Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        src/backend/storage/page/bufpage.c
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/htup_details.h"
  18 #include "access/itup.h"
  19 #include "access/xlog.h"
  20 #include "storage/checksum.h"
  21 #include "utils/memdebug.h"
  22 #include "utils/memutils.h"
  23
  24
  25 /* GUC variable */
  26 bool            ignore_checksum_failure = false;
  27
  28
  29 /* ----------------------------------------------------------------
  30  *                                              Page support functions
  31  * ----------------------------------------------------------------
  32  */
  33
  34 /*
  35  * PageInit
  36  *              Initializes the contents of a page.
  37  *              Note that we don't calculate an initial checksum here; that's not done
  38  *              until it's time to write.
  39  */
  40 void
  41 PageInit(Page page, Size pageSize, Size specialSize)
  42 {
  43         PageHeader      p = (PageHeader) page;
  44
  45         specialSize = MAXALIGN(specialSize);
  46
  47         Assert(pageSize == BLCKSZ);
  48         Assert(pageSize > specialSize + SizeOfPageHeaderData);
  49
  50         /* Make sure all fields of page are zero, as well as unused space */
  51         MemSet(p, 0, pageSize);
  52
  53         p->pd_flags = 0;
  54         p->pd_lower = SizeOfPageHeaderData;
  55         p->pd_upper = pageSize - specialSize;
  56         p->pd_special = pageSize - specialSize;
  57         PageSetPageSizeAndVersion(page, pageSize, PG_PAGE_LAYOUT_VERSION);
  58         /* p->pd_prune_xid = InvalidTransactionId;              done by above MemSet */
  59 }
  60
  61
  62 /*
  63  * PageIsVerified
  64  *              Check that the page header and checksum (if any) appear valid.
  65  *
  66  * This is called when a page has just been read in from disk.  The idea is
  67  * to cheaply detect trashed pages before we go nuts following bogus item
  68  * pointers, testing invalid transaction identifiers, etc.
  69  *
  70  * It turns out to be necessary to allow zeroed pages here too.  Even though
  71  * this routine is *not* called when deliberately adding a page to a relation,
  72  * there are scenarios in which a zeroed page might be found in a table.
  73  * (Example: a backend extends a relation, then crashes before it can write
  74  * any WAL entry about the new page.  The kernel will already have the
  75  * zeroed page in the file, and it will stay that way after restart.)  So we
  76  * allow zeroed pages here, and are careful that the page access macros
  77  * treat such a page as empty and without free space.  Eventually, VACUUM
  78  * will clean up such a page and make it usable.
  79  */
  80 bool
  81 PageIsVerified(Page page, BlockNumber blkno)
  82 {
  83         PageHeader      p = (PageHeader) page;
  84         char       *pagebytes;
  85         int                     i;
  86         bool            checksum_failure = false;
  87         bool            header_sane = false;
  88         bool            all_zeroes = false;
  89         uint16          checksum = 0;
  90
  91         /*
  92          * Don't verify page data unless the page passes basic non-zero test
  93          */
  94         if (!PageIsNew(page))
  95         {
  96                 if (DataChecksumsEnabled())
  97                 {
  98                         checksum = pg_checksum_page((char *) page, blkno);
  99
 100                         if (checksum != p->pd_checksum)
 101                                 checksum_failure = true;
 102                 }
 103
 104                 /*
 105                  * The following checks don't prove the header is correct, only that
 106                  * it looks sane enough to allow into the buffer pool. Later usage of
 107                  * the block can still reveal problems, which is why we offer the
 108                  * checksum option.
 109                  */
 110                 if ((p->pd_flags & ~PD_VALID_FLAG_BITS) == 0 &&
 111                         p->pd_lower <= p->pd_upper &&
 112                         p->pd_upper <= p->pd_special &&
 113                         p->pd_special <= BLCKSZ &&
 114                         p->pd_special == MAXALIGN(p->pd_special))
 115                         header_sane = true;
 116
 117                 if (header_sane && !checksum_failure)
 118                         return true;
 119         }
 120
 121         /* Check all-zeroes case */
 122         all_zeroes = true;
 123         pagebytes = (char *) page;
 124         for (i = 0; i < BLCKSZ; i++)
 125         {
 126                 if (pagebytes[i] != 0)
 127                 {
 128                         all_zeroes = false;
 129                         break;
 130                 }
 131         }
 132
 133         if (all_zeroes)
 134                 return true;
 135
 136         /*
 137          * Throw a WARNING if the checksum fails, but only after we've checked for
 138          * the all-zeroes case.
 139          */
 140         if (checksum_failure)
 141         {
 142                 ereport(WARNING,
 143                                 (ERRCODE_DATA_CORRUPTED,
 144                                  errmsg("page verification failed, calculated checksum %u but expected %u",
 145                                                 checksum, p->pd_checksum)));
 146
 147                 if (header_sane && ignore_checksum_failure)
 148                         return true;
 149         }
 150
 151         return false;
 152 }
 153
 154
 155 /*
 156  *      PageAddItem
 157  *
 158  *      Add an item to a page.  Return value is offset at which it was
 159  *      inserted, or InvalidOffsetNumber if there's not room to insert.
 160  *
 161  *      If overwrite is true, we just store the item at the specified
 162  *      offsetNumber (which must be either a currently-unused item pointer,
 163  *      or one past the last existing item).  Otherwise,
 164  *      if offsetNumber is valid and <= current max offset in the page,
 165  *      insert item into the array at that position by shuffling ItemId's
 166  *      down to make room.
 167  *      If offsetNumber is not valid, then assign one by finding the first
 168  *      one that is both unused and deallocated.
 169  *
 170  *      If is_heap is true, we enforce that there can't be more than
 171  *      MaxHeapTuplesPerPage line pointers on the page.
 172  *
 173  *      !!! EREPORT(ERROR) IS DISALLOWED HERE !!!
 174  */
 175 OffsetNumber
 176 PageAddItem(Page page,
 177                         Item item,
 178                         Size size,
 179                         OffsetNumber offsetNumber,
 180                         bool overwrite,
 181                         bool is_heap)
 182 {
 183         PageHeader      phdr = (PageHeader) page;
 184         Size            alignedSize;
 185         int                     lower;
 186         int                     upper;
 187         ItemId          itemId;
 188         OffsetNumber limit;
 189         bool            needshuffle = false;
 190
 191         /*
 192          * Be wary about corrupted page pointers
 193          */
 194         if (phdr->pd_lower < SizeOfPageHeaderData ||
 195                 phdr->pd_lower > phdr->pd_upper ||
 196                 phdr->pd_upper > phdr->pd_special ||
 197                 phdr->pd_special > BLCKSZ)
 198                 ereport(PANIC,
 199                                 (errcode(ERRCODE_DATA_CORRUPTED),
 200                                  errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
 201                                                 phdr->pd_lower, phdr->pd_upper, phdr->pd_special)));
 202
 203         /*
 204          * Select offsetNumber to place the new item at
 205          */
 206         limit = OffsetNumberNext(PageGetMaxOffsetNumber(page));
 207
 208         /* was offsetNumber passed in? */
 209         if (OffsetNumberIsValid(offsetNumber))
 210         {
 211                 /* yes, check it */
 212                 if (overwrite)
 213                 {
 214                         if (offsetNumber < limit)
 215                         {
 216                                 itemId = PageGetItemId(phdr, offsetNumber);
 217                                 if (ItemIdIsUsed(itemId) || ItemIdHasStorage(itemId))
 218                                 {
 219                                         elog(WARNING, "will not overwrite a used ItemId");
 220                                         return InvalidOffsetNumber;
 221                                 }
 222                         }
 223                 }
 224                 else
 225                 {
 226                         if (offsetNumber < limit)
 227                                 needshuffle = true;             /* need to move existing linp's */
 228                 }
 229         }
 230         else
 231         {
 232                 /* offsetNumber was not passed in, so find a free slot */
 233                 /* if no free slot, we'll put it at limit (1st open slot) */
 234                 if (PageHasFreeLinePointers(phdr))
 235                 {
 236                         /*
 237                          * Look for "recyclable" (unused) ItemId.  We check for no storage
 238                          * as well, just to be paranoid --- unused items should never have
 239                          * storage.
 240                          */
 241                         for (offsetNumber = 1; offsetNumber < limit; offsetNumber++)
 242                         {
 243                                 itemId = PageGetItemId(phdr, offsetNumber);
 244                                 if (!ItemIdIsUsed(itemId) && !ItemIdHasStorage(itemId))
 245                                         break;
 246                         }
 247                         if (offsetNumber >= limit)
 248                         {
 249                                 /* the hint is wrong, so reset it */
 250                                 PageClearHasFreeLinePointers(phdr);
 251                         }
 252                 }
 253                 else
 254                 {
 255                         /* don't bother searching if hint says there's no free slot */
 256                         offsetNumber = limit;
 257                 }
 258         }
 259
 260         if (offsetNumber > limit)
 261         {
 262                 elog(WARNING, "specified item offset is too large");
 263                 return InvalidOffsetNumber;
 264         }
 265
 266         if (is_heap && offsetNumber > MaxHeapTuplesPerPage)
 267         {
 268                 elog(WARNING, "can't put more than MaxHeapTuplesPerPage items in a heap page");
 269                 return InvalidOffsetNumber;
 270         }
 271
 272         /*
 273          * Compute new lower and upper pointers for page, see if it'll fit.
 274          *
 275          * Note: do arithmetic as signed ints, to avoid mistakes if, say,
 276          * alignedSize > pd_upper.
 277          */
 278         if (offsetNumber == limit || needshuffle)
 279                 lower = phdr->pd_lower + sizeof(ItemIdData);
 280         else
 281                 lower = phdr->pd_lower;
 282
 283         alignedSize = MAXALIGN(size);
 284
 285         upper = (int) phdr->pd_upper - (int) alignedSize;
 286
 287         if (lower > upper)
 288                 return InvalidOffsetNumber;
 289
 290         /*
 291          * OK to insert the item.  First, shuffle the existing pointers if needed.
 292          */
 293         itemId = PageGetItemId(phdr, offsetNumber);
 294
 295         if (needshuffle)
 296                 memmove(itemId + 1, itemId,
 297                                 (limit - offsetNumber) * sizeof(ItemIdData));
 298
 299         /* set the item pointer */
 300         ItemIdSetNormal(itemId, upper, size);
 301
 302         /*
 303          * Items normally contain no uninitialized bytes.  Core bufpage consumers
 304          * conform, but this is not a necessary coding rule; a new index AM could
 305          * opt to depart from it.  However, data type input functions and other
 306          * C-language functions that synthesize datums should initialize all
 307          * bytes; datumIsEqual() relies on this.  Testing here, along with the
 308          * similar check in printtup(), helps to catch such mistakes.
 309          *
 310          * Values of the "name" type retrieved via index-only scans may contain
 311          * uninitialized bytes; see comment in btrescan().  Valgrind will report
 312          * this as an error, but it is safe to ignore.
 313          */
 314         VALGRIND_CHECK_MEM_IS_DEFINED(item, size);
 315
 316         /* copy the item's data onto the page */
 317         memcpy((char *) page + upper, item, size);
 318
 319         /* adjust page header */
 320         phdr->pd_lower = (LocationIndex) lower;
 321         phdr->pd_upper = (LocationIndex) upper;
 322
 323         return offsetNumber;
 324 }
 325
 326 /*
 327  * PageGetTempPage
 328  *              Get a temporary page in local memory for special processing.
 329  *              The returned page is not initialized at all; caller must do that.
 330  */
 331 Page
 332 PageGetTempPage(Page page)
 333 {
 334         Size            pageSize;
 335         Page            temp;
 336
 337         pageSize = PageGetPageSize(page);
 338         temp = (Page) palloc(pageSize);
 339
 340         return temp;
 341 }
 342
 343 /*
 344  * PageGetTempPageCopy
 345  *              Get a temporary page in local memory for special processing.
 346  *              The page is initialized by copying the contents of the given page.
 347  */
 348 Page
 349 PageGetTempPageCopy(Page page)
 350 {
 351         Size            pageSize;
 352         Page            temp;
 353
 354         pageSize = PageGetPageSize(page);
 355         temp = (Page) palloc(pageSize);
 356
 357         memcpy(temp, page, pageSize);
 358
 359         return temp;
 360 }
 361
 362 /*
 363  * PageGetTempPageCopySpecial
 364  *              Get a temporary page in local memory for special processing.
 365  *              The page is PageInit'd with the same special-space size as the
 366  *              given page, and the special space is copied from the given page.
 367  */
 368 Page
 369 PageGetTempPageCopySpecial(Page page)
 370 {
 371         Size            pageSize;
 372         Page            temp;
 373
 374         pageSize = PageGetPageSize(page);
 375         temp = (Page) palloc(pageSize);
 376
 377         PageInit(temp, pageSize, PageGetSpecialSize(page));
 378         memcpy(PageGetSpecialPointer(temp),
 379                    PageGetSpecialPointer(page),
 380                    PageGetSpecialSize(page));
 381
 382         return temp;
 383 }
 384
 385 /*
 386  * PageRestoreTempPage
 387  *              Copy temporary page back to permanent page after special processing
 388  *              and release the temporary page.
 389  */
 390 void
 391 PageRestoreTempPage(Page tempPage, Page oldPage)
 392 {
 393         Size            pageSize;
 394
 395         pageSize = PageGetPageSize(tempPage);
 396         memcpy((char *) oldPage, (char *) tempPage, pageSize);
 397
 398         pfree(tempPage);
 399 }
 400
 401 /*
 402  * sorting support for PageRepairFragmentation, PageIndexMultiDelete,
 403  * PageIndexDeleteNoCompact
 404  */
 405 typedef struct itemIdSortData
 406 {
 407         int                     offsetindex;    /* linp array index */
 408         int                     itemoff;                /* page offset of item data */
 409         Size            alignedlen;             /* MAXALIGN(item data len) */
 410         ItemIdData      olditemid;              /* used only in PageIndexMultiDelete */
 411 } itemIdSortData;
 412 typedef itemIdSortData *itemIdSort;
 413
 414 static int
 415 itemoffcompare(const void *itemidp1, const void *itemidp2)
 416 {
 417         /* Sort in decreasing itemoff order */
 418         return ((itemIdSort) itemidp2)->itemoff -
 419                 ((itemIdSort) itemidp1)->itemoff;
 420 }
 421
 422 /*
 423  * PageRepairFragmentation
 424  *
 425  * Frees fragmented space on a page.
 426  * It doesn't remove unused line pointers! Please don't change this.
 427  *
 428  * This routine is usable for heap pages only, but see PageIndexMultiDelete.
 429  *
 430  * As a side effect, the page's PD_HAS_FREE_LINES hint bit is updated.
 431  */
 432 void
 433 PageRepairFragmentation(Page page)
 434 {
 435         Offset          pd_lower = ((PageHeader) page)->pd_lower;
 436         Offset          pd_upper = ((PageHeader) page)->pd_upper;
 437         Offset          pd_special = ((PageHeader) page)->pd_special;
 438         ItemId          lp;
 439         int                     nline,
 440                                 nstorage,
 441                                 nunused;
 442         int                     i;
 443         Size            totallen;
 444         Offset          upper;
 445
 446         /*
 447          * It's worth the trouble to be more paranoid here than in most places,
 448          * because we are about to reshuffle data in (what is usually) a shared
 449          * disk buffer.  If we aren't careful then corrupted pointers, lengths,
 450          * etc could cause us to clobber adjacent disk buffers, spreading the data
 451          * loss further.  So, check everything.
 452          */
 453         if (pd_lower < SizeOfPageHeaderData ||
 454                 pd_lower > pd_upper ||
 455                 pd_upper > pd_special ||
 456                 pd_special > BLCKSZ ||
 457                 pd_special != MAXALIGN(pd_special))
 458                 ereport(ERROR,
 459                                 (errcode(ERRCODE_DATA_CORRUPTED),
 460                                  errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
 461                                                 pd_lower, pd_upper, pd_special)));
 462
 463         nline = PageGetMaxOffsetNumber(page);
 464         nunused = nstorage = 0;
 465         for (i = FirstOffsetNumber; i <= nline; i++)
 466         {
 467                 lp = PageGetItemId(page, i);
 468                 if (ItemIdIsUsed(lp))
 469                 {
 470                         if (ItemIdHasStorage(lp))
 471                                 nstorage++;
 472                 }
 473                 else
 474                 {
 475                         /* Unused entries should have lp_len = 0, but make sure */
 476                         ItemIdSetUnused(lp);
 477                         nunused++;
 478                 }
 479         }
 480
 481         if (nstorage == 0)
 482         {
 483                 /* Page is completely empty, so just reset it quickly */
 484                 ((PageHeader) page)->pd_upper = pd_special;
 485         }
 486         else
 487         {
 488                 /* Need to compact the page the hard way */
 489                 itemIdSortData itemidbase[MaxHeapTuplesPerPage];
 490                 itemIdSort      itemidptr = itemidbase;
 491
 492                 totallen = 0;
 493                 for (i = 0; i < nline; i++)
 494                 {
 495                         lp = PageGetItemId(page, i + 1);
 496                         if (ItemIdHasStorage(lp))
 497                         {
 498                                 itemidptr->offsetindex = i;
 499                                 itemidptr->itemoff = ItemIdGetOffset(lp);
 500                                 if (itemidptr->itemoff < (int) pd_upper ||
 501                                         itemidptr->itemoff >= (int) pd_special)
 502                                         ereport(ERROR,
 503                                                         (errcode(ERRCODE_DATA_CORRUPTED),
 504                                                          errmsg("corrupted item pointer: %u",
 505                                                                         itemidptr->itemoff)));
 506                                 itemidptr->alignedlen = MAXALIGN(ItemIdGetLength(lp));
 507                                 totallen += itemidptr->alignedlen;
 508                                 itemidptr++;
 509                         }
 510                 }
 511
 512                 if (totallen > (Size) (pd_special - pd_lower))
 513                         ereport(ERROR,
 514                                         (errcode(ERRCODE_DATA_CORRUPTED),
 515                            errmsg("corrupted item lengths: total %u, available space %u",
 516                                           (unsigned int) totallen, pd_special - pd_lower)));
 517
 518                 /* sort itemIdSortData array into decreasing itemoff order */
 519                 qsort((char *) itemidbase, nstorage, sizeof(itemIdSortData),
 520                           itemoffcompare);
 521
 522                 /* compactify page */
 523                 upper = pd_special;
 524
 525                 for (i = 0, itemidptr = itemidbase; i < nstorage; i++, itemidptr++)
 526                 {
 527                         lp = PageGetItemId(page, itemidptr->offsetindex + 1);
 528                         upper -= itemidptr->alignedlen;
 529                         memmove((char *) page + upper,
 530                                         (char *) page + itemidptr->itemoff,
 531                                         itemidptr->alignedlen);
 532                         lp->lp_off = upper;
 533                 }
 534
 535                 ((PageHeader) page)->pd_upper = upper;
 536         }
 537
 538         /* Set hint bit for PageAddItem */
 539         if (nunused > 0)
 540                 PageSetHasFreeLinePointers(page);
 541         else
 542                 PageClearHasFreeLinePointers(page);
 543 }
 544
 545 /*
 546  * PageGetFreeSpace
 547  *              Returns the size of the free (allocatable) space on a page,
 548  *              reduced by the space needed for a new line pointer.
 549  *
 550  * Note: this should usually only be used on index pages.  Use
 551  * PageGetHeapFreeSpace on heap pages.
 552  */
 553 Size
 554 PageGetFreeSpace(Page page)
 555 {
 556         int                     space;
 557
 558         /*
 559          * Use signed arithmetic here so that we behave sensibly if pd_lower >
 560          * pd_upper.
 561          */
 562         space = (int) ((PageHeader) page)->pd_upper -
 563                 (int) ((PageHeader) page)->pd_lower;
 564
 565         if (space < (int) sizeof(ItemIdData))
 566                 return 0;
 567         space -= sizeof(ItemIdData);
 568
 569         return (Size) space;
 570 }
 571
 572 /*
 573  * PageGetExactFreeSpace
 574  *              Returns the size of the free (allocatable) space on a page,
 575  *              without any consideration for adding/removing line pointers.
 576  */
 577 Size
 578 PageGetExactFreeSpace(Page page)
 579 {
 580         int                     space;
 581
 582         /*
 583          * Use signed arithmetic here so that we behave sensibly if pd_lower >
 584          * pd_upper.
 585          */
 586         space = (int) ((PageHeader) page)->pd_upper -
 587                 (int) ((PageHeader) page)->pd_lower;
 588
 589         if (space < 0)
 590                 return 0;
 591
 592         return (Size) space;
 593 }
 594
 595
 596 /*
 597  * PageGetHeapFreeSpace
 598  *              Returns the size of the free (allocatable) space on a page,
 599  *              reduced by the space needed for a new line pointer.
 600  *
 601  * The difference between this and PageGetFreeSpace is that this will return
 602  * zero if there are already MaxHeapTuplesPerPage line pointers in the page
 603  * and none are free.  We use this to enforce that no more than
 604  * MaxHeapTuplesPerPage line pointers are created on a heap page.  (Although
 605  * no more tuples than that could fit anyway, in the presence of redirected
 606  * or dead line pointers it'd be possible to have too many line pointers.
 607  * To avoid breaking code that assumes MaxHeapTuplesPerPage is a hard limit
 608  * on the number of line pointers, we make this extra check.)
 609  */
 610 Size
 611 PageGetHeapFreeSpace(Page page)
 612 {
 613         Size            space;
 614
 615         space = PageGetFreeSpace(page);
 616         if (space > 0)
 617         {
 618                 OffsetNumber offnum,
 619                                         nline;
 620
 621                 /*
 622                  * Are there already MaxHeapTuplesPerPage line pointers in the page?
 623                  */
 624                 nline = PageGetMaxOffsetNumber(page);
 625                 if (nline >= MaxHeapTuplesPerPage)
 626                 {
 627                         if (PageHasFreeLinePointers((PageHeader) page))
 628                         {
 629                                 /*
 630                                  * Since this is just a hint, we must confirm that there is
 631                                  * indeed a free line pointer
 632                                  */
 633                                 for (offnum = FirstOffsetNumber; offnum <= nline; offnum = OffsetNumberNext(offnum))
 634                                 {
 635                                         ItemId          lp = PageGetItemId(page, offnum);
 636
 637                                         if (!ItemIdIsUsed(lp))
 638                                                 break;
 639                                 }
 640
 641                                 if (offnum > nline)
 642                                 {
 643                                         /*
 644                                          * The hint is wrong, but we can't clear it here since we
 645                                          * don't have the ability to mark the page dirty.
 646                                          */
 647                                         space = 0;
 648                                 }
 649                         }
 650                         else
 651                         {
 652                                 /*
 653                                  * Although the hint might be wrong, PageAddItem will believe
 654                                  * it anyway, so we must believe it too.
 655                                  */
 656                                 space = 0;
 657                         }
 658                 }
 659         }
 660         return space;
 661 }
 662
 663
 664 /*
 665  * PageIndexTupleDelete
 666  *
 667  * This routine does the work of removing a tuple from an index page.
 668  *
 669  * Unlike heap pages, we compact out the line pointer for the removed tuple.
 670  */
 671 void
 672 PageIndexTupleDelete(Page page, OffsetNumber offnum)
 673 {
 674         PageHeader      phdr = (PageHeader) page;
 675         char       *addr;
 676         ItemId          tup;
 677         Size            size;
 678         unsigned        offset;
 679         int                     nbytes;
 680         int                     offidx;
 681         int                     nline;
 682
 683         /*
 684          * As with PageRepairFragmentation, paranoia seems justified.
 685          */
 686         if (phdr->pd_lower < SizeOfPageHeaderData ||
 687                 phdr->pd_lower > phdr->pd_upper ||
 688                 phdr->pd_upper > phdr->pd_special ||
 689                 phdr->pd_special > BLCKSZ)
 690                 ereport(ERROR,
 691                                 (errcode(ERRCODE_DATA_CORRUPTED),
 692                                  errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
 693                                                 phdr->pd_lower, phdr->pd_upper, phdr->pd_special)));
 694
 695         nline = PageGetMaxOffsetNumber(page);
 696         if ((int) offnum <= 0 || (int) offnum > nline)
 697                 elog(ERROR, "invalid index offnum: %u", offnum);
 698
 699         /* change offset number to offset index */
 700         offidx = offnum - 1;
 701
 702         tup = PageGetItemId(page, offnum);
 703         Assert(ItemIdHasStorage(tup));
 704         size = ItemIdGetLength(tup);
 705         offset = ItemIdGetOffset(tup);
 706
 707         if (offset < phdr->pd_upper || (offset + size) > phdr->pd_special ||
 708                 offset != MAXALIGN(offset) || size != MAXALIGN(size))
 709                 ereport(ERROR,
 710                                 (errcode(ERRCODE_DATA_CORRUPTED),
 711                                  errmsg("corrupted item pointer: offset = %u, size = %u",
 712                                                 offset, (unsigned int) size)));
 713
 714         /*
 715          * First, we want to get rid of the pd_linp entry for the index tuple. We
 716          * copy all subsequent linp's back one slot in the array. We don't use
 717          * PageGetItemId, because we are manipulating the _array_, not individual
 718          * linp's.
 719          */
 720         nbytes = phdr->pd_lower -
 721                 ((char *) &phdr->pd_linp[offidx + 1] - (char *) phdr);
 722
 723         if (nbytes > 0)
 724                 memmove((char *) &(phdr->pd_linp[offidx]),
 725                                 (char *) &(phdr->pd_linp[offidx + 1]),
 726                                 nbytes);
 727
 728         /*
 729          * Now move everything between the old upper bound (beginning of tuple
 730          * space) and the beginning of the deleted tuple forward, so that space in
 731          * the middle of the page is left free.  If we've just deleted the tuple
 732          * at the beginning of tuple space, then there's no need to do the copy
 733          * (and bcopy on some architectures SEGV's if asked to move zero bytes).
 734          */
 735
 736         /* beginning of tuple space */
 737         addr = (char *) page + phdr->pd_upper;
 738
 739         if (offset > phdr->pd_upper)
 740                 memmove(addr + size, addr, (int) (offset - phdr->pd_upper));
 741
 742         /* adjust free space boundary pointers */
 743         phdr->pd_upper += size;
 744         phdr->pd_lower -= sizeof(ItemIdData);
 745
 746         /*
 747          * Finally, we need to adjust the linp entries that remain.
 748          *
 749          * Anything that used to be before the deleted tuple's data was moved
 750          * forward by the size of the deleted tuple.
 751          */
 752         if (!PageIsEmpty(page))
 753         {
 754                 int                     i;
 755
 756                 nline--;                                /* there's one less than when we started */
 757                 for (i = 1; i <= nline; i++)
 758                 {
 759                         ItemId          ii = PageGetItemId(phdr, i);
 760
 761                         Assert(ItemIdHasStorage(ii));
 762                         if (ItemIdGetOffset(ii) <= offset)
 763                                 ii->lp_off += size;
 764                 }
 765         }
 766 }
 767
 768
 769 /*
 770  * PageIndexMultiDelete
 771  *
 772  * This routine handles the case of deleting multiple tuples from an
 773  * index page at once.  It is considerably faster than a loop around
 774  * PageIndexTupleDelete ... however, the caller *must* supply the array
 775  * of item numbers to be deleted in item number order!
 776  */
 777 void
 778 PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems)
 779 {
 780         PageHeader      phdr = (PageHeader) page;
 781         Offset          pd_lower = phdr->pd_lower;
 782         Offset          pd_upper = phdr->pd_upper;
 783         Offset          pd_special = phdr->pd_special;
 784         itemIdSortData itemidbase[MaxIndexTuplesPerPage];
 785         itemIdSort      itemidptr;
 786         ItemId          lp;
 787         int                     nline,
 788                                 nused;
 789         int                     i;
 790         Size            totallen;
 791         Offset          upper;
 792         Size            size;
 793         unsigned        offset;
 794         int                     nextitm;
 795         OffsetNumber offnum;
 796
 797         Assert(nitems < MaxIndexTuplesPerPage);
 798
 799         /*
 800          * If there aren't very many items to delete, then retail
 801          * PageIndexTupleDelete is the best way.  Delete the items in reverse
 802          * order so we don't have to think about adjusting item numbers for
 803          * previous deletions.
 804          *
 805          * TODO: tune the magic number here
 806          */
 807         if (nitems <= 2)
 808         {
 809                 while (--nitems >= 0)
 810                         PageIndexTupleDelete(page, itemnos[nitems]);
 811                 return;
 812         }
 813
 814         /*
 815          * As with PageRepairFragmentation, paranoia seems justified.
 816          */
 817         if (pd_lower < SizeOfPageHeaderData ||
 818                 pd_lower > pd_upper ||
 819                 pd_upper > pd_special ||
 820                 pd_special > BLCKSZ ||
 821                 pd_special != MAXALIGN(pd_special))
 822                 ereport(ERROR,
 823                                 (errcode(ERRCODE_DATA_CORRUPTED),
 824                                  errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
 825                                                 pd_lower, pd_upper, pd_special)));
 826
 827         /*
 828          * Scan the item pointer array and build a list of just the ones we are
 829          * going to keep.  Notice we do not modify the page yet, since we are
 830          * still validity-checking.
 831          */
 832         nline = PageGetMaxOffsetNumber(page);
 833         itemidptr = itemidbase;
 834         totallen = 0;
 835         nused = 0;
 836         nextitm = 0;
 837         for (offnum = FirstOffsetNumber; offnum <= nline; offnum = OffsetNumberNext(offnum))
 838         {
 839                 lp = PageGetItemId(page, offnum);
 840                 Assert(ItemIdHasStorage(lp));
 841                 size = ItemIdGetLength(lp);
 842                 offset = ItemIdGetOffset(lp);
 843                 if (offset < pd_upper ||
 844                         (offset + size) > pd_special ||
 845                         offset != MAXALIGN(offset))
 846                         ereport(ERROR,
 847                                         (errcode(ERRCODE_DATA_CORRUPTED),
 848                                          errmsg("corrupted item pointer: offset = %u, size = %u",
 849                                                         offset, (unsigned int) size)));
 850
 851                 if (nextitm < nitems && offnum == itemnos[nextitm])
 852                 {
 853                         /* skip item to be deleted */
 854                         nextitm++;
 855                 }
 856                 else
 857                 {
 858                         itemidptr->offsetindex = nused;         /* where it will go */
 859                         itemidptr->itemoff = offset;
 860                         itemidptr->olditemid = *lp;
 861                         itemidptr->alignedlen = MAXALIGN(size);
 862                         totallen += itemidptr->alignedlen;
 863                         itemidptr++;
 864                         nused++;
 865                 }
 866         }
 867
 868         /* this will catch invalid or out-of-order itemnos[] */
 869         if (nextitm != nitems)
 870                 elog(ERROR, "incorrect index offsets supplied");
 871
 872         if (totallen > (Size) (pd_special - pd_lower))
 873                 ereport(ERROR,
 874                                 (errcode(ERRCODE_DATA_CORRUPTED),
 875                            errmsg("corrupted item lengths: total %u, available space %u",
 876                                           (unsigned int) totallen, pd_special - pd_lower)));
 877
 878         /* sort itemIdSortData array into decreasing itemoff order */
 879         qsort((char *) itemidbase, nused, sizeof(itemIdSortData),
 880                   itemoffcompare);
 881
 882         /* compactify page and install new itemids */
 883         upper = pd_special;
 884
 885         for (i = 0, itemidptr = itemidbase; i < nused; i++, itemidptr++)
 886         {
 887                 lp = PageGetItemId(page, itemidptr->offsetindex + 1);
 888                 upper -= itemidptr->alignedlen;
 889                 memmove((char *) page + upper,
 890                                 (char *) page + itemidptr->itemoff,
 891                                 itemidptr->alignedlen);
 892                 *lp = itemidptr->olditemid;
 893                 lp->lp_off = upper;
 894         }
 895
 896         phdr->pd_lower = SizeOfPageHeaderData + nused * sizeof(ItemIdData);
 897         phdr->pd_upper = upper;
 898 }
 899
 900 /*
 901  * PageIndexDeleteNoCompact
 902  *              Delete the given items for an index page, and defragment the resulting
 903  *              free space, but do not compact the item pointers array.
 904  *
 905  * itemnos is the array of tuples to delete; nitems is its size.  maxIdxTuples
 906  * is the maximum number of tuples that can exist in a page.
 907  *
 908  * Unused items at the end of the array are removed.
 909  *
 910  * This is used for index AMs that require that existing TIDs of live tuples
 911  * remain unchanged.
 912  */
 913 void
 914 PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos, int nitems)
 915 {
 916         PageHeader      phdr = (PageHeader) page;
 917         LocationIndex pd_lower = phdr->pd_lower;
 918         LocationIndex pd_upper = phdr->pd_upper;
 919         LocationIndex pd_special = phdr->pd_special;
 920         int                     nline;
 921         bool            empty;
 922         OffsetNumber offnum;
 923         int                     nextitm;
 924
 925         /*
 926          * As with PageRepairFragmentation, paranoia seems justified.
 927          */
 928         if (pd_lower < SizeOfPageHeaderData ||
 929                 pd_lower > pd_upper ||
 930                 pd_upper > pd_special ||
 931                 pd_special > BLCKSZ ||
 932                 pd_special != MAXALIGN(pd_special))
 933                 ereport(ERROR,
 934                                 (errcode(ERRCODE_DATA_CORRUPTED),
 935                                  errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
 936                                                 pd_lower, pd_upper, pd_special)));
 937
 938         /*
 939          * Scan the existing item pointer array and mark as unused those that are
 940          * in our kill-list; make sure any non-interesting ones are marked unused
 941          * as well.
 942          */
 943         nline = PageGetMaxOffsetNumber(page);
 944         empty = true;
 945         nextitm = 0;
 946         for (offnum = FirstOffsetNumber; offnum <= nline; offnum = OffsetNumberNext(offnum))
 947         {
 948                 ItemId          lp;
 949                 ItemLength      itemlen;
 950                 ItemOffset      offset;
 951
 952                 lp = PageGetItemId(page, offnum);
 953
 954                 itemlen = ItemIdGetLength(lp);
 955                 offset = ItemIdGetOffset(lp);
 956
 957                 if (ItemIdIsUsed(lp))
 958                 {
 959                         if (offset < pd_upper ||
 960                                 (offset + itemlen) > pd_special ||
 961                                 offset != MAXALIGN(offset))
 962                                 ereport(ERROR,
 963                                                 (errcode(ERRCODE_DATA_CORRUPTED),
 964                                                  errmsg("corrupted item pointer: offset = %u, length = %u",
 965                                                                 offset, (unsigned int) itemlen)));
 966
 967                         if (nextitm < nitems && offnum == itemnos[nextitm])
 968                         {
 969                                 /* this one is on our list to delete, so mark it unused */
 970                                 ItemIdSetUnused(lp);
 971                                 nextitm++;
 972                         }
 973                         else if (ItemIdHasStorage(lp))
 974                         {
 975                                 /* This one's live -- must do the compaction dance */
 976                                 empty = false;
 977                         }
 978                         else
 979                         {
 980                                 /* get rid of this one too */
 981                                 ItemIdSetUnused(lp);
 982                         }
 983                 }
 984         }
 985
 986         /* this will catch invalid or out-of-order itemnos[] */
 987         if (nextitm != nitems)
 988                 elog(ERROR, "incorrect index offsets supplied");
 989
 990         if (empty)
 991         {
 992                 /* Page is completely empty, so just reset it quickly */
 993                 phdr->pd_lower = SizeOfPageHeaderData;
 994                 phdr->pd_upper = pd_special;
 995         }
 996         else
 997         {
 998                 /* There are live items: need to compact the page the hard way */
 999                 itemIdSortData itemidbase[MaxOffsetNumber];
1000                 itemIdSort      itemidptr;
1001                 int                     i;
1002                 Size            totallen;
1003                 Offset          upper;
1004
1005                 /*
1006                  * Scan the page taking note of each item that we need to preserve.
1007                  * This includes both live items (those that contain data) and
1008                  * interspersed unused ones.  It's critical to preserve these unused
1009                  * items, because otherwise the offset numbers for later live items
1010                  * would change, which is not acceptable.  Unused items might get used
1011                  * again later; that is fine.
1012                  */
1013                 itemidptr = itemidbase;
1014                 totallen = 0;
1015                 for (i = 0; i < nline; i++, itemidptr++)
1016                 {
1017                         ItemId          lp;
1018
1019                         itemidptr->offsetindex = i;
1020
1021                         lp = PageGetItemId(page, i + 1);
1022                         if (ItemIdHasStorage(lp))
1023                         {
1024                                 itemidptr->itemoff = ItemIdGetOffset(lp);
1025                                 itemidptr->alignedlen = MAXALIGN(ItemIdGetLength(lp));
1026                                 totallen += itemidptr->alignedlen;
1027                         }
1028                         else
1029                         {
1030                                 itemidptr->itemoff = 0;
1031                                 itemidptr->alignedlen = 0;
1032                         }
1033                 }
1034                 /* By here, there are exactly nline elements in itemidbase array */
1035
1036                 if (totallen > (Size) (pd_special - pd_lower))
1037                         ereport(ERROR,
1038                                         (errcode(ERRCODE_DATA_CORRUPTED),
1039                                          errmsg("corrupted item lengths: total %u, available space %u",
1040                                                         (unsigned int) totallen, pd_special - pd_lower)));
1041
1042                 /* sort itemIdSortData array into decreasing itemoff order */
1043                 qsort((char *) itemidbase, nline, sizeof(itemIdSortData),
1044                           itemoffcompare);
1045
1046                 /*
1047                  * Defragment the data areas of each tuple, being careful to preserve
1048                  * each item's position in the linp array.
1049                  */
1050                 upper = pd_special;
1051                 PageClearHasFreeLinePointers(page);
1052                 for (i = 0, itemidptr = itemidbase; i < nline; i++, itemidptr++)
1053                 {
1054                         ItemId          lp;
1055
1056                         lp = PageGetItemId(page, itemidptr->offsetindex + 1);
1057                         if (itemidptr->alignedlen == 0)
1058                         {
1059                                 PageSetHasFreeLinePointers(page);
1060                                 ItemIdSetUnused(lp);
1061                                 continue;
1062                         }
1063                         upper -= itemidptr->alignedlen;
1064                         memmove((char *) page + upper,
1065                                         (char *) page + itemidptr->itemoff,
1066                                         itemidptr->alignedlen);
1067                         lp->lp_off = upper;
1068                         /* lp_flags and lp_len remain the same as originally */
1069                 }
1070
1071                 /* Set the new page limits */
1072                 phdr->pd_upper = upper;
1073                 phdr->pd_lower = SizeOfPageHeaderData + i * sizeof(ItemIdData);
1074         }
1075 }
1076
1077 /*
1078  * Set checksum for a page in shared buffers.
1079  *
1080  * If checksums are disabled, or if the page is not initialized, just return
1081  * the input.  Otherwise, we must make a copy of the page before calculating
1082  * the checksum, to prevent concurrent modifications (e.g. setting hint bits)
1083  * from making the final checksum invalid.  It doesn't matter if we include or
1084  * exclude hints during the copy, as long as we write a valid page and
1085  * associated checksum.
1086  *
1087  * Returns a pointer to the block-sized data that needs to be written. Uses
1088  * statically-allocated memory, so the caller must immediately write the
1089  * returned page and not refer to it again.
1090  */
1091 char *
1092 PageSetChecksumCopy(Page page, BlockNumber blkno)
1093 {
1094         static char *pageCopy = NULL;
1095
1096         /* If we don't need a checksum, just return the passed-in data */
1097         if (PageIsNew(page) || !DataChecksumsEnabled())
1098                 return (char *) page;
1099
1100         /*
1101          * We allocate the copy space once and use it over on each subsequent
1102          * call.  The point of palloc'ing here, rather than having a static char
1103          * array, is first to ensure adequate alignment for the checksumming code
1104          * and second to avoid wasting space in processes that never call this.
1105          */
1106         if (pageCopy == NULL)
1107                 pageCopy = MemoryContextAlloc(TopMemoryContext, BLCKSZ);
1108
1109         memcpy(pageCopy, (char *) page, BLCKSZ);
1110         ((PageHeader) pageCopy)->pd_checksum = pg_checksum_page(pageCopy, blkno);
1111         return pageCopy;
1112 }
1113
1114 /*
1115  * Set checksum for a page in private memory.
1116  *
1117  * This must only be used when we know that no other process can be modifying
1118  * the page buffer.
1119  */
1120 void
1121 PageSetChecksumInplace(Page page, BlockNumber blkno)
1122 {
1123         /* If we don't need a checksum, just return */
1124         if (PageIsNew(page) || !DataChecksumsEnabled())
1125                 return;
1126
1127         ((PageHeader) page)->pd_checksum = pg_checksum_page((char *) page, blkno);
1128 }