LCOV - code coverage report
Current view: top level - src/backend/access/gist - gist.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13beta1 Lines: 395 568 69.5 %
Date: 2020-05-31 21:07:12 Functions: 12 17 70.6 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * gist.c
       4             :  *    interface routines for the postgres GiST index access method.
       5             :  *
       6             :  *
       7             :  * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
       8             :  * Portions Copyright (c) 1994, Regents of the University of California
       9             :  *
      10             :  * IDENTIFICATION
      11             :  *    src/backend/access/gist/gist.c
      12             :  *
      13             :  *-------------------------------------------------------------------------
      14             :  */
      15             : #include "postgres.h"
      16             : 
      17             : #include "access/gist_private.h"
      18             : #include "access/gistscan.h"
      19             : #include "catalog/pg_collation.h"
      20             : #include "commands/vacuum.h"
      21             : #include "miscadmin.h"
      22             : #include "nodes/execnodes.h"
      23             : #include "storage/lmgr.h"
      24             : #include "storage/predicate.h"
      25             : #include "utils/builtins.h"
      26             : #include "utils/index_selfuncs.h"
      27             : #include "utils/memutils.h"
      28             : #include "utils/rel.h"
      29             : 
      30             : /* non-export function prototypes */
      31             : static void gistfixsplit(GISTInsertState *state, GISTSTATE *giststate);
      32             : static bool gistinserttuple(GISTInsertState *state, GISTInsertStack *stack,
      33             :                             GISTSTATE *giststate, IndexTuple tuple, OffsetNumber oldoffnum);
      34             : static bool gistinserttuples(GISTInsertState *state, GISTInsertStack *stack,
      35             :                              GISTSTATE *giststate,
      36             :                              IndexTuple *tuples, int ntup, OffsetNumber oldoffnum,
      37             :                              Buffer leftchild, Buffer rightchild,
      38             :                              bool unlockbuf, bool unlockleftchild);
      39             : static void gistfinishsplit(GISTInsertState *state, GISTInsertStack *stack,
      40             :                             GISTSTATE *giststate, List *splitinfo, bool unlockbuf);
      41             : static void gistprunepage(Relation rel, Page page, Buffer buffer,
      42             :                           Relation heapRel);
      43             : 
      44             : 
      45             : #define ROTATEDIST(d) do { \
      46             :     SplitedPageLayout *tmp=(SplitedPageLayout*)palloc0(sizeof(SplitedPageLayout)); \
      47             :     tmp->block.blkno = InvalidBlockNumber;   \
      48             :     tmp->buffer = InvalidBuffer; \
      49             :     tmp->next = (d); \
      50             :     (d)=tmp; \
      51             : } while(0)
      52             : 
      53             : 
      54             : /*
      55             :  * GiST handler function: return IndexAmRoutine with access method parameters
      56             :  * and callbacks.
      57             :  */
      58             : Datum
      59        2760 : gisthandler(PG_FUNCTION_ARGS)
      60             : {
      61        2760 :     IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);
      62             : 
      63        2760 :     amroutine->amstrategies = 0;
      64        2760 :     amroutine->amsupport = GISTNProcs;
      65        2760 :     amroutine->amoptsprocnum = GIST_OPTIONS_PROC;
      66        2760 :     amroutine->amcanorder = false;
      67        2760 :     amroutine->amcanorderbyop = true;
      68        2760 :     amroutine->amcanbackward = false;
      69        2760 :     amroutine->amcanunique = false;
      70        2760 :     amroutine->amcanmulticol = true;
      71        2760 :     amroutine->amoptionalkey = true;
      72        2760 :     amroutine->amsearcharray = false;
      73        2760 :     amroutine->amsearchnulls = true;
      74        2760 :     amroutine->amstorage = true;
      75        2760 :     amroutine->amclusterable = true;
      76        2760 :     amroutine->ampredlocks = true;
      77        2760 :     amroutine->amcanparallel = false;
      78        2760 :     amroutine->amcaninclude = true;
      79        2760 :     amroutine->amusemaintenanceworkmem = false;
      80        2760 :     amroutine->amparallelvacuumoptions =
      81             :         VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_COND_CLEANUP;
      82        2760 :     amroutine->amkeytype = InvalidOid;
      83             : 
      84        2760 :     amroutine->ambuild = gistbuild;
      85        2760 :     amroutine->ambuildempty = gistbuildempty;
      86        2760 :     amroutine->aminsert = gistinsert;
      87        2760 :     amroutine->ambulkdelete = gistbulkdelete;
      88        2760 :     amroutine->amvacuumcleanup = gistvacuumcleanup;
      89        2760 :     amroutine->amcanreturn = gistcanreturn;
      90        2760 :     amroutine->amcostestimate = gistcostestimate;
      91        2760 :     amroutine->amoptions = gistoptions;
      92        2760 :     amroutine->amproperty = gistproperty;
      93        2760 :     amroutine->ambuildphasename = NULL;
      94        2760 :     amroutine->amvalidate = gistvalidate;
      95        2760 :     amroutine->ambeginscan = gistbeginscan;
      96        2760 :     amroutine->amrescan = gistrescan;
      97        2760 :     amroutine->amgettuple = gistgettuple;
      98        2760 :     amroutine->amgetbitmap = gistgetbitmap;
      99        2760 :     amroutine->amendscan = gistendscan;
     100        2760 :     amroutine->ammarkpos = NULL;
     101        2760 :     amroutine->amrestrpos = NULL;
     102        2760 :     amroutine->amestimateparallelscan = NULL;
     103        2760 :     amroutine->aminitparallelscan = NULL;
     104        2760 :     amroutine->amparallelrescan = NULL;
     105             : 
     106        2760 :     PG_RETURN_POINTER(amroutine);
     107             : }
     108             : 
     109             : /*
     110             :  * Create and return a temporary memory context for use by GiST. We
     111             :  * _always_ invoke user-provided methods in a temporary memory
     112             :  * context, so that memory leaks in those functions cannot cause
     113             :  * problems. Also, we use some additional temporary contexts in the
     114             :  * GiST code itself, to avoid the need to do some awkward manual
     115             :  * memory management.
     116             :  */
     117             : MemoryContext
     118        2630 : createTempGistContext(void)
     119             : {
     120        2630 :     return AllocSetContextCreate(CurrentMemoryContext,
     121             :                                  "GiST temporary context",
     122             :                                  ALLOCSET_DEFAULT_SIZES);
     123             : }
     124             : 
     125             : /*
     126             :  *  gistbuildempty() -- build an empty gist index in the initialization fork
     127             :  */
     128             : void
     129           0 : gistbuildempty(Relation index)
     130             : {
     131             :     Buffer      buffer;
     132             : 
     133             :     /* Initialize the root page */
     134           0 :     buffer = ReadBufferExtended(index, INIT_FORKNUM, P_NEW, RBM_NORMAL, NULL);
     135           0 :     LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
     136             : 
     137             :     /* Initialize and xlog buffer */
     138           0 :     START_CRIT_SECTION();
     139           0 :     GISTInitBuffer(buffer, F_LEAF);
     140           0 :     MarkBufferDirty(buffer);
     141           0 :     log_newpage_buffer(buffer, true);
     142           0 :     END_CRIT_SECTION();
     143             : 
     144             :     /* Unlock and release the buffer */
     145           0 :     UnlockReleaseBuffer(buffer);
     146           0 : }
     147             : 
     148             : /*
     149             :  *  gistinsert -- wrapper for GiST tuple insertion.
     150             :  *
     151             :  *    This is the public interface routine for tuple insertion in GiSTs.
     152             :  *    It doesn't do any work; just locks the relation and passes the buck.
     153             :  */
     154             : bool
     155      230228 : gistinsert(Relation r, Datum *values, bool *isnull,
     156             :            ItemPointer ht_ctid, Relation heapRel,
     157             :            IndexUniqueCheck checkUnique,
     158             :            IndexInfo *indexInfo)
     159             : {
     160      230228 :     GISTSTATE  *giststate = (GISTSTATE *) indexInfo->ii_AmCache;
     161             :     IndexTuple  itup;
     162             :     MemoryContext oldCxt;
     163             : 
     164             :     /* Initialize GISTSTATE cache if first call in this statement */
     165      230228 :     if (giststate == NULL)
     166             :     {
     167         406 :         oldCxt = MemoryContextSwitchTo(indexInfo->ii_Context);
     168         406 :         giststate = initGISTstate(r);
     169         406 :         giststate->tempCxt = createTempGistContext();
     170         406 :         indexInfo->ii_AmCache = (void *) giststate;
     171         406 :         MemoryContextSwitchTo(oldCxt);
     172             :     }
     173             : 
     174      230228 :     oldCxt = MemoryContextSwitchTo(giststate->tempCxt);
     175             : 
     176      230228 :     itup = gistFormTuple(giststate, r,
     177             :                          values, isnull, true /* size is currently bogus */ );
     178      230228 :     itup->t_tid = *ht_ctid;
     179             : 
     180      230228 :     gistdoinsert(r, itup, 0, giststate, heapRel, false);
     181             : 
     182             :     /* cleanup */
     183      230216 :     MemoryContextSwitchTo(oldCxt);
     184      230216 :     MemoryContextReset(giststate->tempCxt);
     185             : 
     186      230216 :     return false;
     187             : }
     188             : 
     189             : 
     190             : /*
     191             :  * Place tuples from 'itup' to 'buffer'. If 'oldoffnum' is valid, the tuple
     192             :  * at that offset is atomically removed along with inserting the new tuples.
     193             :  * This is used to replace a tuple with a new one.
     194             :  *
     195             :  * If 'leftchildbuf' is valid, we're inserting the downlink for the page
     196             :  * to the right of 'leftchildbuf', or updating the downlink for 'leftchildbuf'.
     197             :  * F_FOLLOW_RIGHT flag on 'leftchildbuf' is cleared and NSN is set.
     198             :  *
     199             :  * If 'markfollowright' is true and the page is split, the left child is
     200             :  * marked with F_FOLLOW_RIGHT flag. That is the normal case. During buffered
     201             :  * index build, however, there is no concurrent access and the page splitting
     202             :  * is done in a slightly simpler fashion, and false is passed.
     203             :  *
     204             :  * If there is not enough room on the page, it is split. All the split
     205             :  * pages are kept pinned and locked and returned in *splitinfo, the caller
     206             :  * is responsible for inserting the downlinks for them. However, if
     207             :  * 'buffer' is the root page and it needs to be split, gistplacetopage()
     208             :  * performs the split as one atomic operation, and *splitinfo is set to NIL.
     209             :  * In that case, we continue to hold the root page locked, and the child
     210             :  * pages are released; note that new tuple(s) are *not* on the root page
     211             :  * but in one of the new child pages.
     212             :  *
     213             :  * If 'newblkno' is not NULL, returns the block number of page the first
     214             :  * new/updated tuple was inserted to. Usually it's the given page, but could
     215             :  * be its right sibling if the page was split.
     216             :  *
     217             :  * Returns 'true' if the page was split, 'false' otherwise.
     218             :  */
     219             : bool
     220     1283072 : gistplacetopage(Relation rel, Size freespace, GISTSTATE *giststate,
     221             :                 Buffer buffer,
     222             :                 IndexTuple *itup, int ntup, OffsetNumber oldoffnum,
     223             :                 BlockNumber *newblkno,
     224             :                 Buffer leftchildbuf,
     225             :                 List **splitinfo,
     226             :                 bool markfollowright,
     227             :                 Relation heapRel,
     228             :                 bool is_build)
     229             : {
     230     1283072 :     BlockNumber blkno = BufferGetBlockNumber(buffer);
     231     1283072 :     Page        page = BufferGetPage(buffer);
     232     1283072 :     bool        is_leaf = (GistPageIsLeaf(page)) ? true : false;
     233             :     XLogRecPtr  recptr;
     234             :     int         i;
     235             :     bool        is_split;
     236             : 
     237             :     /*
     238             :      * Refuse to modify a page that's incompletely split. This should not
     239             :      * happen because we finish any incomplete splits while we walk down the
     240             :      * tree. However, it's remotely possible that another concurrent inserter
     241             :      * splits a parent page, and errors out before completing the split. We
     242             :      * will just throw an error in that case, and leave any split we had in
     243             :      * progress unfinished too. The next insert that comes along will clean up
     244             :      * the mess.
     245             :      */
     246     1283072 :     if (GistFollowRight(page))
     247           0 :         elog(ERROR, "concurrent GiST page split was incomplete");
     248             : 
     249     1283072 :     *splitinfo = NIL;
     250             : 
     251             :     /*
     252             :      * if isupdate, remove old key: This node's key has been modified, either
     253             :      * because a child split occurred or because we needed to adjust our key
     254             :      * for an insert in a child node. Therefore, remove the old version of
     255             :      * this node's key.
     256             :      *
     257             :      * for WAL replay, in the non-split case we handle this by setting up a
     258             :      * one-element todelete array; in the split case, it's handled implicitly
     259             :      * because the tuple vector passed to gistSplit won't include this tuple.
     260             :      */
     261     1283072 :     is_split = gistnospace(page, itup, ntup, oldoffnum, freespace);
     262             : 
     263             :     /*
     264             :      * If leaf page is full, try at first to delete dead tuples. And then
     265             :      * check again.
     266             :      */
     267     1283072 :     if (is_split && GistPageIsLeaf(page) && GistPageHasGarbage(page))
     268             :     {
     269           0 :         gistprunepage(rel, page, buffer, heapRel);
     270           0 :         is_split = gistnospace(page, itup, ntup, oldoffnum, freespace);
     271             :     }
     272             : 
     273     1283072 :     if (is_split)
     274             :     {
     275             :         /* no space for insertion */
     276             :         IndexTuple *itvec;
     277             :         int         tlen;
     278       21924 :         SplitedPageLayout *dist = NULL,
     279             :                    *ptr;
     280       21924 :         BlockNumber oldrlink = InvalidBlockNumber;
     281       21924 :         GistNSN     oldnsn = 0;
     282             :         SplitedPageLayout rootpg;
     283             :         bool        is_rootsplit;
     284             :         int         npage;
     285             : 
     286       21924 :         is_rootsplit = (blkno == GIST_ROOT_BLKNO);
     287             : 
     288             :         /*
     289             :          * Form index tuples vector to split. If we're replacing an old tuple,
     290             :          * remove the old version from the vector.
     291             :          */
     292       21924 :         itvec = gistextractpage(page, &tlen);
     293       21924 :         if (OffsetNumberIsValid(oldoffnum))
     294             :         {
     295             :             /* on inner page we should remove old tuple */
     296        4686 :             int         pos = oldoffnum - FirstOffsetNumber;
     297             : 
     298        4686 :             tlen--;
     299        4686 :             if (pos != tlen)
     300        2894 :                 memmove(itvec + pos, itvec + pos + 1, sizeof(IndexTuple) * (tlen - pos));
     301             :         }
     302       21924 :         itvec = gistjoinvector(itvec, &tlen, itup, ntup);
     303       21924 :         dist = gistSplit(rel, page, itvec, tlen, giststate);
     304             : 
     305             :         /*
     306             :          * Check that split didn't produce too many pages.
     307             :          */
     308       21924 :         npage = 0;
     309       65830 :         for (ptr = dist; ptr; ptr = ptr->next)
     310       43906 :             npage++;
     311             :         /* in a root split, we'll add one more page to the list below */
     312       21924 :         if (is_rootsplit)
     313         330 :             npage++;
     314       21924 :         if (npage > GIST_MAX_SPLIT_PAGES)
     315           0 :             elog(ERROR, "GiST page split into too many halves (%d, maximum %d)",
     316             :                  npage, GIST_MAX_SPLIT_PAGES);
     317             : 
     318             :         /*
     319             :          * Set up pages to work with. Allocate new buffers for all but the
     320             :          * leftmost page. The original page becomes the new leftmost page, and
     321             :          * is just replaced with the new contents.
     322             :          *
     323             :          * For a root-split, allocate new buffers for all child pages, the
     324             :          * original page is overwritten with new root page containing
     325             :          * downlinks to the new child pages.
     326             :          */
     327       21924 :         ptr = dist;
     328       21924 :         if (!is_rootsplit)
     329             :         {
     330             :             /* save old rightlink and NSN */
     331       21594 :             oldrlink = GistPageGetOpaque(page)->rightlink;
     332       21594 :             oldnsn = GistPageGetNSN(page);
     333             : 
     334       21594 :             dist->buffer = buffer;
     335       21594 :             dist->block.blkno = BufferGetBlockNumber(buffer);
     336       21594 :             dist->page = PageGetTempPageCopySpecial(BufferGetPage(buffer));
     337             : 
     338             :             /* clean all flags except F_LEAF */
     339       21594 :             GistPageGetOpaque(dist->page)->flags = (is_leaf) ? F_LEAF : 0;
     340             : 
     341       21594 :             ptr = ptr->next;
     342             :         }
     343       44236 :         for (; ptr; ptr = ptr->next)
     344             :         {
     345             :             /* Allocate new page */
     346       22312 :             ptr->buffer = gistNewBuffer(rel);
     347       22312 :             GISTInitBuffer(ptr->buffer, (is_leaf) ? F_LEAF : 0);
     348       22312 :             ptr->page = BufferGetPage(ptr->buffer);
     349       22312 :             ptr->block.blkno = BufferGetBlockNumber(ptr->buffer);
     350       22312 :             PredicateLockPageSplit(rel,
     351             :                                    BufferGetBlockNumber(buffer),
     352             :                                    BufferGetBlockNumber(ptr->buffer));
     353             :         }
     354             : 
     355             :         /*
     356             :          * Now that we know which blocks the new pages go to, set up downlink
     357             :          * tuples to point to them.
     358             :          */
     359       65830 :         for (ptr = dist; ptr; ptr = ptr->next)
     360             :         {
     361       43906 :             ItemPointerSetBlockNumber(&(ptr->itup->t_tid), ptr->block.blkno);
     362       43906 :             GistTupleSetValid(ptr->itup);
     363             :         }
     364             : 
     365             :         /*
     366             :          * If this is a root split, we construct the new root page with the
     367             :          * downlinks here directly, instead of requiring the caller to insert
     368             :          * them. Add the new root page to the list along with the child pages.
     369             :          */
     370       21924 :         if (is_rootsplit)
     371             :         {
     372             :             IndexTuple *downlinks;
     373         330 :             int         ndownlinks = 0;
     374             :             int         i;
     375             : 
     376         330 :             rootpg.buffer = buffer;
     377         330 :             rootpg.page = PageGetTempPageCopySpecial(BufferGetPage(rootpg.buffer));
     378         330 :             GistPageGetOpaque(rootpg.page)->flags = 0;
     379             : 
     380             :             /* Prepare a vector of all the downlinks */
     381         992 :             for (ptr = dist; ptr; ptr = ptr->next)
     382         662 :                 ndownlinks++;
     383         330 :             downlinks = palloc(sizeof(IndexTuple) * ndownlinks);
     384         992 :             for (i = 0, ptr = dist; ptr; ptr = ptr->next)
     385         662 :                 downlinks[i++] = ptr->itup;
     386             : 
     387         330 :             rootpg.block.blkno = GIST_ROOT_BLKNO;
     388         330 :             rootpg.block.num = ndownlinks;
     389         330 :             rootpg.list = gistfillitupvec(downlinks, ndownlinks,
     390             :                                           &(rootpg.lenlist));
     391         330 :             rootpg.itup = NULL;
     392             : 
     393         330 :             rootpg.next = dist;
     394         330 :             dist = &rootpg;
     395             :         }
     396             :         else
     397             :         {
     398             :             /* Prepare split-info to be returned to caller */
     399       64838 :             for (ptr = dist; ptr; ptr = ptr->next)
     400             :             {
     401       43244 :                 GISTPageSplitInfo *si = palloc(sizeof(GISTPageSplitInfo));
     402             : 
     403       43244 :                 si->buf = ptr->buffer;
     404       43244 :                 si->downlink = ptr->itup;
     405       43244 :                 *splitinfo = lappend(*splitinfo, si);
     406             :             }
     407             :         }
     408             : 
     409             :         /*
     410             :          * Fill all pages. All the pages are new, ie. freshly allocated empty
     411             :          * pages, or a temporary copy of the old page.
     412             :          */
     413       66160 :         for (ptr = dist; ptr; ptr = ptr->next)
     414             :         {
     415       44236 :             char       *data = (char *) (ptr->list);
     416             : 
     417     1440024 :             for (i = 0; i < ptr->block.num; i++)
     418             :             {
     419     1395788 :                 IndexTuple  thistup = (IndexTuple) data;
     420             : 
     421     1395788 :                 if (PageAddItem(ptr->page, (Item) data, IndexTupleSize(thistup), i + FirstOffsetNumber, false, false) == InvalidOffsetNumber)
     422           0 :                     elog(ERROR, "failed to add item to index page in \"%s\"", RelationGetRelationName(rel));
     423             : 
     424             :                 /*
     425             :                  * If this is the first inserted/updated tuple, let the caller
     426             :                  * know which page it landed on.
     427             :                  */
     428     1395788 :                 if (newblkno && ItemPointerEquals(&thistup->t_tid, &(*itup)->t_tid))
     429           0 :                     *newblkno = ptr->block.blkno;
     430             : 
     431     1395788 :                 data += IndexTupleSize(thistup);
     432             :             }
     433             : 
     434             :             /* Set up rightlinks */
     435       44236 :             if (ptr->next && ptr->block.blkno != GIST_ROOT_BLKNO)
     436       43964 :                 GistPageGetOpaque(ptr->page)->rightlink =
     437       21982 :                     ptr->next->block.blkno;
     438             :             else
     439       22254 :                 GistPageGetOpaque(ptr->page)->rightlink = oldrlink;
     440             : 
     441             :             /*
     442             :              * Mark the all but the right-most page with the follow-right
     443             :              * flag. It will be cleared as soon as the downlink is inserted
     444             :              * into the parent, but this ensures that if we error out before
     445             :              * that, the index is still consistent. (in buffering build mode,
     446             :              * any error will abort the index build anyway, so this is not
     447             :              * needed.)
     448             :              */
     449       44236 :             if (ptr->next && !is_rootsplit && markfollowright)
     450       21650 :                 GistMarkFollowRight(ptr->page);
     451             :             else
     452       22586 :                 GistClearFollowRight(ptr->page);
     453             : 
     454             :             /*
     455             :              * Copy the NSN of the original page to all pages. The
     456             :              * F_FOLLOW_RIGHT flags ensure that scans will follow the
     457             :              * rightlinks until the downlinks are inserted.
     458             :              */
     459       44236 :             GistPageSetNSN(ptr->page, oldnsn);
     460             :         }
     461             : 
     462             :         /*
     463             :          * gistXLogSplit() needs to WAL log a lot of pages, prepare WAL
     464             :          * insertion for that. NB: The number of pages and data segments
     465             :          * specified here must match the calculations in gistXLogSplit()!
     466             :          */
     467       21924 :         if (!is_build && RelationNeedsWAL(rel))
     468        2534 :             XLogEnsureRecordSpace(npage, 1 + npage * 2);
     469             : 
     470       21924 :         START_CRIT_SECTION();
     471             : 
     472             :         /*
     473             :          * Must mark buffers dirty before XLogInsert, even though we'll still
     474             :          * be changing their opaque fields below.
     475             :          */
     476       66160 :         for (ptr = dist; ptr; ptr = ptr->next)
     477       44236 :             MarkBufferDirty(ptr->buffer);
     478       21924 :         if (BufferIsValid(leftchildbuf))
     479        4662 :             MarkBufferDirty(leftchildbuf);
     480             : 
     481             :         /*
     482             :          * The first page in the chain was a temporary working copy meant to
     483             :          * replace the old page. Copy it over the old page.
     484             :          */
     485       21924 :         PageRestoreTempPage(dist->page, BufferGetPage(dist->buffer));
     486       21924 :         dist->page = BufferGetPage(dist->buffer);
     487             : 
     488             :         /*
     489             :          * Write the WAL record.
     490             :          *
     491             :          * If we're building a new index, however, we don't WAL-log changes
     492             :          * yet. The LSN-NSN interlock between parent and child requires that
     493             :          * LSNs never move backwards, so set the LSNs to a value that's
     494             :          * smaller than any real or fake unlogged LSN that might be generated
     495             :          * later. (There can't be any concurrent scans during index build, so
     496             :          * we don't need to be able to detect concurrent splits yet.)
     497             :          */
     498       21924 :         if (is_build)
     499       19390 :             recptr = GistBuildLSN;
     500             :         else
     501             :         {
     502        2534 :             if (RelationNeedsWAL(rel))
     503        2534 :                 recptr = gistXLogSplit(is_leaf,
     504             :                                        dist, oldrlink, oldnsn, leftchildbuf,
     505             :                                        markfollowright);
     506             :             else
     507           0 :                 recptr = gistGetFakeLSN(rel);
     508             :         }
     509             : 
     510       66160 :         for (ptr = dist; ptr; ptr = ptr->next)
     511       44236 :             PageSetLSN(ptr->page, recptr);
     512             : 
     513             :         /*
     514             :          * Return the new child buffers to the caller.
     515             :          *
     516             :          * If this was a root split, we've already inserted the downlink
     517             :          * pointers, in the form of a new root page. Therefore we can release
     518             :          * all the new buffers, and keep just the root page locked.
     519             :          */
     520       21924 :         if (is_rootsplit)
     521             :         {
     522         992 :             for (ptr = dist->next; ptr; ptr = ptr->next)
     523         662 :                 UnlockReleaseBuffer(ptr->buffer);
     524             :         }
     525             :     }
     526             :     else
     527             :     {
     528             :         /*
     529             :          * Enough space.  We always get here if ntup==0.
     530             :          */
     531     1261148 :         START_CRIT_SECTION();
     532             : 
     533             :         /*
     534             :          * Delete old tuple if any, then insert new tuple(s) if any.  If
     535             :          * possible, use the fast path of PageIndexTupleOverwrite.
     536             :          */
     537     1261148 :         if (OffsetNumberIsValid(oldoffnum))
     538             :         {
     539      530816 :             if (ntup == 1)
     540             :             {
     541             :                 /* One-for-one replacement, so use PageIndexTupleOverwrite */
     542      513884 :                 if (!PageIndexTupleOverwrite(page, oldoffnum, (Item) *itup,
     543      513884 :                                              IndexTupleSize(*itup)))
     544           0 :                     elog(ERROR, "failed to add item to index page in \"%s\"",
     545             :                          RelationGetRelationName(rel));
     546             :             }
     547             :             else
     548             :             {
     549             :                 /* Delete old, then append new tuple(s) to page */
     550       16932 :                 PageIndexTupleDelete(page, oldoffnum);
     551       16932 :                 gistfillbuffer(page, itup, ntup, InvalidOffsetNumber);
     552             :             }
     553             :         }
     554             :         else
     555             :         {
     556             :             /* Just append new tuples at the end of the page */
     557      730332 :             gistfillbuffer(page, itup, ntup, InvalidOffsetNumber);
     558             :         }
     559             : 
     560     1261148 :         MarkBufferDirty(buffer);
     561             : 
     562     1261148 :         if (BufferIsValid(leftchildbuf))
     563       16988 :             MarkBufferDirty(leftchildbuf);
     564             : 
     565     1261148 :         if (is_build)
     566      882376 :             recptr = GistBuildLSN;
     567             :         else
     568             :         {
     569      378772 :             if (RelationNeedsWAL(rel))
     570      378764 :             {
     571      378764 :                 OffsetNumber ndeloffs = 0,
     572             :                             deloffs[1];
     573             : 
     574      378764 :                 if (OffsetNumberIsValid(oldoffnum))
     575             :                 {
     576      151078 :                     deloffs[0] = oldoffnum;
     577      151078 :                     ndeloffs = 1;
     578             :                 }
     579             : 
     580      378764 :                 recptr = gistXLogUpdate(buffer,
     581             :                                         deloffs, ndeloffs, itup, ntup,
     582             :                                         leftchildbuf);
     583             :             }
     584             :             else
     585           8 :                 recptr = gistGetFakeLSN(rel);
     586             :         }
     587     1261148 :         PageSetLSN(page, recptr);
     588             : 
     589     1261148 :         if (newblkno)
     590           0 :             *newblkno = blkno;
     591             :     }
     592             : 
     593             :     /*
     594             :      * If we inserted the downlink for a child page, set NSN and clear
     595             :      * F_FOLLOW_RIGHT flag on the left child, so that concurrent scans know to
     596             :      * follow the rightlink if and only if they looked at the parent page
     597             :      * before we inserted the downlink.
     598             :      *
     599             :      * Note that we do this *after* writing the WAL record. That means that
     600             :      * the possible full page image in the WAL record does not include these
     601             :      * changes, and they must be replayed even if the page is restored from
     602             :      * the full page image. There's a chicken-and-egg problem: if we updated
     603             :      * the child pages first, we wouldn't know the recptr of the WAL record
     604             :      * we're about to write.
     605             :      */
     606     1283072 :     if (BufferIsValid(leftchildbuf))
     607             :     {
     608       21650 :         Page        leftpg = BufferGetPage(leftchildbuf);
     609             : 
     610       21650 :         GistPageSetNSN(leftpg, recptr);
     611       21650 :         GistClearFollowRight(leftpg);
     612             : 
     613       21650 :         PageSetLSN(leftpg, recptr);
     614             :     }
     615             : 
     616     1283072 :     END_CRIT_SECTION();
     617             : 
     618     1283072 :     return is_split;
     619             : }
     620             : 
     621             : /*
     622             :  * Workhouse routine for doing insertion into a GiST index. Note that
     623             :  * this routine assumes it is invoked in a short-lived memory context,
     624             :  * so it does not bother releasing palloc'd allocations.
     625             :  */
     626             : void
     627      747526 : gistdoinsert(Relation r, IndexTuple itup, Size freespace,
     628             :              GISTSTATE *giststate, Relation heapRel, bool is_build)
     629             : {
     630             :     ItemId      iid;
     631             :     IndexTuple  idxtuple;
     632             :     GISTInsertStack firststack;
     633             :     GISTInsertStack *stack;
     634             :     GISTInsertState state;
     635      747526 :     bool        xlocked = false;
     636             : 
     637      747526 :     memset(&state, 0, sizeof(GISTInsertState));
     638      747526 :     state.freespace = freespace;
     639      747526 :     state.r = r;
     640      747526 :     state.heapRel = heapRel;
     641      747526 :     state.is_build = is_build;
     642             : 
     643             :     /* Start from the root */
     644      747526 :     firststack.blkno = GIST_ROOT_BLKNO;
     645      747526 :     firststack.lsn = 0;
     646      747526 :     firststack.retry_from_parent = false;
     647      747526 :     firststack.parent = NULL;
     648      747526 :     firststack.downlinkoffnum = InvalidOffsetNumber;
     649      747526 :     state.stack = stack = &firststack;
     650             : 
     651             :     /*
     652             :      * Walk down along the path of smallest penalty, updating the parent
     653             :      * pointers with the key we're inserting as we go. If we crash in the
     654             :      * middle, the tree is consistent, although the possible parent updates
     655             :      * were a waste.
     656             :      */
     657             :     for (;;)
     658             :     {
     659             :         /*
     660             :          * If we split an internal page while descending the tree, we have to
     661             :          * retry at the parent. (Normally, the LSN-NSN interlock below would
     662             :          * also catch this and cause us to retry. But LSNs are not updated
     663             :          * during index build.)
     664             :          */
     665     1709186 :         while (stack->retry_from_parent)
     666             :         {
     667           2 :             if (xlocked)
     668           0 :                 LockBuffer(stack->buffer, GIST_UNLOCK);
     669           2 :             xlocked = false;
     670           2 :             ReleaseBuffer(stack->buffer);
     671           2 :             state.stack = stack = stack->parent;
     672             :         }
     673             : 
     674     1709184 :         if (XLogRecPtrIsInvalid(stack->lsn))
     675     1709160 :             stack->buffer = ReadBuffer(state.r, stack->blkno);
     676             : 
     677             :         /*
     678             :          * Be optimistic and grab shared lock first. Swap it for an exclusive
     679             :          * lock later if we need to update the page.
     680             :          */
     681     1709184 :         if (!xlocked)
     682             :         {
     683     1709184 :             LockBuffer(stack->buffer, GIST_SHARE);
     684     1709184 :             gistcheckpage(state.r, stack->buffer);
     685             :         }
     686             : 
     687     1709184 :         stack->page = (Page) BufferGetPage(stack->buffer);
     688     1709184 :         stack->lsn = xlocked ?
     689     1709184 :             PageGetLSN(stack->page) : BufferGetLSNAtomic(stack->buffer);
     690             :         Assert(!RelationNeedsWAL(state.r) || !XLogRecPtrIsInvalid(stack->lsn));
     691             : 
     692             :         /*
     693             :          * If this page was split but the downlink was never inserted to the
     694             :          * parent because the inserting backend crashed before doing that, fix
     695             :          * that now.
     696             :          */
     697     1709184 :         if (GistFollowRight(stack->page))
     698             :         {
     699           0 :             if (!xlocked)
     700             :             {
     701           0 :                 LockBuffer(stack->buffer, GIST_UNLOCK);
     702           0 :                 LockBuffer(stack->buffer, GIST_EXCLUSIVE);
     703           0 :                 xlocked = true;
     704             :                 /* someone might've completed the split when we unlocked */
     705           0 :                 if (!GistFollowRight(stack->page))
     706           0 :                     continue;
     707             :             }
     708           0 :             gistfixsplit(&state, giststate);
     709             : 
     710           0 :             UnlockReleaseBuffer(stack->buffer);
     711           0 :             xlocked = false;
     712           0 :             state.stack = stack = stack->parent;
     713           0 :             continue;
     714             :         }
     715             : 
     716     1709184 :         if ((stack->blkno != GIST_ROOT_BLKNO &&
     717      961634 :              stack->parent->lsn < GistPageGetNSN(stack->page)) ||
     718     1709184 :             GistPageIsDeleted(stack->page))
     719             :         {
     720             :             /*
     721             :              * Concurrent split or page deletion detected. There's no
     722             :              * guarantee that the downlink for this page is consistent with
     723             :              * the tuple we're inserting anymore, so go back to parent and
     724             :              * rechoose the best child.
     725             :              */
     726           0 :             UnlockReleaseBuffer(stack->buffer);
     727           0 :             xlocked = false;
     728           0 :             state.stack = stack = stack->parent;
     729           0 :             continue;
     730             :         }
     731             : 
     732     1709184 :         if (!GistPageIsLeaf(stack->page))
     733             :         {
     734             :             /*
     735             :              * This is an internal page so continue to walk down the tree.
     736             :              * Find the child node that has the minimum insertion penalty.
     737             :              */
     738             :             BlockNumber childblkno;
     739             :             IndexTuple  newtup;
     740             :             GISTInsertStack *item;
     741             :             OffsetNumber downlinkoffnum;
     742             : 
     743      961658 :             downlinkoffnum = gistchoose(state.r, stack->page, itup, giststate);
     744      961658 :             iid = PageGetItemId(stack->page, downlinkoffnum);
     745      961658 :             idxtuple = (IndexTuple) PageGetItem(stack->page, iid);
     746      961658 :             childblkno = ItemPointerGetBlockNumber(&(idxtuple->t_tid));
     747             : 
     748             :             /*
     749             :              * Check that it's not a leftover invalid tuple from pre-9.1
     750             :              */
     751      961658 :             if (GistTupleIsInvalid(idxtuple))
     752           0 :                 ereport(ERROR,
     753             :                         (errmsg("index \"%s\" contains an inner tuple marked as invalid",
     754             :                                 RelationGetRelationName(r)),
     755             :                          errdetail("This is caused by an incomplete page split at crash recovery before upgrading to PostgreSQL 9.1."),
     756             :                          errhint("Please REINDEX it.")));
     757             : 
     758             :             /*
     759             :              * Check that the key representing the target child node is
     760             :              * consistent with the key we're inserting. Update it if it's not.
     761             :              */
     762      961658 :             newtup = gistgetadjusted(state.r, idxtuple, itup, giststate);
     763      961658 :             if (newtup)
     764             :             {
     765             :                 /*
     766             :                  * Swap shared lock for an exclusive one. Beware, the page may
     767             :                  * change while we unlock/lock the page...
     768             :                  */
     769      513908 :                 if (!xlocked)
     770             :                 {
     771      513908 :                     LockBuffer(stack->buffer, GIST_UNLOCK);
     772      513908 :                     LockBuffer(stack->buffer, GIST_EXCLUSIVE);
     773      513908 :                     xlocked = true;
     774      513908 :                     stack->page = (Page) BufferGetPage(stack->buffer);
     775             : 
     776      513908 :                     if (PageGetLSN(stack->page) != stack->lsn)
     777             :                     {
     778             :                         /* the page was changed while we unlocked it, retry */
     779           0 :                         continue;
     780             :                     }
     781             :                 }
     782             : 
     783             :                 /*
     784             :                  * Update the tuple.
     785             :                  *
     786             :                  * We still hold the lock after gistinserttuple(), but it
     787             :                  * might have to split the page to make the updated tuple fit.
     788             :                  * In that case the updated tuple might migrate to the other
     789             :                  * half of the split, so we have to go back to the parent and
     790             :                  * descend back to the half that's a better fit for the new
     791             :                  * tuple.
     792             :                  */
     793      513908 :                 if (gistinserttuple(&state, stack, giststate, newtup,
     794             :                                     downlinkoffnum))
     795             :                 {
     796             :                     /*
     797             :                      * If this was a root split, the root page continues to be
     798             :                      * the parent and the updated tuple went to one of the
     799             :                      * child pages, so we just need to retry from the root
     800             :                      * page.
     801             :                      */
     802          24 :                     if (stack->blkno != GIST_ROOT_BLKNO)
     803             :                     {
     804          24 :                         UnlockReleaseBuffer(stack->buffer);
     805          24 :                         xlocked = false;
     806          24 :                         state.stack = stack = stack->parent;
     807             :                     }
     808          24 :                     continue;
     809             :                 }
     810             :             }
     811      961634 :             LockBuffer(stack->buffer, GIST_UNLOCK);
     812      961634 :             xlocked = false;
     813             : 
     814             :             /* descend to the chosen child */
     815      961634 :             item = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));
     816      961634 :             item->blkno = childblkno;
     817      961634 :             item->parent = stack;
     818      961634 :             item->downlinkoffnum = downlinkoffnum;
     819      961634 :             state.stack = stack = item;
     820             :         }
     821             :         else
     822             :         {
     823             :             /*
     824             :              * Leaf page. Insert the new key. We've already updated all the
     825             :              * parents on the way down, but we might have to split the page if
     826             :              * it doesn't fit. gistinserttuple() will take care of that.
     827             :              */
     828             : 
     829             :             /*
     830             :              * Swap shared lock for an exclusive one. Be careful, the page may
     831             :              * change while we unlock/lock the page...
     832             :              */
     833      747526 :             if (!xlocked)
     834             :             {
     835      747526 :                 LockBuffer(stack->buffer, GIST_UNLOCK);
     836      747526 :                 LockBuffer(stack->buffer, GIST_EXCLUSIVE);
     837      747526 :                 xlocked = true;
     838      747526 :                 stack->page = (Page) BufferGetPage(stack->buffer);
     839      747526 :                 stack->lsn = PageGetLSN(stack->page);
     840             : 
     841      747526 :                 if (stack->blkno == GIST_ROOT_BLKNO)
     842             :                 {
     843             :                     /*
     844             :                      * the only page that can become inner instead of leaf is
     845             :                      * the root page, so for root we should recheck it
     846             :                      */
     847       46062 :                     if (!GistPageIsLeaf(stack->page))
     848             :                     {
     849             :                         /*
     850             :                          * very rare situation: during unlock/lock index with
     851             :                          * number of pages = 1 was increased
     852             :                          */
     853           0 :                         LockBuffer(stack->buffer, GIST_UNLOCK);
     854           0 :                         xlocked = false;
     855           0 :                         continue;
     856             :                     }
     857             : 
     858             :                     /*
     859             :                      * we don't need to check root split, because checking
     860             :                      * leaf/inner is enough to recognize split for root
     861             :                      */
     862             :                 }
     863      701464 :                 else if ((GistFollowRight(stack->page) ||
     864      701464 :                           stack->parent->lsn < GistPageGetNSN(stack->page)) &&
     865           0 :                          GistPageIsDeleted(stack->page))
     866             :                 {
     867             :                     /*
     868             :                      * The page was split or deleted while we momentarily
     869             :                      * unlocked the page. Go back to parent.
     870             :                      */
     871           0 :                     UnlockReleaseBuffer(stack->buffer);
     872           0 :                     xlocked = false;
     873           0 :                     state.stack = stack = stack->parent;
     874           0 :                     continue;
     875             :                 }
     876             :             }
     877             : 
     878             :             /* now state.stack->(page, buffer and blkno) points to leaf page */
     879             : 
     880      747526 :             gistinserttuple(&state, stack, giststate, itup,
     881             :                             InvalidOffsetNumber);
     882      747514 :             LockBuffer(stack->buffer, GIST_UNLOCK);
     883             : 
     884             :             /* Release any pins we might still hold before exiting */
     885     2456624 :             for (; stack; stack = stack->parent)
     886     1709110 :                 ReleaseBuffer(stack->buffer);
     887      747514 :             break;
     888             :         }
     889             :     }
     890      747514 : }
     891             : 
     892             : /*
     893             :  * Traverse the tree to find path from root page to specified "child" block.
     894             :  *
     895             :  * returns a new insertion stack, starting from the parent of "child", up
     896             :  * to the root. *downlinkoffnum is set to the offset of the downlink in the
     897             :  * direct parent of child.
     898             :  *
     899             :  * To prevent deadlocks, this should lock only one page at a time.
     900             :  */
     901             : static GISTInsertStack *
     902           0 : gistFindPath(Relation r, BlockNumber child, OffsetNumber *downlinkoffnum)
     903             : {
     904             :     Page        page;
     905             :     Buffer      buffer;
     906             :     OffsetNumber i,
     907             :                 maxoff;
     908             :     ItemId      iid;
     909             :     IndexTuple  idxtuple;
     910             :     List       *fifo;
     911             :     GISTInsertStack *top,
     912             :                *ptr;
     913             :     BlockNumber blkno;
     914             : 
     915           0 :     top = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));
     916           0 :     top->blkno = GIST_ROOT_BLKNO;
     917           0 :     top->downlinkoffnum = InvalidOffsetNumber;
     918             : 
     919           0 :     fifo = list_make1(top);
     920           0 :     while (fifo != NIL)
     921             :     {
     922             :         /* Get next page to visit */
     923           0 :         top = linitial(fifo);
     924           0 :         fifo = list_delete_first(fifo);
     925             : 
     926           0 :         buffer = ReadBuffer(r, top->blkno);
     927           0 :         LockBuffer(buffer, GIST_SHARE);
     928           0 :         gistcheckpage(r, buffer);
     929           0 :         page = (Page) BufferGetPage(buffer);
     930             : 
     931           0 :         if (GistPageIsLeaf(page))
     932             :         {
     933             :             /*
     934             :              * Because we scan the index top-down, all the rest of the pages
     935             :              * in the queue must be leaf pages as well.
     936             :              */
     937           0 :             UnlockReleaseBuffer(buffer);
     938           0 :             break;
     939             :         }
     940             : 
     941             :         /* currently, internal pages are never deleted */
     942             :         Assert(!GistPageIsDeleted(page));
     943             : 
     944           0 :         top->lsn = BufferGetLSNAtomic(buffer);
     945             : 
     946             :         /*
     947             :          * If F_FOLLOW_RIGHT is set, the page to the right doesn't have a
     948             :          * downlink. This should not normally happen..
     949             :          */
     950           0 :         if (GistFollowRight(page))
     951           0 :             elog(ERROR, "concurrent GiST page split was incomplete");
     952             : 
     953           0 :         if (top->parent && top->parent->lsn < GistPageGetNSN(page) &&
     954           0 :             GistPageGetOpaque(page)->rightlink != InvalidBlockNumber /* sanity check */ )
     955             :         {
     956             :             /*
     957             :              * Page was split while we looked elsewhere. We didn't see the
     958             :              * downlink to the right page when we scanned the parent, so add
     959             :              * it to the queue now.
     960             :              *
     961             :              * Put the right page ahead of the queue, so that we visit it
     962             :              * next. That's important, because if this is the lowest internal
     963             :              * level, just above leaves, we might already have queued up some
     964             :              * leaf pages, and we assume that there can't be any non-leaf
     965             :              * pages behind leaf pages.
     966             :              */
     967           0 :             ptr = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));
     968           0 :             ptr->blkno = GistPageGetOpaque(page)->rightlink;
     969           0 :             ptr->downlinkoffnum = InvalidOffsetNumber;
     970           0 :             ptr->parent = top->parent;
     971             : 
     972           0 :             fifo = lcons(ptr, fifo);
     973             :         }
     974             : 
     975           0 :         maxoff = PageGetMaxOffsetNumber(page);
     976             : 
     977           0 :         for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
     978             :         {
     979           0 :             iid = PageGetItemId(page, i);
     980           0 :             idxtuple = (IndexTuple) PageGetItem(page, iid);
     981           0 :             blkno = ItemPointerGetBlockNumber(&(idxtuple->t_tid));
     982           0 :             if (blkno == child)
     983             :             {
     984             :                 /* Found it! */
     985           0 :                 UnlockReleaseBuffer(buffer);
     986           0 :                 *downlinkoffnum = i;
     987           0 :                 return top;
     988             :             }
     989             :             else
     990             :             {
     991             :                 /* Append this child to the list of pages to visit later */
     992           0 :                 ptr = (GISTInsertStack *) palloc0(sizeof(GISTInsertStack));
     993           0 :                 ptr->blkno = blkno;
     994           0 :                 ptr->downlinkoffnum = i;
     995           0 :                 ptr->parent = top;
     996             : 
     997           0 :                 fifo = lappend(fifo, ptr);
     998             :             }
     999             :         }
    1000             : 
    1001           0 :         UnlockReleaseBuffer(buffer);
    1002             :     }
    1003             : 
    1004           0 :     elog(ERROR, "failed to re-find parent of a page in index \"%s\", block %u",
    1005             :          RelationGetRelationName(r), child);
    1006             :     return NULL;                /* keep compiler quiet */
    1007             : }
    1008             : 
    1009             : /*
    1010             :  * Updates the stack so that child->parent is the correct parent of the
    1011             :  * child. child->parent must be exclusively locked on entry, and will
    1012             :  * remain so at exit, but it might not be the same page anymore.
    1013             :  */
    1014             : static void
    1015       21650 : gistFindCorrectParent(Relation r, GISTInsertStack *child)
    1016             : {
    1017       21650 :     GISTInsertStack *parent = child->parent;
    1018             : 
    1019       21650 :     gistcheckpage(r, parent->buffer);
    1020       21650 :     parent->page = (Page) BufferGetPage(parent->buffer);
    1021             : 
    1022             :     /* here we don't need to distinguish between split and page update */
    1023       21650 :     if (child->downlinkoffnum == InvalidOffsetNumber ||
    1024       21650 :         parent->lsn != PageGetLSN(parent->page))
    1025             :     {
    1026             :         /* parent is changed, look child in right links until found */
    1027             :         OffsetNumber i,
    1028             :                     maxoff;
    1029             :         ItemId      iid;
    1030             :         IndexTuple  idxtuple;
    1031             :         GISTInsertStack *ptr;
    1032             : 
    1033             :         while (true)
    1034             :         {
    1035        2326 :             maxoff = PageGetMaxOffsetNumber(parent->page);
    1036      100450 :             for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
    1037             :             {
    1038      100450 :                 iid = PageGetItemId(parent->page, i);
    1039      100450 :                 idxtuple = (IndexTuple) PageGetItem(parent->page, iid);
    1040      100450 :                 if (ItemPointerGetBlockNumber(&(idxtuple->t_tid)) == child->blkno)
    1041             :                 {
    1042             :                     /* yes!!, found */
    1043        2326 :                     child->downlinkoffnum = i;
    1044        2326 :                     return;
    1045             :                 }
    1046             :             }
    1047             : 
    1048           0 :             parent->blkno = GistPageGetOpaque(parent->page)->rightlink;
    1049           0 :             UnlockReleaseBuffer(parent->buffer);
    1050           0 :             if (parent->blkno == InvalidBlockNumber)
    1051             :             {
    1052             :                 /*
    1053             :                  * End of chain and still didn't find parent. It's a very-very
    1054             :                  * rare situation when root splitted.
    1055             :                  */
    1056           0 :                 break;
    1057             :             }
    1058           0 :             parent->buffer = ReadBuffer(r, parent->blkno);
    1059           0 :             LockBuffer(parent->buffer, GIST_EXCLUSIVE);
    1060           0 :             gistcheckpage(r, parent->buffer);
    1061           0 :             parent->page = (Page) BufferGetPage(parent->buffer);
    1062             :         }
    1063             : 
    1064             :         /*
    1065             :          * awful!!, we need search tree to find parent ... , but before we
    1066             :          * should release all old parent
    1067             :          */
    1068             : 
    1069           0 :         ptr = child->parent->parent;  /* child->parent already released
    1070             :                                          * above */
    1071           0 :         while (ptr)
    1072             :         {
    1073           0 :             ReleaseBuffer(ptr->buffer);
    1074           0 :             ptr = ptr->parent;
    1075             :         }
    1076             : 
    1077             :         /* ok, find new path */
    1078           0 :         ptr = parent = gistFindPath(r, child->blkno, &child->downlinkoffnum);
    1079             : 
    1080             :         /* read all buffers as expected by caller */
    1081             :         /* note we don't lock them or gistcheckpage them here! */
    1082           0 :         while (ptr)
    1083             :         {
    1084           0 :             ptr->buffer = ReadBuffer(r, ptr->blkno);
    1085           0 :             ptr->page = (Page) BufferGetPage(ptr->buffer);
    1086           0 :             ptr = ptr->parent;
    1087             :         }
    1088             : 
    1089             :         /* install new chain of parents to stack */
    1090           0 :         child->parent = parent;
    1091             : 
    1092             :         /* make recursive call to normal processing */
    1093           0 :         LockBuffer(child->parent->buffer, GIST_EXCLUSIVE);
    1094           0 :         gistFindCorrectParent(r, child);
    1095             :     }
    1096             : }
    1097             : 
    1098             : /*
    1099             :  * Form a downlink pointer for the page in 'buf'.
    1100             :  */
    1101             : static IndexTuple
    1102           0 : gistformdownlink(Relation rel, Buffer buf, GISTSTATE *giststate,
    1103             :                  GISTInsertStack *stack)
    1104             : {
    1105           0 :     Page        page = BufferGetPage(buf);
    1106             :     OffsetNumber maxoff;
    1107             :     OffsetNumber offset;
    1108           0 :     IndexTuple  downlink = NULL;
    1109             : 
    1110           0 :     maxoff = PageGetMaxOffsetNumber(page);
    1111           0 :     for (offset = FirstOffsetNumber; offset <= maxoff; offset = OffsetNumberNext(offset))
    1112             :     {
    1113           0 :         IndexTuple  ituple = (IndexTuple)
    1114           0 :         PageGetItem(page, PageGetItemId(page, offset));
    1115             : 
    1116           0 :         if (downlink == NULL)
    1117           0 :             downlink = CopyIndexTuple(ituple);
    1118             :         else
    1119             :         {
    1120             :             IndexTuple  newdownlink;
    1121             : 
    1122           0 :             newdownlink = gistgetadjusted(rel, downlink, ituple,
    1123             :                                           giststate);
    1124           0 :             if (newdownlink)
    1125           0 :                 downlink = newdownlink;
    1126             :         }
    1127             :     }
    1128             : 
    1129             :     /*
    1130             :      * If the page is completely empty, we can't form a meaningful downlink
    1131             :      * for it. But we have to insert a downlink for the page. Any key will do,
    1132             :      * as long as its consistent with the downlink of parent page, so that we
    1133             :      * can legally insert it to the parent. A minimal one that matches as few
    1134             :      * scans as possible would be best, to keep scans from doing useless work,
    1135             :      * but we don't know how to construct that. So we just use the downlink of
    1136             :      * the original page that was split - that's as far from optimal as it can
    1137             :      * get but will do..
    1138             :      */
    1139           0 :     if (!downlink)
    1140             :     {
    1141             :         ItemId      iid;
    1142             : 
    1143           0 :         LockBuffer(stack->parent->buffer, GIST_EXCLUSIVE);
    1144           0 :         gistFindCorrectParent(rel, stack);
    1145           0 :         iid = PageGetItemId(stack->parent->page, stack->downlinkoffnum);
    1146           0 :         downlink = (IndexTuple) PageGetItem(stack->parent->page, iid);
    1147           0 :         downlink = CopyIndexTuple(downlink);
    1148           0 :         LockBuffer(stack->parent->buffer, GIST_UNLOCK);
    1149             :     }
    1150             : 
    1151           0 :     ItemPointerSetBlockNumber(&(downlink->t_tid), BufferGetBlockNumber(buf));
    1152           0 :     GistTupleSetValid(downlink);
    1153             : 
    1154           0 :     return downlink;
    1155             : }
    1156             : 
    1157             : 
    1158             : /*
    1159             :  * Complete the incomplete split of state->stack->page.
    1160             :  */
    1161             : static void
    1162           0 : gistfixsplit(GISTInsertState *state, GISTSTATE *giststate)
    1163             : {
    1164           0 :     GISTInsertStack *stack = state->stack;
    1165             :     Buffer      buf;
    1166             :     Page        page;
    1167           0 :     List       *splitinfo = NIL;
    1168             : 
    1169           0 :     elog(LOG, "fixing incomplete split in index \"%s\", block %u",
    1170             :          RelationGetRelationName(state->r), stack->blkno);
    1171             : 
    1172             :     Assert(GistFollowRight(stack->page));
    1173             :     Assert(OffsetNumberIsValid(stack->downlinkoffnum));
    1174             : 
    1175           0 :     buf = stack->buffer;
    1176             : 
    1177             :     /*
    1178             :      * Read the chain of split pages, following the rightlinks. Construct a
    1179             :      * downlink tuple for each page.
    1180             :      */
    1181             :     for (;;)
    1182           0 :     {
    1183           0 :         GISTPageSplitInfo *si = palloc(sizeof(GISTPageSplitInfo));
    1184             :         IndexTuple  downlink;
    1185             : 
    1186           0 :         page = BufferGetPage(buf);
    1187             : 
    1188             :         /* Form the new downlink tuples to insert to parent */
    1189           0 :         downlink = gistformdownlink(state->r, buf, giststate, stack);
    1190             : 
    1191           0 :         si->buf = buf;
    1192           0 :         si->downlink = downlink;
    1193             : 
    1194           0 :         splitinfo = lappend(splitinfo, si);
    1195             : 
    1196           0 :         if (GistFollowRight(page))
    1197             :         {
    1198             :             /* lock next page */
    1199           0 :             buf = ReadBuffer(state->r, GistPageGetOpaque(page)->rightlink);
    1200           0 :             LockBuffer(buf, GIST_EXCLUSIVE);
    1201             :         }
    1202             :         else
    1203           0 :             break;
    1204             :     }
    1205             : 
    1206             :     /* Insert the downlinks */
    1207           0 :     gistfinishsplit(state, stack, giststate, splitinfo, false);
    1208           0 : }
    1209             : 
    1210             : /*
    1211             :  * Insert or replace a tuple in stack->buffer. If 'oldoffnum' is valid, the
    1212             :  * tuple at 'oldoffnum' is replaced, otherwise the tuple is inserted as new.
    1213             :  * 'stack' represents the path from the root to the page being updated.
    1214             :  *
    1215             :  * The caller must hold an exclusive lock on stack->buffer.  The lock is still
    1216             :  * held on return, but the page might not contain the inserted tuple if the
    1217             :  * page was split. The function returns true if the page was split, false
    1218             :  * otherwise.
    1219             :  */
    1220             : static bool
    1221     1261434 : gistinserttuple(GISTInsertState *state, GISTInsertStack *stack,
    1222             :                 GISTSTATE *giststate, IndexTuple tuple, OffsetNumber oldoffnum)
    1223             : {
    1224     1261434 :     return gistinserttuples(state, stack, giststate, &tuple, 1, oldoffnum,
    1225             :                             InvalidBuffer, InvalidBuffer, false, false);
    1226             : }
    1227             : 
    1228             : /* ----------------
    1229             :  * An extended workhorse version of gistinserttuple(). This version allows
    1230             :  * inserting multiple tuples, or replacing a single tuple with multiple tuples.
    1231             :  * This is used to recursively update the downlinks in the parent when a page
    1232             :  * is split.
    1233             :  *
    1234             :  * If leftchild and rightchild are valid, we're inserting/replacing the
    1235             :  * downlink for rightchild, and leftchild is its left sibling. We clear the
    1236             :  * F_FOLLOW_RIGHT flag and update NSN on leftchild, atomically with the
    1237             :  * insertion of the downlink.
    1238             :  *
    1239             :  * To avoid holding locks for longer than necessary, when recursing up the
    1240             :  * tree to update the parents, the locking is a bit peculiar here. On entry,
    1241             :  * the caller must hold an exclusive lock on stack->buffer, as well as
    1242             :  * leftchild and rightchild if given. On return:
    1243             :  *
    1244             :  *  - Lock on stack->buffer is released, if 'unlockbuf' is true. The page is
    1245             :  *    always kept pinned, however.
    1246             :  *  - Lock on 'leftchild' is released, if 'unlockleftchild' is true. The page
    1247             :  *    is kept pinned.
    1248             :  *  - Lock and pin on 'rightchild' are always released.
    1249             :  *
    1250             :  * Returns 'true' if the page had to be split. Note that if the page was
    1251             :  * split, the inserted/updated tuples might've been inserted to a right
    1252             :  * sibling of stack->buffer instead of stack->buffer itself.
    1253             :  */
    1254             : static bool
    1255     1283084 : gistinserttuples(GISTInsertState *state, GISTInsertStack *stack,
    1256             :                  GISTSTATE *giststate,
    1257             :                  IndexTuple *tuples, int ntup, OffsetNumber oldoffnum,
    1258             :                  Buffer leftchild, Buffer rightchild,
    1259             :                  bool unlockbuf, bool unlockleftchild)
    1260             : {
    1261             :     List       *splitinfo;
    1262             :     bool        is_split;
    1263             : 
    1264             :     /*
    1265             :      * Check for any rw conflicts (in serializable isolation level) just
    1266             :      * before we intend to modify the page
    1267             :      */
    1268     1283084 :     CheckForSerializableConflictIn(state->r, NULL, BufferGetBlockNumber(stack->buffer));
    1269             : 
    1270             :     /* Insert the tuple(s) to the page, splitting the page if necessary */
    1271     1283072 :     is_split = gistplacetopage(state->r, state->freespace, giststate,
    1272             :                                stack->buffer,
    1273             :                                tuples, ntup,
    1274             :                                oldoffnum, NULL,
    1275             :                                leftchild,
    1276             :                                &splitinfo,
    1277             :                                true,
    1278             :                                state->heapRel,
    1279     1283072 :                                state->is_build);
    1280             : 
    1281             :     /*
    1282             :      * Before recursing up in case the page was split, release locks on the
    1283             :      * child pages. We don't need to keep them locked when updating the
    1284             :      * parent.
    1285             :      */
    1286     1283072 :     if (BufferIsValid(rightchild))
    1287       21650 :         UnlockReleaseBuffer(rightchild);
    1288     1283072 :     if (BufferIsValid(leftchild) && unlockleftchild)
    1289        4582 :         LockBuffer(leftchild, GIST_UNLOCK);
    1290             : 
    1291             :     /*
    1292             :      * If we had to split, insert/update the downlinks in the parent. If the
    1293             :      * caller requested us to release the lock on stack->buffer, tell
    1294             :      * gistfinishsplit() to do that as soon as it's safe to do so. If we
    1295             :      * didn't have to split, release it ourselves.
    1296             :      */
    1297     1283072 :     if (splitinfo)
    1298       21594 :         gistfinishsplit(state, stack, giststate, splitinfo, unlockbuf);
    1299     1261478 :     else if (unlockbuf)
    1300       17012 :         LockBuffer(stack->buffer, GIST_UNLOCK);
    1301             : 
    1302     1283072 :     return is_split;
    1303             : }
    1304             : 
    1305             : /*
    1306             :  * Finish an incomplete split by inserting/updating the downlinks in parent
    1307             :  * page. 'splitinfo' contains all the child pages involved in the split,
    1308             :  * from left-to-right.
    1309             :  *
    1310             :  * On entry, the caller must hold a lock on stack->buffer and all the child
    1311             :  * pages in 'splitinfo'. If 'unlockbuf' is true, the lock on stack->buffer is
    1312             :  * released on return. The child pages are always unlocked and unpinned.
    1313             :  */
    1314             : static void
    1315       21594 : gistfinishsplit(GISTInsertState *state, GISTInsertStack *stack,
    1316             :                 GISTSTATE *giststate, List *splitinfo, bool unlockbuf)
    1317             : {
    1318             :     GISTPageSplitInfo *right;
    1319             :     GISTPageSplitInfo *left;
    1320             :     IndexTuple  tuples[2];
    1321             : 
    1322             :     /* A split always contains at least two halves */
    1323             :     Assert(list_length(splitinfo) >= 2);
    1324             : 
    1325             :     /*
    1326             :      * We need to insert downlinks for each new page, and update the downlink
    1327             :      * for the original (leftmost) page in the split. Begin at the rightmost
    1328             :      * page, inserting one downlink at a time until there's only two pages
    1329             :      * left. Finally insert the downlink for the last new page and update the
    1330             :      * downlink for the original page as one operation.
    1331             :      */
    1332       21594 :     LockBuffer(stack->parent->buffer, GIST_EXCLUSIVE);
    1333             : 
    1334             :     /*
    1335             :      * Insert downlinks for the siblings from right to left, until there are
    1336             :      * only two siblings left.
    1337             :      */
    1338       21650 :     for (int pos = list_length(splitinfo) - 1; pos > 1; pos--)
    1339             :     {
    1340          56 :         right = (GISTPageSplitInfo *) list_nth(splitinfo, pos);
    1341          56 :         left = (GISTPageSplitInfo *) list_nth(splitinfo, pos - 1);
    1342             : 
    1343          56 :         gistFindCorrectParent(state->r, stack);
    1344          56 :         if (gistinserttuples(state, stack->parent, giststate,
    1345             :                              &right->downlink, 1,
    1346             :                              InvalidOffsetNumber,
    1347             :                              left->buf, right->buf, false, false))
    1348             :         {
    1349             :             /*
    1350             :              * If the parent page was split, the existing downlink might have
    1351             :              * moved.
    1352             :              */
    1353           0 :             stack->downlinkoffnum = InvalidOffsetNumber;
    1354             :         }
    1355             :         /* gistinserttuples() released the lock on right->buf. */
    1356             :     }
    1357             : 
    1358       21594 :     right = (GISTPageSplitInfo *) lsecond(splitinfo);
    1359       21594 :     left = (GISTPageSplitInfo *) linitial(splitinfo);
    1360             : 
    1361             :     /*
    1362             :      * Finally insert downlink for the remaining right page and update the
    1363             :      * downlink for the original page to not contain the tuples that were
    1364             :      * moved to the new pages.
    1365             :      */
    1366       21594 :     tuples[0] = left->downlink;
    1367       21594 :     tuples[1] = right->downlink;
    1368       21594 :     gistFindCorrectParent(state->r, stack);
    1369       21594 :     if (gistinserttuples(state, stack->parent, giststate,
    1370             :                          tuples, 2,
    1371       21594 :                          stack->downlinkoffnum,
    1372             :                          left->buf, right->buf,
    1373             :                          true,  /* Unlock parent */
    1374             :                          unlockbuf  /* Unlock stack->buffer if caller wants
    1375             :                                      * that */
    1376             :                          ))
    1377             :     {
    1378             :         /*
    1379             :          * If the parent page was split, the downlink might have moved.
    1380             :          */
    1381        4662 :         stack->downlinkoffnum = InvalidOffsetNumber;
    1382             :     }
    1383             : 
    1384             :     Assert(left->buf == stack->buffer);
    1385             : 
    1386             :     /*
    1387             :      * If we split the page because we had to adjust the downlink on an
    1388             :      * internal page, while descending the tree for inserting a new tuple,
    1389             :      * then this might no longer be the correct page for the new tuple. The
    1390             :      * downlink to this page might not cover the new tuple anymore, it might
    1391             :      * need to go to the newly-created right sibling instead. Tell the caller
    1392             :      * to walk back up the stack, to re-check at the parent which page to
    1393             :      * insert to.
    1394             :      *
    1395             :      * Normally, the LSN-NSN interlock during the tree descend would also
    1396             :      * detect that a concurrent split happened (by ourselves), and cause us to
    1397             :      * retry at the parent. But that mechanism doesn't work during index
    1398             :      * build, because we don't do WAL-logging, and don't update LSNs, during
    1399             :      * index build.
    1400             :      */
    1401       21594 :     stack->retry_from_parent = true;
    1402       21594 : }
    1403             : 
    1404             : /*
    1405             :  * gistSplit -- split a page in the tree and fill struct
    1406             :  * used for XLOG and real writes buffers. Function is recursive, ie
    1407             :  * it will split page until keys will fit in every page.
    1408             :  */
    1409             : SplitedPageLayout *
    1410       21982 : gistSplit(Relation r,
    1411             :           Page page,
    1412             :           IndexTuple *itup,     /* contains compressed entry */
    1413             :           int len,
    1414             :           GISTSTATE *giststate)
    1415             : {
    1416             :     IndexTuple *lvectup,
    1417             :                *rvectup;
    1418             :     GistSplitVector v;
    1419             :     int         i;
    1420       21982 :     SplitedPageLayout *res = NULL;
    1421             : 
    1422             :     /* this should never recurse very deeply, but better safe than sorry */
    1423       21982 :     check_stack_depth();
    1424             : 
    1425             :     /* there's no point in splitting an empty page */
    1426             :     Assert(len > 0);
    1427             : 
    1428             :     /*
    1429             :      * If a single tuple doesn't fit on a page, no amount of splitting will
    1430             :      * help.
    1431             :      */
    1432       21982 :     if (len == 1)
    1433           0 :         ereport(ERROR,
    1434             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
    1435             :                  errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
    1436             :                         IndexTupleSize(itup[0]), GiSTPageSize,
    1437             :                         RelationGetRelationName(r))));
    1438             : 
    1439       21982 :     memset(v.spl_lisnull, true,
    1440       21982 :            sizeof(bool) * giststate->nonLeafTupdesc->natts);
    1441       21982 :     memset(v.spl_risnull, true,
    1442       21982 :            sizeof(bool) * giststate->nonLeafTupdesc->natts);
    1443       21982 :     gistSplitByKey(r, page, itup, len, giststate, &v, 0);
    1444             : 
    1445             :     /* form left and right vector */
    1446       21982 :     lvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * (len + 1));
    1447       21982 :     rvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * (len + 1));
    1448             : 
    1449      679378 :     for (i = 0; i < v.splitVector.spl_nleft; i++)
    1450      657396 :         lvectup[i] = itup[v.splitVector.spl_left[i] - 1];
    1451             : 
    1452      759944 :     for (i = 0; i < v.splitVector.spl_nright; i++)
    1453      737962 :         rvectup[i] = itup[v.splitVector.spl_right[i] - 1];
    1454             : 
    1455             :     /* finalize splitting (may need another split) */
    1456       21982 :     if (!gistfitpage(rvectup, v.splitVector.spl_nright))
    1457             :     {
    1458           0 :         res = gistSplit(r, page, rvectup, v.splitVector.spl_nright, giststate);
    1459             :     }
    1460             :     else
    1461             :     {
    1462       21982 :         ROTATEDIST(res);
    1463       21982 :         res->block.num = v.splitVector.spl_nright;
    1464       21982 :         res->list = gistfillitupvec(rvectup, v.splitVector.spl_nright, &(res->lenlist));
    1465       21982 :         res->itup = gistFormTuple(giststate, r, v.spl_rattr, v.spl_risnull, false);
    1466             :     }
    1467             : 
    1468       21982 :     if (!gistfitpage(lvectup, v.splitVector.spl_nleft))
    1469             :     {
    1470             :         SplitedPageLayout *resptr,
    1471             :                    *subres;
    1472             : 
    1473          58 :         resptr = subres = gistSplit(r, page, lvectup, v.splitVector.spl_nleft, giststate);
    1474             : 
    1475             :         /* install on list's tail */
    1476         116 :         while (resptr->next)
    1477          58 :             resptr = resptr->next;
    1478             : 
    1479          58 :         resptr->next = res;
    1480          58 :         res = subres;
    1481             :     }
    1482             :     else
    1483             :     {
    1484       21924 :         ROTATEDIST(res);
    1485       21924 :         res->block.num = v.splitVector.spl_nleft;
    1486       21924 :         res->list = gistfillitupvec(lvectup, v.splitVector.spl_nleft, &(res->lenlist));
    1487       21924 :         res->itup = gistFormTuple(giststate, r, v.spl_lattr, v.spl_lisnull, false);
    1488             :     }
    1489             : 
    1490       21982 :     return res;
    1491             : }
    1492             : 
    1493             : /*
    1494             :  * Create a GISTSTATE and fill it with information about the index
    1495             :  */
    1496             : GISTSTATE *
    1497        2434 : initGISTstate(Relation index)
    1498             : {
    1499             :     GISTSTATE  *giststate;
    1500             :     MemoryContext scanCxt;
    1501             :     MemoryContext oldCxt;
    1502             :     int         i;
    1503             : 
    1504             :     /* safety check to protect fixed-size arrays in GISTSTATE */
    1505        2434 :     if (index->rd_att->natts > INDEX_MAX_KEYS)
    1506           0 :         elog(ERROR, "numberOfAttributes %d > %d",
    1507             :              index->rd_att->natts, INDEX_MAX_KEYS);
    1508             : 
    1509             :     /* Create the memory context that will hold the GISTSTATE */
    1510        2434 :     scanCxt = AllocSetContextCreate(CurrentMemoryContext,
    1511             :                                     "GiST scan context",
    1512             :                                     ALLOCSET_DEFAULT_SIZES);
    1513        2434 :     oldCxt = MemoryContextSwitchTo(scanCxt);
    1514             : 
    1515             :     /* Create and fill in the GISTSTATE */
    1516        2434 :     giststate = (GISTSTATE *) palloc(sizeof(GISTSTATE));
    1517             : 
    1518        2434 :     giststate->scanCxt = scanCxt;
    1519        2434 :     giststate->tempCxt = scanCxt;    /* caller must change this if needed */
    1520        2434 :     giststate->leafTupdesc = index->rd_att;
    1521             : 
    1522             :     /*
    1523             :      * The truncated tupdesc for non-leaf index tuples, which doesn't contain
    1524             :      * the INCLUDE attributes.
    1525             :      *
    1526             :      * It is used to form tuples during tuple adjustment and page split.
    1527             :      * B-tree creates shortened tuple descriptor for every truncated tuple,
    1528             :      * because it is doing this less often: it does not have to form truncated
    1529             :      * tuples during page split.  Also, B-tree is not adjusting tuples on
    1530             :      * internal pages the way GiST does.
    1531             :      */
    1532        2434 :     giststate->nonLeafTupdesc = CreateTupleDescCopyConstr(index->rd_att);
    1533        4868 :     giststate->nonLeafTupdesc->natts =
    1534        4868 :         IndexRelationGetNumberOfKeyAttributes(index);
    1535             : 
    1536        5058 :     for (i = 0; i < IndexRelationGetNumberOfKeyAttributes(index); i++)
    1537             :     {
    1538        2624 :         fmgr_info_copy(&(giststate->consistentFn[i]),
    1539        2624 :                        index_getprocinfo(index, i + 1, GIST_CONSISTENT_PROC),
    1540             :                        scanCxt);
    1541        2624 :         fmgr_info_copy(&(giststate->unionFn[i]),
    1542        2624 :                        index_getprocinfo(index, i + 1, GIST_UNION_PROC),
    1543             :                        scanCxt);
    1544             : 
    1545             :         /* opclasses are not required to provide a Compress method */
    1546        2624 :         if (OidIsValid(index_getprocid(index, i + 1, GIST_COMPRESS_PROC)))
    1547        2060 :             fmgr_info_copy(&(giststate->compressFn[i]),
    1548        2060 :                            index_getprocinfo(index, i + 1, GIST_COMPRESS_PROC),
    1549             :                            scanCxt);
    1550             :         else
    1551         564 :             giststate->compressFn[i].fn_oid = InvalidOid;
    1552             : 
    1553             :         /* opclasses are not required to provide a Decompress method */
    1554        2624 :         if (OidIsValid(index_getprocid(index, i + 1, GIST_DECOMPRESS_PROC)))
    1555        1150 :             fmgr_info_copy(&(giststate->decompressFn[i]),
    1556        1150 :                            index_getprocinfo(index, i + 1, GIST_DECOMPRESS_PROC),
    1557             :                            scanCxt);
    1558             :         else
    1559        1474 :             giststate->decompressFn[i].fn_oid = InvalidOid;
    1560             : 
    1561        2624 :         fmgr_info_copy(&(giststate->penaltyFn[i]),
    1562        2624 :                        index_getprocinfo(index, i + 1, GIST_PENALTY_PROC),
    1563             :                        scanCxt);
    1564        2624 :         fmgr_info_copy(&(giststate->picksplitFn[i]),
    1565        2624 :                        index_getprocinfo(index, i + 1, GIST_PICKSPLIT_PROC),
    1566             :                        scanCxt);
    1567        2624 :         fmgr_info_copy(&(giststate->equalFn[i]),
    1568        2624 :                        index_getprocinfo(index, i + 1, GIST_EQUAL_PROC),
    1569             :                        scanCxt);
    1570             : 
    1571             :         /* opclasses are not required to provide a Distance method */
    1572        2624 :         if (OidIsValid(index_getprocid(index, i + 1, GIST_DISTANCE_PROC)))
    1573        1380 :             fmgr_info_copy(&(giststate->distanceFn[i]),
    1574        1380 :                            index_getprocinfo(index, i + 1, GIST_DISTANCE_PROC),
    1575             :                            scanCxt);
    1576             :         else
    1577        1244 :             giststate->distanceFn[i].fn_oid = InvalidOid;
    1578             : 
    1579             :         /* opclasses are not required to provide a Fetch method */
    1580        2624 :         if (OidIsValid(index_getprocid(index, i + 1, GIST_FETCH_PROC)))
    1581        1112 :             fmgr_info_copy(&(giststate->fetchFn[i]),
    1582        1112 :                            index_getprocinfo(index, i + 1, GIST_FETCH_PROC),
    1583             :                            scanCxt);
    1584             :         else
    1585        1512 :             giststate->fetchFn[i].fn_oid = InvalidOid;
    1586             : 
    1587             :         /*
    1588             :          * If the index column has a specified collation, we should honor that
    1589             :          * while doing comparisons.  However, we may have a collatable storage
    1590             :          * type for a noncollatable indexed data type.  If there's no index
    1591             :          * collation then specify default collation in case the support
    1592             :          * functions need collation.  This is harmless if the support
    1593             :          * functions don't care about collation, so we just do it
    1594             :          * unconditionally.  (We could alternatively call get_typcollation,
    1595             :          * but that seems like expensive overkill --- there aren't going to be
    1596             :          * any cases where a GiST storage type has a nondefault collation.)
    1597             :          */
    1598        2624 :         if (OidIsValid(index->rd_indcollation[i]))
    1599         168 :             giststate->supportCollation[i] = index->rd_indcollation[i];
    1600             :         else
    1601        2456 :             giststate->supportCollation[i] = DEFAULT_COLLATION_OID;
    1602             :     }
    1603             : 
    1604             :     /* No opclass information for INCLUDE attributes */
    1605        2766 :     for (; i < index->rd_att->natts; i++)
    1606             :     {
    1607         332 :         giststate->consistentFn[i].fn_oid = InvalidOid;
    1608         332 :         giststate->unionFn[i].fn_oid = InvalidOid;
    1609         332 :         giststate->compressFn[i].fn_oid = InvalidOid;
    1610         332 :         giststate->decompressFn[i].fn_oid = InvalidOid;
    1611         332 :         giststate->penaltyFn[i].fn_oid = InvalidOid;
    1612         332 :         giststate->picksplitFn[i].fn_oid = InvalidOid;
    1613         332 :         giststate->equalFn[i].fn_oid = InvalidOid;
    1614         332 :         giststate->distanceFn[i].fn_oid = InvalidOid;
    1615         332 :         giststate->fetchFn[i].fn_oid = InvalidOid;
    1616         332 :         giststate->supportCollation[i] = InvalidOid;
    1617             :     }
    1618             : 
    1619        2434 :     MemoryContextSwitchTo(oldCxt);
    1620             : 
    1621        2434 :     return giststate;
    1622             : }
    1623             : 
    1624             : void
    1625        1994 : freeGISTstate(GISTSTATE *giststate)
    1626             : {
    1627             :     /* It's sufficient to delete the scanCxt */
    1628        1994 :     MemoryContextDelete(giststate->scanCxt);
    1629        1994 : }
    1630             : 
    1631             : /*
    1632             :  * gistprunepage() -- try to remove LP_DEAD items from the given page.
    1633             :  * Function assumes that buffer is exclusively locked.
    1634             :  */
    1635             : static void
    1636           0 : gistprunepage(Relation rel, Page page, Buffer buffer, Relation heapRel)
    1637             : {
    1638             :     OffsetNumber deletable[MaxIndexTuplesPerPage];
    1639           0 :     int         ndeletable = 0;
    1640             :     OffsetNumber offnum,
    1641             :                 maxoff;
    1642           0 :     TransactionId latestRemovedXid = InvalidTransactionId;
    1643             : 
    1644             :     Assert(GistPageIsLeaf(page));
    1645             : 
    1646             :     /*
    1647             :      * Scan over all items to see which ones need to be deleted according to
    1648             :      * LP_DEAD flags.
    1649             :      */
    1650           0 :     maxoff = PageGetMaxOffsetNumber(page);
    1651           0 :     for (offnum = FirstOffsetNumber;
    1652             :          offnum <= maxoff;
    1653           0 :          offnum = OffsetNumberNext(offnum))
    1654             :     {
    1655           0 :         ItemId      itemId = PageGetItemId(page, offnum);
    1656             : 
    1657           0 :         if (ItemIdIsDead(itemId))
    1658           0 :             deletable[ndeletable++] = offnum;
    1659             :     }
    1660             : 
    1661           0 :     if (XLogStandbyInfoActive() && RelationNeedsWAL(rel))
    1662             :         latestRemovedXid =
    1663           0 :             index_compute_xid_horizon_for_tuples(rel, heapRel, buffer,
    1664             :                                                  deletable, ndeletable);
    1665             : 
    1666           0 :     if (ndeletable > 0)
    1667             :     {
    1668           0 :         START_CRIT_SECTION();
    1669             : 
    1670           0 :         PageIndexMultiDelete(page, deletable, ndeletable);
    1671             : 
    1672             :         /*
    1673             :          * Mark the page as not containing any LP_DEAD items.  This is not
    1674             :          * certainly true (there might be some that have recently been marked,
    1675             :          * but weren't included in our target-item list), but it will almost
    1676             :          * always be true and it doesn't seem worth an additional page scan to
    1677             :          * check it. Remember that F_HAS_GARBAGE is only a hint anyway.
    1678             :          */
    1679           0 :         GistClearPageHasGarbage(page);
    1680             : 
    1681           0 :         MarkBufferDirty(buffer);
    1682             : 
    1683             :         /* XLOG stuff */
    1684           0 :         if (RelationNeedsWAL(rel))
    1685           0 :         {
    1686             :             XLogRecPtr  recptr;
    1687             : 
    1688           0 :             recptr = gistXLogDelete(buffer,
    1689             :                                     deletable, ndeletable,
    1690             :                                     latestRemovedXid);
    1691             : 
    1692           0 :             PageSetLSN(page, recptr);
    1693             :         }
    1694             :         else
    1695           0 :             PageSetLSN(page, gistGetFakeLSN(rel));
    1696             : 
    1697           0 :         END_CRIT_SECTION();
    1698             :     }
    1699             : 
    1700             :     /*
    1701             :      * Note: if we didn't find any LP_DEAD items, then the page's
    1702             :      * F_HAS_GARBAGE hint bit is falsely set.  We do not bother expending a
    1703             :      * separate write to clear it, however.  We will clear it when we split
    1704             :      * the page.
    1705             :      */
    1706           0 : }

Generated by: LCOV version 1.13