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

 *

 * hash_xlog.c

 *    WAL replay logic for hash index.

 *

 *

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

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

 *

 * IDENTIFICATION

 *    src/backend/access/hash/hash_xlog.c

 *

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

 */

#include "postgres.h"

#include "access/bufmask.h"

#include "access/hash.h"

#include "access/hash_xlog.h"

#include "access/xlogutils.h"

#include "storage/standby.h"

/*

 * replay a hash index meta page

 */

static void

hash_xlog_init_meta_page(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  Page    page;

  Buffer    metabuf;

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

  /* create the index' metapage */

  metabuf = XLogInitBufferForRedo(record, 0);

  Assert(BufferIsValid(metabuf));

  _hash_init_metabuffer(metabuf, xlrec->num_tuples, xlrec->procid,

              xlrec->ffactor, true);

  page = BufferGetPage(metabuf);

  PageSetLSN(page, lsn);

  MarkBufferDirty(metabuf);

  XLogFlushBufferForRedoIfInit(record, 0, metabuf);

  /* all done */

  UnlockReleaseBuffer(metabuf);

}

/*

 * replay a hash index bitmap page

 */

static void

hash_xlog_init_bitmap_page(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  Buffer    bitmapbuf;

  Buffer    metabuf;

  Page    page;

  HashMetaPage metap;

  uint32    num_buckets;

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

  /*

   * Initialize bitmap page

   */

  bitmapbuf = XLogInitBufferForRedo(record, 0);

  _hash_initbitmapbuffer(bitmapbuf, xlrec->bmsize, true);

  PageSetLSN(BufferGetPage(bitmapbuf), lsn);

  MarkBufferDirty(bitmapbuf);

  XLogFlushBufferForRedoIfInit(record, 0, bitmapbuf);

  UnlockReleaseBuffer(bitmapbuf);

  /* add the new bitmap page to the metapage's list of bitmaps */

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

  {

    /*

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

     * holding lock on the bitmap page.  But during replay it's not

     * necessary to hold that lock, since nobody can see it yet; the

     * creating transaction hasn't yet committed.

     */

    page = BufferGetPage(metabuf);

    metap = HashPageGetMeta(page);

    num_buckets = metap->hashm_maxbucket + 1;

    metap->hashm_mapp[metap->hashm_nmaps] = num_buckets + 1;

    metap->hashm_nmaps++;

    PageSetLSN(page, lsn);

    MarkBufferDirty(metabuf);

    XLogFlushBufferForRedoIfInit(record, 1, metabuf);

  }

  if (BufferIsValid(metabuf))

    UnlockReleaseBuffer(metabuf);

}

/*

 * replay a hash index insert without split

 */

static void

hash_xlog_insert(XLogReaderState *record)

{

  HashMetaPage metap;

  XLogRecPtr  lsn = record->EndRecPtr;

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

  Buffer    buffer;

  Page    page;

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

  {

    Size    datalen;

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

    page = BufferGetPage(buffer);

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

      elog(PANIC, "hash_xlog_insert: failed to add item");

    PageSetLSN(page, lsn);

    MarkBufferDirty(buffer);

  }

  if (BufferIsValid(buffer))

    UnlockReleaseBuffer(buffer);

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

  {

    /*

     * 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.

     */

    page = BufferGetPage(buffer);

    metap = HashPageGetMeta(page);

    metap->hashm_ntuples += 1;

    PageSetLSN(page, lsn);

    MarkBufferDirty(buffer);

  }

  if (BufferIsValid(buffer))

    UnlockReleaseBuffer(buffer);

}

/*

 * replay addition of overflow page for hash index

 */

static void

