/*-------------------------------------------------------------------------

 *

 * nbtxlog.c

 *    WAL replay logic for btrees.

 *

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 * IDENTIFICATION

 *    src/backend/access/nbtree/nbtxlog.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "access/bufmask.h"

#include "access/nbtree.h"

#include "access/nbtxlog.h"

#include "access/transam.h"

#include "access/xlogutils.h"

#include "storage/standby.h"

#include "utils/memutils.h"

static MemoryContext opCtx;   /* working memory for operations */

/*

 * _bt_restore_page -- re-enter all the index tuples on a page

 *

 * The page is freshly init'd, and *from (length len) is a copy of what

 * had been its upper part (pd_upper to pd_special).  We assume that the

 * tuples had been added to the page in item-number order, and therefore

 * the one with highest item number appears first (lowest on the page).

 */

static void

_bt_restore_page(Page page, char *from, int len)

{

  IndexTupleData itupdata;

  Size    itemsz;

  char     *end = from + len;

  void     *items[MaxIndexTuplesPerPage];

  uint16    itemsizes[MaxIndexTuplesPerPage];

  int     i;

  int     nitems;

  /*

   * To get the items back in the original order, we add them to the page in

   * reverse.  To figure out where one tuple ends and another begins, we

   * have to scan them in forward order first.

   */

  i = 0;

  while (from < end)

  {

    /*

     * As we step through the items, 'from' won't always be properly

     * aligned, so we need to use memcpy().  Further, we use void * here

     * for our items array for the same reason; wouldn't want the compiler

     * or anyone thinking that an item is aligned when it isn't.

     */

    memcpy(&itupdata, from, sizeof(IndexTupleData));

    itemsz = IndexTupleSize(&itupdata);

    itemsz = MAXALIGN(itemsz);

    items[i] = from;

    itemsizes[i] = itemsz;

    i++;

    from += itemsz;

  }

  nitems = i;

  for (i = nitems - 1; i >= 0; i--)

  {

    if (PageAddItem(page, items[i], itemsizes[i], nitems - i, false, false) == InvalidOffsetNumber)

      elog(PANIC, "_bt_restore_page: cannot add item to page");

  }

}

static void

_bt_restore_meta(XLogReaderState *record, uint8 block_id)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  Buffer    metabuf;

  Page    metapg;

  BTMetaPageData *md;

  BTPageOpaque pageop;

  xl_btree_metadata *xlrec;

  char     *ptr;

  Size    len;

  metabuf = XLogInitBufferForRedo(record, block_id);

  ptr = XLogRecGetBlockData(record, block_id, &len);

  Assert(len == sizeof(xl_btree_metadata));

  Assert(BufferGetBlockNumber(metabuf) == BTREE_METAPAGE);

  xlrec = (xl_btree_metadata *) ptr;

  metapg = BufferGetPage(metabuf);

  _bt_pageinit(metapg, BufferGetPageSize(metabuf));

  md = BTPageGetMeta(metapg);

  md->btm_magic = BTREE_MAGIC;

  md->btm_version = xlrec->version;

  md->btm_root = xlrec->root;

  md->btm_level = xlrec->level;

  md->btm_fastroot = xlrec->fastroot;

  md->btm_fastlevel = xlrec->fastlevel;

  /* Cannot log BTREE_MIN_VERSION index metapage without upgrade */

  Assert(md->btm_version >= BTREE_NOVAC_VERSION);

  md->btm_last_cleanup_num_delpages = xlrec->last_cleanup_num_delpages;

  md->btm_last_cleanup_num_heap_tuples = -1.0;

  md->btm_allequalimage = xlrec->allequalimage;

  pageop = BTPageGetOpaque(metapg);

  pageop->btpo_flags = BTP_META;

  /*

   * Set pd_lower just past the end of the metadata.  This is essential,

   * because without doing so, metadata will be lost if xlog.c compresses

   * the page.

   */

  ((PageHeader) metapg)->pd_lower =

    ((char *) md + sizeof(BTMetaPageData)) - (char *) metapg;

  PageSetLSN(metapg, lsn);

  MarkBufferDirty(metabuf);

  UnlockReleaseBuffer(metabuf);

}

