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

 *

 * gist.c

 *    interface routines for the postgres GiST index access method.

 *

 *

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

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

 *

 * IDENTIFICATION

 *    src/backend/access/gist/gist.c

 *

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

 */

#include "postgres.h"

#include "access/gist_private.h"

#include "access/gistscan.h"

#include "access/xloginsert.h"

#include "catalog/pg_collation.h"

#include "commands/vacuum.h"

#include "miscadmin.h"

#include "nodes/execnodes.h"

#include "storage/predicate.h"

#include "utils/fmgrprotos.h"

#include "utils/index_selfuncs.h"

#include "utils/memutils.h"

#include "utils/rel.h"

/* non-export function prototypes */

static void gistfixsplit(GISTInsertState *state, GISTSTATE *giststate);

static bool gistinserttuple(GISTInsertState *state, GISTInsertStack *stack,

              GISTSTATE *giststate, IndexTuple tuple, OffsetNumber oldoffnum);

static bool gistinserttuples(GISTInsertState *state, GISTInsertStack *stack,

               GISTSTATE *giststate,

               IndexTuple *tuples, int ntup, OffsetNumber oldoffnum,

               Buffer leftchild, Buffer rightchild,

               bool unlockbuf, bool unlockleftchild);

static void gistfinishsplit(GISTInsertState *state, GISTInsertStack *stack,

              GISTSTATE *giststate, List *splitinfo, bool unlockbuf);

static void gistprunepage(Relation rel, Page page, Buffer buffer,

              Relation heapRel);

#define ROTATEDIST(d) do { \

  SplitPageLayout *tmp = (SplitPageLayout *) palloc0(sizeof(SplitPageLayout)); \

  tmp->block.blkno = InvalidBlockNumber; \

  tmp->buffer = InvalidBuffer; \

  tmp->next = (d); \

  (d)=tmp; \

} while(0)

/*

 * GiST handler function: return IndexAmRoutine with access method parameters

 * and callbacks.

 */

Datum

gisthandler(PG_FUNCTION_ARGS)

{

  IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);

  amroutine->amstrategies = 0;

  amroutine->amsupport = GISTNProcs;

  amroutine->amoptsprocnum = GIST_OPTIONS_PROC;

  amroutine->amcanorder = false;

  amroutine->amcanorderbyop = true;

  amroutine->amcanhash = false;

  amroutine->amconsistentequality = false;

  amroutine->amconsistentordering = false;

  amroutine->amcanbackward = false;

  amroutine->amcanunique = false;

  amroutine->amcanmulticol = true;

  amroutine->amoptionalkey = true;

  amroutine->amsearcharray = false;

  amroutine->amsearchnulls = true;

  amroutine->amstorage = true;

  amroutine->amclusterable = true;

  amroutine->ampredlocks = true;

  amroutine->amcanparallel = false;

  amroutine->amcanbuildparallel = false;

  amroutine->amcaninclude = true;

  amroutine->amusemaintenanceworkmem = false;

  amroutine->amsummarizing = false;

  amroutine->amparallelvacuumoptions =

    VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_COND_CLEANUP;

  amroutine->amkeytype = InvalidOid;

  amroutine->ambuild = gistbuild;

  amroutine->ambuildempty = gistbuildempty;

  amroutine->aminsert = gistinsert;

  amroutine->aminsertcleanup = NULL;

  amroutine->ambulkdelete = gistbulkdelete;

  amroutine->amvacuumcleanup = gistvacuumcleanup;

  amroutine->amcanreturn = gistcanreturn;

  amroutine->amcostestimate = gistcostestimate;

  amroutine->amgettreeheight = NULL;

  amroutine->amoptions = gistoptions;

  amroutine->amproperty = gistproperty;

  amroutine->ambuildphasename = NULL;

  amroutine->amvalidate = gistvalidate;

  amroutine->amadjustmembers = gistadjustmembers;

  amroutine->ambeginscan = gistbeginscan;

  amroutine->amrescan = gistrescan;

  amroutine->amgettuple = gistgettuple;

  amroutine->amgetbitmap = gistgetbitmap;

  amroutine->amendscan = gistendscan;

  amroutine->ammarkpos = NULL;

  amroutine->amrestrpos = NULL;

  amroutine->amestimateparallelscan = NULL;

  amroutine->aminitparallelscan = NULL;

  amroutine->amparallelrescan = NULL;

  amroutine->amtranslatestrategy = NULL;

  amroutine->amtranslatecmptype = gisttranslatecmptype;

  PG_RETURN_POINTER(amroutine);

}