hash_xlog_add_ovfl_page(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

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

  Buffer    leftbuf;

  Buffer    ovflbuf;

  Buffer    metabuf;

  BlockNumber leftblk;

  BlockNumber rightblk;

  BlockNumber newmapblk = InvalidBlockNumber;

  Page    ovflpage;

  HashPageOpaque ovflopaque;

  uint32     *num_bucket;

  char     *data;

  Size    datalen PG_USED_FOR_ASSERTS_ONLY;

  bool    new_bmpage = false;

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

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

  ovflbuf = XLogInitBufferForRedo(record, 0);

  Assert(BufferIsValid(ovflbuf));

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

  num_bucket = (uint32 *) data;

  Assert(datalen == sizeof(uint32));

  _hash_initbuf(ovflbuf, InvalidBlockNumber, *num_bucket, LH_OVERFLOW_PAGE,

          true);

  /* update backlink */

  ovflpage = BufferGetPage(ovflbuf);

  ovflopaque = HashPageGetOpaque(ovflpage);

  ovflopaque->hasho_prevblkno = leftblk;

  PageSetLSN(ovflpage, lsn);

  MarkBufferDirty(ovflbuf);

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

  {

    Page    leftpage;

    HashPageOpaque leftopaque;

    leftpage = BufferGetPage(leftbuf);

    leftopaque = HashPageGetOpaque(leftpage);

    leftopaque->hasho_nextblkno = rightblk;

    PageSetLSN(leftpage, lsn);

    MarkBufferDirty(leftbuf);

  }

  if (BufferIsValid(leftbuf))

    UnlockReleaseBuffer(leftbuf);

  UnlockReleaseBuffer(ovflbuf);

  /*

   * Note: in normal operation, we'd update the bitmap and meta page while

   * still holding lock on the overflow pages.  But during replay it's not

   * necessary to hold those locks, since no other index updates can be

   * happening concurrently.

   */

  if (XLogRecHasBlockRef(record, 2))

  {

    Buffer    mapbuffer;

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

    {

      Page    mappage = BufferGetPage(mapbuffer);

      uint32     *freep = NULL;

      uint32     *bitmap_page_bit;

      freep = HashPageGetBitmap(mappage);

      data = XLogRecGetBlockData(record, 2, &datalen);

      bitmap_page_bit = (uint32 *) data;

      SETBIT(freep, *bitmap_page_bit);

      PageSetLSN(mappage, lsn);

      MarkBufferDirty(mapbuffer);

    }

    if (BufferIsValid(mapbuffer))

      UnlockReleaseBuffer(mapbuffer);

  }

  if (XLogRecHasBlockRef(record, 3))

  {

    Buffer    newmapbuf;

    newmapbuf = XLogInitBufferForRedo(record, 3);

    _hash_initbitmapbuffer(newmapbuf, xlrec->bmsize, true);

    new_bmpage = true;

    newmapblk = BufferGetBlockNumber(newmapbuf);

    MarkBufferDirty(newmapbuf);

    PageSetLSN(BufferGetPage(newmapbuf), lsn);

    UnlockReleaseBuffer(newmapbuf);

  }

  if (XLogReadBufferForRedo(record, 4, &metabuf) == BLK_NEEDS_REDO)

  {

    HashMetaPage metap;

    Page    page;

    uint32     *firstfree_ovflpage;

    data = XLogRecGetBlockData(record, 4, &datalen);

    firstfree_ovflpage = (uint32 *) data;

    page = BufferGetPage(metabuf);

    metap = HashPageGetMeta(page);

    metap->hashm_firstfree = *firstfree_ovflpage;

    if (!xlrec->bmpage_found)

    {

      metap->hashm_spares[metap->hashm_ovflpoint]++;

      if (new_bmpage)

      {

        Assert(BlockNumberIsValid(newmapblk));

        metap->hashm_mapp[metap->hashm_nmaps] = newmapblk;

        metap->hashm_nmaps++;

        metap->hashm_spares[metap->hashm_ovflpoint]++;

      }

    }

    PageSetLSN(page, lsn);

    MarkBufferDirty(metabuf);

  }

  if (BufferIsValid(metabuf))

    UnlockReleaseBuffer(metabuf);

}