/*

 * _bt_clear_incomplete_split -- clear INCOMPLETE_SPLIT flag on a page

 *

 * This is a common subroutine of the redo functions of all the WAL record

 * types that can insert a downlink: insert, split, and newroot.

 */

static void

_bt_clear_incomplete_split(XLogReaderState *record, uint8 block_id)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  Buffer    buf;

  if (XLogReadBufferForRedo(record, block_id, &buf) == BLK_NEEDS_REDO)

  {

    Page    page = BufferGetPage(buf);

    BTPageOpaque pageop = BTPageGetOpaque(page);

    Assert(P_INCOMPLETE_SPLIT(pageop));

    pageop->btpo_flags &= ~BTP_INCOMPLETE_SPLIT;

    PageSetLSN(page, lsn);

    MarkBufferDirty(buf);

  }

  if (BufferIsValid(buf))

    UnlockReleaseBuffer(buf);

}

static void

btree_xlog_insert(bool isleaf, bool ismeta, bool posting,

          XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_insert *xlrec = (xl_btree_insert *) XLogRecGetData(record);

  Buffer    buffer;

  Page    page;

  /*

   * Insertion to an internal page finishes an incomplete split at the child

   * level.  Clear the incomplete-split flag in the child.  Note: during

   * normal operation, the child and parent pages are locked at the same

   * time (the locks are coupled), so that clearing the flag and inserting

   * the downlink appear atomic to other backends.  We don't bother with

   * that during replay, because readers don't care about the

   * incomplete-split flag and there cannot be updates happening.

   */

  if (!isleaf)

    _bt_clear_incomplete_split(record, 1);

  if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)

  {

    Size    datalen;

    char     *datapos = XLogRecGetBlockData(record, 0, &datalen);

    page = BufferGetPage(buffer);

    if (!posting)

    {

      /* Simple retail insertion */

      if (PageAddItem(page, datapos, datalen, xlrec->offnum, false, false) == InvalidOffsetNumber)

        elog(PANIC, "failed to add new item");

    }

    else

    {

      ItemId    itemid;

      IndexTuple  oposting,

            newitem,

            nposting;

      uint16    postingoff;

      /*

       * A posting list split occurred during leaf page insertion.  WAL

       * record data will start with an offset number representing the

       * point in an existing posting list that a split occurs at.

       *

       * Use _bt_swap_posting() to repeat posting list split steps from

       * primary.  Note that newitem from WAL record is 'orignewitem',

       * not the final version of newitem that is actually inserted on

       * page.

       */

      postingoff = *((uint16 *) datapos);

      datapos += sizeof(uint16);

      datalen -= sizeof(uint16);

      itemid = PageGetItemId(page, OffsetNumberPrev(xlrec->offnum));

      oposting = (IndexTuple) PageGetItem(page, itemid);

      /* Use mutable, aligned newitem copy in _bt_swap_posting() */

      Assert(isleaf && postingoff > 0);

      newitem = CopyIndexTuple((IndexTuple) datapos);

      nposting = _bt_swap_posting(newitem, oposting, postingoff);

      /* Replace existing posting list with post-split version */

      memcpy(oposting, nposting, MAXALIGN(IndexTupleSize(nposting)));

      /* Insert "final" new item (not orignewitem from WAL stream) */

      Assert(IndexTupleSize(newitem) == datalen);

      if (PageAddItem(page, newitem, datalen, xlrec->offnum, false, false) == InvalidOffsetNumber)

        elog(PANIC, "failed to add posting split new item");

    }

    PageSetLSN(page, lsn);

    MarkBufferDirty(buffer);

  }

  if (BufferIsValid(buffer))

    UnlockReleaseBuffer(buffer);

  /*

   * Note: in normal operation, we'd update the metapage while still holding

   * lock on the page we inserted into.  But during replay it's not

   * necessary to hold that lock, since no other index updates can be

   * happening concurrently, and readers will cope fine with following an

   * obsolete link from the metapage.

   */

  if (ismeta)

    _bt_restore_meta(record, 2);

}

