/*

 * brin_pageops.c

 *    Page-handling routines for BRIN indexes

 *

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

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

 *

 * IDENTIFICATION

 *    src/backend/access/brin/brin_pageops.c

 */

#include "postgres.h"

#include "access/brin_page.h"

#include "access/brin_pageops.h"

#include "access/brin_revmap.h"

#include "access/brin_xlog.h"

#include "access/xloginsert.h"

#include "miscadmin.h"

#include "storage/bufmgr.h"

#include "storage/freespace.h"

#include "storage/lmgr.h"

#include "utils/rel.h"

/*

 * Maximum size of an entry in a BRIN_PAGETYPE_REGULAR page.  We can tolerate

 * a single item per page, unlike other index AMs.

 */

#define BrinMaxItemSize \

  MAXALIGN_DOWN(BLCKSZ - \

          (MAXALIGN(SizeOfPageHeaderData + \

              sizeof(ItemIdData)) + \

           MAXALIGN(sizeof(BrinSpecialSpace))))

static Buffer brin_getinsertbuffer(Relation irel, Buffer oldbuf, Size itemsz,

                   bool *extended);

static Size br_page_get_freespace(Page page);

static void brin_initialize_empty_new_buffer(Relation idxrel, Buffer buffer);

/*

 * Update tuple origtup (size origsz), located in offset oldoff of buffer

 * oldbuf, to newtup (size newsz) as summary tuple for the page range starting

 * at heapBlk.  oldbuf must not be locked on entry, and is not locked at exit.

 *

 * If samepage is true, attempt to put the new tuple in the same page, but if

 * there's no room, use some other one.

 *

 * If the update is successful, return true; the revmap is updated to point to

 * the new tuple.  If the update is not done for whatever reason, return false.

 * Caller may retry the update if this happens.

 */

bool

brin_doupdate(Relation idxrel, BlockNumber pagesPerRange,

        BrinRevmap *revmap, BlockNumber heapBlk,

        Buffer oldbuf, OffsetNumber oldoff,

        const BrinTuple *origtup, Size origsz,

        const BrinTuple *newtup, Size newsz,

        bool samepage)