/*

 * Create and return a temporary memory context for use by GiST. We

 * _always_ invoke user-provided methods in a temporary memory

 * context, so that memory leaks in those functions cannot cause

 * problems. Also, we use some additional temporary contexts in the

 * GiST code itself, to avoid the need to do some awkward manual

 * memory management.

 */

MemoryContext

createTempGistContext(void)

{

  return AllocSetContextCreate(CurrentMemoryContext,

                 "GiST temporary context",

                 ALLOCSET_DEFAULT_SIZES);

}

/*

 *  gistbuildempty() -- build an empty gist index in the initialization fork

 */

void

gistbuildempty(Relation index)

{

  Buffer    buffer;

  /* Initialize the root page */

  buffer = ExtendBufferedRel(BMR_REL(index), INIT_FORKNUM, NULL,

                 EB_SKIP_EXTENSION_LOCK | EB_LOCK_FIRST);

  /* Initialize and xlog buffer */

  START_CRIT_SECTION();

  GISTInitBuffer(buffer, F_LEAF);

  MarkBufferDirty(buffer);

  log_newpage_buffer(buffer, true);

  END_CRIT_SECTION();

  /* Unlock and release the buffer */

  UnlockReleaseBuffer(buffer);

}

/*

 *  gistinsert -- wrapper for GiST tuple insertion.

 *

 *    This is the public interface routine for tuple insertion in GiSTs.

 *    It doesn't do any work; just locks the relation and passes the buck.

 */

bool

gistinsert(Relation r, Datum *values, bool *isnull,

       ItemPointer ht_ctid, Relation heapRel,

       IndexUniqueCheck checkUnique,

       bool indexUnchanged,

       IndexInfo *indexInfo)

{

  GISTSTATE  *giststate = (GISTSTATE *) indexInfo->ii_AmCache;

  IndexTuple  itup;

  MemoryContext oldCxt;

  /* Initialize GISTSTATE cache if first call in this statement */

  if (giststate == NULL)

  {

    oldCxt = MemoryContextSwitchTo(indexInfo->ii_Context);

    giststate = initGISTstate(r);

    giststate->tempCxt = createTempGistContext();

    indexInfo->ii_AmCache = giststate;

    MemoryContextSwitchTo(oldCxt);

  }

  oldCxt = MemoryContextSwitchTo(giststate->tempCxt);

  itup = gistFormTuple(giststate, r, values, isnull, true);

  itup->t_tid = *ht_ctid;

  gistdoinsert(r, itup, 0, giststate, heapRel, false);

  /* cleanup */

  MemoryContextSwitchTo(oldCxt);

  MemoryContextReset(giststate->tempCxt);

  return false;

}

/*

 * Place tuples from 'itup' to 'buffer'. If 'oldoffnum' is valid, the tuple

 * at that offset is atomically removed along with inserting the new tuples.

 * This is used to replace a tuple with a new one.

 *

 * If 'leftchildbuf' is valid, we're inserting the downlink for the page

 * to the right of 'leftchildbuf', or updating the downlink for 'leftchildbuf'.

 * F_FOLLOW_RIGHT flag on 'leftchildbuf' is cleared and NSN is set.

 *

 * If 'markfollowright' is true and the page is split, the left child is

 * marked with F_FOLLOW_RIGHT flag. That is the normal case. During buffered

 * index build, however, there is no concurrent access and the page splitting

 * is done in a slightly simpler fashion, and false is passed.

 *

 * If there is not enough room on the page, it is split. All the split

 * pages are kept pinned and locked and returned in *splitinfo, the caller

 * is responsible for inserting the downlinks for them. However, if

 * 'buffer' is the root page and it needs to be split, gistplacetopage()

 * performs the split as one atomic operation, and *splitinfo is set to NIL.

 * In that case, we continue to hold the root page locked, and the child

 * pages are released; note that new tuple(s) are *not* on the root page

 * but in one of the new child pages.

 *

 * If 'newblkno' is not NULL, returns the block number of page the first

 * new/updated tuple was inserted to. Usually it's the given page, but could

 * be its right sibling if the page was split.

 *

 * Returns 'true' if the page was split, 'false' otherwise.

 */