static void

btree_xlog_split(bool newitemonleft, XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_split *xlrec = (xl_btree_split *) XLogRecGetData(record);

  bool    isleaf = (xlrec->level == 0);

  Buffer    buf;

  Buffer    rbuf;

  Page    rpage;

  BTPageOpaque ropaque;

  char     *datapos;

  Size    datalen;

  BlockNumber origpagenumber;

  BlockNumber rightpagenumber;

  BlockNumber spagenumber;

  XLogRecGetBlockTag(record, 0, NULL, NULL, &origpagenumber);

  XLogRecGetBlockTag(record, 1, NULL, NULL, &rightpagenumber);

  if (!XLogRecGetBlockTagExtended(record, 2, NULL, NULL, &spagenumber, NULL))

    spagenumber = P_NONE;

  /*

   * Clear the incomplete split flag on the appropriate child page one level

   * down when origpage/buf is an internal page (there must have been

   * cascading page splits during original execution in the event of an

   * internal page split).  This is like the corresponding btree_xlog_insert

   * call for internal pages.  We're not clearing the incomplete split flag

   * for the current page split here (you can think of this as part of the

   * insert of newitem that the page split action needs to perform in

   * passing).

   *

   * Like in btree_xlog_insert, this can be done before locking other pages.

   * We never need to couple cross-level locks in REDO routines.

   */

  if (!isleaf)

    _bt_clear_incomplete_split(record, 3);

  /* Reconstruct right (new) sibling page from scratch */

  rbuf = XLogInitBufferForRedo(record, 1);

  datapos = XLogRecGetBlockData(record, 1, &datalen);

  rpage = BufferGetPage(rbuf);

  _bt_pageinit(rpage, BufferGetPageSize(rbuf));

  ropaque = BTPageGetOpaque(rpage);

  ropaque->btpo_prev = origpagenumber;

  ropaque->btpo_next = spagenumber;

  ropaque->btpo_level = xlrec->level;

  ropaque->btpo_flags = isleaf ? BTP_LEAF : 0;

  ropaque->btpo_cycleid = 0;

  _bt_restore_page(rpage, datapos, datalen);

  PageSetLSN(rpage, lsn);

  MarkBufferDirty(rbuf);

  /* Now reconstruct original page (left half of split) */

  if (XLogReadBufferForRedo(record, 0, &buf) == BLK_NEEDS_REDO)

  {

    /*

     * To retain the same physical order of the tuples that they had, we

     * initialize a temporary empty page for the left page and add all the

     * items to that in item number order.  This mirrors how _bt_split()

     * works.  Retaining the same physical order makes WAL consistency

     * checking possible.  See also _bt_restore_page(), which does the

     * same for the right page.

     */

    Page    origpage = BufferGetPage(buf);

    BTPageOpaque oopaque = BTPageGetOpaque(origpage);

    OffsetNumber off;

    IndexTuple  newitem = NULL,

          left_hikey = NULL,

          nposting = NULL;

    Size    newitemsz = 0,

          left_hikeysz = 0;

    Page    leftpage;

    OffsetNumber leftoff,

          replacepostingoff = InvalidOffsetNumber;

    datapos = XLogRecGetBlockData(record, 0, &datalen);

    if (newitemonleft || xlrec->postingoff != 0)

    {

      newitem = (IndexTuple) datapos;

      newitemsz = MAXALIGN(IndexTupleSize(newitem));

      datapos += newitemsz;

      datalen -= newitemsz;

      if (xlrec->postingoff != 0)

      {

        ItemId    itemid;

        IndexTuple  oposting;

        /* Posting list must be at offset number before new item's */

        replacepostingoff = OffsetNumberPrev(xlrec->newitemoff);

        /* Use mutable, aligned newitem copy in _bt_swap_posting() */

        newitem = CopyIndexTuple(newitem);

        itemid = PageGetItemId(origpage, replacepostingoff);

        oposting = (IndexTuple) PageGetItem(origpage, itemid);

        nposting = _bt_swap_posting(newitem, oposting,

                      xlrec->postingoff);

      }

    }

    /*

     * Extract left hikey and its size.  We assume that 16-bit alignment

     * is enough to apply IndexTupleSize (since it's fetching from a

     * uint16 field).

     */

    left_hikey = (IndexTuple) datapos;

    left_hikeysz = MAXALIGN(IndexTupleSize(left_hikey));

    datapos += left_hikeysz;

    datalen -= left_hikeysz;

    Assert(datalen == 0);

    leftpage = PageGetTempPageCopySpecial(origpage);

    /* Add high key tuple from WAL record to temp page */

    leftoff = P_HIKEY;

    if (PageAddItem(leftpage, left_hikey, left_hikeysz, P_HIKEY, false, false) == InvalidOffsetNumber)

      elog(ERROR, "failed to add high key to left page after split");

    leftoff = OffsetNumberNext(leftoff);

    for (off = P_FIRSTDATAKEY(oopaque); off < xlrec->firstrightoff; off++)

    {

      ItemId    itemid;

      Size    itemsz;

      IndexTuple  item;

      /* Add replacement posting list when required */

      if (off == replacepostingoff)

      {

        Assert(newitemonleft ||

             xlrec->firstrightoff == xlrec->newitemoff);

        if (PageAddItem(leftpage, nposting, MAXALIGN(IndexTupleSize(nposting)), leftoff, false, false) == InvalidOffsetNumber)

          elog(ERROR, "failed to add new posting list item to left page after split");

        leftoff = OffsetNumberNext(leftoff);

        continue;   /* don't insert oposting */

      }

      /* add the new item if it was inserted on left page */

      else if (newitemonleft && off == xlrec->newitemoff)

      {

        if (PageAddItem(leftpage, newitem, newitemsz, leftoff, false, false) == InvalidOffsetNumber)

          elog(ERROR, "failed to add new item to left page after split");

        leftoff = OffsetNumberNext(leftoff);

      }

      itemid = PageGetItemId(origpage, off);

      itemsz = ItemIdGetLength(itemid);

      item = (IndexTuple) PageGetItem(origpage, itemid);

      if (PageAddItem(leftpage, item, itemsz, leftoff, false, false) == InvalidOffsetNumber)

        elog(ERROR, "failed to add old item to left page after split");

      leftoff = OffsetNumberNext(leftoff);

    }

    /* cope with possibility that newitem goes at the end */

    if (newitemonleft && off == xlrec->newitemoff)

    {

      if (PageAddItem(leftpage, newitem, newitemsz, leftoff, false, false) == InvalidOffsetNumber)

        elog(ERROR, "failed to add new item to left page after split");

      leftoff = OffsetNumberNext(leftoff);

    }

    PageRestoreTempPage(leftpage, origpage);

    /* Fix opaque fields */

    oopaque->btpo_flags = BTP_INCOMPLETE_SPLIT;

    if (isleaf)

      oopaque->btpo_flags |= BTP_LEAF;

    oopaque->btpo_next = rightpagenumber;

    oopaque->btpo_cycleid = 0;

    PageSetLSN(origpage, lsn);

    MarkBufferDirty(buf);

  }

  /* Fix left-link of the page to the right of the new right sibling */

  if (spagenumber != P_NONE)

  {

    Buffer    sbuf;

    if (XLogReadBufferForRedo(record, 2, &sbuf) == BLK_NEEDS_REDO)

    {

      Page    spage = BufferGetPage(sbuf);

      BTPageOpaque spageop = BTPageGetOpaque(spage);

      spageop->btpo_prev = rightpagenumber;

      PageSetLSN(spage, lsn);

      MarkBufferDirty(sbuf);

    }

    if (BufferIsValid(sbuf))

      UnlockReleaseBuffer(sbuf);

  }

  /*

   * Finally, release the remaining buffers.  sbuf, rbuf, and buf must be

   * released together, so that readers cannot observe inconsistencies.

   */

  UnlockReleaseBuffer(rbuf);

  if (BufferIsValid(buf))

    UnlockReleaseBuffer(buf);

}