{

  Page    oldpage;

  ItemId    oldlp;

  BrinTuple  *oldtup;

  Size    oldsz;

  Buffer    newbuf;

  BlockNumber newblk = InvalidBlockNumber;

  bool    extended;

  Assert(newsz == MAXALIGN(newsz));

  /* If the item is oversized, don't bother. */

  if (newsz > BrinMaxItemSize)

  {

    ereport(ERROR,

        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),

         errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",

            newsz, BrinMaxItemSize, RelationGetRelationName(idxrel))));

    return false;     /* keep compiler quiet */

  }

  /* make sure the revmap is long enough to contain the entry we need */

  brinRevmapExtend(revmap, heapBlk);

  if (!samepage)

  {

    /* need a page on which to put the item */

    newbuf = brin_getinsertbuffer(idxrel, oldbuf, newsz, &extended);

    if (!BufferIsValid(newbuf))

    {

      Assert(!extended);

      return false;

    }

    /*

     * Note: it's possible (though unlikely) that the returned newbuf is

     * the same as oldbuf, if brin_getinsertbuffer determined that the old

     * buffer does in fact have enough space.

     */

    if (newbuf == oldbuf)

    {

      Assert(!extended);

      newbuf = InvalidBuffer;

    }

    else

      newblk = BufferGetBlockNumber(newbuf);

  }

  else

  {

    LockBuffer(oldbuf, BUFFER_LOCK_EXCLUSIVE);

    newbuf = InvalidBuffer;

    extended = false;

  }

  oldpage = BufferGetPage(oldbuf);

  oldlp = PageGetItemId(oldpage, oldoff);

  /*

   * Check that the old tuple wasn't updated concurrently: it might have

   * moved someplace else entirely, and for that matter the whole page

   * might've become a revmap page.  Note that in the first two cases

   * checked here, the "oldlp" we just calculated is garbage; but

   * PageGetItemId() is simple enough that it was safe to do that

   * calculation anyway.

   */

  if (!BRIN_IS_REGULAR_PAGE(oldpage) ||

    oldoff > PageGetMaxOffsetNumber(oldpage) ||

    !ItemIdIsNormal(oldlp))

  {

    LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

    /*

     * If this happens, and the new buffer was obtained by extending the

     * relation, then we need to ensure we don't leave it uninitialized or

     * forget about it.

     */

    if (BufferIsValid(newbuf))

    {

      if (extended)

        brin_initialize_empty_new_buffer(idxrel, newbuf);

      UnlockReleaseBuffer(newbuf);

      if (extended)

        FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);

    }

    return false;

  }

  oldsz = ItemIdGetLength(oldlp);

  oldtup = (BrinTuple *) PageGetItem(oldpage, oldlp);

  /*

   * ... or it might have been updated in place to different contents.

   */

  if (!brin_tuples_equal(oldtup, oldsz, origtup, origsz))

  {

    LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

    if (BufferIsValid(newbuf))

    {

      /* As above, initialize and record new page if we got one */

      if (extended)

        brin_initialize_empty_new_buffer(idxrel, newbuf);

      UnlockReleaseBuffer(newbuf);

      if (extended)

        FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);

    }

    return false;

  }

  /*

   * Great, the old tuple is intact.  We can proceed with the update.

   *

   * If there's enough room in the old page for the new tuple, replace it.

   *

   * Note that there might now be enough space on the page even though the

   * caller told us there isn't, if a concurrent update moved another tuple

   * elsewhere or replaced a tuple with a smaller one.

   */

  if (((BrinPageFlags(oldpage) & BRIN_EVACUATE_PAGE) == 0) &&

    brin_can_do_samepage_update(oldbuf, origsz, newsz))

  {

    START_CRIT_SECTION();

    if (!PageIndexTupleOverwrite(oldpage, oldoff, newtup, newsz))

      elog(ERROR, "failed to replace BRIN tuple");

    MarkBufferDirty(oldbuf);

    /* XLOG stuff */

    if (RelationNeedsWAL(idxrel))

    {

      xl_brin_samepage_update xlrec;

      XLogRecPtr  recptr;

      uint8   info = XLOG_BRIN_SAMEPAGE_UPDATE;

      xlrec.offnum = oldoff;

      XLogBeginInsert();

      XLogRegisterData(&xlrec, SizeOfBrinSamepageUpdate);

      XLogRegisterBuffer(0, oldbuf, REGBUF_STANDARD);

      XLogRegisterBufData(0, newtup, newsz);

      recptr = XLogInsert(RM_BRIN_ID, info);

      PageSetLSN(oldpage, recptr);

    }

    END_CRIT_SECTION();

    LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

    if (BufferIsValid(newbuf))

    {

      /* As above, initialize and record new page if we got one */

      if (extended)

        brin_initialize_empty_new_buffer(idxrel, newbuf);

      UnlockReleaseBuffer(newbuf);

      if (extended)

        FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);

    }

    return true;

  }

  else if (newbuf == InvalidBuffer)

  {

    /*

     * Not enough space, but caller said that there was. Tell them to

     * start over.

     */

    LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

    return false;

  }

  else

  {

    /*

     * Not enough free space on the oldpage. Put the new tuple on the new

     * page, and update the revmap.

     */

    Page    newpage = BufferGetPage(newbuf);

    Buffer    revmapbuf;

    ItemPointerData newtid;

    OffsetNumber newoff;

    Size    freespace = 0;

    revmapbuf = brinLockRevmapPageForUpdate(revmap, heapBlk);

    START_CRIT_SECTION();

    /*

     * We need to initialize the page if it's newly obtained.  Note we

     * will WAL-log the initialization as part of the update, so we don't

     * need to do that here.

     */

    if (extended)

      brin_page_init(newpage, BRIN_PAGETYPE_REGULAR);

    PageIndexTupleDeleteNoCompact(oldpage, oldoff);

    newoff = PageAddItem(newpage, newtup, newsz, InvalidOffsetNumber, false, false);

    if (newoff == InvalidOffsetNumber)

      elog(ERROR, "failed to add BRIN tuple to new page");

    MarkBufferDirty(oldbuf);

    MarkBufferDirty(newbuf);

    /* needed to update FSM below */

    if (extended)

      freespace = br_page_get_freespace(newpage);

    ItemPointerSet(&newtid, newblk, newoff);

    brinSetHeapBlockItemptr(revmapbuf, pagesPerRange, heapBlk, newtid);

    MarkBufferDirty(revmapbuf);

    /* XLOG stuff */

    if (RelationNeedsWAL(idxrel))

    {

      xl_brin_update xlrec;

      XLogRecPtr  recptr;

      uint8   info;

      info = XLOG_BRIN_UPDATE | (extended ? XLOG_BRIN_INIT_PAGE : 0);

      xlrec.insert.offnum = newoff;

      xlrec.insert.heapBlk = heapBlk;

      xlrec.insert.pagesPerRange = pagesPerRange;

      xlrec.oldOffnum = oldoff;

      XLogBeginInsert();

      /* new page */

      XLogRegisterData(&xlrec, SizeOfBrinUpdate);

      XLogRegisterBuffer(0, newbuf, REGBUF_STANDARD | (extended ? REGBUF_WILL_INIT : 0));

      XLogRegisterBufData(0, newtup, newsz);

      /* revmap page */

      XLogRegisterBuffer(1, revmapbuf, 0);

      /* old page */

      XLogRegisterBuffer(2, oldbuf, REGBUF_STANDARD);

      recptr = XLogInsert(RM_BRIN_ID, info);

      PageSetLSN(oldpage, recptr);

      PageSetLSN(newpage, recptr);

      PageSetLSN(BufferGetPage(revmapbuf), recptr);

    }

    END_CRIT_SECTION();

    LockBuffer(revmapbuf, BUFFER_LOCK_UNLOCK);

    LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

    UnlockReleaseBuffer(newbuf);

    if (extended)

    {

      RecordPageWithFreeSpace(idxrel, newblk, freespace);

      FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);

    }

    return true;

  }

}