/*

 * replay allocation of page for split operation

 */

static void

hash_xlog_split_allocate_page(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

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

  Buffer    oldbuf;

  Buffer    newbuf;

  Buffer    metabuf;

  XLogRedoAction action;

  /*

   * To be consistent with normal operation, here we take cleanup locks on

   * both the old and new buckets even though there can't be any concurrent

   * inserts.

   */

  /* replay the record for old bucket */

  action = XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &oldbuf);

  /*

   * Note that we still update the page even if it was restored from a full

   * page image, because the special space is not included in the image.

   */

  if (action == BLK_NEEDS_REDO || action == BLK_RESTORED)

  {

    Page    oldpage;

    HashPageOpaque oldopaque;

    oldpage = BufferGetPage(oldbuf);

    oldopaque = HashPageGetOpaque(oldpage);

    oldopaque->hasho_flag = xlrec->old_bucket_flag;

    oldopaque->hasho_prevblkno = xlrec->new_bucket;

    PageSetLSN(oldpage, lsn);

    MarkBufferDirty(oldbuf);

  }

  /* replay the record for new bucket */

  XLogReadBufferForRedoExtended(record, 1, RBM_ZERO_AND_CLEANUP_LOCK, true,

                  &newbuf);

  _hash_initbuf(newbuf, xlrec->new_bucket, xlrec->new_bucket,

          xlrec->new_bucket_flag, true);

  MarkBufferDirty(newbuf);

  PageSetLSN(BufferGetPage(newbuf), lsn);

  /*

   * We can release the lock on old bucket early as well but doing here to

   * consistent with normal operation.

   */

  if (BufferIsValid(oldbuf))

    UnlockReleaseBuffer(oldbuf);

  if (BufferIsValid(newbuf))

    UnlockReleaseBuffer(newbuf);

  /*

   * Note: in normal operation, we'd update the meta page while still

   * holding lock on the old and new bucket pages.  But during replay it's

   * not necessary to hold those locks, since no other bucket splits can be

   * happening concurrently.

   */

  /* replay the record for metapage changes */

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

  {

    Page    page;

    HashMetaPage metap;

    Size    datalen;

    char     *data;

    uint32     *uidata;

    int     uidatacount;

    page = BufferGetPage(metabuf);

    metap = HashPageGetMeta(page);

    metap->hashm_maxbucket = xlrec->new_bucket;

    data = XLogRecGetBlockData(record, 2, &datalen);

    /*

     * This cast is ok because XLogRecGetBlockData() returns a MAXALIGNed

     * buffer.

     */

    uidata = (uint32 *) data;

    uidatacount = 0;

    if (xlrec->flags & XLH_SPLIT_META_UPDATE_MASKS)

    {

      uint32    lowmask = uidata[uidatacount++];

      uint32    highmask = uidata[uidatacount++];

      /* update metapage */

      metap->hashm_lowmask = lowmask;

      metap->hashm_highmask = highmask;

    }

    if (xlrec->flags & XLH_SPLIT_META_UPDATE_SPLITPOINT)

    {

      uint32    ovflpoint = uidata[uidatacount++];

      uint32    ovflpages = uidata[uidatacount++];

      /* update metapage */

      metap->hashm_ovflpoint = ovflpoint;

      metap->hashm_spares[ovflpoint] = ovflpages;

    }

    MarkBufferDirty(metabuf);

    PageSetLSN(BufferGetPage(metabuf), lsn);

  }

  if (BufferIsValid(metabuf))

    UnlockReleaseBuffer(metabuf);

}

/*

 * replay of split operation

 */

static void

hash_xlog_split_page(XLogReaderState *record)

{

  Buffer    buf;

  if (XLogReadBufferForRedo(record, 0, &buf) != BLK_RESTORED)

    elog(ERROR, "Hash split record did not contain a full-page image");

  UnlockReleaseBuffer(buf);

}