static void

btree_xlog_dedup(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_dedup *xlrec = (xl_btree_dedup *) XLogRecGetData(record);

  Buffer    buf;

  if (XLogReadBufferForRedo(record, 0, &buf) == BLK_NEEDS_REDO)

  {

    char     *ptr = XLogRecGetBlockData(record, 0, NULL);

    Page    page = BufferGetPage(buf);

    BTPageOpaque opaque = BTPageGetOpaque(page);

    OffsetNumber offnum,

          minoff,

          maxoff;

    BTDedupState state;

    BTDedupInterval *intervals;

    Page    newpage;

    state = palloc_object(BTDedupStateData);

    state->deduplicate = true; /* unused */

    state->nmaxitems = 0; /* unused */

    /* Conservatively use larger maxpostingsize than primary */

    state->maxpostingsize = BTMaxItemSize;

    state->base = NULL;

    state->baseoff = InvalidOffsetNumber;

    state->basetupsize = 0;

    state->htids = palloc(state->maxpostingsize);

    state->nhtids = 0;

    state->nitems = 0;

    state->phystupsize = 0;

    state->nintervals = 0;

    minoff = P_FIRSTDATAKEY(opaque);

    maxoff = PageGetMaxOffsetNumber(page);

    newpage = PageGetTempPageCopySpecial(page);

    if (!P_RIGHTMOST(opaque))

    {

      ItemId    itemid = PageGetItemId(page, P_HIKEY);

      Size    itemsz = ItemIdGetLength(itemid);

      IndexTuple  item = (IndexTuple) PageGetItem(page, itemid);

      if (PageAddItem(newpage, item, itemsz, P_HIKEY, false, false) == InvalidOffsetNumber)

        elog(ERROR, "deduplication failed to add highkey");

    }

    intervals = (BTDedupInterval *) ptr;

    for (offnum = minoff;

       offnum <= maxoff;

       offnum = OffsetNumberNext(offnum))

    {

      ItemId    itemid = PageGetItemId(page, offnum);

      IndexTuple  itup = (IndexTuple) PageGetItem(page, itemid);

      if (offnum == minoff)

        _bt_dedup_start_pending(state, itup, offnum);

      else if (state->nintervals < xlrec->nintervals &&

           state->baseoff == intervals[state->nintervals].baseoff &&

           state->nitems < intervals[state->nintervals].nitems)

      {

        if (!_bt_dedup_save_htid(state, itup))

          elog(ERROR, "deduplication failed to add heap tid to pending posting list");

      }

      else

      {

        _bt_dedup_finish_pending(newpage, state);

        _bt_dedup_start_pending(state, itup, offnum);

      }

    }

    _bt_dedup_finish_pending(newpage, state);

    Assert(state->nintervals == xlrec->nintervals);

    Assert(memcmp(state->intervals, intervals,

            state->nintervals * sizeof(BTDedupInterval)) == 0);

    if (P_HAS_GARBAGE(opaque))

    {

      BTPageOpaque nopaque = BTPageGetOpaque(newpage);

      nopaque->btpo_flags &= ~BTP_HAS_GARBAGE;

    }

    PageRestoreTempPage(newpage, page);

    PageSetLSN(page, lsn);

    MarkBufferDirty(buf);

  }

  if (BufferIsValid(buf))

    UnlockReleaseBuffer(buf);

}

