LCOV - code coverage report
Current view: top level - src/backend/access/spgist - spgdoinsert.c (source / functions) Hit Total Coverage
Test: PostgreSQL 12beta2 Lines: 716 783 91.4 %
Date: 2019-06-19 16:07:09 Functions: 15 15 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * spgdoinsert.c
       4             :  *    implementation of insert algorithm
       5             :  *
       6             :  *
       7             :  * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
       8             :  * Portions Copyright (c) 1994, Regents of the University of California
       9             :  *
      10             :  * IDENTIFICATION
      11             :  *          src/backend/access/spgist/spgdoinsert.c
      12             :  *
      13             :  *-------------------------------------------------------------------------
      14             :  */
      15             : 
      16             : #include "postgres.h"
      17             : 
      18             : #include "access/genam.h"
      19             : #include "access/spgist_private.h"
      20             : #include "access/spgxlog.h"
      21             : #include "access/xloginsert.h"
      22             : #include "miscadmin.h"
      23             : #include "storage/bufmgr.h"
      24             : #include "utils/rel.h"
      25             : 
      26             : 
      27             : /*
      28             :  * SPPageDesc tracks all info about a page we are inserting into.  In some
      29             :  * situations it actually identifies a tuple, or even a specific node within
      30             :  * an inner tuple.  But any of the fields can be invalid.  If the buffer
      31             :  * field is valid, it implies we hold pin and exclusive lock on that buffer.
      32             :  * page pointer should be valid exactly when buffer is.
      33             :  */
      34             : typedef struct SPPageDesc
      35             : {
      36             :     BlockNumber blkno;          /* block number, or InvalidBlockNumber */
      37             :     Buffer      buffer;         /* page's buffer number, or InvalidBuffer */
      38             :     Page        page;           /* pointer to page buffer, or NULL */
      39             :     OffsetNumber offnum;        /* offset of tuple, or InvalidOffsetNumber */
      40             :     int         node;           /* node number within inner tuple, or -1 */
      41             : } SPPageDesc;
      42             : 
      43             : 
      44             : /*
      45             :  * Set the item pointer in the nodeN'th entry in inner tuple tup.  This
      46             :  * is used to update the parent inner tuple's downlink after a move or
      47             :  * split operation.
      48             :  */
      49             : void
      50        6496 : spgUpdateNodeLink(SpGistInnerTuple tup, int nodeN,
      51             :                   BlockNumber blkno, OffsetNumber offset)
      52             : {
      53             :     int         i;
      54             :     SpGistNodeTuple node;
      55             : 
      56       34982 :     SGITITERATE(tup, i, node)
      57             :     {
      58       34982 :         if (i == nodeN)
      59             :         {
      60        6496 :             ItemPointerSet(&node->t_tid, blkno, offset);
      61        6496 :             return;
      62             :         }
      63             :     }
      64             : 
      65           0 :     elog(ERROR, "failed to find requested node %d in SPGiST inner tuple",
      66             :          nodeN);
      67             : }
      68             : 
      69             : /*
      70             :  * Form a new inner tuple containing one more node than the given one, with
      71             :  * the specified label datum, inserted at offset "offset" in the node array.
      72             :  * The new tuple's prefix is the same as the old one's.
      73             :  *
      74             :  * Note that the new node initially has an invalid downlink.  We'll find a
      75             :  * page to point it to later.
      76             :  */
      77             : static SpGistInnerTuple
      78         896 : addNode(SpGistState *state, SpGistInnerTuple tuple, Datum label, int offset)
      79             : {
      80             :     SpGistNodeTuple node,
      81             :                *nodes;
      82             :     int         i;
      83             : 
      84             :     /* if offset is negative, insert at end */
      85         896 :     if (offset < 0)
      86           0 :         offset = tuple->nNodes;
      87         896 :     else if (offset > tuple->nNodes)
      88           0 :         elog(ERROR, "invalid offset for adding node to SPGiST inner tuple");
      89             : 
      90         896 :     nodes = palloc(sizeof(SpGistNodeTuple) * (tuple->nNodes + 1));
      91        5136 :     SGITITERATE(tuple, i, node)
      92             :     {
      93        4240 :         if (i < offset)
      94        4200 :             nodes[i] = node;
      95             :         else
      96          40 :             nodes[i + 1] = node;
      97             :     }
      98             : 
      99         896 :     nodes[offset] = spgFormNodeTuple(state, label, false);
     100             : 
     101        3144 :     return spgFormInnerTuple(state,
     102         896 :                              (tuple->prefixSize > 0),
     103        1352 :                              SGITDATUM(tuple, state),
     104         896 :                              tuple->nNodes + 1,
     105             :                              nodes);
     106             : }
     107             : 
     108             : /* qsort comparator for sorting OffsetNumbers */
     109             : static int
     110     3134200 : cmpOffsetNumbers(const void *a, const void *b)
     111             : {
     112     3134200 :     if (*(const OffsetNumber *) a == *(const OffsetNumber *) b)
     113           0 :         return 0;
     114     3134200 :     return (*(const OffsetNumber *) a > *(const OffsetNumber *) b) ? 1 : -1;
     115             : }
     116             : 
     117             : /*
     118             :  * Delete multiple tuples from an index page, preserving tuple offset numbers.
     119             :  *
     120             :  * The first tuple in the given list is replaced with a dead tuple of type
     121             :  * "firststate" (REDIRECT/DEAD/PLACEHOLDER); the remaining tuples are replaced
     122             :  * with dead tuples of type "reststate".  If either firststate or reststate
     123             :  * is REDIRECT, blkno/offnum specify where to link to.
     124             :  *
     125             :  * NB: this is used during WAL replay, so beware of trying to make it too
     126             :  * smart.  In particular, it shouldn't use "state" except for calling
     127             :  * spgFormDeadTuple().  This is also used in a critical section, so no
     128             :  * pallocs either!
     129             :  */
     130             : void
     131        5026 : spgPageIndexMultiDelete(SpGistState *state, Page page,
     132             :                         OffsetNumber *itemnos, int nitems,
     133             :                         int firststate, int reststate,
     134             :                         BlockNumber blkno, OffsetNumber offnum)
     135             : {
     136             :     OffsetNumber firstItem;
     137             :     OffsetNumber sortednos[MaxIndexTuplesPerPage];
     138        5026 :     SpGistDeadTuple tuple = NULL;
     139             :     int         i;
     140             : 
     141        5026 :     if (nitems == 0)
     142         268 :         return;                 /* nothing to do */
     143             : 
     144             :     /*
     145             :      * For efficiency we want to use PageIndexMultiDelete, which requires the
     146             :      * targets to be listed in sorted order, so we have to sort the itemnos
     147             :      * array.  (This also greatly simplifies the math for reinserting the
     148             :      * replacement tuples.)  However, we must not scribble on the caller's
     149             :      * array, so we have to make a copy.
     150             :      */
     151        4758 :     memcpy(sortednos, itemnos, sizeof(OffsetNumber) * nitems);
     152        4758 :     if (nitems > 1)
     153        4496 :         qsort(sortednos, nitems, sizeof(OffsetNumber), cmpOffsetNumbers);
     154             : 
     155        4758 :     PageIndexMultiDelete(page, sortednos, nitems);
     156             : 
     157        4758 :     firstItem = itemnos[0];
     158             : 
     159      458096 :     for (i = 0; i < nitems; i++)
     160             :     {
     161      453338 :         OffsetNumber itemno = sortednos[i];
     162             :         int         tupstate;
     163             : 
     164      453338 :         tupstate = (itemno == firstItem) ? firststate : reststate;
     165      453338 :         if (tuple == NULL || tuple->tupstate != tupstate)
     166        7084 :             tuple = spgFormDeadTuple(state, tupstate, blkno, offnum);
     167             : 
     168      453338 :         if (PageAddItem(page, (Item) tuple, tuple->size,
     169             :                         itemno, false, false) != itemno)
     170           0 :             elog(ERROR, "failed to add item of size %u to SPGiST index page",
     171             :                  tuple->size);
     172             : 
     173      453338 :         if (tupstate == SPGIST_REDIRECT)
     174        1186 :             SpGistPageGetOpaque(page)->nRedirection++;
     175      452152 :         else if (tupstate == SPGIST_PLACEHOLDER)
     176      452152 :             SpGistPageGetOpaque(page)->nPlaceholder++;
     177             :     }
     178             : }
     179             : 
     180             : /*
     181             :  * Update the parent inner tuple's downlink, and mark the parent buffer
     182             :  * dirty (this must be the last change to the parent page in the current
     183             :  * WAL action).
     184             :  */
     185             : static void
     186        5656 : saveNodeLink(Relation index, SPPageDesc *parent,
     187             :              BlockNumber blkno, OffsetNumber offnum)
     188             : {
     189             :     SpGistInnerTuple innerTuple;
     190             : 
     191        5656 :     innerTuple = (SpGistInnerTuple) PageGetItem(parent->page,
     192             :                                                 PageGetItemId(parent->page, parent->offnum));
     193             : 
     194        5656 :     spgUpdateNodeLink(innerTuple, parent->node, blkno, offnum);
     195             : 
     196        5656 :     MarkBufferDirty(parent->buffer);
     197        5656 : }
     198             : 
     199             : /*
     200             :  * Add a leaf tuple to a leaf page where there is known to be room for it
     201             :  */
     202             : static void
     203      467088 : addLeafTuple(Relation index, SpGistState *state, SpGistLeafTuple leafTuple,
     204             :              SPPageDesc *current, SPPageDesc *parent, bool isNulls, bool isNew)
     205             : {
     206             :     spgxlogAddLeaf xlrec;
     207             : 
     208      467088 :     xlrec.newPage = isNew;
     209      467088 :     xlrec.storesNulls = isNulls;
     210             : 
     211             :     /* these will be filled below as needed */
     212      467088 :     xlrec.offnumLeaf = InvalidOffsetNumber;
     213      467088 :     xlrec.offnumHeadLeaf = InvalidOffsetNumber;
     214      467088 :     xlrec.offnumParent = InvalidOffsetNumber;
     215      467088 :     xlrec.nodeI = 0;
     216             : 
     217      467088 :     START_CRIT_SECTION();
     218             : 
     219      932948 :     if (current->offnum == InvalidOffsetNumber ||
     220      923676 :         SpGistBlockIsRoot(current->blkno))
     221             :     {
     222             :         /* Tuple is not part of a chain */
     223        9320 :         leafTuple->nextOffset = InvalidOffsetNumber;
     224        9320 :         current->offnum = SpGistPageAddNewItem(state, current->page,
     225        9320 :                                                (Item) leafTuple, leafTuple->size,
     226             :                                                NULL, false);
     227             : 
     228        9320 :         xlrec.offnumLeaf = current->offnum;
     229             : 
     230             :         /* Must update parent's downlink if any */
     231       18640 :         if (parent->buffer != InvalidBuffer)
     232             :         {
     233        1228 :             xlrec.offnumParent = parent->offnum;
     234        1228 :             xlrec.nodeI = parent->node;
     235             : 
     236        1228 :             saveNodeLink(index, parent, current->blkno, current->offnum);
     237             :         }
     238             :     }
     239             :     else
     240             :     {
     241             :         /*
     242             :          * Tuple must be inserted into existing chain.  We mustn't change the
     243             :          * chain's head address, but we don't need to chase the entire chain
     244             :          * to put the tuple at the end; we can insert it second.
     245             :          *
     246             :          * Also, it's possible that the "chain" consists only of a DEAD tuple,
     247             :          * in which case we should replace the DEAD tuple in-place.
     248             :          */
     249             :         SpGistLeafTuple head;
     250             :         OffsetNumber offnum;
     251             : 
     252      457768 :         head = (SpGistLeafTuple) PageGetItem(current->page,
     253             :                                              PageGetItemId(current->page, current->offnum));
     254      457768 :         if (head->tupstate == SPGIST_LIVE)
     255             :         {
     256      457768 :             leafTuple->nextOffset = head->nextOffset;
     257      457768 :             offnum = SpGistPageAddNewItem(state, current->page,
     258      457768 :                                           (Item) leafTuple, leafTuple->size,
     259             :                                           NULL, false);
     260             : 
     261             :             /*
     262             :              * re-get head of list because it could have been moved on page,
     263             :              * and set new second element
     264             :              */
     265      457768 :             head = (SpGistLeafTuple) PageGetItem(current->page,
     266             :                                                  PageGetItemId(current->page, current->offnum));
     267      457768 :             head->nextOffset = offnum;
     268             : 
     269      457768 :             xlrec.offnumLeaf = offnum;
     270      457768 :             xlrec.offnumHeadLeaf = current->offnum;
     271             :         }
     272           0 :         else if (head->tupstate == SPGIST_DEAD)
     273             :         {
     274           0 :             leafTuple->nextOffset = InvalidOffsetNumber;
     275           0 :             PageIndexTupleDelete(current->page, current->offnum);
     276           0 :             if (PageAddItem(current->page,
     277             :                             (Item) leafTuple, leafTuple->size,
     278           0 :                             current->offnum, false, false) != current->offnum)
     279           0 :                 elog(ERROR, "failed to add item of size %u to SPGiST index page",
     280             :                      leafTuple->size);
     281             : 
     282             :             /* WAL replay distinguishes this case by equal offnums */
     283           0 :             xlrec.offnumLeaf = current->offnum;
     284           0 :             xlrec.offnumHeadLeaf = current->offnum;
     285             :         }
     286             :         else
     287           0 :             elog(ERROR, "unexpected SPGiST tuple state: %d", head->tupstate);
     288             :     }
     289             : 
     290      467088 :     MarkBufferDirty(current->buffer);
     291             : 
     292      467088 :     if (RelationNeedsWAL(index) && !state->isBuild)
     293             :     {
     294             :         XLogRecPtr  recptr;
     295             :         int         flags;
     296             : 
     297      148050 :         XLogBeginInsert();
     298      148050 :         XLogRegisterData((char *) &xlrec, sizeof(xlrec));
     299      148050 :         XLogRegisterData((char *) leafTuple, leafTuple->size);
     300             : 
     301      148050 :         flags = REGBUF_STANDARD;
     302      148050 :         if (xlrec.newPage)
     303          24 :             flags |= REGBUF_WILL_INIT;
     304      148050 :         XLogRegisterBuffer(0, current->buffer, flags);
     305      148050 :         if (xlrec.offnumParent != InvalidOffsetNumber)
     306         500 :             XLogRegisterBuffer(1, parent->buffer, REGBUF_STANDARD);
     307             : 
     308      148050 :         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_ADD_LEAF);
     309             : 
     310      148050 :         PageSetLSN(current->page, recptr);
     311             : 
     312             :         /* update parent only if we actually changed it */
     313      148050 :         if (xlrec.offnumParent != InvalidOffsetNumber)
     314             :         {
     315         500 :             PageSetLSN(parent->page, recptr);
     316             :         }
     317             :     }
     318             : 
     319      467088 :     END_CRIT_SECTION();
     320      467088 : }
     321             : 
     322             : /*
     323             :  * Count the number and total size of leaf tuples in the chain starting at
     324             :  * current->offnum.  Return number into *nToSplit and total size as function
     325             :  * result.
     326             :  *
     327             :  * Klugy special case when considering the root page (i.e., root is a leaf
     328             :  * page, but we're about to split for the first time): return fake large
     329             :  * values to force spgdoinsert() to take the doPickSplit rather than
     330             :  * moveLeafs code path.  moveLeafs is not prepared to deal with root page.
     331             :  */
     332             : static int
     333        4468 : checkSplitConditions(Relation index, SpGistState *state,
     334             :                      SPPageDesc *current, int *nToSplit)
     335             : {
     336             :     int         i,
     337        4468 :                 n = 0,
     338        4468 :                 totalSize = 0;
     339             : 
     340        4468 :     if (SpGistBlockIsRoot(current->blkno))
     341             :     {
     342             :         /* return impossible values to force split */
     343          44 :         *nToSplit = BLCKSZ;
     344          44 :         return BLCKSZ;
     345             :     }
     346             : 
     347        4424 :     i = current->offnum;
     348      469944 :     while (i != InvalidOffsetNumber)
     349             :     {
     350             :         SpGistLeafTuple it;
     351             : 
     352             :         Assert(i >= FirstOffsetNumber &&
     353             :                i <= PageGetMaxOffsetNumber(current->page));
     354      461096 :         it = (SpGistLeafTuple) PageGetItem(current->page,
     355             :                                            PageGetItemId(current->page, i));
     356      461096 :         if (it->tupstate == SPGIST_LIVE)
     357             :         {
     358      461096 :             n++;
     359      461096 :             totalSize += it->size + sizeof(ItemIdData);
     360             :         }
     361           0 :         else if (it->tupstate == SPGIST_DEAD)
     362             :         {
     363             :             /* We could see a DEAD tuple as first/only chain item */
     364             :             Assert(i == current->offnum);
     365             :             Assert(it->nextOffset == InvalidOffsetNumber);
     366             :             /* Don't count it in result, because it won't go to other page */
     367             :         }
     368             :         else
     369           0 :             elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
     370             : 
     371      461096 :         i = it->nextOffset;
     372             :     }
     373             : 
     374        4424 :     *nToSplit = n;
     375             : 
     376        4424 :     return totalSize;
     377             : }
     378             : 
     379             : /*
     380             :  * current points to a leaf-tuple chain that we wanted to add newLeafTuple to,
     381             :  * but the chain has to be moved because there's not enough room to add
     382             :  * newLeafTuple to its page.  We use this method when the chain contains
     383             :  * very little data so a split would be inefficient.  We are sure we can
     384             :  * fit the chain plus newLeafTuple on one other page.
     385             :  */
     386             : static void
     387        1084 : moveLeafs(Relation index, SpGistState *state,
     388             :           SPPageDesc *current, SPPageDesc *parent,
     389             :           SpGistLeafTuple newLeafTuple, bool isNulls)
     390             : {
     391             :     int         i,
     392             :                 nDelete,
     393             :                 nInsert,
     394             :                 size;
     395             :     Buffer      nbuf;
     396             :     Page        npage;
     397             :     SpGistLeafTuple it;
     398        1084 :     OffsetNumber r = InvalidOffsetNumber,
     399        1084 :                 startOffset = InvalidOffsetNumber;
     400        1084 :     bool        replaceDead = false;
     401             :     OffsetNumber *toDelete;
     402             :     OffsetNumber *toInsert;
     403             :     BlockNumber nblkno;
     404             :     spgxlogMoveLeafs xlrec;
     405             :     char       *leafdata,
     406             :                *leafptr;
     407             : 
     408             :     /* This doesn't work on root page */
     409             :     Assert(parent->buffer != InvalidBuffer);
     410             :     Assert(parent->buffer != current->buffer);
     411             : 
     412             :     /* Locate the tuples to be moved, and count up the space needed */
     413        1084 :     i = PageGetMaxOffsetNumber(current->page);
     414        1084 :     toDelete = (OffsetNumber *) palloc(sizeof(OffsetNumber) * i);
     415        1084 :     toInsert = (OffsetNumber *) palloc(sizeof(OffsetNumber) * (i + 1));
     416             : 
     417        1084 :     size = newLeafTuple->size + sizeof(ItemIdData);
     418             : 
     419        1084 :     nDelete = 0;
     420        1084 :     i = current->offnum;
     421       46068 :     while (i != InvalidOffsetNumber)
     422             :     {
     423             :         SpGistLeafTuple it;
     424             : 
     425             :         Assert(i >= FirstOffsetNumber &&
     426             :                i <= PageGetMaxOffsetNumber(current->page));
     427       43900 :         it = (SpGistLeafTuple) PageGetItem(current->page,
     428             :                                            PageGetItemId(current->page, i));
     429             : 
     430       43900 :         if (it->tupstate == SPGIST_LIVE)
     431             :         {
     432       43900 :             toDelete[nDelete] = i;
     433       43900 :             size += it->size + sizeof(ItemIdData);
     434       43900 :             nDelete++;
     435             :         }
     436           0 :         else if (it->tupstate == SPGIST_DEAD)
     437             :         {
     438             :             /* We could see a DEAD tuple as first/only chain item */
     439             :             Assert(i == current->offnum);
     440             :             Assert(it->nextOffset == InvalidOffsetNumber);
     441             :             /* We don't want to move it, so don't count it in size */
     442           0 :             toDelete[nDelete] = i;
     443           0 :             nDelete++;
     444           0 :             replaceDead = true;
     445             :         }
     446             :         else
     447           0 :             elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
     448             : 
     449       43900 :         i = it->nextOffset;
     450             :     }
     451             : 
     452             :     /* Find a leaf page that will hold them */
     453        1084 :     nbuf = SpGistGetBuffer(index, GBUF_LEAF | (isNulls ? GBUF_NULLS : 0),
     454             :                            size, &xlrec.newPage);
     455        1084 :     npage = BufferGetPage(nbuf);
     456        1084 :     nblkno = BufferGetBlockNumber(nbuf);
     457             :     Assert(nblkno != current->blkno);
     458             : 
     459        1084 :     leafdata = leafptr = palloc(size);
     460             : 
     461        1084 :     START_CRIT_SECTION();
     462             : 
     463             :     /* copy all the old tuples to new page, unless they're dead */
     464        1084 :     nInsert = 0;
     465        1084 :     if (!replaceDead)
     466             :     {
     467       44984 :         for (i = 0; i < nDelete; i++)
     468             :         {
     469       43900 :             it = (SpGistLeafTuple) PageGetItem(current->page,
     470             :                                                PageGetItemId(current->page, toDelete[i]));
     471             :             Assert(it->tupstate == SPGIST_LIVE);
     472             : 
     473             :             /*
     474             :              * Update chain link (notice the chain order gets reversed, but we
     475             :              * don't care).  We're modifying the tuple on the source page
     476             :              * here, but it's okay since we're about to delete it.
     477             :              */
     478       43900 :             it->nextOffset = r;
     479             : 
     480       43900 :             r = SpGistPageAddNewItem(state, npage, (Item) it, it->size,
     481             :                                      &startOffset, false);
     482             : 
     483       43900 :             toInsert[nInsert] = r;
     484       43900 :             nInsert++;
     485             : 
     486             :             /* save modified tuple into leafdata as well */
     487       43900 :             memcpy(leafptr, it, it->size);
     488       43900 :             leafptr += it->size;
     489             :         }
     490             :     }
     491             : 
     492             :     /* add the new tuple as well */
     493        1084 :     newLeafTuple->nextOffset = r;
     494        1084 :     r = SpGistPageAddNewItem(state, npage,
     495        1084 :                              (Item) newLeafTuple, newLeafTuple->size,
     496             :                              &startOffset, false);
     497        1084 :     toInsert[nInsert] = r;
     498        1084 :     nInsert++;
     499        1084 :     memcpy(leafptr, newLeafTuple, newLeafTuple->size);
     500        1084 :     leafptr += newLeafTuple->size;
     501             : 
     502             :     /*
     503             :      * Now delete the old tuples, leaving a redirection pointer behind for the
     504             :      * first one, unless we're doing an index build; in which case there can't
     505             :      * be any concurrent scan so we need not provide a redirect.
     506             :      */
     507        2168 :     spgPageIndexMultiDelete(state, current->page, toDelete, nDelete,
     508        1084 :                             state->isBuild ? SPGIST_PLACEHOLDER : SPGIST_REDIRECT,
     509             :                             SPGIST_PLACEHOLDER,
     510             :                             nblkno, r);
     511             : 
     512             :     /* Update parent's downlink and mark parent page dirty */
     513        1084 :     saveNodeLink(index, parent, nblkno, r);
     514             : 
     515             :     /* Mark the leaf pages too */
     516        1084 :     MarkBufferDirty(current->buffer);
     517        1084 :     MarkBufferDirty(nbuf);
     518             : 
     519        1084 :     if (RelationNeedsWAL(index) && !state->isBuild)
     520             :     {
     521             :         XLogRecPtr  recptr;
     522             : 
     523             :         /* prepare WAL info */
     524         328 :         STORE_STATE(state, xlrec.stateSrc);
     525             : 
     526         328 :         xlrec.nMoves = nDelete;
     527         328 :         xlrec.replaceDead = replaceDead;
     528         328 :         xlrec.storesNulls = isNulls;
     529             : 
     530         328 :         xlrec.offnumParent = parent->offnum;
     531         328 :         xlrec.nodeI = parent->node;
     532             : 
     533         328 :         XLogBeginInsert();
     534         328 :         XLogRegisterData((char *) &xlrec, SizeOfSpgxlogMoveLeafs);
     535         328 :         XLogRegisterData((char *) toDelete,
     536         328 :                          sizeof(OffsetNumber) * nDelete);
     537         328 :         XLogRegisterData((char *) toInsert,
     538         328 :                          sizeof(OffsetNumber) * nInsert);
     539         328 :         XLogRegisterData((char *) leafdata, leafptr - leafdata);
     540             : 
     541         328 :         XLogRegisterBuffer(0, current->buffer, REGBUF_STANDARD);
     542         328 :         XLogRegisterBuffer(1, nbuf, REGBUF_STANDARD | (xlrec.newPage ? REGBUF_WILL_INIT : 0));
     543         328 :         XLogRegisterBuffer(2, parent->buffer, REGBUF_STANDARD);
     544             : 
     545         328 :         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_MOVE_LEAFS);
     546             : 
     547         328 :         PageSetLSN(current->page, recptr);
     548         328 :         PageSetLSN(npage, recptr);
     549         328 :         PageSetLSN(parent->page, recptr);
     550             :     }
     551             : 
     552        1084 :     END_CRIT_SECTION();
     553             : 
     554             :     /* Update local free-space cache and release new buffer */
     555        1084 :     SpGistSetLastUsedPage(index, nbuf);
     556        1084 :     UnlockReleaseBuffer(nbuf);
     557        1084 : }
     558             : 
     559             : /*
     560             :  * Update previously-created redirection tuple with appropriate destination
     561             :  *
     562             :  * We use this when it's not convenient to know the destination first.
     563             :  * The tuple should have been made with the "impossible" destination of
     564             :  * the metapage.
     565             :  */
     566             : static void
     567         858 : setRedirectionTuple(SPPageDesc *current, OffsetNumber position,
     568             :                     BlockNumber blkno, OffsetNumber offnum)
     569             : {
     570             :     SpGistDeadTuple dt;
     571             : 
     572         858 :     dt = (SpGistDeadTuple) PageGetItem(current->page,
     573             :                                        PageGetItemId(current->page, position));
     574             :     Assert(dt->tupstate == SPGIST_REDIRECT);
     575             :     Assert(ItemPointerGetBlockNumber(&dt->pointer) == SPGIST_METAPAGE_BLKNO);
     576         858 :     ItemPointerSet(&dt->pointer, blkno, offnum);
     577         858 : }
     578             : 
     579             : /*
     580             :  * Test to see if the user-defined picksplit function failed to do its job,
     581             :  * ie, it put all the leaf tuples into the same node.
     582             :  * If so, randomly divide the tuples into several nodes (all with the same
     583             :  * label) and return true to select allTheSame mode for this inner tuple.
     584             :  *
     585             :  * (This code is also used to forcibly select allTheSame mode for nulls.)
     586             :  *
     587             :  * If we know that the leaf tuples wouldn't all fit on one page, then we
     588             :  * exclude the last tuple (which is the incoming new tuple that forced a split)
     589             :  * from the check to see if more than one node is used.  The reason for this
     590             :  * is that if the existing tuples are put into only one chain, then even if
     591             :  * we move them all to an empty page, there would still not be room for the
     592             :  * new tuple, so we'd get into an infinite loop of picksplit attempts.
     593             :  * Forcing allTheSame mode dodges this problem by ensuring the old tuples will
     594             :  * be split across pages.  (Exercise for the reader: figure out why this
     595             :  * fixes the problem even when there is only one old tuple.)
     596             :  */
     597             : static bool
     598        3384 : checkAllTheSame(spgPickSplitIn *in, spgPickSplitOut *out, bool tooBig,
     599             :                 bool *includeNew)
     600             : {
     601             :     int         theNode;
     602             :     int         limit;
     603             :     int         i;
     604             : 
     605             :     /* For the moment, assume we can include the new leaf tuple */
     606        3384 :     *includeNew = true;
     607             : 
     608             :     /* If there's only the new leaf tuple, don't select allTheSame mode */
     609        3384 :     if (in->nTuples <= 1)
     610           0 :         return false;
     611             : 
     612             :     /* If tuple set doesn't fit on one page, ignore the new tuple in test */
     613        3384 :     limit = tooBig ? in->nTuples - 1 : in->nTuples;
     614             : 
     615             :     /* Check to see if more than one node is populated */
     616        3384 :     theNode = out->mapTuplesToNodes[0];
     617       83602 :     for (i = 1; i < limit; i++)
     618             :     {
     619       83422 :         if (out->mapTuplesToNodes[i] != theNode)
     620        3204 :             return false;
     621             :     }
     622             : 
     623             :     /* Nope, so override the picksplit function's decisions */
     624             : 
     625             :     /* If the new tuple is in its own node, it can't be included in split */
     626         180 :     if (tooBig && out->mapTuplesToNodes[in->nTuples - 1] != theNode)
     627           0 :         *includeNew = false;
     628             : 
     629         180 :     out->nNodes = 8;         /* arbitrary number of child nodes */
     630             : 
     631             :     /* Random assignment of tuples to nodes (note we include new tuple) */
     632       19300 :     for (i = 0; i < in->nTuples; i++)
     633       19120 :         out->mapTuplesToNodes[i] = i % out->nNodes;
     634             : 
     635             :     /* The opclass may not use node labels, but if it does, duplicate 'em */
     636         180 :     if (out->nodeLabels)
     637             :     {
     638          34 :         Datum       theLabel = out->nodeLabels[theNode];
     639             : 
     640          34 :         out->nodeLabels = (Datum *) palloc(sizeof(Datum) * out->nNodes);
     641         306 :         for (i = 0; i < out->nNodes; i++)
     642         272 :             out->nodeLabels[i] = theLabel;
     643             :     }
     644             : 
     645             :     /* We don't touch the prefix or the leaf tuple datum assignments */
     646             : 
     647         180 :     return true;
     648             : }
     649             : 
     650             : /*
     651             :  * current points to a leaf-tuple chain that we wanted to add newLeafTuple to,
     652             :  * but the chain has to be split because there's not enough room to add
     653             :  * newLeafTuple to its page.
     654             :  *
     655             :  * This function splits the leaf tuple set according to picksplit's rules,
     656             :  * creating one or more new chains that are spread across the current page
     657             :  * and an additional leaf page (we assume that two leaf pages will be
     658             :  * sufficient).  A new inner tuple is created, and the parent downlink
     659             :  * pointer is updated to point to that inner tuple instead of the leaf chain.
     660             :  *
     661             :  * On exit, current contains the address of the new inner tuple.
     662             :  *
     663             :  * Returns true if we successfully inserted newLeafTuple during this function,
     664             :  * false if caller still has to do it (meaning another picksplit operation is
     665             :  * probably needed).  Failure could occur if the picksplit result is fairly
     666             :  * unbalanced, or if newLeafTuple is just plain too big to fit on a page.
     667             :  * Because we force the picksplit result to be at least two chains, each
     668             :  * cycle will get rid of at least one leaf tuple from the chain, so the loop
     669             :  * will eventually terminate if lack of balance is the issue.  If the tuple
     670             :  * is too big, we assume that repeated picksplit operations will eventually
     671             :  * make it small enough by repeated prefix-stripping.  A broken opclass could
     672             :  * make this an infinite loop, though.
     673             :  */
     674             : static bool
     675        3384 : doPickSplit(Relation index, SpGistState *state,
     676             :             SPPageDesc *current, SPPageDesc *parent,
     677             :             SpGistLeafTuple newLeafTuple,
     678             :             int level, bool isNulls, bool isNew)
     679             : {
     680        3384 :     bool        insertedNew = false;
     681             :     spgPickSplitIn in;
     682             :     spgPickSplitOut out;
     683             :     FmgrInfo   *procinfo;
     684             :     bool        includeNew;
     685             :     int         i,
     686             :                 max,
     687             :                 n;
     688             :     SpGistInnerTuple innerTuple;
     689             :     SpGistNodeTuple node,
     690             :                *nodes;
     691             :     Buffer      newInnerBuffer,
     692             :                 newLeafBuffer;
     693             :     ItemPointerData *heapPtrs;
     694             :     uint8      *leafPageSelect;
     695             :     int        *leafSizes;
     696             :     OffsetNumber *toDelete;
     697             :     OffsetNumber *toInsert;
     698        3384 :     OffsetNumber redirectTuplePos = InvalidOffsetNumber;
     699             :     OffsetNumber startOffsets[2];
     700             :     SpGistLeafTuple *newLeafs;
     701             :     int         spaceToDelete;
     702             :     int         currentFreeSpace;
     703             :     int         totalLeafSizes;
     704             :     bool        allTheSame;
     705             :     spgxlogPickSplit xlrec;
     706             :     char       *leafdata,
     707             :                *leafptr;
     708             :     SPPageDesc  saveCurrent;
     709             :     int         nToDelete,
     710             :                 nToInsert,
     711             :                 maxToInclude;
     712             : 
     713        3384 :     in.level = level;
     714             : 
     715             :     /*
     716             :      * Allocate per-leaf-tuple work arrays with max possible size
     717             :      */
     718        3384 :     max = PageGetMaxOffsetNumber(current->page);
     719        3384 :     n = max + 1;
     720        3384 :     in.datums = (Datum *) palloc(sizeof(Datum) * n);
     721        3384 :     heapPtrs = (ItemPointerData *) palloc(sizeof(ItemPointerData) * n);
     722        3384 :     toDelete = (OffsetNumber *) palloc(sizeof(OffsetNumber) * n);
     723        3384 :     toInsert = (OffsetNumber *) palloc(sizeof(OffsetNumber) * n);
     724        3384 :     newLeafs = (SpGistLeafTuple *) palloc(sizeof(SpGistLeafTuple) * n);
     725        3384 :     leafPageSelect = (uint8 *) palloc(sizeof(uint8) * n);
     726             : 
     727        3384 :     STORE_STATE(state, xlrec.stateSrc);
     728             : 
     729             :     /*
     730             :      * Form list of leaf tuples which will be distributed as split result;
     731             :      * also, count up the amount of space that will be freed from current.
     732             :      * (Note that in the non-root case, we won't actually delete the old
     733             :      * tuples, only replace them with redirects or placeholders.)
     734             :      *
     735             :      * Note: the SGLTDATUM calls here are safe even when dealing with a nulls
     736             :      * page.  For a pass-by-value data type we will fetch a word that must
     737             :      * exist even though it may contain garbage (because of the fact that leaf
     738             :      * tuples must have size at least SGDTSIZE).  For a pass-by-reference type
     739             :      * we are just computing a pointer that isn't going to get dereferenced.
     740             :      * So it's not worth guarding the calls with isNulls checks.
     741             :      */
     742        3384 :     nToInsert = 0;
     743        3384 :     nToDelete = 0;
     744        3384 :     spaceToDelete = 0;
     745        3428 :     if (SpGistBlockIsRoot(current->blkno))
     746             :     {
     747             :         /*
     748             :          * We are splitting the root (which up to now is also a leaf page).
     749             :          * Its tuples are not linked, so scan sequentially to get them all. We
     750             :          * ignore the original value of current->offnum.
     751             :          */
     752        7800 :         for (i = FirstOffsetNumber; i <= max; i++)
     753             :         {
     754             :             SpGistLeafTuple it;
     755             : 
     756        7756 :             it = (SpGistLeafTuple) PageGetItem(current->page,
     757             :                                                PageGetItemId(current->page, i));
     758        7756 :             if (it->tupstate == SPGIST_LIVE)
     759             :             {
     760        7756 :                 in.datums[nToInsert] = SGLTDATUM(it, state);
     761        7756 :                 heapPtrs[nToInsert] = it->heapPtr;
     762        7756 :                 nToInsert++;
     763        7756 :                 toDelete[nToDelete] = i;
     764        7756 :                 nToDelete++;
     765             :                 /* we will delete the tuple altogether, so count full space */
     766        7756 :                 spaceToDelete += it->size + sizeof(ItemIdData);
     767             :             }
     768             :             else                /* tuples on root should be live */
     769           0 :                 elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
     770             :         }
     771             :     }
     772             :     else
     773             :     {
     774             :         /* Normal case, just collect the leaf tuples in the chain */
     775        3340 :         i = current->offnum;
     776      423876 :         while (i != InvalidOffsetNumber)
     777             :         {
     778             :             SpGistLeafTuple it;
     779             : 
     780             :             Assert(i >= FirstOffsetNumber && i <= max);
     781      417196 :             it = (SpGistLeafTuple) PageGetItem(current->page,
     782             :                                                PageGetItemId(current->page, i));
     783      417196 :             if (it->tupstate == SPGIST_LIVE)
     784             :             {
     785      417196 :                 in.datums[nToInsert] = SGLTDATUM(it, state);
     786      417196 :                 heapPtrs[nToInsert] = it->heapPtr;
     787      417196 :                 nToInsert++;
     788      417196 :                 toDelete[nToDelete] = i;
     789      417196 :                 nToDelete++;
     790             :                 /* we will not delete the tuple, only replace with dead */
     791             :                 Assert(it->size >= SGDTSIZE);
     792      417196 :                 spaceToDelete += it->size - SGDTSIZE;
     793             :             }
     794           0 :             else if (it->tupstate == SPGIST_DEAD)
     795             :             {
     796             :                 /* We could see a DEAD tuple as first/only chain item */
     797             :                 Assert(i == current->offnum);
     798             :                 Assert(it->nextOffset == InvalidOffsetNumber);
     799           0 :                 toDelete[nToDelete] = i;
     800           0 :                 nToDelete++;
     801             :                 /* replacing it with redirect will save no space */
     802             :             }
     803             :             else
     804           0 :                 elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
     805             : 
     806      417196 :             i = it->nextOffset;
     807             :         }
     808             :     }
     809        3384 :     in.nTuples = nToInsert;
     810             : 
     811             :     /*
     812             :      * We may not actually insert new tuple because another picksplit may be
     813             :      * necessary due to too large value, but we will try to allocate enough
     814             :      * space to include it; and in any case it has to be included in the input
     815             :      * for the picksplit function.  So don't increment nToInsert yet.
     816             :      */
     817        3384 :     in.datums[in.nTuples] = SGLTDATUM(newLeafTuple, state);
     818        3384 :     heapPtrs[in.nTuples] = newLeafTuple->heapPtr;
     819        3384 :     in.nTuples++;
     820             : 
     821        3384 :     memset(&out, 0, sizeof(out));
     822             : 
     823        3384 :     if (!isNulls)
     824             :     {
     825             :         /*
     826             :          * Perform split using user-defined method.
     827             :          */
     828        3384 :         procinfo = index_getprocinfo(index, 1, SPGIST_PICKSPLIT_PROC);
     829        6768 :         FunctionCall2Coll(procinfo,
     830        3384 :                           index->rd_indcollation[0],
     831             :                           PointerGetDatum(&in),
     832             :                           PointerGetDatum(&out));
     833             : 
     834             :         /*
     835             :          * Form new leaf tuples and count up the total space needed.
     836             :          */
     837        3384 :         totalLeafSizes = 0;
     838      431720 :         for (i = 0; i < in.nTuples; i++)
     839             :         {
     840      856672 :             newLeafs[i] = spgFormLeafTuple(state, heapPtrs + i,
     841      428336 :                                            out.leafTupleDatums[i],
     842             :                                            false);
     843      428336 :             totalLeafSizes += newLeafs[i]->size + sizeof(ItemIdData);
     844             :         }
     845             :     }
     846             :     else
     847             :     {
     848             :         /*
     849             :          * Perform dummy split that puts all tuples into one node.
     850             :          * checkAllTheSame will override this and force allTheSame mode.
     851             :          */
     852           0 :         out.hasPrefix = false;
     853           0 :         out.nNodes = 1;
     854           0 :         out.nodeLabels = NULL;
     855           0 :         out.mapTuplesToNodes = palloc0(sizeof(int) * in.nTuples);
     856             : 
     857             :         /*
     858             :          * Form new leaf tuples and count up the total space needed.
     859             :          */
     860           0 :         totalLeafSizes = 0;
     861           0 :         for (i = 0; i < in.nTuples; i++)
     862             :         {
     863           0 :             newLeafs[i] = spgFormLeafTuple(state, heapPtrs + i,
     864             :                                            (Datum) 0,
     865             :                                            true);
     866           0 :             totalLeafSizes += newLeafs[i]->size + sizeof(ItemIdData);
     867             :         }
     868             :     }
     869             : 
     870             :     /*
     871             :      * Check to see if the picksplit function failed to separate the values,
     872             :      * ie, it put them all into the same child node.  If so, select allTheSame
     873             :      * mode and create a random split instead.  See comments for
     874             :      * checkAllTheSame as to why we need to know if the new leaf tuples could
     875             :      * fit on one page.
     876             :      */
     877        3384 :     allTheSame = checkAllTheSame(&in, &out,
     878             :                                  totalLeafSizes > SPGIST_PAGE_CAPACITY,
     879             :                                  &includeNew);
     880             : 
     881             :     /*
     882             :      * If checkAllTheSame decided we must exclude the new tuple, don't
     883             :      * consider it any further.
     884             :      */
     885        3384 :     if (includeNew)
     886        3384 :         maxToInclude = in.nTuples;
     887             :     else
     888             :     {
     889           0 :         maxToInclude = in.nTuples - 1;
     890           0 :         totalLeafSizes -= newLeafs[in.nTuples - 1]->size + sizeof(ItemIdData);
     891             :     }
     892             : 
     893             :     /*
     894             :      * Allocate per-node work arrays.  Since checkAllTheSame could replace
     895             :      * out.nNodes with a value larger than the number of tuples on the input
     896             :      * page, we can't allocate these arrays before here.
     897             :      */
     898        3384 :     nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) * out.nNodes);
     899        3384 :     leafSizes = (int *) palloc0(sizeof(int) * out.nNodes);
     900             : 
     901             :     /*
     902             :      * Form nodes of inner tuple and inner tuple itself
     903             :      */
     904       27216 :     for (i = 0; i < out.nNodes; i++)
     905             :     {
     906       23832 :         Datum       label = (Datum) 0;
     907       23832 :         bool        labelisnull = (out.nodeLabels == NULL);
     908             : 
     909       23832 :         if (!labelisnull)
     910        1928 :             label = out.nodeLabels[i];
     911       23832 :         nodes[i] = spgFormNodeTuple(state, label, labelisnull);
     912             :     }
     913        6768 :     innerTuple = spgFormInnerTuple(state,
     914        3384 :                                    out.hasPrefix, out.prefixDatum,
     915             :                                    out.nNodes, nodes);
     916        3384 :     innerTuple->allTheSame = allTheSame;
     917             : 
     918             :     /*
     919             :      * Update nodes[] array to point into the newly formed innerTuple, so that
     920             :      * we can adjust their downlinks below.
     921             :      */
     922       27216 :     SGITITERATE(innerTuple, i, node)
     923             :     {
     924       23832 :         nodes[i] = node;
     925             :     }
     926             : 
     927             :     /*
     928             :      * Re-scan new leaf tuples and count up the space needed under each node.
     929             :      */
     930      431720 :     for (i = 0; i < maxToInclude; i++)
     931             :     {
     932      428336 :         n = out.mapTuplesToNodes[i];
     933      428336 :         if (n < 0 || n >= out.nNodes)
     934           0 :             elog(ERROR, "inconsistent result of SPGiST picksplit function");
     935      428336 :         leafSizes[n] += newLeafs[i]->size + sizeof(ItemIdData);
     936             :     }
     937             : 
     938             :     /*
     939             :      * To perform the split, we must insert a new inner tuple, which can't go
     940             :      * on a leaf page; and unless we are splitting the root page, we must then
     941             :      * update the parent tuple's downlink to point to the inner tuple.  If
     942             :      * there is room, we'll put the new inner tuple on the same page as the
     943             :      * parent tuple, otherwise we need another non-leaf buffer. But if the
     944             :      * parent page is the root, we can't add the new inner tuple there,
     945             :      * because the root page must have only one inner tuple.
     946             :      */
     947        3384 :     xlrec.initInner = false;
     948        6724 :     if (parent->buffer != InvalidBuffer &&
     949        9660 :         !SpGistBlockIsRoot(parent->blkno) &&
     950        3160 :         (SpGistPageGetFreeSpace(parent->page, 1) >=
     951        3160 :          innerTuple->size + sizeof(ItemIdData)))
     952             :     {
     953             :         /* New inner tuple will fit on parent page */
     954        2888 :         newInnerBuffer = parent->buffer;
     955             :     }
     956         496 :     else if (parent->buffer != InvalidBuffer)
     957             :     {
     958             :         /* Send tuple to page with next triple parity (see README) */
     959         904 :         newInnerBuffer = SpGistGetBuffer(index,
     960         904 :                                          GBUF_INNER_PARITY(parent->blkno + 1) |
     961         452 :                                          (isNulls ? GBUF_NULLS : 0),
     962         452 :                                          innerTuple->size + sizeof(ItemIdData),
     963             :                                          &xlrec.initInner);
     964             :     }
     965             :     else
     966             :     {
     967             :         /* Root page split ... inner tuple will go to root page */
     968          44 :         newInnerBuffer = InvalidBuffer;
     969             :     }
     970             : 
     971             :     /*
     972             :      * The new leaf tuples converted from the existing ones should require the
     973             :      * same or less space, and therefore should all fit onto one page
     974             :      * (although that's not necessarily the current page, since we can't
     975             :      * delete the old tuples but only replace them with placeholders).
     976             :      * However, the incoming new tuple might not also fit, in which case we
     977             :      * might need another picksplit cycle to reduce it some more.
     978             :      *
     979             :      * If there's not room to put everything back onto the current page, then
     980             :      * we decide on a per-node basis which tuples go to the new page. (We do
     981             :      * it like that because leaf tuple chains can't cross pages, so we must
     982             :      * place all leaf tuples belonging to the same parent node on the same
     983             :      * page.)
     984             :      *
     985             :      * If we are splitting the root page (turning it from a leaf page into an
     986             :      * inner page), then no leaf tuples can go back to the current page; they
     987             :      * must all go somewhere else.
     988             :      */
     989        3384 :     if (!SpGistBlockIsRoot(current->blkno))
     990        3340 :         currentFreeSpace = PageGetExactFreeSpace(current->page) + spaceToDelete;
     991             :     else
     992          44 :         currentFreeSpace = 0;   /* prevent assigning any tuples to current */
     993             : 
     994        3384 :     xlrec.initDest = false;
     995             : 
     996        3384 :     if (totalLeafSizes <= currentFreeSpace)
     997             :     {
     998             :         /* All the leaf tuples will fit on current page */
     999          16 :         newLeafBuffer = InvalidBuffer;
    1000             :         /* mark new leaf tuple as included in insertions, if allowed */
    1001          16 :         if (includeNew)
    1002             :         {
    1003          16 :             nToInsert++;
    1004          16 :             insertedNew = true;
    1005             :         }
    1006        1980 :         for (i = 0; i < nToInsert; i++)
    1007        1964 :             leafPageSelect[i] = 0;  /* signifies current page */
    1008             :     }
    1009        3368 :     else if (in.nTuples == 1 && totalLeafSizes > SPGIST_PAGE_CAPACITY)
    1010             :     {
    1011             :         /*
    1012             :          * We're trying to split up a long value by repeated suffixing, but
    1013             :          * it's not going to fit yet.  Don't bother allocating a second leaf
    1014             :          * buffer that we won't be able to use.
    1015             :          */
    1016           0 :         newLeafBuffer = InvalidBuffer;
    1017             :         Assert(includeNew);
    1018           0 :         Assert(nToInsert == 0);
    1019             :     }
    1020             :     else
    1021             :     {
    1022             :         /* We will need another leaf page */
    1023             :         uint8      *nodePageSelect;
    1024             :         int         curspace;
    1025             :         int         newspace;
    1026             : 
    1027        3368 :         newLeafBuffer = SpGistGetBuffer(index,
    1028             :                                         GBUF_LEAF | (isNulls ? GBUF_NULLS : 0),
    1029        3368 :                                         Min(totalLeafSizes,
    1030             :                                             SPGIST_PAGE_CAPACITY),
    1031             :                                         &xlrec.initDest);
    1032             : 
    1033             :         /*
    1034             :          * Attempt to assign node groups to the two pages.  We might fail to
    1035             :          * do so, even if totalLeafSizes is less than the available space,
    1036             :          * because we can't split a group across pages.
    1037             :          */
    1038        3368 :         nodePageSelect = (uint8 *) palloc(sizeof(uint8) * out.nNodes);
    1039             : 
    1040        3368 :         curspace = currentFreeSpace;
    1041        3368 :         newspace = PageGetExactFreeSpace(BufferGetPage(newLeafBuffer));
    1042       27168 :         for (i = 0; i < out.nNodes; i++)
    1043             :         {
    1044       23800 :             if (leafSizes[i] <= curspace)
    1045             :             {
    1046       16064 :                 nodePageSelect[i] = 0;  /* signifies current page */
    1047       16064 :                 curspace -= leafSizes[i];
    1048             :             }
    1049             :             else
    1050             :             {
    1051        7736 :                 nodePageSelect[i] = 1;  /* signifies new leaf page */
    1052        7736 :                 newspace -= leafSizes[i];
    1053             :             }
    1054             :         }
    1055        3368 :         if (curspace >= 0 && newspace >= 0)
    1056             :         {
    1057             :             /* Successful assignment, so we can include the new leaf tuple */
    1058        6472 :             if (includeNew)
    1059             :             {
    1060        3236 :                 nToInsert++;
    1061        3236 :                 insertedNew = true;
    1062             :             }
    1063             :         }
    1064         132 :         else if (includeNew)
    1065             :         {
    1066             :             /* We must exclude the new leaf tuple from the split */
    1067         132 :             int         nodeOfNewTuple = out.mapTuplesToNodes[in.nTuples - 1];
    1068             : 
    1069         264 :             leafSizes[nodeOfNewTuple] -=
    1070         132 :                 newLeafs[in.nTuples - 1]->size + sizeof(ItemIdData);
    1071             : 
    1072             :             /* Repeat the node assignment process --- should succeed now */
    1073         132 :             curspace = currentFreeSpace;
    1074         132 :             newspace = PageGetExactFreeSpace(BufferGetPage(newLeafBuffer));
    1075         636 :             for (i = 0; i < out.nNodes; i++)
    1076             :             {
    1077         504 :                 if (leafSizes[i] <= curspace)
    1078             :                 {
    1079         152 :                     nodePageSelect[i] = 0;  /* signifies current page */
    1080         152 :                     curspace -= leafSizes[i];
    1081             :                 }
    1082             :                 else
    1083             :                 {
    1084         352 :                     nodePageSelect[i] = 1;  /* signifies new leaf page */
    1085         352 :                     newspace -= leafSizes[i];
    1086             :                 }
    1087             :             }
    1088         132 :             if (curspace < 0 || newspace < 0)
    1089           0 :                 elog(ERROR, "failed to divide leaf tuple groups across pages");
    1090             :         }
    1091             :         else
    1092             :         {
    1093             :             /* oops, we already excluded new tuple ... should not get here */
    1094           0 :             elog(ERROR, "failed to divide leaf tuple groups across pages");
    1095             :         }
    1096             :         /* Expand the per-node assignments to be shown per leaf tuple */
    1097      429608 :         for (i = 0; i < nToInsert; i++)
    1098             :         {
    1099      426240 :             n = out.mapTuplesToNodes[i];
    1100      426240 :             leafPageSelect[i] = nodePageSelect[n];
    1101             :         }
    1102             :     }
    1103             : 
    1104             :     /* Start preparing WAL record */
    1105        3384 :     xlrec.nDelete = 0;
    1106        3384 :     xlrec.initSrc = isNew;
    1107        3384 :     xlrec.storesNulls = isNulls;
    1108        3384 :     xlrec.isRootSplit = SpGistBlockIsRoot(current->blkno);
    1109             : 
    1110        3384 :     leafdata = leafptr = (char *) palloc(totalLeafSizes);
    1111             : 
    1112             :     /* Here we begin making the changes to the target pages */
    1113        3384 :     START_CRIT_SECTION();
    1114             : 
    1115             :     /*
    1116             :      * Delete old leaf tuples from current buffer, except when we're splitting
    1117             :      * the root; in that case there's no need because we'll re-init the page
    1118             :      * below.  We do this first to make room for reinserting new leaf tuples.
    1119             :      */
    1120        3384 :     if (!SpGistBlockIsRoot(current->blkno))
    1121             :     {
    1122             :         /*
    1123             :          * Init buffer instead of deleting individual tuples, but only if
    1124             :          * there aren't any other live tuples and only during build; otherwise
    1125             :          * we need to set a redirection tuple for concurrent scans.
    1126             :          */
    1127        5822 :         if (state->isBuild &&
    1128        2482 :             nToDelete + SpGistPageGetOpaque(current->page)->nPlaceholder ==
    1129        2482 :             PageGetMaxOffsetNumber(current->page))
    1130             :         {
    1131         196 :             SpGistInitBuffer(current->buffer,
    1132             :                              SPGIST_LEAF | (isNulls ? SPGIST_NULLS : 0));
    1133         196 :             xlrec.initSrc = true;
    1134             :         }
    1135        3144 :         else if (isNew)
    1136             :         {
    1137             :             /* don't expose the freshly init'd buffer as a backup block */
    1138             :             Assert(nToDelete == 0);
    1139             :         }
    1140             :         else
    1141             :         {
    1142        3144 :             xlrec.nDelete = nToDelete;
    1143             : 
    1144        3144 :             if (!state->isBuild)
    1145             :             {
    1146             :                 /*
    1147             :                  * Need to create redirect tuple (it will point to new inner
    1148             :                  * tuple) but right now the new tuple's location is not known
    1149             :                  * yet.  So, set the redirection pointer to "impossible" value
    1150             :                  * and remember its position to update tuple later.
    1151             :                  */
    1152         858 :                 if (nToDelete > 0)
    1153         858 :                     redirectTuplePos = toDelete[0];
    1154         858 :                 spgPageIndexMultiDelete(state, current->page,
    1155             :                                         toDelete, nToDelete,
    1156             :                                         SPGIST_REDIRECT,
    1157             :                                         SPGIST_PLACEHOLDER,
    1158             :                                         SPGIST_METAPAGE_BLKNO,
    1159             :                                         FirstOffsetNumber);
    1160             :             }
    1161             :             else
    1162             :             {
    1163             :                 /*
    1164             :                  * During index build there is not concurrent searches, so we
    1165             :                  * don't need to create redirection tuple.
    1166             :                  */
    1167        2286 :                 spgPageIndexMultiDelete(state, current->page,
    1168             :                                         toDelete, nToDelete,
    1169             :                                         SPGIST_PLACEHOLDER,
    1170             :                                         SPGIST_PLACEHOLDER,
    1171             :                                         InvalidBlockNumber,
    1172             :                                         InvalidOffsetNumber);
    1173             :             }
    1174             :         }
    1175             :     }
    1176             : 
    1177             :     /*
    1178             :      * Put leaf tuples on proper pages, and update downlinks in innerTuple's
    1179             :      * nodes.
    1180             :      */
    1181        3384 :     startOffsets[0] = startOffsets[1] = InvalidOffsetNumber;
    1182      431588 :     for (i = 0; i < nToInsert; i++)
    1183             :     {
    1184      428204 :         SpGistLeafTuple it = newLeafs[i];
    1185             :         Buffer      leafBuffer;
    1186             :         BlockNumber leafBlock;
    1187             :         OffsetNumber newoffset;
    1188             : 
    1189             :         /* Which page is it going to? */
    1190      428204 :         leafBuffer = leafPageSelect[i] ? newLeafBuffer : current->buffer;
    1191      428204 :         leafBlock = BufferGetBlockNumber(leafBuffer);
    1192             : 
    1193             :         /* Link tuple into correct chain for its node */
    1194      428204 :         n = out.mapTuplesToNodes[i];
    1195             : 
    1196      428204 :         if (ItemPointerIsValid(&nodes[n]->t_tid))
    1197             :         {
    1198             :             Assert(ItemPointerGetBlockNumber(&nodes[n]->t_tid) == leafBlock);
    1199      415488 :             it->nextOffset = ItemPointerGetOffsetNumber(&nodes[n]->t_tid);
    1200             :         }
    1201             :         else
    1202       12716 :             it->nextOffset = InvalidOffsetNumber;
    1203             : 
    1204             :         /* Insert it on page */
    1205      856408 :         newoffset = SpGistPageAddNewItem(state, BufferGetPage(leafBuffer),
    1206      428204 :                                          (Item) it, it->size,
    1207      428204 :                                          &startOffsets[leafPageSelect[i]],
    1208             :                                          false);
    1209      428204 :         toInsert[i] = newoffset;
    1210             : 
    1211             :         /* ... and complete the chain linking */
    1212      428204 :         ItemPointerSet(&nodes[n]->t_tid, leafBlock, newoffset);
    1213             : 
    1214             :         /* Also copy leaf tuple into WAL data */
    1215      428204 :         memcpy(leafptr, newLeafs[i], newLeafs[i]->size);
    1216      428204 :         leafptr += newLeafs[i]->size;
    1217             :     }
    1218             : 
    1219             :     /*
    1220             :      * We're done modifying the other leaf buffer (if any), so mark it dirty.
    1221             :      * current->buffer will be marked below, after we're entirely done
    1222             :      * modifying it.
    1223             :      */
    1224        3384 :     if (newLeafBuffer != InvalidBuffer)
    1225             :     {
    1226        3368 :         MarkBufferDirty(newLeafBuffer);
    1227             :     }
    1228             : 
    1229             :     /* Remember current buffer, since we're about to change "current" */
    1230        3384 :     saveCurrent = *current;
    1231             : 
    1232             :     /*
    1233             :      * Store the new innerTuple
    1234             :      */
    1235        3384 :     if (newInnerBuffer == parent->buffer && newInnerBuffer != InvalidBuffer)
    1236             :     {
    1237             :         /*
    1238             :          * new inner tuple goes to parent page
    1239             :          */
    1240             :         Assert(current->buffer != parent->buffer);
    1241             : 
    1242             :         /* Repoint "current" at the new inner tuple */
    1243        2888 :         current->blkno = parent->blkno;
    1244        2888 :         current->buffer = parent->buffer;
    1245        2888 :         current->page = parent->page;
    1246        2888 :         xlrec.offnumInner = current->offnum =
    1247        2888 :             SpGistPageAddNewItem(state, current->page,
    1248        2888 :                                  (Item) innerTuple, innerTuple->size,
    1249             :                                  NULL, false);
    1250             : 
    1251             :         /*
    1252             :          * Update parent node link and mark parent page dirty
    1253             :          */
    1254        2888 :         xlrec.innerIsParent = true;
    1255        2888 :         xlrec.offnumParent = parent->offnum;
    1256        2888 :         xlrec.nodeI = parent->node;
    1257        2888 :         saveNodeLink(index, parent, current->blkno, current->offnum);
    1258             : 
    1259             :         /*
    1260             :          * Update redirection link (in old current buffer)
    1261             :          */
    1262        2888 :         if (redirectTuplePos != InvalidOffsetNumber)
    1263         798 :             setRedirectionTuple(&saveCurrent, redirectTuplePos,
    1264         798 :                                 current->blkno, current->offnum);
    1265             : 
    1266             :         /* Done modifying old current buffer, mark it dirty */
    1267        2888 :         MarkBufferDirty(saveCurrent.buffer);
    1268             :     }
    1269         496 :     else if (parent->buffer != InvalidBuffer)
    1270             :     {
    1271             :         /*
    1272             :          * new inner tuple will be stored on a new page
    1273             :          */
    1274             :         Assert(newInnerBuffer != InvalidBuffer);
    1275             : 
    1276             :         /* Repoint "current" at the new inner tuple */
    1277         452 :         current->buffer = newInnerBuffer;
    1278         452 :         current->blkno = BufferGetBlockNumber(current->buffer);
    1279         452 :         current->page = BufferGetPage(current->buffer);
    1280         452 :         xlrec.offnumInner = current->offnum =
    1281         452 :             SpGistPageAddNewItem(state, current->page,
    1282         452 :                                  (Item) innerTuple, innerTuple->size,
    1283             :                                  NULL, false);
    1284             : 
    1285             :         /* Done modifying new current buffer, mark it dirty */
    1286         452 :         MarkBufferDirty(current->buffer);
    1287             : 
    1288             :         /*
    1289             :          * Update parent node link and mark parent page dirty
    1290             :          */
    1291         452 :         xlrec.innerIsParent = (parent->buffer == current->buffer);
    1292         452 :         xlrec.offnumParent = parent->offnum;
    1293         452 :         xlrec.nodeI = parent->node;
    1294         452 :         saveNodeLink(index, parent, current->blkno, current->offnum);
    1295             : 
    1296             :         /*
    1297             :          * Update redirection link (in old current buffer)
    1298             :          */
    1299         452 :         if (redirectTuplePos != InvalidOffsetNumber)
    1300          60 :             setRedirectionTuple(&saveCurrent, redirectTuplePos,
    1301          60 :                                 current->blkno, current->offnum);
    1302             : 
    1303             :         /* Done modifying old current buffer, mark it dirty */
    1304         452 :         MarkBufferDirty(saveCurrent.buffer);
    1305             :     }
    1306             :     else
    1307             :     {
    1308             :         /*
    1309             :          * Splitting root page, which was a leaf but now becomes inner page
    1310             :          * (and so "current" continues to point at it)
    1311             :          */
    1312             :         Assert(SpGistBlockIsRoot(current->blkno));
    1313             :         Assert(redirectTuplePos == InvalidOffsetNumber);
    1314             : 
    1315          44 :         SpGistInitBuffer(current->buffer, (isNulls ? SPGIST_NULLS : 0));
    1316          44 :         xlrec.initInner = true;
    1317          44 :         xlrec.innerIsParent = false;
    1318             : 
    1319          44 :         xlrec.offnumInner = current->offnum =
    1320          44 :             PageAddItem(current->page, (Item) innerTuple, innerTuple->size,
    1321             :                         InvalidOffsetNumber, false, false);
    1322          44 :         if (current->offnum != FirstOffsetNumber)
    1323           0 :             elog(ERROR, "failed to add item of size %u to SPGiST index page",
    1324             :                  innerTuple->size);
    1325             : 
    1326             :         /* No parent link to update, nor redirection to do */
    1327          44 :         xlrec.offnumParent = InvalidOffsetNumber;
    1328          44 :         xlrec.nodeI = 0;
    1329             : 
    1330             :         /* Done modifying new current buffer, mark it dirty */
    1331          44 :         MarkBufferDirty(current->buffer);
    1332             : 
    1333             :         /* saveCurrent doesn't represent a different buffer */
    1334          44 :         saveCurrent.buffer = InvalidBuffer;
    1335             :     }
    1336             : 
    1337        3384 :     if (RelationNeedsWAL(index) && !state->isBuild)
    1338             :     {
    1339             :         XLogRecPtr  recptr;
    1340             :         int         flags;
    1341             : 
    1342         870 :         XLogBeginInsert();
    1343             : 
    1344         870 :         xlrec.nInsert = nToInsert;
    1345         870 :         XLogRegisterData((char *) &xlrec, SizeOfSpgxlogPickSplit);
    1346             : 
    1347         870 :         XLogRegisterData((char *) toDelete,
    1348         870 :                          sizeof(OffsetNumber) * xlrec.nDelete);
    1349         870 :         XLogRegisterData((char *) toInsert,
    1350         870 :                          sizeof(OffsetNumber) * xlrec.nInsert);
    1351         870 :         XLogRegisterData((char *) leafPageSelect,
    1352         870 :                          sizeof(uint8) * xlrec.nInsert);
    1353         870 :         XLogRegisterData((char *) innerTuple, innerTuple->size);
    1354         870 :         XLogRegisterData(leafdata, leafptr - leafdata);
    1355             : 
    1356             :         /* Old leaf page */
    1357         870 :         if (BufferIsValid(saveCurrent.buffer))
    1358             :         {
    1359         858 :             flags = REGBUF_STANDARD;
    1360         858 :             if (xlrec.initSrc)
    1361           0 :                 flags |= REGBUF_WILL_INIT;
    1362         858 :             XLogRegisterBuffer(0, saveCurrent.buffer, flags);
    1363             :         }
    1364             : 
    1365             :         /* New leaf page */
    1366         870 :         if (BufferIsValid(newLeafBuffer))
    1367             :         {
    1368         870 :             flags = REGBUF_STANDARD;
    1369         870 :             if (xlrec.initDest)
    1370         728 :                 flags |= REGBUF_WILL_INIT;
    1371         870 :             XLogRegisterBuffer(1, newLeafBuffer, flags);
    1372             :         }
    1373             : 
    1374             :         /* Inner page */
    1375         870 :         flags = REGBUF_STANDARD;
    1376         870 :         if (xlrec.initInner)
    1377          24 :             flags |= REGBUF_WILL_INIT;
    1378         870 :         XLogRegisterBuffer(2, current->buffer, flags);
    1379             : 
    1380             :         /* Parent page, if different from inner page */
    1381         870 :         if (parent->buffer != InvalidBuffer)
    1382             :         {
    1383         858 :             if (parent->buffer != current->buffer)
    1384          60 :                 XLogRegisterBuffer(3, parent->buffer, REGBUF_STANDARD);
    1385             :             else
    1386             :                 Assert(xlrec.innerIsParent);
    1387             :         }
    1388             : 
    1389             :         /* Issue the WAL record */
    1390         870 :         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_PICKSPLIT);
    1391             : 
    1392             :         /* Update page LSNs on all affected pages */
    1393         870 :         if (newLeafBuffer != InvalidBuffer)
    1394             :         {
    1395         870 :             Page        page = BufferGetPage(newLeafBuffer);
    1396             : 
    1397         870 :             PageSetLSN(page, recptr);
    1398             :         }
    1399             : 
    1400         870 :         if (saveCurrent.buffer != InvalidBuffer)
    1401             :         {
    1402         858 :             Page        page = BufferGetPage(saveCurrent.buffer);
    1403             : 
    1404         858 :             PageSetLSN(page, recptr);
    1405             :         }
    1406             : 
    1407         870 :         PageSetLSN(current->page, recptr);
    1408             : 
    1409         870 :         if (parent->buffer != InvalidBuffer)
    1410             :         {
    1411         858 :             PageSetLSN(parent->page, recptr);
    1412             :         }
    1413             :     }
    1414             : 
    1415        3384 :     END_CRIT_SECTION();
    1416             : 
    1417             :     /* Update local free-space cache and unlock buffers */
    1418        3384 :     if (newLeafBuffer != InvalidBuffer)
    1419             :     {
    1420        3368 :         SpGistSetLastUsedPage(index, newLeafBuffer);
    1421        3368 :         UnlockReleaseBuffer(newLeafBuffer);
    1422             :     }
    1423        3384 :     if (saveCurrent.buffer != InvalidBuffer)
    1424             :     {
    1425        3340 :         SpGistSetLastUsedPage(index, saveCurrent.buffer);
    1426        3340 :         UnlockReleaseBuffer(saveCurrent.buffer);
    1427             :     }
    1428             : 
    1429        3384 :     return insertedNew;
    1430             : }
    1431             : 
    1432             : /*
    1433             :  * spgMatchNode action: descend to N'th child node of current inner tuple
    1434             :  */
    1435             : static void
    1436    12498708 : spgMatchNodeAction(Relation index, SpGistState *state,
    1437             :                    SpGistInnerTuple innerTuple,
    1438             :                    SPPageDesc *current, SPPageDesc *parent, int nodeN)
    1439             : {
    1440             :     int         i;
    1441             :     SpGistNodeTuple node;
    1442             : 
    1443             :     /* Release previous parent buffer if any */
    1444    24534088 :     if (parent->buffer != InvalidBuffer &&
    1445    12035380 :         parent->buffer != current->buffer)
    1446             :     {
    1447      465944 :         SpGistSetLastUsedPage(index, parent->buffer);
    1448      465944 :         UnlockReleaseBuffer(parent->buffer);
    1449             :     }
    1450             : 
    1451             :     /* Repoint parent to specified node of current inner tuple */
    1452    12498708 :     parent->blkno = current->blkno;
    1453    12498708 :     parent->buffer = current->buffer;
    1454    12498708 :     parent->page = current->page;
    1455    12498708 :     parent->offnum = current->offnum;
    1456    12498708 :     parent->node = nodeN;
    1457             : 
    1458             :     /* Locate that node */
    1459    20500570 :     SGITITERATE(innerTuple, i, node)
    1460             :     {
    1461    20500570 :         if (i == nodeN)
    1462    12498708 :             break;
    1463             :     }
    1464             : 
    1465    12498708 :     if (i != nodeN)
    1466           0 :         elog(ERROR, "failed to find requested node %d in SPGiST inner tuple",
    1467             :              nodeN);
    1468             : 
    1469             :     /* Point current to the downlink location, if any */
    1470    12498708 :     if (ItemPointerIsValid(&node->t_tid))
    1471             :     {
    1472    12497480 :         current->blkno = ItemPointerGetBlockNumber(&node->t_tid);
    1473    12497480 :         current->offnum = ItemPointerGetOffsetNumber(&node->t_tid);
    1474             :     }
    1475             :     else
    1476             :     {
    1477             :         /* Downlink is empty, so we'll need to find a new page */
    1478        1228 :         current->blkno = InvalidBlockNumber;
    1479        1228 :         current->offnum = InvalidOffsetNumber;
    1480             :     }
    1481             : 
    1482    12498708 :     current->buffer = InvalidBuffer;
    1483    12498708 :     current->page = NULL;
    1484    12498708 : }
    1485             : 
    1486             : /*
    1487             :  * spgAddNode action: add a node to the inner tuple at current
    1488             :  */
    1489             : static void
    1490         896 : spgAddNodeAction(Relation index, SpGistState *state,
    1491             :                  SpGistInnerTuple innerTuple,
    1492             :                  SPPageDesc *current, SPPageDesc *parent,
    1493             :                  int nodeN, Datum nodeLabel)
    1494             : {
    1495             :     SpGistInnerTuple newInnerTuple;
    1496             :     spgxlogAddNode xlrec;
    1497             : 
    1498             :     /* Should not be applied to nulls */
    1499             :     Assert(!SpGistPageStoresNulls(current->page));
    1500             : 
    1501             :     /* Construct new inner tuple with additional node */
    1502         896 :     newInnerTuple = addNode(state, innerTuple, nodeLabel, nodeN);
    1503             : 
    1504             :     /* Prepare WAL record */
    1505         896 :     STORE_STATE(state, xlrec.stateSrc);
    1506         896 :     xlrec.offnum = current->offnum;
    1507             : 
    1508             :     /* we don't fill these unless we need to change the parent downlink */
    1509         896 :     xlrec.parentBlk = -1;
    1510         896 :     xlrec.offnumParent = InvalidOffsetNumber;
    1511         896 :     xlrec.nodeI = 0;
    1512             : 
    1513             :     /* we don't fill these unless tuple has to be moved */
    1514         896 :     xlrec.offnumNew = InvalidOffsetNumber;
    1515         896 :     xlrec.newPage = false;
    1516             : 
    1517        1792 :     if (PageGetExactFreeSpace(current->page) >=
    1518         896 :         newInnerTuple->size - innerTuple->size)
    1519             :     {
    1520             :         /*
    1521             :          * We can replace the inner tuple by new version in-place
    1522             :          */
    1523         892 :         START_CRIT_SECTION();
    1524             : 
    1525         892 :         PageIndexTupleDelete(current->page, current->offnum);
    1526        1784 :         if (PageAddItem(current->page,
    1527             :                         (Item) newInnerTuple, newInnerTuple->size,
    1528         892 :                         current->offnum, false, false) != current->offnum)
    1529           0 :             elog(ERROR, "failed to add item of size %u to SPGiST index page",
    1530             :                  newInnerTuple->size);
    1531             : 
    1532         892 :         MarkBufferDirty(current->buffer);
    1533             : 
    1534         892 :         if (RelationNeedsWAL(index) && !state->isBuild)
    1535             :         {
    1536             :             XLogRecPtr  recptr;
    1537             : 
    1538         400 :             XLogBeginInsert();
    1539         400 :             XLogRegisterData((char *) &xlrec, sizeof(xlrec));
    1540         400 :             XLogRegisterData((char *) newInnerTuple, newInnerTuple->size);
    1541             : 
    1542         400 :             XLogRegisterBuffer(0, current->buffer, REGBUF_STANDARD);
    1543             : 
    1544         400 :             recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_ADD_NODE);
    1545             : 
    1546         400 :             PageSetLSN(current->page, recptr);
    1547             :         }
    1548             : 
    1549         892 :         END_CRIT_SECTION();
    1550             :     }
    1551             :     else
    1552             :     {
    1553             :         /*
    1554             :          * move inner tuple to another page, and update parent
    1555             :          */
    1556             :         SpGistDeadTuple dt;
    1557             :         SPPageDesc  saveCurrent;
    1558             : 
    1559             :         /*
    1560             :          * It should not be possible to get here for the root page, since we
    1561             :          * allow only one inner tuple on the root page, and spgFormInnerTuple
    1562             :          * always checks that inner tuples don't exceed the size of a page.
    1563             :          */
    1564           4 :         if (SpGistBlockIsRoot(current->blkno))
    1565           0 :             elog(ERROR, "cannot enlarge root tuple any more");
    1566             :         Assert(parent->buffer != InvalidBuffer);
    1567             : 
    1568           4 :         saveCurrent = *current;
    1569             : 
    1570           4 :         xlrec.offnumParent = parent->offnum;
    1571           4 :         xlrec.nodeI = parent->node;
    1572             : 
    1573             :         /*
    1574             :          * obtain new buffer with the same parity as current, since it will be
    1575             :          * a child of same parent tuple
    1576             :          */
    1577           8 :         current->buffer = SpGistGetBuffer(index,
    1578           4 :                                           GBUF_INNER_PARITY(current->blkno),
    1579           4 :                                           newInnerTuple->size + sizeof(ItemIdData),
    1580             :                                           &xlrec.newPage);
    1581           4 :         current->blkno = BufferGetBlockNumber(current->buffer);
    1582           4 :         current->page = BufferGetPage(current->buffer);
    1583             : 
    1584             :         /*
    1585             :          * Let's just make real sure new current isn't same as old.  Right now
    1586             :          * that's impossible, but if SpGistGetBuffer ever got smart enough to
    1587             :          * delete placeholder tuples before checking space, maybe it wouldn't
    1588             :          * be impossible.  The case would appear to work except that WAL
    1589             :          * replay would be subtly wrong, so I think a mere assert isn't enough
    1590             :          * here.
    1591             :          */
    1592           4 :         if (current->blkno == saveCurrent.blkno)
    1593           0 :             elog(ERROR, "SPGiST new buffer shouldn't be same as old buffer");
    1594             : 
    1595             :         /*
    1596             :          * New current and parent buffer will both be modified; but note that
    1597             :          * parent buffer could be same as either new or old current.
    1598             :          */
    1599           4 :         if (parent->buffer == saveCurrent.buffer)
    1600           0 :             xlrec.parentBlk = 0;
    1601           4 :         else if (parent->buffer == current->buffer)
    1602           0 :             xlrec.parentBlk = 1;
    1603             :         else
    1604           4 :             xlrec.parentBlk = 2;
    1605             : 
    1606           4 :         START_CRIT_SECTION();
    1607             : 
    1608             :         /* insert new ... */
    1609           4 :         xlrec.offnumNew = current->offnum =
    1610           4 :             SpGistPageAddNewItem(state, current->page,
    1611           4 :                                  (Item) newInnerTuple, newInnerTuple->size,
    1612             :                                  NULL, false);
    1613             : 
    1614           4 :         MarkBufferDirty(current->buffer);
    1615             : 
    1616             :         /* update parent's downlink and mark parent page dirty */
    1617           4 :         saveNodeLink(index, parent, current->blkno, current->offnum);
    1618             : 
    1619             :         /*
    1620             :          * Replace old tuple with a placeholder or redirection tuple.  Unless
    1621             :          * doing an index build, we have to insert a redirection tuple for
    1622             :          * possible concurrent scans.  We can't just delete it in any case,
    1623             :          * because that could change the offsets of other tuples on the page,
    1624             :          * breaking downlinks from their parents.
    1625             :          */
    1626           4 :         if (state->isBuild)
    1627           0 :             dt = spgFormDeadTuple(state, SPGIST_PLACEHOLDER,
    1628             :                                   InvalidBlockNumber, InvalidOffsetNumber);
    1629             :         else
    1630           4 :             dt = spgFormDeadTuple(state, SPGIST_REDIRECT,
    1631           4 :                                   current->blkno, current->offnum);
    1632             : 
    1633           4 :         PageIndexTupleDelete(saveCurrent.page, saveCurrent.offnum);
    1634           8 :         if (PageAddItem(saveCurrent.page, (Item) dt, dt->size,
    1635             :                         saveCurrent.offnum,
    1636           4 :                         false, false) != saveCurrent.offnum)
    1637           0 :             elog(ERROR, "failed to add item of size %u to SPGiST index page",
    1638             :                  dt->size);
    1639             : 
    1640           4 :         if (state->isBuild)
    1641           0 :             SpGistPageGetOpaque(saveCurrent.page)->nPlaceholder++;
    1642             :         else
    1643           4 :             SpGistPageGetOpaque(saveCurrent.page)->nRedirection++;
    1644             : 
    1645           4 :         MarkBufferDirty(saveCurrent.buffer);
    1646             : 
    1647           4 :         if (RelationNeedsWAL(index) && !state->isBuild)
    1648             :         {
    1649             :             XLogRecPtr  recptr;
    1650             :             int         flags;
    1651             : 
    1652           4 :             XLogBeginInsert();
    1653             : 
    1654             :             /* orig page */
    1655           4 :             XLogRegisterBuffer(0, saveCurrent.buffer, REGBUF_STANDARD);
    1656             :             /* new page */
    1657           4 :             flags = REGBUF_STANDARD;
    1658           4 :             if (xlrec.newPage)
    1659           4 :                 flags |= REGBUF_WILL_INIT;
    1660           4 :             XLogRegisterBuffer(1, current->buffer, flags);
    1661             :             /* parent page (if different from orig and new) */
    1662           4 :             if (xlrec.parentBlk == 2)
    1663           4 :                 XLogRegisterBuffer(2, parent->buffer, REGBUF_STANDARD);
    1664             : 
    1665           4 :             XLogRegisterData((char *) &xlrec, sizeof(xlrec));
    1666           4 :             XLogRegisterData((char *) newInnerTuple, newInnerTuple->size);
    1667             : 
    1668           4 :             recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_ADD_NODE);
    1669             : 
    1670             :             /* we don't bother to check if any of these are redundant */
    1671           4 :             PageSetLSN(current->page, recptr);
    1672           4 :             PageSetLSN(parent->page, recptr);
    1673           4 :             PageSetLSN(saveCurrent.page, recptr);
    1674             :         }
    1675             : 
    1676           4 :         END_CRIT_SECTION();
    1677             : 
    1678             :         /* Release saveCurrent if it's not same as current or parent */
    1679           8 :         if (saveCurrent.buffer != current->buffer &&
    1680           4 :             saveCurrent.buffer != parent->buffer)
    1681             :         {
    1682           4 :             SpGistSetLastUsedPage(index, saveCurrent.buffer);
    1683           4 :             UnlockReleaseBuffer(saveCurrent.buffer);
    1684             :         }
    1685             :     }
    1686         896 : }
    1687             : 
    1688             : /*
    1689             :  * spgSplitNode action: split inner tuple at current into prefix and postfix
    1690             :  */
    1691             : static void
    1692         420 : spgSplitNodeAction(Relation index, SpGistState *state,
    1693             :                    SpGistInnerTuple innerTuple,
    1694             :                    SPPageDesc *current, spgChooseOut *out)
    1695             : {
    1696             :     SpGistInnerTuple prefixTuple,
    1697             :                 postfixTuple;
    1698             :     SpGistNodeTuple node,
    1699             :                *nodes;
    1700             :     BlockNumber postfixBlkno;
    1701             :     OffsetNumber postfixOffset;
    1702             :     int         i;
    1703             :     spgxlogSplitTuple xlrec;
    1704         420 :     Buffer      newBuffer = InvalidBuffer;
    1705             : 
    1706             :     /* Should not be applied to nulls */
    1707             :     Assert(!SpGistPageStoresNulls(current->page));
    1708             : 
    1709             :     /* Check opclass gave us sane values */
    1710         840 :     if (out->result.splitTuple.prefixNNodes <= 0 ||
    1711         420 :         out->result.splitTuple.prefixNNodes > SGITMAXNNODES)
    1712           0 :         elog(ERROR, "invalid number of prefix nodes: %d",
    1713             :              out->result.splitTuple.prefixNNodes);
    1714         840 :     if (out->result.splitTuple.childNodeN < 0 ||
    1715         420 :         out->result.splitTuple.childNodeN >=
    1716         420 :         out->result.splitTuple.prefixNNodes)
    1717           0 :         elog(ERROR, "invalid child node number: %d",
    1718             :              out->result.splitTuple.childNodeN);
    1719             : 
    1720             :     /*
    1721             :      * Construct new prefix tuple with requested number of nodes.  We'll fill
    1722             :      * in the childNodeN'th node's downlink below.
    1723             :      */
    1724         420 :     nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) *
    1725         420 :                                        out->result.splitTuple.prefixNNodes);
    1726             : 
    1727         840 :     for (i = 0; i < out->result.splitTuple.prefixNNodes; i++)
    1728             :     {
    1729         420 :         Datum       label = (Datum) 0;
    1730             :         bool        labelisnull;
    1731             : 
    1732         420 :         labelisnull = (out->result.splitTuple.prefixNodeLabels == NULL);
    1733         420 :         if (!labelisnull)
    1734         420 :             label = out->result.splitTuple.prefixNodeLabels[i];
    1735         420 :         nodes[i] = spgFormNodeTuple(state, label, labelisnull);
    1736             :     }
    1737             : 
    1738         840 :     prefixTuple = spgFormInnerTuple(state,
    1739         420 :                                     out->result.splitTuple.prefixHasPrefix,
    1740             :                                     out->result.splitTuple.prefixPrefixDatum,
    1741             :                                     out->result.splitTuple.prefixNNodes,
    1742             :                                     nodes);
    1743             : 
    1744             :     /* it must fit in the space that innerTuple now occupies */
    1745         420 :     if (prefixTuple->size > innerTuple->size)
    1746           0 :         elog(ERROR, "SPGiST inner-tuple split must not produce longer prefix");
    1747             : 
    1748             :     /*
    1749             :      * Construct new postfix tuple, containing all nodes of innerTuple with
    1750             :      * same node datums, but with the prefix specified by the picksplit
    1751             :      * function.
    1752             :      */
    1753         420 :     nodes = palloc(sizeof(SpGistNodeTuple) * innerTuple->nNodes);
    1754        1428 :     SGITITERATE(innerTuple, i, node)
    1755             :     {
    1756        1008 :         nodes[i] = node;
    1757             :     }
    1758             : 
    1759         840 :     postfixTuple = spgFormInnerTuple(state,
    1760         420 :                                      out->result.splitTuple.postfixHasPrefix,
    1761             :                                      out->result.splitTuple.postfixPrefixDatum,
    1762         420 :                                      innerTuple->nNodes, nodes);
    1763             : 
    1764             :     /* Postfix tuple is allTheSame if original tuple was */
    1765         420 :     postfixTuple->allTheSame = innerTuple->allTheSame;
    1766             : 
    1767             :     /* prep data for WAL record */
    1768         420 :     xlrec.newPage = false;
    1769             : 
    1770             :     /*
    1771             :      * If we can't fit both tuples on the current page, get a new page for the
    1772             :      * postfix tuple.  In particular, can't split to the root page.
    1773             :      *
    1774             :      * For the space calculation, note that prefixTuple replaces innerTuple
    1775             :      * but postfixTuple will be a new entry.
    1776             :      */
    1777         832 :     if (SpGistBlockIsRoot(current->blkno) ||
    1778         412 :         SpGistPageGetFreeSpace(current->page, 1) + innerTuple->size <
    1779         412 :         prefixTuple->size + postfixTuple->size + sizeof(ItemIdData))
    1780             :     {
    1781             :         /*
    1782             :          * Choose page with next triple parity, because postfix tuple is a
    1783             :          * child of prefix one
    1784             :          */
    1785         224 :         newBuffer = SpGistGetBuffer(index,
    1786         112 :                                     GBUF_INNER_PARITY(current->blkno + 1),
    1787         112 :                                     postfixTuple->size + sizeof(ItemIdData),
    1788             :                                     &xlrec.newPage);
    1789             :     }
    1790             : 
    1791         420 :     START_CRIT_SECTION();
    1792             : 
    1793             :     /*
    1794             :      * Replace old tuple by prefix tuple
    1795             :      */
    1796         420 :     PageIndexTupleDelete(current->page, current->offnum);
    1797         420 :     xlrec.offnumPrefix = PageAddItem(current->page,
    1798             :                                      (Item) prefixTuple, prefixTuple->size,
    1799             :                                      current->offnum, false, false);
    1800         420 :     if (xlrec.offnumPrefix != current->offnum)
    1801           0 :         elog(ERROR, "failed to add item of size %u to SPGiST index page",
    1802             :              prefixTuple->size);
    1803             : 
    1804             :     /*
    1805             :      * put postfix tuple into appropriate page
    1806             :      */
    1807         420 :     if (newBuffer == InvalidBuffer)
    1808             :     {
    1809         308 :         postfixBlkno = current->blkno;
    1810         308 :         xlrec.offnumPostfix = postfixOffset =
    1811         308 :             SpGistPageAddNewItem(state, current->page,
    1812         308 :                                  (Item) postfixTuple, postfixTuple->size,
    1813             :                                  NULL, false);
    1814         308 :         xlrec.postfixBlkSame = true;
    1815             :     }
    1816             :     else
    1817             :     {
    1818         112 :         postfixBlkno = BufferGetBlockNumber(newBuffer);
    1819         112 :         xlrec.offnumPostfix = postfixOffset =
    1820         112 :             SpGistPageAddNewItem(state, BufferGetPage(newBuffer),
    1821         112 :                                  (Item) postfixTuple, postfixTuple->size,
    1822             :                                  NULL, false);
    1823         112 :         MarkBufferDirty(newBuffer);
    1824         112 :         xlrec.postfixBlkSame = false;
    1825             :     }
    1826             : 
    1827             :     /*
    1828             :      * And set downlink pointer in the prefix tuple to point to postfix tuple.
    1829             :      * (We can't avoid this step by doing the above two steps in opposite
    1830             :      * order, because there might not be enough space on the page to insert
    1831             :      * the postfix tuple first.)  We have to update the local copy of the
    1832             :      * prefixTuple too, because that's what will be written to WAL.
    1833             :      */
    1834         420 :     spgUpdateNodeLink(prefixTuple, out->result.splitTuple.childNodeN,
    1835             :                       postfixBlkno, postfixOffset);
    1836         420 :     prefixTuple = (SpGistInnerTuple) PageGetItem(current->page,
    1837             :                                                  PageGetItemId(current->page, current->offnum));
    1838         420 :     spgUpdateNodeLink(prefixTuple, out->result.splitTuple.childNodeN,
    1839             :                       postfixBlkno, postfixOffset);
    1840             : 
    1841         420 :     MarkBufferDirty(current->buffer);
    1842             : 
    1843         420 :     if (RelationNeedsWAL(index) && !state->isBuild)
    1844             :     {
    1845             :         XLogRecPtr  recptr;
    1846             : 
    1847         404 :         XLogBeginInsert();
    1848         404 :         XLogRegisterData((char *) &xlrec, sizeof(xlrec));
    1849         404 :         XLogRegisterData((char *) prefixTuple, prefixTuple->size);
    1850         404 :         XLogRegisterData((char *) postfixTuple, postfixTuple->size);
    1851             : 
    1852         404 :         XLogRegisterBuffer(0, current->buffer, REGBUF_STANDARD);
    1853         404 :         if (newBuffer != InvalidBuffer)
    1854             :         {
    1855             :             int         flags;
    1856             : 
    1857         108 :             flags = REGBUF_STANDARD;
    1858         108 :             if (xlrec.newPage)
    1859           4 :                 flags |= REGBUF_WILL_INIT;
    1860         108 :             XLogRegisterBuffer(1, newBuffer, flags);
    1861             :         }
    1862             : 
    1863         404 :         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_SPLIT_TUPLE);
    1864             : 
    1865         404 :         PageSetLSN(current->page, recptr);
    1866             : 
    1867         404 :         if (newBuffer != InvalidBuffer)
    1868             :         {
    1869         108 :             PageSetLSN(BufferGetPage(newBuffer), recptr);
    1870             :         }
    1871             :     }
    1872             : 
    1873         420 :     END_CRIT_SECTION();
    1874             : 
    1875             :     /* Update local free-space cache and release buffer */
    1876         420 :     if (newBuffer != InvalidBuffer)
    1877             :     {
    1878         112 :         SpGistSetLastUsedPage(index, newBuffer);
    1879         112 :         UnlockReleaseBuffer(newBuffer);
    1880             :     }
    1881         420 : }
    1882             : 
    1883             : /*
    1884             :  * Insert one item into the index.
    1885             :  *
    1886             :  * Returns true on success, false if we failed to complete the insertion
    1887             :  * because of conflict with a concurrent insert.  In the latter case,
    1888             :  * caller should re-call spgdoinsert() with the same args.
    1889             :  */
    1890             : bool
    1891      471424 : spgdoinsert(Relation index, SpGistState *state,
    1892             :             ItemPointer heapPtr, Datum datum, bool isnull)
    1893             : {
    1894      471424 :     int         level = 0;
    1895             :     Datum       leafDatum;
    1896             :     int         leafSize;
    1897             :     SPPageDesc  current,
    1898             :                 parent;
    1899      471424 :     FmgrInfo   *procinfo = NULL;
    1900             : 
    1901             :     /*
    1902             :      * Look up FmgrInfo of the user-defined choose function once, to save
    1903             :      * cycles in the loop below.
    1904             :      */
    1905      471424 :     if (!isnull)
    1906      471376 :         procinfo = index_getprocinfo(index, 1, SPGIST_CHOOSE_PROC);
    1907             : 
    1908             :     /*
    1909             :      * Prepare the leaf datum to insert.
    1910             :      *
    1911             :      * If an optional "compress" method is provided, then call it to form the
    1912             :      * leaf datum from the input datum.  Otherwise store the input datum as
    1913             :      * is.  Since we don't use index_form_tuple in this AM, we have to make
    1914             :      * sure value to be inserted is not toasted; FormIndexDatum doesn't
    1915             :      * guarantee that.  But we assume the "compress" method to return an
    1916             :      * untoasted value.
    1917             :      */
    1918      471424 :     if (!isnull)
    1919             :     {
    1920      471376 :         if (OidIsValid(index_getprocid(index, 1, SPGIST_COMPRESS_PROC)))
    1921             :         {
    1922       44000 :             FmgrInfo   *compressProcinfo = NULL;
    1923             : 
    1924       44000 :             compressProcinfo = index_getprocinfo(index, 1, SPGIST_COMPRESS_PROC);
    1925       44000 :             leafDatum = FunctionCall1Coll(compressProcinfo,
    1926       44000 :                                           index->rd_indcollation[0],
    1927             :                                           datum);
    1928             :         }
    1929             :         else
    1930             :         {
    1931             :             Assert(state->attLeafType.type == state->attType.type);
    1932             : 
    1933      427376 :             if (state->attType.attlen == -1)
    1934      119072 :                 leafDatum = PointerGetDatum(PG_DETOAST_DATUM(datum));
    1935             :             else
    1936      308304 :                 leafDatum = datum;
    1937             :         }
    1938             :     }
    1939             :     else
    1940          48 :         leafDatum = (Datum) 0;
    1941             : 
    1942             :     /*
    1943             :      * Compute space needed for a leaf tuple containing the given datum.
    1944             :      *
    1945             :      * If it isn't gonna fit, and the opclass can't reduce the datum size by
    1946             :      * suffixing, bail out now rather than getting into an endless loop.
    1947             :      */
    1948      471424 :     if (!isnull)
    1949      471376 :         leafSize = SGLTHDRSZ + sizeof(ItemIdData) +
    1950      471376 :             SpGistGetTypeSize(&state->attLeafType, leafDatum);
    1951             :     else
    1952          48 :         leafSize = SGDTSIZE + sizeof(ItemIdData);
    1953             : 
    1954      471424 :     if (leafSize > SPGIST_PAGE_CAPACITY && !state->config.longValuesOK)
    1955           0 :         ereport(ERROR,
    1956             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
    1957             :                  errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
    1958             :                         leafSize - sizeof(ItemIdData),
    1959             :                         SPGIST_PAGE_CAPACITY - sizeof(ItemIdData),
    1960             :                         RelationGetRelationName(index)),
    1961             :                  errhint("Values larger than a buffer page cannot be indexed.")));
    1962             : 
    1963             :     /* Initialize "current" to the appropriate root page */
    1964      471424 :     current.blkno = isnull ? SPGIST_NULL_BLKNO : SPGIST_ROOT_BLKNO;
    1965      471424 :     current.buffer = InvalidBuffer;
    1966      471424 :     current.page = NULL;
    1967      471424 :     current.offnum = FirstOffsetNumber;
    1968      471424 :     current.node = -1;
    1969             : 
    1970             :     /* "parent" is invalid for the moment */
    1971      471424 :     parent.blkno = InvalidBlockNumber;
    1972      471424 :     parent.buffer = InvalidBuffer;
    1973      471424 :     parent.page = NULL;
    1974      471424 :     parent.offnum = InvalidOffsetNumber;
    1975      471424 :     parent.node = -1;
    1976             : 
    1977             :     for (;;)
    1978    12498708 :     {
    1979    12970132 :         bool        isNew = false;
    1980             : 
    1981             :         /*
    1982             :          * Bail out if query cancel is pending.  We must have this somewhere
    1983             :          * in the loop since a broken opclass could produce an infinite
    1984             :          * picksplit loop.
    1985             :          */
    1986    12970132 :         CHECK_FOR_INTERRUPTS();
    1987             : 
    1988    12970132 :         if (current.blkno == InvalidBlockNumber)
    1989             :         {
    1990             :             /*
    1991             :              * Create a leaf page.  If leafSize is too large to fit on a page,
    1992             :              * we won't actually use the page yet, but it simplifies the API
    1993             :              * for doPickSplit to always have a leaf page at hand; so just
    1994             :              * quietly limit our request to a page size.
    1995             :              */
    1996        1228 :             current.buffer =
    1997        1228 :                 SpGistGetBuffer(index,
    1998             :                                 GBUF_LEAF | (isnull ? GBUF_NULLS : 0),
    1999        1228 :                                 Min(leafSize, SPGIST_PAGE_CAPACITY),
    2000             :                                 &isNew);
    2001        1228 :             current.blkno = BufferGetBlockNumber(current.buffer);
    2002             :         }
    2003    12968904 :         else if (parent.buffer == InvalidBuffer)
    2004             :         {
    2005             :             /* we hold no parent-page lock, so no deadlock is possible */
    2006      471424 :             current.buffer = ReadBuffer(index, current.blkno);
    2007      471424 :             LockBuffer(current.buffer, BUFFER_LOCK_EXCLUSIVE);
    2008             :         }
    2009    12497480 :         else if (current.blkno != parent.blkno)
    2010             :         {
    2011             :             /* descend to a new child page */
    2012      928136 :             current.buffer = ReadBuffer(index, current.blkno);
    2013             : 
    2014             :             /*
    2015             :              * Attempt to acquire lock on child page.  We must beware of
    2016             :              * deadlock against another insertion process descending from that
    2017             :              * page to our parent page (see README).  If we fail to get lock,
    2018             :              * abandon the insertion and tell our caller to start over.
    2019             :              *
    2020             :              * XXX this could be improved, because failing to get lock on a
    2021             :              * buffer is not proof of a deadlock situation; the lock might be
    2022             :              * held by a reader, or even just background writer/checkpointer
    2023             :              * process.  Perhaps it'd be worth retrying after sleeping a bit?
    2024             :              */
    2025      928136 :             if (!ConditionalLockBuffer(current.buffer))
    2026             :             {
    2027           0 :                 ReleaseBuffer(current.buffer);
    2028           0 :                 UnlockReleaseBuffer(parent.buffer);
    2029           0 :                 return false;
    2030             :             }
    2031             :         }
    2032             :         else
    2033             :         {
    2034             :             /* inner tuple can be stored on the same page as parent one */
    2035    11569344 :             current.buffer = parent.buffer;
    2036             :         }
    2037    12970132 :         current.page = BufferGetPage(current.buffer);
    2038             : 
    2039             :         /* should not arrive at a page of the wrong type */
    2040    25940216 :         if (isnull ? !SpGistPageStoresNulls(current.page) :
    2041    12970084 :             SpGistPageStoresNulls(current.page))
    2042           0 :             elog(ERROR, "SPGiST index page %u has wrong nulls flag",
    2043             :                  current.blkno);
    2044             : 
    2045    12970132 :         if (SpGistPageIsLeaf(current.page))
    2046             :         {
    2047             :             SpGistLeafTuple leafTuple;
    2048             :             int         nToSplit,
    2049             :                         sizeToSplit;
    2050             : 
    2051      471556 :             leafTuple = spgFormLeafTuple(state, heapPtr, leafDatum, isnull);
    2052      943112 :             if (leafTuple->size + sizeof(ItemIdData) <=
    2053      471556 :                 SpGistPageGetFreeSpace(current.page, 1))
    2054             :             {
    2055             :                 /* it fits on page, so insert it and we're done */
    2056      467088 :                 addLeafTuple(index, state, leafTuple,
    2057             :                              &current, &parent, isnull, isNew);
    2058      467088 :                 break;
    2059             :             }
    2060        4468 :             else if ((sizeToSplit =
    2061             :                       checkSplitConditions(index, state, &current,
    2062        2034 :                                            &nToSplit)) < SPGIST_PAGE_CAPACITY / 2 &&
    2063        3118 :                      nToSplit < 64 &&
    2064        1084 :                      leafTuple->size + sizeof(ItemIdData) + sizeToSplit <= SPGIST_PAGE_CAPACITY)
    2065             :             {
    2066             :                 /*
    2067             :                  * the amount of data is pretty small, so just move the whole
    2068             :                  * chain to another leaf page rather than splitting it.
    2069             :                  */
    2070             :                 Assert(!isNew);
    2071        1084 :                 moveLeafs(index, state, &current, &parent, leafTuple, isnull);
    2072        1084 :                 break;          /* we're done */
    2073             :             }
    2074             :             else
    2075             :             {
    2076             :                 /* picksplit */
    2077        3384 :                 if (doPickSplit(index, state, &current, &parent,
    2078             :                                 leafTuple, level, isnull, isNew))
    2079        3252 :                     break;      /* doPickSplit installed new tuples */
    2080             : 
    2081             :                 /* leaf tuple will not be inserted yet */
    2082         132 :                 pfree(leafTuple);
    2083             : 
    2084             :                 /*
    2085             :                  * current now describes new inner tuple, go insert into it
    2086             :                  */
    2087             :                 Assert(!SpGistPageIsLeaf(current.page));
    2088         132 :                 goto process_inner_tuple;
    2089             :             }
    2090             :         }
    2091             :         else                    /* non-leaf page */
    2092             :         {
    2093             :             /*
    2094             :              * Apply the opclass choose function to figure out how to insert
    2095             :              * the given datum into the current inner tuple.
    2096             :              */
    2097             :             SpGistInnerTuple innerTuple;
    2098             :             spgChooseIn in;
    2099             :             spgChooseOut out;
    2100             : 
    2101             :             /*
    2102             :              * spgAddNode and spgSplitTuple cases will loop back to here to
    2103             :              * complete the insertion operation.  Just in case the choose
    2104             :              * function is broken and produces add or split requests
    2105             :              * repeatedly, check for query cancel.
    2106             :              */
    2107             :     process_inner_tuple:
    2108    12500024 :             CHECK_FOR_INTERRUPTS();
    2109             : 
    2110    12500024 :             innerTuple = (SpGistInnerTuple) PageGetItem(current.page,
    2111             :                                                         PageGetItemId(current.page, current.offnum));
    2112             : 
    2113    12500024 :             in.datum = datum;
    2114    12500024 :             in.leafDatum = leafDatum;
    2115    12500024 :             in.level = level;
    2116    12500024 :             in.allTheSame = innerTuple->allTheSame;
    2117    12500024 :             in.hasPrefix = (innerTuple->prefixSize > 0);
    2118    12500024 :             in.prefixDatum = SGITDATUM(innerTuple, state);
    2119    12500024 :             in.nNodes = innerTuple->nNodes;
    2120    12500024 :             in.nodeLabels = spgExtractNodeLabels(state, innerTuple);
    2121             : 
    2122    12500024 :             memset(&out, 0, sizeof(out));
    2123             : 
    2124    12500024 :             if (!isnull)
    2125             :             {
    2126             :                 /* use user-defined choose method */
    2127    25000048 :                 FunctionCall2Coll(procinfo,
    2128    12500024 :                                   index->rd_indcollation[0],
    2129             :                                   PointerGetDatum(&in),
    2130             :                                   PointerGetDatum(&out));
    2131             :             }
    2132             :             else
    2133             :             {
    2134             :                 /* force "match" action (to insert to random subnode) */
    2135           0 :                 out.resultType = spgMatchNode;
    2136             :             }
    2137             : 
    2138    12500024 :             if (innerTuple->allTheSame)
    2139             :             {
    2140             :                 /*
    2141             :                  * It's not allowed to do an AddNode at an allTheSame tuple.
    2142             :                  * Opclass must say "match", in which case we choose a random
    2143             :                  * one of the nodes to descend into, or "split".
    2144             :                  */
    2145       25692 :                 if (out.resultType == spgAddNode)
    2146           0 :                     elog(ERROR, "cannot add a node to an allTheSame inner tuple");
    2147       25692 :                 else if (out.resultType == spgMatchNode)
    2148       25684 :                     out.result.matchNode.nodeN = random() % innerTuple->nNodes;
    2149             :             }
    2150             : 
    2151    12500024 :             switch (out.resultType)
    2152             :             {
    2153             :                 case spgMatchNode:
    2154             :                     /* Descend to N'th child node */
    2155    12498708 :                     spgMatchNodeAction(index, state, innerTuple,
    2156             :                                        &current, &parent,
    2157             :                                        out.result.matchNode.nodeN);
    2158             :                     /* Adjust level as per opclass request */
    2159    12498708 :                     level += out.result.matchNode.levelAdd;
    2160             :                     /* Replace leafDatum and recompute leafSize */
    2161    12498708 :                     if (!isnull)
    2162             :                     {
    2163    12498708 :                         leafDatum = out.result.matchNode.restDatum;
    2164    12498708 :                         leafSize = SGLTHDRSZ + sizeof(ItemIdData) +
    2165    12498708 :                             SpGistGetTypeSize(&state->attLeafType, leafDatum);
    2166             :                     }
    2167             : 
    2168             :                     /*
    2169             :                      * Loop around and attempt to insert the new leafDatum at
    2170             :                      * "current" (which might reference an existing child
    2171             :                      * tuple, or might be invalid to force us to find a new
    2172             :                      * page for the tuple).
    2173             :                      *
    2174             :                      * Note: if the opclass sets longValuesOK, we rely on the
    2175             :                      * choose function to eventually shorten the leafDatum
    2176             :                      * enough to fit on a page.  We could add a test here to
    2177             :                      * complain if the datum doesn't get visibly shorter each
    2178             :                      * time, but that could get in the way of opclasses that
    2179             :                      * "simplify" datums in a way that doesn't necessarily
    2180             :                      * lead to physical shortening on every cycle.
    2181             :                      */
    2182    12498708 :                     break;
    2183             :                 case spgAddNode:
    2184             :                     /* AddNode is not sensible if nodes don't have labels */
    2185         896 :                     if (in.nodeLabels == NULL)
    2186           0 :                         elog(ERROR, "cannot add a node to an inner tuple without node labels");
    2187             :                     /* Add node to inner tuple, per request */
    2188         896 :                     spgAddNodeAction(index, state, innerTuple,
    2189             :                                      &current, &parent,
    2190             :                                      out.result.addNode.nodeN,
    2191             :                                      out.result.addNode.nodeLabel);
    2192             : 
    2193             :                     /*
    2194             :                      * Retry insertion into the enlarged node.  We assume that
    2195             :                      * we'll get a MatchNode result this time.
    2196             :                      */
    2197         896 :                     goto process_inner_tuple;
    2198             :                     break;
    2199             :                 case spgSplitTuple:
    2200             :                     /* Split inner tuple, per request */
    2201         420 :                     spgSplitNodeAction(index, state, innerTuple,
    2202             :                                        &current, &out);
    2203             : 
    2204             :                     /* Retry insertion into the split node */
    2205         420 :                     goto process_inner_tuple;
    2206             :                     break;
    2207             :                 default:
    2208           0 :                     elog(ERROR, "unrecognized SPGiST choose result: %d",
    2209             :                          (int) out.resultType);
    2210             :                     break;
    2211             :             }
    2212             :         }
    2213             :     }                           /* end loop */
    2214             : 
    2215             :     /*
    2216             :      * Release any buffers we're still holding.  Beware of possibility that
    2217             :      * current and parent reference same buffer.
    2218             :      */
    2219      471424 :     if (current.buffer != InvalidBuffer)
    2220             :     {
    2221      471424 :         SpGistSetLastUsedPage(index, current.buffer);
    2222      471424 :         UnlockReleaseBuffer(current.buffer);
    2223             :     }
    2224      934752 :     if (parent.buffer != InvalidBuffer &&
    2225      463328 :         parent.buffer != current.buffer)
    2226             :     {
    2227      460532 :         SpGistSetLastUsedPage(index, parent.buffer);
    2228      460532 :         UnlockReleaseBuffer(parent.buffer);
    2229             :     }
    2230             : 
    2231      471424 :     return true;
    2232             : }

Generated by: LCOV version 1.13