/*

 * Return whether brin_doupdate can do a samepage update.

 */

bool

brin_can_do_samepage_update(Buffer buffer, Size origsz, Size newsz)

{

  return

    ((newsz <= origsz) ||

     PageGetExactFreeSpace(BufferGetPage(buffer)) >= (newsz - origsz));

}

/*

 * Insert an index tuple into the index relation.  The revmap is updated to

 * mark the range containing the given page as pointing to the inserted entry.

 * A WAL record is written.

 *

 * The buffer, if valid, is first checked for free space to insert the new

 * entry; if there isn't enough, a new buffer is obtained and pinned.  No

 * buffer lock must be held on entry, no buffer lock is held on exit.

 *

 * Return value is the offset number where the tuple was inserted.

 */

OffsetNumber

brin_doinsert(Relation idxrel, BlockNumber pagesPerRange,

        BrinRevmap *revmap, Buffer *buffer, BlockNumber heapBlk,

        const BrinTuple *tup, Size itemsz)

{

  Page    page;

  BlockNumber blk;

  OffsetNumber off;

  Size    freespace = 0;

  Buffer    revmapbuf;

  ItemPointerData tid;

  bool    extended;

  Assert(itemsz == MAXALIGN(itemsz));

  /* If the item is oversized, don't even bother. */

  if (itemsz > BrinMaxItemSize)

  {

    ereport(ERROR,

        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),

         errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",

            itemsz, BrinMaxItemSize, RelationGetRelationName(idxrel))));

    return InvalidOffsetNumber; /* keep compiler quiet */

  }

  /* Make sure the revmap is long enough to contain the entry we need */

  brinRevmapExtend(revmap, heapBlk);

  /*

   * Acquire lock on buffer supplied by caller, if any.  If it doesn't have

   * enough space, unpin it to obtain a new one below.

   */

  if (BufferIsValid(*buffer))

  {

    /*

     * It's possible that another backend (or ourselves!) extended the

     * revmap over the page we held a pin on, so we cannot assume that

     * it's still a regular page.

     */

    LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);

    if (br_page_get_freespace(BufferGetPage(*buffer)) < itemsz)

    {

      UnlockReleaseBuffer(*buffer);

      *buffer = InvalidBuffer;

    }

  }

  /*

   * If we still don't have a usable buffer, have brin_getinsertbuffer

   * obtain one for us.

   */

  if (!BufferIsValid(*buffer))

  {

    do

      *buffer = brin_getinsertbuffer(idxrel, InvalidBuffer, itemsz, &extended);

    while (!BufferIsValid(*buffer));

  }

  else

    extended = false;

  /* Now obtain lock on revmap buffer */

  revmapbuf = brinLockRevmapPageForUpdate(revmap, heapBlk);

  page = BufferGetPage(*buffer);

  blk = BufferGetBlockNumber(*buffer);

  /* Execute the actual insertion */

  START_CRIT_SECTION();

  if (extended)

    brin_page_init(page, BRIN_PAGETYPE_REGULAR);

  off = PageAddItem(page, tup, itemsz, InvalidOffsetNumber, false, false);

  if (off == InvalidOffsetNumber)

    elog(ERROR, "failed to add BRIN tuple to new page");

  MarkBufferDirty(*buffer);

  /* needed to update FSM below */

  if (extended)

    freespace = br_page_get_freespace(page);

  ItemPointerSet(&tid, blk, off);

  brinSetHeapBlockItemptr(revmapbuf, pagesPerRange, heapBlk, tid);

  MarkBufferDirty(revmapbuf);

  /* XLOG stuff */

  if (RelationNeedsWAL(idxrel))

  {

    xl_brin_insert xlrec;

    XLogRecPtr  recptr;

    uint8   info;

    info = XLOG_BRIN_INSERT | (extended ? XLOG_BRIN_INIT_PAGE : 0);

    xlrec.heapBlk = heapBlk;

    xlrec.pagesPerRange = pagesPerRange;

    xlrec.offnum = off;

    XLogBeginInsert();

    XLogRegisterData(&xlrec, SizeOfBrinInsert);

    XLogRegisterBuffer(0, *buffer, REGBUF_STANDARD | (extended ? REGBUF_WILL_INIT : 0));

    XLogRegisterBufData(0, tup, itemsz);

    XLogRegisterBuffer(1, revmapbuf, 0);

    recptr = XLogInsert(RM_BRIN_ID, info);

    PageSetLSN(page, recptr);

    PageSetLSN(BufferGetPage(revmapbuf), recptr);

  }

  END_CRIT_SECTION();

  /* Tuple is firmly on buffer; we can release our locks */

  LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);

  LockBuffer(revmapbuf, BUFFER_LOCK_UNLOCK);

  BRIN_elog((DEBUG2, "inserted tuple (%u,%u) for range starting at %u",

         blk, off, heapBlk));

  if (extended)

  {

    RecordPageWithFreeSpace(idxrel, blk, freespace);

    FreeSpaceMapVacuumRange(idxrel, blk, blk + 1);

  }

  return off;

}