static void

btree_xlog_updates(Page page, OffsetNumber *updatedoffsets,

           xl_btree_update *updates, int nupdated)

{

  BTVacuumPosting vacposting;

  IndexTuple  origtuple;

  ItemId    itemid;

  Size    itemsz;

  for (int i = 0; i < nupdated; i++)

  {

    itemid = PageGetItemId(page, updatedoffsets[i]);

    origtuple = (IndexTuple) PageGetItem(page, itemid);

    vacposting = palloc(offsetof(BTVacuumPostingData, deletetids) +

              updates->ndeletedtids * sizeof(uint16));

    vacposting->updatedoffset = updatedoffsets[i];

    vacposting->itup = origtuple;

    vacposting->ndeletedtids = updates->ndeletedtids;

    memcpy(vacposting->deletetids,

         (char *) updates + SizeOfBtreeUpdate,

         updates->ndeletedtids * sizeof(uint16));

    _bt_update_posting(vacposting);

    /* Overwrite updated version of tuple */

    itemsz = MAXALIGN(IndexTupleSize(vacposting->itup));

    if (!PageIndexTupleOverwrite(page, updatedoffsets[i], vacposting->itup, itemsz))

      elog(PANIC, "failed to update partially dead item");

    pfree(vacposting->itup);

    pfree(vacposting);

    /* advance to next xl_btree_update from array */

    updates = (xl_btree_update *)

      ((char *) updates + SizeOfBtreeUpdate +

       updates->ndeletedtids * sizeof(uint16));

  }

}