bool

gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,

        Buffer buffer,

        IndexTuple *itup, int ntup, OffsetNumber oldoffnum,

        BlockNumber *newblkno,

        Buffer leftchildbuf,

        List **splitinfo,

        bool markfollowright,

        Relation heapRel,

        bool is_build)

{

  BlockNumber blkno = BufferGetBlockNumber(buffer);

  Page    page = BufferGetPage(buffer);

  bool    is_leaf = (GistPageIsLeaf(page)) ? true : false;

  XLogRecPtr  recptr;

  bool    is_split;

  /*

   * Refuse to modify a page that's incompletely split. This should not

   * happen because we finish any incomplete splits while we walk down the

   * tree. However, it's remotely possible that another concurrent inserter

   * splits a parent page, and errors out before completing the split. We

   * will just throw an error in that case, and leave any split we had in

   * progress unfinished too. The next insert that comes along will clean up

   * the mess.

   */

  if (GistFollowRight(page))

    elog(ERROR, "concurrent GiST page split was incomplete");

  /* should never try to insert to a deleted page */

  Assert(!GistPageIsDeleted(page));

  *splitinfo = NIL;

  /*

   * if isupdate, remove old key: This node's key has been modified, either

   * because a child split occurred or because we needed to adjust our key

   * for an insert in a child node. Therefore, remove the old version of

   * this node's key.

   *

   * for WAL replay, in the non-split case we handle this by setting up a

   * one-element todelete array; in the split case, it's handled implicitly

   * because the tuple vector passed to gistSplit won't include this tuple.

   */

  is_split = gistnospace(page, itup, ntup, oldoffnum, freespace);

  /*

   * If leaf page is full, try at first to delete dead tuples. And then

   * check again.

   */

  if (is_split && GistPageIsLeaf(page) && GistPageHasGarbage(page))

  {

    gistprunepage(rel, page, buffer, heapRel);

    is_split = gistnospace(page, itup, ntup, oldoffnum, freespace);

  }

  if (is_split)

  {

    /* no space for insertion */

    IndexTuple *itvec;

    int     tlen;

    SplitPageLayout *dist = NULL,

           *ptr;

    BlockNumber oldrlink = InvalidBlockNumber;

    GistNSN   oldnsn = 0;

    SplitPageLayout rootpg;

    bool    is_rootsplit;

    int     npage;

    is_rootsplit = (blkno == GIST_ROOT_BLKNO);

    /*

     * Form index tuples vector to split. If we're replacing an old tuple,

     * remove the old version from the vector.

     */

    itvec = gistextractpage(page, &tlen);

    if (OffsetNumberIsValid(oldoffnum))

    {

      /* on inner page we should remove old tuple */

      int     pos = oldoffnum - FirstOffsetNumber;

      tlen--;

      if (pos != tlen)

        memmove(itvec + pos, itvec + pos + 1, sizeof(IndexTuple) * (tlen - pos));

    }

    itvec = gistjoinvector(itvec, &tlen, itup, ntup);

    dist = gistSplit(rel, page, itvec, tlen, giststate);

    /*

     * Check that split didn't produce too many pages.

     */

    npage = 0;

    for (ptr = dist; ptr; ptr = ptr->next)

      npage++;

    /* in a root split, we'll add one more page to the list below */

    if (is_rootsplit)

      npage++;

    if (npage > GIST_MAX_SPLIT_PAGES)

      elog(ERROR, "GiST page split into too many halves (%d, maximum %d)",

         npage, GIST_MAX_SPLIT_PAGES);

    /*

     * Set up pages to work with. Allocate new buffers for all but the

     * leftmost page. The original page becomes the new leftmost page, and

     * is just replaced with the new contents.

     *

     * For a root-split, allocate new buffers for all child pages, the

     * original page is overwritten with new root page containing

     * downlinks to the new child pages.

     */

    ptr = dist;

    if (!is_rootsplit)

    {

      /* save old rightlink and NSN */

      oldrlink = GistPageGetOpaque(page)->rightlink;

      oldnsn = GistPageGetNSN(page);

      dist->buffer = buffer;

      dist->block.blkno = BufferGetBlockNumber(buffer);

      dist->page = PageGetTempPageCopySpecial(BufferGetPage(buffer));

      /* clean all flags except F_LEAF */

      GistPageGetOpaque(dist->page)->flags = (is_leaf) ? F_LEAF : 0;

      ptr = ptr->next;

    }

    for (; ptr; ptr = ptr->next)

    {

      /* Allocate new page */

      ptr->buffer = gistNewBuffer(rel, heapRel);

      GISTInitBuffer(ptr->buffer, (is_leaf) ? F_LEAF : 0);

      ptr->page = BufferGetPage(ptr->buffer);

      ptr->block.blkno = BufferGetBlockNumber(ptr->buffer);

      PredicateLockPageSplit(rel,

                   BufferGetBlockNumber(buffer),

                   BufferGetBlockNumber(ptr->buffer));

    }

    /*

     * Now that we know which blocks the new pages go to, set up downlink

     * tuples to point to them.

     */

    for (ptr = dist; ptr; ptr = ptr->next)

    {

      ItemPointerSetBlockNumber(&(ptr->itup->t_tid), ptr->block.blkno);

      GistTupleSetValid(ptr->itup);

    }

    /*

     * If this is a root split, we construct the new root page with the

     * downlinks here directly, instead of requiring the caller to insert

     * them. Add the new root page to the list along with the child pages.

     */

    if (is_rootsplit)

    {

      IndexTuple *downlinks;

      int     ndownlinks = 0;

      int     i;

      rootpg.buffer = buffer;

      rootpg.page = PageGetTempPageCopySpecial(BufferGetPage(rootpg.buffer));

      GistPageGetOpaque(rootpg.page)->flags = 0;

      /* Prepare a vector of all the downlinks */

      for (ptr = dist; ptr; ptr = ptr->next)

        ndownlinks++;

      downlinks = palloc(sizeof(IndexTuple) * ndownlinks);

      for (i = 0, ptr = dist; ptr; ptr = ptr->next)

        downlinks[i++] = ptr->itup;

      rootpg.block.blkno = GIST_ROOT_BLKNO;

      rootpg.block.num = ndownlinks;

      rootpg.list = gistfillitupvec(downlinks, ndownlinks,

                      &(rootpg.lenlist));

      rootpg.itup = NULL;

      rootpg.next = dist;

      dist = &rootpg;

    }

    else

    {

      /* Prepare split-info to be returned to caller */

      for (ptr = dist; ptr; ptr = ptr->next)

      {

        GISTPageSplitInfo *si = palloc(sizeof(GISTPageSplitInfo));

        si->buf = ptr->buffer;

        si->downlink = ptr->itup;

        *splitinfo = lappend(*splitinfo, si);

      }

    }

    /*

     * Fill all pages. All the pages are new, ie. freshly allocated empty

     * pages, or a temporary copy of the old page.

     */

    for (ptr = dist; ptr; ptr = ptr->next)

    {

      char     *data = (char *) (ptr->list);

      for (int i = 0; i < ptr->block.num; i++)

      {

        IndexTuple  thistup = (IndexTuple) data;

        if (PageAddItem(ptr->page, (Item) data, IndexTupleSize(thistup), i + FirstOffsetNumber, false, false) == InvalidOffsetNumber)

          elog(ERROR, "failed to add item to index page in \"%s\"", RelationGetRelationName(rel));

        /*

         * If this is the first inserted/updated tuple, let the caller

         * know which page it landed on.

         */

        if (newblkno && ItemPointerEquals(&thistup->t_tid, &(*itup)->t_tid))

          *newblkno = ptr->block.blkno;

        data += IndexTupleSize(thistup);

      }

      /* Set up rightlinks */

      if (ptr->next && ptr->block.blkno != GIST_ROOT_BLKNO)

        GistPageGetOpaque(ptr->page)->rightlink =

          ptr->next->block.blkno;

      else

        GistPageGetOpaque(ptr->page)->rightlink = oldrlink;

      /*

       * Mark the all but the right-most page with the follow-right

       * flag. It will be cleared as soon as the downlink is inserted

       * into the parent, but this ensures that if we error out before

       * that, the index is still consistent. (in buffering build mode,

       * any error will abort the index build anyway, so this is not

       * needed.)

       */

      if (ptr->next && !is_rootsplit && markfollowright)

        GistMarkFollowRight(ptr->page);

      else

        GistClearFollowRight(ptr->page);

      /*

       * Copy the NSN of the original page to all pages. The

       * F_FOLLOW_RIGHT flags ensure that scans will follow the

       * rightlinks until the downlinks are inserted.

       */

      GistPageSetNSN(ptr->page, oldnsn);

    }

    /*

     * gistXLogSplit() needs to WAL log a lot of pages, prepare WAL

     * insertion for that. NB: The number of pages and data segments

     * specified here must match the calculations in gistXLogSplit()!

     */

    if (!is_build && RelationNeedsWAL(rel))

      XLogEnsureRecordSpace(npage, 1 + npage * 2);

    START_CRIT_SECTION();

    /*

     * Must mark buffers dirty before XLogInsert, even though we'll still

     * be changing their opaque fields below.

     */

    for (ptr = dist; ptr; ptr = ptr->next)

      MarkBufferDirty(ptr->buffer);

    if (BufferIsValid(leftchildbuf))

      MarkBufferDirty(leftchildbuf);

    /*

     * The first page in the chain was a temporary working copy meant to

     * replace the old page. Copy it over the old page.

     */

    PageRestoreTempPage(dist->page, BufferGetPage(dist->buffer));

    dist->page = BufferGetPage(dist->buffer);

    /*

     * Write the WAL record.

     *

     * If we're building a new index, however, we don't WAL-log changes

     * yet. The LSN-NSN interlock between parent and child requires that

     * LSNs never move backwards, so set the LSNs to a value that's

     * smaller than any real or fake unlogged LSN that might be generated

     * later. (There can't be any concurrent scans during index build, so

     * we don't need to be able to detect concurrent splits yet.)

     */

    if (is_build)

      recptr = GistBuildLSN;

    else

    {

      if (RelationNeedsWAL(rel))

        recptr = gistXLogSplit(is_leaf,

                     dist, oldrlink, oldnsn, leftchildbuf,

                     markfollowright);

      else

        recptr = gistGetFakeLSN(rel);

    }

    for (ptr = dist; ptr; ptr = ptr->next)

      PageSetLSN(ptr->page, recptr);

    /*

     * Return the new child buffers to the caller.

     *

     * If this was a root split, we've already inserted the downlink

     * pointers, in the form of a new root page. Therefore we can release

     * all the new buffers, and keep just the root page locked.

     */

    if (is_rootsplit)

    {

      for (ptr = dist->next; ptr; ptr = ptr->next)

        UnlockReleaseBuffer(ptr->buffer);

    }

  }

  else

  {

    /*

     * Enough space.  We always get here if ntup==0.

     */

    START_CRIT_SECTION();

    /*

     * Delete old tuple if any, then insert new tuple(s) if any.  If

     * possible, use the fast path of PageIndexTupleOverwrite.

     */

    if (OffsetNumberIsValid(oldoffnum))

    {

      if (ntup == 1)

      {

        /* One-for-one replacement, so use PageIndexTupleOverwrite */

        if (!PageIndexTupleOverwrite(page, oldoffnum, (Item) *itup,

                       IndexTupleSize(*itup)))

          elog(ERROR, "failed to add item to index page in \"%s\"",

             RelationGetRelationName(rel));

      }

      else

      {

        /* Delete old, then append new tuple(s) to page */

        PageIndexTupleDelete(page, oldoffnum);

        gistfillbuffer(page, itup, ntup, InvalidOffsetNumber);

      }

    }

    else

    {

      /* Just append new tuples at the end of the page */

      gistfillbuffer(page, itup, ntup, InvalidOffsetNumber);

    }

    MarkBufferDirty(buffer);

    if (BufferIsValid(leftchildbuf))

      MarkBufferDirty(leftchildbuf);

    if (is_build)

      recptr = GistBuildLSN;

    else

    {

      if (RelationNeedsWAL(rel))

      {

        OffsetNumber ndeloffs = 0,

              deloffs[1];

        if (OffsetNumberIsValid(oldoffnum))

        {

          deloffs[0] = oldoffnum;

          ndeloffs = 1;

        }

        recptr = gistXLogUpdate(buffer,

                    deloffs, ndeloffs, itup, ntup,

                    leftchildbuf);

      }

      else

        recptr = gistGetFakeLSN(rel);

    }

    PageSetLSN(page, recptr);

    if (newblkno)

      *newblkno = blkno;

  }

  /*

   * If we inserted the downlink for a child page, set NSN and clear

   * F_FOLLOW_RIGHT flag on the left child, so that concurrent scans know to

   * follow the rightlink if and only if they looked at the parent page

   * before we inserted the downlink.

   *

   * Note that we do this *after* writing the WAL record. That means that

   * the possible full page image in the WAL record does not include these

   * changes, and they must be replayed even if the page is restored from

   * the full page image. There's a chicken-and-egg problem: if we updated

   * the child pages first, we wouldn't know the recptr of the WAL record

   * we're about to write.

   */

  if (BufferIsValid(leftchildbuf))

  {

    Page    leftpg = BufferGetPage(leftchildbuf);

    GistPageSetNSN(leftpg, recptr);

    GistClearFollowRight(leftpg);

    PageSetLSN(leftpg, recptr);

  }

  END_CRIT_SECTION();

  return is_split;

}