/*

 * Initialize a page with the given type.

 *

 * Caller is responsible for marking it dirty, as appropriate.

 */

void

brin_page_init(Page page, uint16 type)

{

  PageInit(page, BLCKSZ, sizeof(BrinSpecialSpace));

  BrinPageType(page) = type;

}

/*

 * Initialize a new BRIN index's metapage.

 */

void

brin_metapage_init(Page page, BlockNumber pagesPerRange, uint16 version)

{

  BrinMetaPageData *metadata;

  brin_page_init(page, BRIN_PAGETYPE_META);

  metadata = (BrinMetaPageData *) PageGetContents(page);

  metadata->brinMagic = BRIN_META_MAGIC;

  metadata->brinVersion = version;

  metadata->pagesPerRange = pagesPerRange;

  /*

   * Note we cheat here a little.  0 is not a valid revmap block number

   * (because it's the metapage buffer), but doing this enables the first

   * revmap page to be created when the index is.

   */

  metadata->lastRevmapPage = 0;

  /*

   * 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) page)->pd_lower =

    ((char *) metadata + sizeof(BrinMetaPageData)) - (char *) page;

}

/*

 * Initiate page evacuation protocol.

 *

 * The page must be locked in exclusive mode by the caller.

 *

 * If the page is not yet initialized or empty, return false without doing

 * anything; it can be used for revmap without any further changes.  If it

 * contains tuples, mark it for evacuation and return true.

 */

bool

brin_start_evacuating_page(Relation idxRel, Buffer buf)

{

  OffsetNumber off;

  OffsetNumber maxoff;

  Page    page;

  page = BufferGetPage(buf);

  if (PageIsNew(page))

    return false;

  maxoff = PageGetMaxOffsetNumber(page);

  for (off = FirstOffsetNumber; off <= maxoff; off++)

  {

    ItemId    lp;

    lp = PageGetItemId(page, off);

    if (ItemIdIsUsed(lp))

    {

      /*

       * Prevent other backends from adding more stuff to this page:

       * BRIN_EVACUATE_PAGE informs br_page_get_freespace that this page

       * can no longer be used to add new tuples.  Note that this flag

       * is not WAL-logged, except accidentally.

       */

      BrinPageFlags(page) |= BRIN_EVACUATE_PAGE;

      MarkBufferDirtyHint(buf, true);

      return true;

    }

  }

  return false;

}

/*

 * Move all tuples out of a page.

 *

 * The caller must hold lock on the page. The lock and pin are released.

 */

void

brin_evacuate_page(Relation idxRel, BlockNumber pagesPerRange,

           BrinRevmap *revmap, Buffer buf)

{

  OffsetNumber off;

  OffsetNumber maxoff;

  Page    page;

  BrinTuple  *btup = NULL;

  Size    btupsz = 0;

  page = BufferGetPage(buf);

  Assert(BrinPageFlags(page) & BRIN_EVACUATE_PAGE);

  maxoff = PageGetMaxOffsetNumber(page);

  for (off = FirstOffsetNumber; off <= maxoff; off++)

  {

    BrinTuple  *tup;

    Size    sz;

    ItemId    lp;

    CHECK_FOR_INTERRUPTS();

    lp = PageGetItemId(page, off);

    if (ItemIdIsUsed(lp))

    {

      sz = ItemIdGetLength(lp);

      tup = (BrinTuple *) PageGetItem(page, lp);

      tup = brin_copy_tuple(tup, sz, btup, &btupsz);

      LockBuffer(buf, BUFFER_LOCK_UNLOCK);

      if (!brin_doupdate(idxRel, pagesPerRange, revmap, tup->bt_blkno,

                 buf, off, tup, sz, tup, sz, false))

        off--;     /* retry */

      LockBuffer(buf, BUFFER_LOCK_SHARE);

      /* It's possible that someone extended the revmap over this page */

      if (!BRIN_IS_REGULAR_PAGE(page))

        break;

    }

  }

  UnlockReleaseBuffer(buf);

}

/*

 * Given a BRIN index page, initialize it if necessary, and record its

 * current free space in the FSM.

 *

 * The main use for this is when, during vacuuming, an uninitialized page is

 * found, which could be the result of relation extension followed by a crash

 * before the page can be used.

 *

 * Here, we don't bother to update upper FSM pages, instead expecting that our

 * caller (brin_vacuum_scan) will fix them at the end of the scan.  Elsewhere

 * in this file, it's generally a good idea to propagate additions of free

 * space into the upper FSM pages immediately.

 */

void

brin_page_cleanup(Relation idxrel, Buffer buf)

{

  Page    page = BufferGetPage(buf);

  /*

   * If a page was left uninitialized, initialize it now; also record it in

   * FSM.

   *

   * Somebody else might be extending the relation concurrently.  To avoid

   * re-initializing the page before they can grab the buffer lock, we

   * acquire the extension lock momentarily.  Since they hold the extension

   * lock from before getting the page and after its been initialized, we're

   * sure to see their initialization.

   */

  if (PageIsNew(page))

  {

    LockRelationForExtension(idxrel, ShareLock);

    UnlockRelationForExtension(idxrel, ShareLock);

    LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);

    if (PageIsNew(page))

    {

      brin_initialize_empty_new_buffer(idxrel, buf);

      LockBuffer(buf, BUFFER_LOCK_UNLOCK);

      return;

    }

    LockBuffer(buf, BUFFER_LOCK_UNLOCK);

  }

  /* Nothing to be done for non-regular index pages */

  if (BRIN_IS_META_PAGE(BufferGetPage(buf)) ||

    BRIN_IS_REVMAP_PAGE(BufferGetPage(buf)))

    return;

  /* Measure free space and record it */

  RecordPageWithFreeSpace(idxrel, BufferGetBlockNumber(buf),

              br_page_get_freespace(page));

}