static void

btree_xlog_vacuum(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_vacuum *xlrec = (xl_btree_vacuum *) XLogRecGetData(record);

  Buffer    buffer;

  Page    page;

  BTPageOpaque opaque;

  /*

   * We need to take a cleanup lock here, just like btvacuumpage(). However,

   * it isn't necessary to exhaustively get a cleanup lock on every block in

   * the index during recovery (just getting a cleanup lock on pages with

   * items to kill suffices).  See nbtree/README for details.

   */

  if (XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &buffer)

    == BLK_NEEDS_REDO)

  {

    char     *ptr = XLogRecGetBlockData(record, 0, NULL);

    page = BufferGetPage(buffer);

    if (xlrec->nupdated > 0)

    {

      OffsetNumber *updatedoffsets;

      xl_btree_update *updates;

      updatedoffsets = (OffsetNumber *)

        (ptr + xlrec->ndeleted * sizeof(OffsetNumber));

      updates = (xl_btree_update *) ((char *) updatedoffsets +

                       xlrec->nupdated *

                       sizeof(OffsetNumber));

      btree_xlog_updates(page, updatedoffsets, updates, xlrec->nupdated);

    }

    if (xlrec->ndeleted > 0)

      PageIndexMultiDelete(page, (OffsetNumber *) ptr, xlrec->ndeleted);

    /*

     * Clear the vacuum cycle ID, and mark the page as not containing any

     * LP_DEAD items

     */

    opaque = BTPageGetOpaque(page);

    opaque->btpo_cycleid = 0;

    opaque->btpo_flags &= ~BTP_HAS_GARBAGE;

    PageSetLSN(page, lsn);

    MarkBufferDirty(buffer);

  }

  if (BufferIsValid(buffer))

    UnlockReleaseBuffer(buffer);

}

static void

btree_xlog_delete(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_delete *xlrec = (xl_btree_delete *) XLogRecGetData(record);

  Buffer    buffer;

  Page    page;

  BTPageOpaque opaque;

  /*

   * If we have any conflict processing to do, it must happen before we

   * update the page

   */

  if (InHotStandby)

  {

    RelFileLocator rlocator;

    XLogRecGetBlockTag(record, 0, &rlocator, NULL, NULL);

    ResolveRecoveryConflictWithSnapshot(xlrec->snapshotConflictHorizon,

                      xlrec->isCatalogRel,

                      rlocator);

  }

  /*

   * We don't need to take a cleanup lock to apply these changes. See

   * nbtree/README for details.

   */

  if (XLogReadBufferForRedo(record, 0, &buffer) == BLK_NEEDS_REDO)

  {

    char     *ptr = XLogRecGetBlockData(record, 0, NULL);

    page = BufferGetPage(buffer);

    if (xlrec->nupdated > 0)

    {

      OffsetNumber *updatedoffsets;

      xl_btree_update *updates;

      updatedoffsets = (OffsetNumber *)

        (ptr + xlrec->ndeleted * sizeof(OffsetNumber));

      updates = (xl_btree_update *) ((char *) updatedoffsets +

                       xlrec->nupdated *

                       sizeof(OffsetNumber));

      btree_xlog_updates(page, updatedoffsets, updates, xlrec->nupdated);

    }

    if (xlrec->ndeleted > 0)

      PageIndexMultiDelete(page, (OffsetNumber *) ptr, xlrec->ndeleted);

    /*

     * Do *not* clear the vacuum cycle ID, but do mark the page as not

     * containing any LP_DEAD items

     */

    opaque = BTPageGetOpaque(page);

    opaque->btpo_flags &= ~BTP_HAS_GARBAGE;

    PageSetLSN(page, lsn);

    MarkBufferDirty(buffer);

  }

  if (BufferIsValid(buffer))

    UnlockReleaseBuffer(buffer);

}