/*

 * replay completion of split operation

 */

static void

hash_xlog_split_complete(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

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

  Buffer    oldbuf;

  Buffer    newbuf;

  XLogRedoAction action;

  /* replay the record for old bucket */

  action = XLogReadBufferForRedo(record, 0, &oldbuf);

  /*

   * Note that we still update the page even if it was restored from a full

   * page image, because the bucket flag is not included in the image.

   */

  if (action == BLK_NEEDS_REDO || action == BLK_RESTORED)

  {

    Page    oldpage;

    HashPageOpaque oldopaque;

    oldpage = BufferGetPage(oldbuf);

    oldopaque = HashPageGetOpaque(oldpage);

    oldopaque->hasho_flag = xlrec->old_bucket_flag;

    PageSetLSN(oldpage, lsn);

    MarkBufferDirty(oldbuf);

  }

  if (BufferIsValid(oldbuf))

    UnlockReleaseBuffer(oldbuf);

  /* replay the record for new bucket */

  action = XLogReadBufferForRedo(record, 1, &newbuf);

  /*

   * Note that we still update the page even if it was restored from a full

   * page image, because the bucket flag is not included in the image.

   */

  if (action == BLK_NEEDS_REDO || action == BLK_RESTORED)

  {

    Page    newpage;

    HashPageOpaque nopaque;

    newpage = BufferGetPage(newbuf);

    nopaque = HashPageGetOpaque(newpage);

    nopaque->hasho_flag = xlrec->new_bucket_flag;

    PageSetLSN(newpage, lsn);

    MarkBufferDirty(newbuf);

  }

  if (BufferIsValid(newbuf))

    UnlockReleaseBuffer(newbuf);

}

/*

 * replay move of page contents for squeeze operation of hash index

 */

static void

hash_xlog_move_page_contents(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_hash_move_page_contents *xldata = (xl_hash_move_page_contents *) XLogRecGetData(record);

  Buffer    bucketbuf = InvalidBuffer;

  Buffer    writebuf = InvalidBuffer;

  Buffer    deletebuf = InvalidBuffer;

  XLogRedoAction action;

  /*

   * Ensure we have a cleanup lock on primary bucket page before we start

   * with the actual replay operation.  This is to ensure that neither a

   * scan can start nor a scan can be already-in-progress during the replay

   * of this operation.  If we allow scans during this operation, then they

   * can miss some records or show the same record multiple times.

   */

  if (xldata->is_prim_bucket_same_wrt)

    action = XLogReadBufferForRedoExtended(record, 1, RBM_NORMAL, true, &writebuf);

  else

  {

    /*

     * we don't care for return value as the purpose of reading bucketbuf

     * is to ensure a cleanup lock on primary bucket page.

     */

    (void) XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &bucketbuf);

    action = XLogReadBufferForRedo(record, 1, &writebuf);

  }

  /* replay the record for adding entries in overflow buffer */

  if (action == BLK_NEEDS_REDO)

  {

    Page    writepage;

    char     *begin;

    char     *data;

    Size    datalen;

    uint16    ninserted = 0;

    data = begin = XLogRecGetBlockData(record, 1, &datalen);

    writepage = BufferGetPage(writebuf);

    if (xldata->ntups > 0)

    {

      OffsetNumber *towrite = (OffsetNumber *) data;

      data += sizeof(OffsetNumber) * xldata->ntups;

      while (data - begin < datalen)

      {

        IndexTuple  itup = (IndexTuple) data;

        Size    itemsz;

        OffsetNumber l;

        itemsz = IndexTupleSize(itup);

        itemsz = MAXALIGN(itemsz);

        data += itemsz;

        l = PageAddItem(writepage, itup, itemsz, towrite[ninserted], false, false);

        if (l == InvalidOffsetNumber)

          elog(ERROR, "hash_xlog_move_page_contents: failed to add item to hash index page, size %zu bytes", itemsz);

        ninserted++;

      }

    }

    /*

     * number of tuples inserted must be same as requested in REDO record.

     */

    Assert(ninserted == xldata->ntups);

    PageSetLSN(writepage, lsn);

    MarkBufferDirty(writebuf);

  }

  /* replay the record for deleting entries from overflow buffer */

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

  {

    Page    page;

    char     *ptr;

    Size    len;

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

    page = BufferGetPage(deletebuf);

    if (len > 0)

    {

      OffsetNumber *unused;

      OffsetNumber *unend;

      unused = (OffsetNumber *) ptr;

      unend = (OffsetNumber *) (ptr + len);

      if ((unend - unused) > 0)

        PageIndexMultiDelete(page, unused, unend - unused);

    }

    PageSetLSN(page, lsn);

    MarkBufferDirty(deletebuf);

  }

  /*

   * Replay is complete, now we can release the buffers. We release locks at

   * end of replay operation to ensure that we hold lock on primary bucket

   * page till end of operation.  We can optimize by releasing the lock on

   * write buffer as soon as the operation for same is complete, if it is

   * not same as primary bucket page, but that doesn't seem to be worth

   * complicating the code.

   */

  if (BufferIsValid(deletebuf))

    UnlockReleaseBuffer(deletebuf);

  if (BufferIsValid(writebuf))

    UnlockReleaseBuffer(writebuf);

  if (BufferIsValid(bucketbuf))

    UnlockReleaseBuffer(bucketbuf);

}