/*

 * Return a pinned and exclusively locked buffer which can be used to insert an

 * index item of size itemsz (caller must ensure not to request sizes

 * impossible to fulfill).  If oldbuf is a valid buffer, it is also locked (in

 * an order determined to avoid deadlocks).

 *

 * If we find that the old page is no longer a regular index page (because

 * of a revmap extension), the old buffer is unlocked and we return

 * InvalidBuffer.

 *

 * If there's no existing page with enough free space to accommodate the new

 * item, the relation is extended.  If this happens, *extended is set to true,

 * and it is the caller's responsibility to initialize the page (and WAL-log

 * that fact) prior to use.  The caller should also update the FSM with the

 * page's remaining free space after the insertion.

 *

 * Note that the caller is not expected to update FSM unless *extended is set

 * true.  This policy means that we'll update FSM when a page is created, and

 * when it's found to have too little space for a desired tuple insertion,

 * but not every single time we add a tuple to the page.

 *

 * Note that in some corner cases it is possible for this routine to extend

 * the relation and then not return the new page.  It is this routine's

 * responsibility to WAL-log the page initialization and to record the page in

 * FSM if that happens, since the caller certainly can't do it.

 */

static Buffer

brin_getinsertbuffer(Relation irel, Buffer oldbuf, Size itemsz,

           bool *extended)