static void

btree_xlog_mark_page_halfdead(uint8 info, XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_mark_page_halfdead *xlrec = (xl_btree_mark_page_halfdead *) XLogRecGetData(record);

  Buffer    buffer;

  Page    page;

  BTPageOpaque pageop;

  IndexTupleData trunctuple;

  /*

   * In normal operation, we would lock all the pages this WAL record

   * touches before changing any of them.  In WAL replay, it should be okay

   * to lock just one page at a time, since no concurrent index updates can

   * be happening, and readers should not care whether they arrive at the

   * target page or not (since it's surely empty).

   */

  /* to-be-deleted subtree's parent page */

  if (XLogReadBufferForRedo(record, 1, &buffer) == BLK_NEEDS_REDO)

  {

    OffsetNumber poffset;

    ItemId    itemid;

    IndexTuple  itup;

    OffsetNumber nextoffset;

    BlockNumber rightsib;

    page = BufferGetPage(buffer);

    pageop = BTPageGetOpaque(page);

    poffset = xlrec->poffset;

    nextoffset = OffsetNumberNext(poffset);

    itemid = PageGetItemId(page, nextoffset);

    itup = (IndexTuple) PageGetItem(page, itemid);

    rightsib = BTreeTupleGetDownLink(itup);

    itemid = PageGetItemId(page, poffset);

    itup = (IndexTuple) PageGetItem(page, itemid);

    BTreeTupleSetDownLink(itup, rightsib);

    nextoffset = OffsetNumberNext(poffset);

    PageIndexTupleDelete(page, nextoffset);

    PageSetLSN(page, lsn);

    MarkBufferDirty(buffer);

  }

  /*

   * Don't need to couple cross-level locks in REDO routines, so release

   * lock on internal page immediately

   */

  if (BufferIsValid(buffer))

    UnlockReleaseBuffer(buffer);

  /* Rewrite the leaf page as a halfdead page */

  buffer = XLogInitBufferForRedo(record, 0);

  page = BufferGetPage(buffer);

  _bt_pageinit(page, BufferGetPageSize(buffer));

  pageop = BTPageGetOpaque(page);

  pageop->btpo_prev = xlrec->leftblk;

  pageop->btpo_next = xlrec->rightblk;

  pageop->btpo_level = 0;

  pageop->btpo_flags = BTP_HALF_DEAD | BTP_LEAF;

  pageop->btpo_cycleid = 0;

  /*

   * Construct a dummy high key item that points to top parent page (value

   * is InvalidBlockNumber when the top parent page is the leaf page itself)

   */

  MemSet(&trunctuple, 0, sizeof(IndexTupleData));

  trunctuple.t_info = sizeof(IndexTupleData);

  BTreeTupleSetTopParent(&trunctuple, xlrec->topparent);

  if (PageAddItem(page, &trunctuple, sizeof(IndexTupleData), P_HIKEY, false, false) == InvalidOffsetNumber)

    elog(ERROR, "could not add dummy high key to half-dead page");

  PageSetLSN(page, lsn);

  MarkBufferDirty(buffer);

  UnlockReleaseBuffer(buffer);

}