/*

 * replay squeeze page operation of hash index

 */

static void

hash_xlog_squeeze_page(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_hash_squeeze_page *xldata = (xl_hash_squeeze_page *) XLogRecGetData(record);

  Buffer    bucketbuf = InvalidBuffer;

  Buffer    writebuf = InvalidBuffer;

  Buffer    ovflbuf;

  Buffer    prevbuf = InvalidBuffer;

  Buffer    mapbuf;

  XLogRedoAction action;

  /*

   * Ensure we have a cleanup lock on primary bucket page before we start

   * with the actual replay operation.  This is to ensure that neither a

   * scan can start nor a scan can be already-in-progress during the replay

   * of this operation.  If we allow scans during this operation, then they

   * can miss some records or show the same record multiple times.

   */

  if (xldata->is_prim_bucket_same_wrt)

    action = XLogReadBufferForRedoExtended(record, 1, RBM_NORMAL, true, &writebuf);

  else

  {

    /*

     * we don't care for return value as the purpose of reading bucketbuf

     * is to ensure a cleanup lock on primary bucket page.

     */

    (void) XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &bucketbuf);

    if (xldata->ntups > 0 || xldata->is_prev_bucket_same_wrt)

      action = XLogReadBufferForRedo(record, 1, &writebuf);

    else

      action = BLK_NOTFOUND;

  }

  /* replay the record for adding entries in overflow buffer */

  if (action == BLK_NEEDS_REDO)

  {

    Page    writepage;

    char     *begin;

    char     *data;

    Size    datalen;

    uint16    ninserted = 0;

    bool    mod_wbuf = false;

    data = begin = XLogRecGetBlockData(record, 1, &datalen);

    writepage = BufferGetPage(writebuf);

    if (xldata->ntups > 0)

    {

      OffsetNumber *towrite = (OffsetNumber *) data;

      data += sizeof(OffsetNumber) * xldata->ntups;

      while (data - begin < datalen)

      {

        IndexTuple  itup = (IndexTuple) data;

        Size    itemsz;

        OffsetNumber l;

        itemsz = IndexTupleSize(itup);

        itemsz = MAXALIGN(itemsz);

        data += itemsz;

        l = PageAddItem(writepage, itup, itemsz, towrite[ninserted], false, false);

        if (l == InvalidOffsetNumber)

          elog(ERROR, "hash_xlog_squeeze_page: failed to add item to hash index page, size %zu bytes", itemsz);

        ninserted++;

      }

      mod_wbuf = true;

    }

    else

    {

      /*

       * Ensure that the required flags are set when there are no

       * tuples.  See _hash_freeovflpage().

       */

      Assert(xldata->is_prim_bucket_same_wrt ||

           xldata->is_prev_bucket_same_wrt);

    }

    /*

     * number of tuples inserted must be same as requested in REDO record.

     */

    Assert(ninserted == xldata->ntups);

    /*

     * if the page on which are adding tuples is a page previous to freed

     * overflow page, then update its nextblkno.

     */

    if (xldata->is_prev_bucket_same_wrt)

    {

      HashPageOpaque writeopaque = HashPageGetOpaque(writepage);

      writeopaque->hasho_nextblkno = xldata->nextblkno;

      mod_wbuf = true;

    }

    /* Set LSN and mark writebuf dirty iff it is modified */

    if (mod_wbuf)

    {

      PageSetLSN(writepage, lsn);

      MarkBufferDirty(writebuf);

    }

  }

  /* replay the record for initializing overflow buffer */

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

  {

    Page    ovflpage;

    HashPageOpaque ovflopaque;

    ovflpage = BufferGetPage(ovflbuf);

    _hash_pageinit(ovflpage, BufferGetPageSize(ovflbuf));

    ovflopaque = HashPageGetOpaque(ovflpage);

    ovflopaque->hasho_prevblkno = InvalidBlockNumber;

    ovflopaque->hasho_nextblkno = InvalidBlockNumber;

    ovflopaque->hasho_bucket = InvalidBucket;

    ovflopaque->hasho_flag = LH_UNUSED_PAGE;

    ovflopaque->hasho_page_id = HASHO_PAGE_ID;

    PageSetLSN(ovflpage, lsn);

    MarkBufferDirty(ovflbuf);

  }

  if (BufferIsValid(ovflbuf))

    UnlockReleaseBuffer(ovflbuf);

  /* replay the record for page previous to the freed overflow page */

  if (!xldata->is_prev_bucket_same_wrt &&

    XLogReadBufferForRedo(record, 3, &prevbuf) == BLK_NEEDS_REDO)

  {

    Page    prevpage = BufferGetPage(prevbuf);

    HashPageOpaque prevopaque = HashPageGetOpaque(prevpage);

    prevopaque->hasho_nextblkno = xldata->nextblkno;

    PageSetLSN(prevpage, lsn);

    MarkBufferDirty(prevbuf);

  }

  if (BufferIsValid(prevbuf))

    UnlockReleaseBuffer(prevbuf);

  /* replay the record for page next to the freed overflow page */

  if (XLogRecHasBlockRef(record, 4))

  {

    Buffer    nextbuf;

    if (XLogReadBufferForRedo(record, 4, &nextbuf) == BLK_NEEDS_REDO)

    {

      Page    nextpage = BufferGetPage(nextbuf);

      HashPageOpaque nextopaque = HashPageGetOpaque(nextpage);

      nextopaque->hasho_prevblkno = xldata->prevblkno;

      PageSetLSN(nextpage, lsn);

      MarkBufferDirty(nextbuf);

    }

    if (BufferIsValid(nextbuf))

      UnlockReleaseBuffer(nextbuf);

  }

  if (BufferIsValid(writebuf))

    UnlockReleaseBuffer(writebuf);

  if (BufferIsValid(bucketbuf))

    UnlockReleaseBuffer(bucketbuf);

  /*

   * Note: in normal operation, we'd update the bitmap and meta page while

   * still holding lock on the primary bucket page and overflow pages.  But

   * during replay it's not necessary to hold those locks, since no other

   * index updates can be happening concurrently.

   */

  /* replay the record for bitmap page */

  if (XLogReadBufferForRedo(record, 5, &mapbuf) == BLK_NEEDS_REDO)

  {

    Page    mappage = BufferGetPage(mapbuf);

    uint32     *freep = NULL;

    char     *data;

    uint32     *bitmap_page_bit;

    Size    datalen;

    freep = HashPageGetBitmap(mappage);

    data = XLogRecGetBlockData(record, 5, &datalen);

    bitmap_page_bit = (uint32 *) data;

    CLRBIT(freep, *bitmap_page_bit);

    PageSetLSN(mappage, lsn);

    MarkBufferDirty(mapbuf);

  }

  if (BufferIsValid(mapbuf))

    UnlockReleaseBuffer(mapbuf);

  /* replay the record for meta page */

  if (XLogRecHasBlockRef(record, 6))

  {

    Buffer    metabuf;

    if (XLogReadBufferForRedo(record, 6, &metabuf) == BLK_NEEDS_REDO)

    {

      HashMetaPage metap;

      Page    page;

      char     *data;

      uint32     *firstfree_ovflpage;

      Size    datalen;

      data = XLogRecGetBlockData(record, 6, &datalen);

      firstfree_ovflpage = (uint32 *) data;

      page = BufferGetPage(metabuf);

      metap = HashPageGetMeta(page);

      metap->hashm_firstfree = *firstfree_ovflpage;

      PageSetLSN(page, lsn);

      MarkBufferDirty(metabuf);

    }

    if (BufferIsValid(metabuf))

      UnlockReleaseBuffer(metabuf);

  }

}