{

  BlockNumber oldblk;

  BlockNumber newblk;

  Page    page;

  Size    freespace;

  /* callers must have checked */

  Assert(itemsz <= BrinMaxItemSize);

  if (BufferIsValid(oldbuf))

    oldblk = BufferGetBlockNumber(oldbuf);

  else

    oldblk = InvalidBlockNumber;

  /* Choose initial target page, re-using existing target if known */

  newblk = RelationGetTargetBlock(irel);

  if (newblk == InvalidBlockNumber)

    newblk = GetPageWithFreeSpace(irel, itemsz);

  /*

   * Loop until we find a page with sufficient free space.  By the time we

   * return to caller out of this loop, both buffers are valid and locked;

   * if we have to restart here, neither page is locked and newblk isn't

   * pinned (if it's even valid).

   */

  for (;;)

  {

    Buffer    buf;

    bool    extensionLockHeld = false;

    CHECK_FOR_INTERRUPTS();

    *extended = false;

    if (newblk == InvalidBlockNumber)

    {

      /*

       * There's not enough free space in any existing index page,

       * according to the FSM: extend the relation to obtain a shiny new

       * page.

       *

       * XXX: It's likely possible to use RBM_ZERO_AND_LOCK here,

       * which'd avoid the need to hold the extension lock during buffer

       * reclaim.

       */

      if (!RELATION_IS_LOCAL(irel))

      {

        LockRelationForExtension(irel, ExclusiveLock);

        extensionLockHeld = true;

      }

      buf = ReadBuffer(irel, P_NEW);

      newblk = BufferGetBlockNumber(buf);

      *extended = true;

      BRIN_elog((DEBUG2, "brin_getinsertbuffer: extending to page %u",

             BufferGetBlockNumber(buf)));

    }

    else if (newblk == oldblk)

    {

      /*

       * There's an odd corner-case here where the FSM is out-of-date,

       * and gave us the old page.

       */

      buf = oldbuf;

    }

    else

    {

      buf = ReadBuffer(irel, newblk);

    }

    /*

     * We lock the old buffer first, if it's earlier than the new one; but

     * then we need to check that it hasn't been turned into a revmap page

     * concurrently.  If we detect that that happened, give up and tell

     * caller to start over.

     */

    if (BufferIsValid(oldbuf) && oldblk < newblk)

    {

      LockBuffer(oldbuf, BUFFER_LOCK_EXCLUSIVE);

      if (!BRIN_IS_REGULAR_PAGE(BufferGetPage(oldbuf)))

      {

        LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

        /*

         * It is possible that the new page was obtained from

         * extending the relation.  In that case, we must be sure to

         * record it in the FSM before leaving, because otherwise the

         * space would be lost forever.  However, we cannot let an

         * uninitialized page get in the FSM, so we need to initialize

         * it first.

         */

        if (*extended)

          brin_initialize_empty_new_buffer(irel, buf);

        if (extensionLockHeld)

          UnlockRelationForExtension(irel, ExclusiveLock);

        ReleaseBuffer(buf);

        if (*extended)

        {

          FreeSpaceMapVacuumRange(irel, newblk, newblk + 1);

          /* shouldn't matter, but don't confuse caller */

          *extended = false;

        }

        return InvalidBuffer;

      }

    }

    LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);

    if (extensionLockHeld)

      UnlockRelationForExtension(irel, ExclusiveLock);

    page = BufferGetPage(buf);

    /*

     * We have a new buffer to insert into.  Check that the new page has

     * enough free space, and return it if it does; otherwise start over.

     * (br_page_get_freespace also checks that the FSM didn't hand us a

     * page that has since been repurposed for the revmap.)

     */

    freespace = *extended ?

      BrinMaxItemSize : br_page_get_freespace(page);

    if (freespace >= itemsz)

    {

      RelationSetTargetBlock(irel, newblk);

      /*

       * Lock the old buffer if not locked already.  Note that in this

       * case we know for sure it's a regular page: it's later than the

       * new page we just got, which is not a revmap page, and revmap

       * pages are always consecutive.

       */

      if (BufferIsValid(oldbuf) && oldblk > newblk)

      {

        LockBuffer(oldbuf, BUFFER_LOCK_EXCLUSIVE);

        Assert(BRIN_IS_REGULAR_PAGE(BufferGetPage(oldbuf)));

      }

      return buf;

    }

    /* This page is no good. */

    /*

     * If an entirely new page does not contain enough free space for the

     * new item, then surely that item is oversized.  Complain loudly; but

     * first make sure we initialize the page and record it as free, for

     * next time.

     */

    if (*extended)

    {

      brin_initialize_empty_new_buffer(irel, buf);

      /* since this should not happen, skip FreeSpaceMapVacuum */

      ereport(ERROR,

          (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),

           errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",

              itemsz, freespace, RelationGetRelationName(irel))));

      return InvalidBuffer; /* keep compiler quiet */

    }

    if (newblk != oldblk)

      UnlockReleaseBuffer(buf);

    if (BufferIsValid(oldbuf) && oldblk <= newblk)

      LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);

    /*

     * Update the FSM with the new, presumably smaller, freespace value

     * for this page, then search for a new target page.

     */

    newblk = RecordAndGetPageWithFreeSpace(irel, newblk, freespace, itemsz);

  }

}