/*

 * Workhorse routine for doing insertion into a GiST index. Note that

 * this routine assumes it is invoked in a short-lived memory context,

 * so it does not bother releasing palloc'd allocations.

 */

void

gistdoinsert(Relation r, IndexTuple itup, Size freespace,

       GISTSTATE *giststate, Relation heapRel, bool is_build)

{

  ItemId    iid;

  IndexTuple  idxtuple;

  GISTInsertStack firststack;

  GISTInsertStack *stack;

  GISTInsertState state;

  bool    xlocked = false;

  memset(&state, 0, sizeof(GISTInsertState));

  state.freespace = freespace;

  state.r = r;

  state.heapRel = heapRel;

  state.is_build = is_build;

  /* Start from the root */

  firststack.blkno = GIST_ROOT_BLKNO;

  firststack.lsn = 0;

  firststack.retry_from_parent = false;

  firststack.parent = NULL;

  firststack.downlinkoffnum = InvalidOffsetNumber;

  state.stack = stack = &firststack;

  /*

   * Walk down along the path of smallest penalty, updating the parent

   * pointers with the key we're inserting as we go. If we crash in the

   * middle, the tree is consistent, although the possible parent updates

   * were a waste.

   */

  for (;;)

  {

    /*

     * If we split an internal page while descending the tree, we have to

     * retry at the parent. (Normally, the LSN-NSN interlock below would

     * also catch this and cause us to retry. But LSNs are not updated

     * during index build.)

     */

    while (stack->retry_from_parent)

    {

      if (xlocked)

        LockBuffer(stack->buffer, GIST_UNLOCK);

      xlocked = false;

      ReleaseBuffer(stack->buffer);

      state.stack = stack = stack->parent;

    }

    if (XLogRecPtrIsInvalid(stack->lsn))

      stack->buffer = ReadBuffer(state.r, stack->blkno);

    /*

     * Be optimistic and grab shared lock first. Swap it for an exclusive

     * lock later if we need to update the page.

     */

    if (!xlocked)

    {

      LockBuffer(stack->buffer, GIST_SHARE);

      gistcheckpage(state.r, stack->buffer);

    }

    stack->page = (Page) BufferGetPage(stack->buffer);

    stack->lsn = xlocked ?

      PageGetLSN(stack->page) : BufferGetLSNAtomic(stack->buffer);

    Assert(!RelationNeedsWAL(state.r) || !XLogRecPtrIsInvalid(stack->lsn));

    /*

     * If this page was split but the downlink was never inserted to the

     * parent because the inserting backend crashed before doing that, fix

     * that now.

     */

    if (GistFollowRight(stack->page))

    {

      if (!xlocked)

      {

        LockBuffer(stack->buffer, GIST_UNLOCK);

        LockBuffer(stack->buffer, GIST_EXCLUSIVE);

        xlocked = true;

        /* someone might've completed the split when we unlocked */

        if (!GistFollowRight(stack->page))

          continue;

      }

      gistfixsplit(&state, giststate);

      UnlockReleaseBuffer(stack->buffer);

      xlocked = false;

      state.stack = stack = stack->parent;

      continue;

    }

    if ((stack->blkno != GIST_ROOT_BLKNO &&

       stack->parent->lsn < GistPageGetNSN(stack->page)) ||

      GistPageIsDeleted(stack->page))

    {

      /*

       * Concurrent split or page deletion detected. There's no

       * guarantee that the downlink for this page is consistent with

       * the tuple we're inserting anymore, so go back to parent and

       * rechoose the best child.

       */

      UnlockReleaseBuffer(stack->buffer);

      xlocked = false;

      state.stack = stack = stack->parent;

      continue;

    }

    if (!GistPageIsLeaf(stack->page))

    {

      /*

       * This is an internal page so continue to walk down the tree.

       * Find the child node that has the minimum insertion penalty.

       */

      BlockNumber childblkno;

      IndexTuple  newtup;

      GISTInsertStack *item;

      OffsetNumber downlinkoffnum;

      downlinkoffnum = gistchoose(state.r, stack->page, itup, giststate);

      iid = PageGetItemId(stack->page, downlinkoffnum);

      idxtuple = (IndexTuple) PageGetItem(stack->page, iid);

      childblkno = ItemPointerGetBlockNumber(&(idxtuple->t_tid));

      /*

       * Check that it's not a leftover invalid tuple from pre-9.1

       */

      if (GistTupleIsInvalid(idxtuple))

        ereport(ERROR,

            (errmsg("index \"%s\" contains an inner tuple marked as invalid",

                RelationGetRelationName(r)),

             errdetail("This is caused by an incomplete page split at crash recovery before upgrading to PostgreSQL 9.1."),

             errhint("Please REINDEX it.")));

      /*

       * Check that the key representing the target child node is

       * consistent with the key we're inserting. Update it if it's not.

       */

      newtup = gistgetadjusted(state.r, idxtuple, itup, giststate);

      if (newtup)

      {

        /*

         * Swap shared lock for an exclusive one. Beware, the page may

         * change while we unlock/lock the page...

         */

        if (!xlocked)

        {

          LockBuffer(stack->buffer, GIST_UNLOCK);

          LockBuffer(stack->buffer, GIST_EXCLUSIVE);

          xlocked = true;

          stack->page = (Page) BufferGetPage(stack->buffer);

          if (PageGetLSN(stack->page) != stack->lsn)

          {

            /* the page was changed while we unlocked it, retry */

            continue;

          }

        }

        /*

         * Update the tuple.

         *

         * We still hold the lock after gistinserttuple(), but it

         * might have to split the page to make the updated tuple fit.

         * In that case the updated tuple might migrate to the other

         * half of the split, so we have to go back to the parent and

         * descend back to the half that's a better fit for the new

         * tuple.

         */

        if (gistinserttuple(&state, stack, giststate, newtup,

                  downlinkoffnum))

        {

          /*

           * If this was a root split, the root page continues to be

           * the parent and the updated tuple went to one of the

           * child pages, so we just need to retry from the root

           * page.

           */

          if (stack->blkno != GIST_ROOT_BLKNO)

          {

            UnlockReleaseBuffer(stack->buffer);

            xlocked = false;

            state.stack = stack = stack->parent;

          }

          continue;

        }

      }

      LockBuffer(stack->buffer, GIST_UNLOCK);

      xlocked = false;

      /* descend to the chosen child */

      item = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));

      item->blkno = childblkno;

      item->parent = stack;

      item->downlinkoffnum = downlinkoffnum;

      state.stack = stack = item;

    }

    else

    {

      /*

       * Leaf page. Insert the new key. We've already updated all the

       * parents on the way down, but we might have to split the page if

       * it doesn't fit. gistinserttuple() will take care of that.

       */

      /*

       * Swap shared lock for an exclusive one. Be careful, the page may

       * change while we unlock/lock the page...

       */

      if (!xlocked)

      {

        LockBuffer(stack->buffer, GIST_UNLOCK);

        LockBuffer(stack->buffer, GIST_EXCLUSIVE);

        xlocked = true;

        stack->page = (Page) BufferGetPage(stack->buffer);

        stack->lsn = PageGetLSN(stack->page);

        if (stack->blkno == GIST_ROOT_BLKNO)

        {

          /*

           * the only page that can become inner instead of leaf is

           * the root page, so for root we should recheck it

           */

          if (!GistPageIsLeaf(stack->page))

          {

            /*

             * very rare situation: during unlock/lock index with

             * number of pages = 1 was increased

             */

            LockBuffer(stack->buffer, GIST_UNLOCK);

            xlocked = false;

            continue;

          }

          /*

           * we don't need to check root split, because checking

           * leaf/inner is enough to recognize split for root

           */

        }

        else if ((GistFollowRight(stack->page) ||

              stack->parent->lsn < GistPageGetNSN(stack->page)) ||

             GistPageIsDeleted(stack->page))

        {

          /*

           * The page was split or deleted while we momentarily

           * unlocked the page. Go back to parent.

           */

          UnlockReleaseBuffer(stack->buffer);

          xlocked = false;

          state.stack = stack = stack->parent;

          continue;

        }

      }

      /* now state.stack->(page, buffer and blkno) points to leaf page */

      gistinserttuple(&state, stack, giststate, itup,

              InvalidOffsetNumber);

      LockBuffer(stack->buffer, GIST_UNLOCK);

      /* Release any pins we might still hold before exiting */

      for (; stack; stack = stack->parent)

        ReleaseBuffer(stack->buffer);

      break;

    }

  }

}

/*

 * Traverse the tree to find path from root page to specified "child" block.

 *

 * returns a new insertion stack, starting from the parent of "child", up

 * to the root. *downlinkoffnum is set to the offset of the downlink in the

 * direct parent of child.

 *

 * To prevent deadlocks, this should lock only one page at a time.

 */

static GISTInsertStack *

gistFindPath(Relation r, BlockNumber child, OffsetNumber *downlinkoffnum)

{

  Page    page;

  Buffer    buffer;

  OffsetNumber i,

        maxoff;

  ItemId    iid;

  IndexTuple  idxtuple;

  List     *fifo;

  GISTInsertStack *top,

         *ptr;

  BlockNumber blkno;

  top = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));

  top->blkno = GIST_ROOT_BLKNO;

  top->downlinkoffnum = InvalidOffsetNumber;

  fifo = list_make1(top);