/*

 * replay delete operation of hash index

 */

static void

hash_xlog_delete(XLogReaderState *record)

{

  XLogRecPtr  lsn = record->EndRecPtr;

  xl_hash_delete *xldata = (xl_hash_delete *) XLogRecGetData(record);

  Buffer    bucketbuf = InvalidBuffer;

  Buffer    deletebuf;

  Page    page;

  XLogRedoAction action;

  /*

   * Ensure we have a cleanup lock on primary bucket page before we start

   * with the actual replay operation.  This is to ensure that neither a

   * scan can start nor a scan can be already-in-progress during the replay

   * of this operation.  If we allow scans during this operation, then they

   * can miss some records or show the same record multiple times.

   */

  if (xldata->is_primary_bucket_page)

    action = XLogReadBufferForRedoExtended(record, 1, RBM_NORMAL, true, &deletebuf);

  else

  {

    /*

     * we don't care for return value as the purpose of reading bucketbuf

     * is to ensure a cleanup lock on primary bucket page.

     */

    (void) XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, true, &bucketbuf);

    action = XLogReadBufferForRedo(record, 1, &deletebuf);

  }

  /* replay the record for deleting entries in bucket page */

  if (action == BLK_NEEDS_REDO)

  {

    char     *ptr;

    Size    len;

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

    page = BufferGetPage(deletebuf);

    if (len > 0)

    {

      OffsetNumber *unused;

      OffsetNumber *unend;

      unused = (OffsetNumber *) ptr;

      unend = (OffsetNumber *) (ptr + len);

      if ((unend - unused) > 0)

        PageIndexMultiDelete(page, unused, unend - unused);

    }

    /*

     * Mark the page as not containing any LP_DEAD items only if

     * clear_dead_marking flag is set to true. See comments in

     * hashbucketcleanup() for details.

     */

    if (xldata->clear_dead_marking)

    {

      HashPageOpaque pageopaque;

      pageopaque = HashPageGetOpaque(page);

      pageopaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;

    }

    PageSetLSN(page, lsn);

    MarkBufferDirty(deletebuf);

  }

  if (BufferIsValid(deletebuf))

    UnlockReleaseBuffer(deletebuf);

  if (BufferIsValid(bucketbuf))

    UnlockReleaseBuffer(bucketbuf);

}

/*

 * replay split cleanup flag operation for primary bucket page.