/*

 * Initialize a page as an empty regular BRIN page, WAL-log this, and record

 * the page in FSM.

 *

 * There are several corner situations in which we extend the relation to

 * obtain a new page and later find that we cannot use it immediately.  When

 * that happens, we don't want to leave the page go unrecorded in FSM, because

 * there is no mechanism to get the space back and the index would bloat.

 * Also, because we would not WAL-log the action that would initialize the

 * page, the page would go uninitialized in a standby (or after recovery).

 *

 * While we record the page in FSM here, caller is responsible for doing FSM

 * upper-page update if that seems appropriate.

 */

static void

brin_initialize_empty_new_buffer(Relation idxrel, Buffer buffer)

{

  Page    page;

  BRIN_elog((DEBUG2,

         "brin_initialize_empty_new_buffer: initializing blank page %u",

         BufferGetBlockNumber(buffer)));

  START_CRIT_SECTION();

  page = BufferGetPage(buffer);

  brin_page_init(page, BRIN_PAGETYPE_REGULAR);

  MarkBufferDirty(buffer);

  /* XLOG stuff */

  if (RelationNeedsWAL(idxrel))

    log_newpage_buffer(buffer, true);

  END_CRIT_SECTION();

  /*

   * We update the FSM for this page, but this is not WAL-logged.  This is

   * acceptable because VACUUM will scan the index and update the FSM with

   * pages whose FSM records were forgotten in a crash.

   */

  RecordPageWithFreeSpace(idxrel, BufferGetBlockNumber(buffer),

              br_page_get_freespace(page));

}

/*

 * Return the amount of free space on a regular BRIN index page.

 *

 * If the page is not a regular page, or has been marked with the

 * BRIN_EVACUATE_PAGE flag, returns 0.

 */

static Size

br_page_get_freespace(Page page)

{

  if (!BRIN_IS_REGULAR_PAGE(page) ||

    (BrinPageFlags(page) & BRIN_EVACUATE_PAGE) != 0)

    return 0;

  else

    return PageGetFreeSpace(page);

}