static void

btree_xlog_unlink_page(uint8 info, XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_btree_unlink_page *xlrec = (xl_btree_unlink_page *) XLogRecGetData(record);

  BlockNumber leftsib;

  BlockNumber rightsib;

  uint32    level;

  bool    isleaf;

  FullTransactionId safexid;

  Buffer    leftbuf;

  Buffer    target;

  Buffer    rightbuf;

  Page    page;

  BTPageOpaque pageop;

  leftsib = xlrec->leftsib;

  rightsib = xlrec->rightsib;

  level = xlrec->level;

  isleaf = (level == 0);

  safexid = xlrec->safexid;

  /* No leaftopparent for level 0 (leaf page) or level 1 target */

  Assert(!BlockNumberIsValid(xlrec->leaftopparent) || level > 1);

  /*

   * In normal operation, we would lock all the pages this WAL record

   * touches before changing any of them.  In WAL replay, we at least lock

   * the pages in the same standard left-to-right order (leftsib, target,

   * rightsib), and don't release the sibling locks until the target is

   * marked deleted.

   */

  /* Fix right-link of left sibling, if any */

  if (leftsib != P_NONE)

  {

    if (XLogReadBufferForRedo(record, 1, &leftbuf) == BLK_NEEDS_REDO)

    {

      page = BufferGetPage(leftbuf);

      pageop = BTPageGetOpaque(page);

      pageop->btpo_next = rightsib;

      PageSetLSN(page, lsn);

      MarkBufferDirty(leftbuf);

    }

  }

  else

    leftbuf = InvalidBuffer;

  /* Rewrite target page as empty deleted page */

  target = XLogInitBufferForRedo(record, 0);

  page = BufferGetPage(target);

  _bt_pageinit(page, BufferGetPageSize(target));

  pageop = BTPageGetOpaque(page);

  pageop->btpo_prev = leftsib;

  pageop->btpo_next = rightsib;

  pageop->btpo_level = level;

  BTPageSetDeleted(page, safexid);

  if (isleaf)

    pageop->btpo_flags |= BTP_LEAF;

  pageop->btpo_cycleid = 0;

  PageSetLSN(page, lsn);

  MarkBufferDirty(target);

  /* Fix left-link of right sibling */

  if (XLogReadBufferForRedo(record, 2, &rightbuf) == BLK_NEEDS_REDO)

  {

    page = BufferGetPage(rightbuf);

    pageop = BTPageGetOpaque(page);

    pageop->btpo_prev = leftsib;

    PageSetLSN(page, lsn);

    MarkBufferDirty(rightbuf);

  }

  /* Release siblings */

  if (BufferIsValid(leftbuf))

    UnlockReleaseBuffer(leftbuf);

  if (BufferIsValid(rightbuf))

    UnlockReleaseBuffer(rightbuf);

  /* Release target */

  UnlockReleaseBuffer(target);

  /*

   * If we deleted a parent of the targeted leaf page, instead of the leaf

   * itself, update the leaf to point to the next remaining child in the

   * to-be-deleted subtree

   */

  if (XLogRecHasBlockRef(record, 3))

  {

    /*

     * There is no real data on the page, so we just re-create it from

     * scratch using the information from the WAL record.

     *

     * Note that we don't end up here when the target page is also the

     * leafbuf page.  There is no need to add a dummy hikey item with a

     * top parent link when deleting leafbuf because it's the last page

     * we'll delete in the subtree undergoing deletion.

     */

    Buffer    leafbuf;

    IndexTupleData trunctuple;

    Assert(!isleaf);

    leafbuf = XLogInitBufferForRedo(record, 3);

    page = BufferGetPage(leafbuf);

    _bt_pageinit(page, BufferGetPageSize(leafbuf));

    pageop = BTPageGetOpaque(page);

    pageop->btpo_flags = BTP_HALF_DEAD | BTP_LEAF;

    pageop->btpo_prev = xlrec->leafleftsib;