#include "postgres.h"
#include "access/genam.h"
#include "access/spgist_private.h"
#include "access/spgxlog.h"
#include "access/xloginsert.h"
#include "common/int.h"
#include "common/pg_prng.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "utils/rel.h"

Include dependency graph for spgdoinsert.c:

Data Structures
struct	SPPageDesc

Typedefs
typedef struct SPPageDesc	SPPageDesc

Functions
void	spgUpdateNodeLink (SpGistInnerTuple tup, int nodeN, BlockNumber blkno, OffsetNumber offset)

static SpGistInnerTuple	addNode (SpGistState *state, SpGistInnerTuple tuple, Datum label, int offset)

static int	cmpOffsetNumbers (const void a, const void b)

void	spgPageIndexMultiDelete (SpGistState state, Page page, OffsetNumber itemnos, int nitems, int firststate, int reststate, BlockNumber blkno, OffsetNumber offnum)

static void	saveNodeLink (Relation index, SPPageDesc *parent, BlockNumber blkno, OffsetNumber offnum)

static void	addLeafTuple (Relation index, SpGistState state, SpGistLeafTuple leafTuple, SPPageDesc current, SPPageDesc *parent, bool isNulls, bool isNew)

static int	checkSplitConditions (Relation index, SpGistState state, SPPageDesc current, int *nToSplit)

static void	moveLeafs (Relation index, SpGistState state, SPPageDesc current, SPPageDesc *parent, SpGistLeafTuple newLeafTuple, bool isNulls)

static void	setRedirectionTuple (SPPageDesc *current, OffsetNumber position, BlockNumber blkno, OffsetNumber offnum)

static bool	checkAllTheSame (spgPickSplitIn in, spgPickSplitOut out, bool tooBig, bool *includeNew)

static bool	doPickSplit (Relation index, SpGistState state, SPPageDesc current, SPPageDesc *parent, SpGistLeafTuple newLeafTuple, int level, bool isNulls, bool isNew)

static void	spgMatchNodeAction (Relation index, SpGistState state, SpGistInnerTuple innerTuple, SPPageDesc current, SPPageDesc *parent, int nodeN)

static void	spgAddNodeAction (Relation index, SpGistState state, SpGistInnerTuple innerTuple, SPPageDesc current, SPPageDesc *parent, int nodeN, Datum nodeLabel)

static void	spgSplitNodeAction (Relation index, SpGistState state, SpGistInnerTuple innerTuple, SPPageDesc current, spgChooseOut *out)

bool	spgdoinsert (Relation index, SpGistState state, ItemPointer heapPtr, Datum datums, bool *isnulls)

Typedef Documentation

◆ SPPageDesc

typedef struct SPPageDesc SPPageDesc

Function Documentation

◆ addLeafTuple()

static void addLeafTuple	(	Relation	index,
		SpGistState *	state,
		SpGistLeafTuple	leafTuple,
		SPPageDesc *	current,
		SPPageDesc *	parent,
		bool	isNulls,
		bool	isNew
	)

static

Definition at line 203 of file spgdoinsert.c.

 {
     spgxlogAddLeaf xlrec;
  
     xlrec.newPage = isNew;
     xlrec.storesNulls = isNulls;
  
     /* these will be filled below as needed */
     xlrec.offnumLeaf = InvalidOffsetNumber;
     xlrec.offnumHeadLeaf = InvalidOffsetNumber;
     xlrec.offnumParent = InvalidOffsetNumber;
     xlrec.nodeI = 0;
  
     START_CRIT_SECTION();
  
     if (current->offnum == InvalidOffsetNumber ||
         SpGistBlockIsRoot(current->blkno))
     {
         /* Tuple is not part of a chain */
         SGLT_SET_NEXTOFFSET(leafTuple, InvalidOffsetNumber);
         current->offnum = SpGistPageAddNewItem(state, current->page,
                                                (Item) leafTuple, leafTuple->size,
                                                NULL, false);
  
         xlrec.offnumLeaf = current->offnum;
  
         /* Must update parent's downlink if any */
         if (parent->buffer != InvalidBuffer)
         {
             xlrec.offnumParent = parent->offnum;
             xlrec.nodeI = parent->node;
  
             saveNodeLink(index, parent, current->blkno, current->offnum);
         }
     }
     else
     {
         /*
          * Tuple must be inserted into existing chain.  We mustn't change the
          * chain's head address, but we don't need to chase the entire chain
          * to put the tuple at the end; we can insert it second.
          *
          * Also, it's possible that the "chain" consists only of a DEAD tuple,
          * in which case we should replace the DEAD tuple in-place.
          */
         SpGistLeafTuple head;
         OffsetNumber offnum;
  
         head = (SpGistLeafTuple) PageGetItem(current->page,
                                              PageGetItemId(current->page, current->offnum));
         if (head->tupstate == SPGIST_LIVE)
         {
             SGLT_SET_NEXTOFFSET(leafTuple, SGLT_GET_NEXTOFFSET(head));
             offnum = SpGistPageAddNewItem(state, current->page,
                                           (Item) leafTuple, leafTuple->size,
                                           NULL, false);
  
             /*
              * re-get head of list because it could have been moved on page,
              * and set new second element
              */
             head = (SpGistLeafTuple) PageGetItem(current->page,
                                                  PageGetItemId(current->page, current->offnum));
             SGLT_SET_NEXTOFFSET(head, offnum);
  
             xlrec.offnumLeaf = offnum;
             xlrec.offnumHeadLeaf = current->offnum;
         }
         else if (head->tupstate == SPGIST_DEAD)
         {
             SGLT_SET_NEXTOFFSET(leafTuple, InvalidOffsetNumber);
             PageIndexTupleDelete(current->page, current->offnum);
             if (PageAddItem(current->page,
                             (Item) leafTuple, leafTuple->size,
                             current->offnum, false, false) != current->offnum)
                 elog(ERROR, "failed to add item of size %u to SPGiST index page",
                      leafTuple->size);
  
             /* WAL replay distinguishes this case by equal offnums */
             xlrec.offnumLeaf = current->offnum;
             xlrec.offnumHeadLeaf = current->offnum;
         }
         else
             elog(ERROR, "unexpected SPGiST tuple state: %d", head->tupstate);
     }
  
     MarkBufferDirty(current->buffer);
  
     if (RelationNeedsWAL(index) && !state->isBuild)
     {
         XLogRecPtr  recptr;
         int         flags;
  
         XLogBeginInsert();
         XLogRegisterData((char *) &xlrec, sizeof(xlrec));
         XLogRegisterData((char *) leafTuple, leafTuple->size);
  
         flags = REGBUF_STANDARD;
         if (xlrec.newPage)
             flags |= REGBUF_WILL_INIT;
         XLogRegisterBuffer(0, current->buffer, flags);
         if (xlrec.offnumParent != InvalidOffsetNumber)
             XLogRegisterBuffer(1, parent->buffer, REGBUF_STANDARD);
  
         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_ADD_LEAF);
  
         PageSetLSN(current->page, recptr);
  
         /* update parent only if we actually changed it */
         if (xlrec.offnumParent != InvalidOffsetNumber)
         {
             PageSetLSN(parent->page, recptr);
         }
     }
  
     END_CRIT_SECTION();
 }

References SPPageDesc::blkno, SPPageDesc::buffer, elog, END_CRIT_SECTION, ERROR, InvalidBuffer, InvalidOffsetNumber, MarkBufferDirty(), spgxlogAddLeaf::newPage, SPPageDesc::node, spgxlogAddLeaf::nodeI, SPPageDesc::offnum, spgxlogAddLeaf::offnumHeadLeaf, spgxlogAddLeaf::offnumLeaf, spgxlogAddLeaf::offnumParent, SPPageDesc::page, PageAddItem, PageGetItem(), PageGetItemId(), PageIndexTupleDelete(), PageSetLSN(), REGBUF_STANDARD, REGBUF_WILL_INIT, RelationNeedsWAL, saveNodeLink(), SGLT_GET_NEXTOFFSET, SGLT_SET_NEXTOFFSET, SpGistLeafTupleData::size, SPGIST_DEAD, SPGIST_LIVE, SpGistBlockIsRoot, SpGistPageAddNewItem(), START_CRIT_SECTION, spgxlogAddLeaf::storesNulls, SpGistLeafTupleData::tupstate, XLOG_SPGIST_ADD_LEAF, XLogBeginInsert(), XLogInsert(), XLogRegisterBuffer(), and XLogRegisterData().

Referenced by spgdoinsert().

◆ addNode()

static SpGistInnerTuple addNode	(	SpGistState *	state,
		SpGistInnerTuple	tuple,
		Datum	label,
		int	offset
	)

static

Definition at line 80 of file spgdoinsert.c.

 {
     SpGistNodeTuple node,
                *nodes;
     int         i;
  
     /* if offset is negative, insert at end */
     if (offset < 0)
         offset = tuple->nNodes;
     else if (offset > tuple->nNodes)
         elog(ERROR, "invalid offset for adding node to SPGiST inner tuple");
  
     nodes = palloc(sizeof(SpGistNodeTuple) * (tuple->nNodes + 1));
     SGITITERATE(tuple, i, node)
     {
         if (i < offset)
             nodes[i] = node;
         else
             nodes[i + 1] = node;
     }
  
     nodes[offset] = spgFormNodeTuple(state, label, false);
  
     return spgFormInnerTuple(state,
                              (tuple->prefixSize > 0),
                              SGITDATUM(tuple, state),
                              tuple->nNodes + 1,
                              nodes);
 }

References elog, ERROR, i, label, SpGistInnerTupleData::nNodes, palloc(), SpGistInnerTupleData::prefixSize, SGITDATUM, SGITITERATE, spgFormInnerTuple(), and spgFormNodeTuple().

Referenced by spgAddNodeAction().

◆ checkAllTheSame()

static bool checkAllTheSame	(	spgPickSplitIn *	in,
		spgPickSplitOut *	out,
		bool	tooBig,
		bool *	includeNew
	)

static

Definition at line 599 of file spgdoinsert.c.

 {
     int         theNode;
     int         limit;
     int         i;
  
     /* For the moment, assume we can include the new leaf tuple */
     *includeNew = true;
  
     /* If there's only the new leaf tuple, don't select allTheSame mode */
     if (in->nTuples <= 1)
         return false;
  
     /* If tuple set doesn't fit on one page, ignore the new tuple in test */
     limit = tooBig ? in->nTuples - 1 : in->nTuples;
  
     /* Check to see if more than one node is populated */
     theNode = out->mapTuplesToNodes[0];
     for (i = 1; i < limit; i++)
     {
         if (out->mapTuplesToNodes[i] != theNode)
             return false;
     }
  
     /* Nope, so override the picksplit function's decisions */
  
     /* If the new tuple is in its own node, it can't be included in split */
     if (tooBig && out->mapTuplesToNodes[in->nTuples - 1] != theNode)
         *includeNew = false;
  
     out->nNodes = 8;            /* arbitrary number of child nodes */
  
     /* Random assignment of tuples to nodes (note we include new tuple) */
     for (i = 0; i < in->nTuples; i++)
         out->mapTuplesToNodes[i] = i % out->nNodes;
  
     /* The opclass may not use node labels, but if it does, duplicate 'em */
     if (out->nodeLabels)
     {
         Datum       theLabel = out->nodeLabels[theNode];
  
         out->nodeLabels = (Datum *) palloc(sizeof(Datum) * out->nNodes);
         for (i = 0; i < out->nNodes; i++)
             out->nodeLabels[i] = theLabel;
     }
  
     /* We don't touch the prefix or the leaf tuple datum assignments */
  
     return true;
 }

References i, if(), spgPickSplitOut::mapTuplesToNodes, spgPickSplitOut::nNodes, spgPickSplitOut::nodeLabels, spgPickSplitIn::nTuples, and palloc().

Referenced by doPickSplit().

◆ checkSplitConditions()

static int checkSplitConditions	(	Relation	index,
		SpGistState *	state,
		SPPageDesc *	current,
		int *	nToSplit
	)

static

Definition at line 333 of file spgdoinsert.c.

 {
     int         i,
                 n = 0,
                 totalSize = 0;
  
     if (SpGistBlockIsRoot(current->blkno))
     {
         /* return impossible values to force split */
         *nToSplit = BLCKSZ;
         return BLCKSZ;
     }
  
     i = current->offnum;
     while (i != InvalidOffsetNumber)
     {
         SpGistLeafTuple it;
  
         Assert(i >= FirstOffsetNumber &&
                i <= PageGetMaxOffsetNumber(current->page));
         it = (SpGistLeafTuple) PageGetItem(current->page,
                                            PageGetItemId(current->page, i));
         if (it->tupstate == SPGIST_LIVE)
         {
             n++;
             totalSize += it->size + sizeof(ItemIdData);
         }
         else if (it->tupstate == SPGIST_DEAD)
         {
             /* We could see a DEAD tuple as first/only chain item */
             Assert(i == current->offnum);
             Assert(SGLT_GET_NEXTOFFSET(it) == InvalidOffsetNumber);
             /* Don't count it in result, because it won't go to other page */
         }
         else
             elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
  
         i = SGLT_GET_NEXTOFFSET(it);
     }
  
     *nToSplit = n;
  
     return totalSize;
 }

References Assert, SPPageDesc::blkno, elog, ERROR, FirstOffsetNumber, i, InvalidOffsetNumber, SPPageDesc::offnum, SPPageDesc::page, PageGetItem(), PageGetItemId(), PageGetMaxOffsetNumber(), SGLT_GET_NEXTOFFSET, SpGistLeafTupleData::size, SPGIST_DEAD, SPGIST_LIVE, SpGistBlockIsRoot, and SpGistLeafTupleData::tupstate.

Referenced by spgdoinsert().

◆ cmpOffsetNumbers()

static int cmpOffsetNumbers	(	const void *	a,
		const void *	b
	)

static

Definition at line 112 of file spgdoinsert.c.

 {
     return pg_cmp_u16(*(const OffsetNumber *) a, *(const OffsetNumber *) b);
 }

References a, b, and pg_cmp_u16().

Referenced by spgPageIndexMultiDelete().

◆ doPickSplit()

static bool doPickSplit	(	Relation	index,
		SpGistState *	state,
		SPPageDesc *	current,
		SPPageDesc *	parent,
		SpGistLeafTuple	newLeafTuple,
		int	level,
		bool	isNulls,
		bool	isNew
	)

static

Definition at line 677 of file spgdoinsert.c.

 {
     bool        insertedNew = false;
     spgPickSplitIn in;
     spgPickSplitOut out;
     FmgrInfo   *procinfo;
     bool        includeNew;
     int         i,
                 max,
                 n;
     SpGistInnerTuple innerTuple;
     SpGistNodeTuple node,
                *nodes;
     Buffer      newInnerBuffer,
                 newLeafBuffer;
     uint8      *leafPageSelect;
     int        *leafSizes;
     OffsetNumber *toDelete;
     OffsetNumber *toInsert;
     OffsetNumber redirectTuplePos = InvalidOffsetNumber;
     OffsetNumber startOffsets[2];
     SpGistLeafTuple *oldLeafs;
     SpGistLeafTuple *newLeafs;
     Datum       leafDatums[INDEX_MAX_KEYS];
     bool        leafIsnulls[INDEX_MAX_KEYS];
     int         spaceToDelete;
     int         currentFreeSpace;
     int         totalLeafSizes;
     bool        allTheSame;
     spgxlogPickSplit xlrec;
     char       *leafdata,
                *leafptr;
     SPPageDesc  saveCurrent;
     int         nToDelete,
                 nToInsert,
                 maxToInclude;
  
     in.level = level;
  
     /*
      * Allocate per-leaf-tuple work arrays with max possible size
      */
     max = PageGetMaxOffsetNumber(current->page);
     n = max + 1;
     in.datums = (Datum *) palloc(sizeof(Datum) * n);
     toDelete = (OffsetNumber *) palloc(sizeof(OffsetNumber) * n);
     toInsert = (OffsetNumber *) palloc(sizeof(OffsetNumber) * n);
     oldLeafs = (SpGistLeafTuple *) palloc(sizeof(SpGistLeafTuple) * n);
     newLeafs = (SpGistLeafTuple *) palloc(sizeof(SpGistLeafTuple) * n);
     leafPageSelect = (uint8 *) palloc(sizeof(uint8) * n);
  
     STORE_STATE(state, xlrec.stateSrc);
  
     /*
      * Form list of leaf tuples which will be distributed as split result;
      * also, count up the amount of space that will be freed from current.
      * (Note that in the non-root case, we won't actually delete the old
      * tuples, only replace them with redirects or placeholders.)
      */
     nToInsert = 0;
     nToDelete = 0;
     spaceToDelete = 0;
     if (SpGistBlockIsRoot(current->blkno))
     {
         /*
          * We are splitting the root (which up to now is also a leaf page).
          * Its tuples are not linked, so scan sequentially to get them all. We
          * ignore the original value of current->offnum.
          */
         for (i = FirstOffsetNumber; i <= max; i++)
         {
             SpGistLeafTuple it;
  
             it = (SpGistLeafTuple) PageGetItem(current->page,
                                                PageGetItemId(current->page, i));
             if (it->tupstate == SPGIST_LIVE)
             {
                 in.datums[nToInsert] =
                     isNulls ? (Datum) 0 : SGLTDATUM(it, state);
                 oldLeafs[nToInsert] = it;
                 nToInsert++;
                 toDelete[nToDelete] = i;
                 nToDelete++;
                 /* we will delete the tuple altogether, so count full space */
                 spaceToDelete += it->size + sizeof(ItemIdData);
             }
             else                /* tuples on root should be live */
                 elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
         }
     }
     else
     {
         /* Normal case, just collect the leaf tuples in the chain */
         i = current->offnum;
         while (i != InvalidOffsetNumber)
         {
             SpGistLeafTuple it;
  
             Assert(i >= FirstOffsetNumber && i <= max);
             it = (SpGistLeafTuple) PageGetItem(current->page,
                                                PageGetItemId(current->page, i));
             if (it->tupstate == SPGIST_LIVE)
             {
                 in.datums[nToInsert] =
                     isNulls ? (Datum) 0 : SGLTDATUM(it, state);
                 oldLeafs[nToInsert] = it;
                 nToInsert++;
                 toDelete[nToDelete] = i;
                 nToDelete++;
                 /* we will not delete the tuple, only replace with dead */
                 Assert(it->size >= SGDTSIZE);
                 spaceToDelete += it->size - SGDTSIZE;
             }
             else if (it->tupstate == SPGIST_DEAD)
             {
                 /* We could see a DEAD tuple as first/only chain item */
                 Assert(i == current->offnum);
                 Assert(SGLT_GET_NEXTOFFSET(it) == InvalidOffsetNumber);
                 toDelete[nToDelete] = i;
                 nToDelete++;
                 /* replacing it with redirect will save no space */
             }
             else
                 elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
  
             i = SGLT_GET_NEXTOFFSET(it);
         }
     }
     in.nTuples = nToInsert;
  
     /*
      * We may not actually insert new tuple because another picksplit may be
      * necessary due to too large value, but we will try to allocate enough
      * space to include it; and in any case it has to be included in the input
      * for the picksplit function.  So don't increment nToInsert yet.
      */
     in.datums[in.nTuples] =
         isNulls ? (Datum) 0 : SGLTDATUM(newLeafTuple, state);
     oldLeafs[in.nTuples] = newLeafTuple;
     in.nTuples++;
  
     memset(&out, 0, sizeof(out));
  
     if (!isNulls)
     {
         /*
          * Perform split using user-defined method.
          */
         procinfo = index_getprocinfo(index, 1, SPGIST_PICKSPLIT_PROC);
         FunctionCall2Coll(procinfo,
                           index->rd_indcollation[0],
                           PointerGetDatum(&in),
                           PointerGetDatum(&out));
  
         /*
          * Form new leaf tuples and count up the total space needed.
          */
         totalLeafSizes = 0;
         for (i = 0; i < in.nTuples; i++)
         {
             if (state->leafTupDesc->natts > 1)
                 spgDeformLeafTuple(oldLeafs[i],
                                    state->leafTupDesc,
                                    leafDatums,
                                    leafIsnulls,
                                    isNulls);
  
             leafDatums[spgKeyColumn] = out.leafTupleDatums[i];
             leafIsnulls[spgKeyColumn] = false;
  
             newLeafs[i] = spgFormLeafTuple(state, &oldLeafs[i]->heapPtr,
                                            leafDatums,
                                            leafIsnulls);
             totalLeafSizes += newLeafs[i]->size + sizeof(ItemIdData);
         }
     }
     else
     {
         /*
          * Perform dummy split that puts all tuples into one node.
          * checkAllTheSame will override this and force allTheSame mode.
          */
         out.hasPrefix = false;
         out.nNodes = 1;
         out.nodeLabels = NULL;
         out.mapTuplesToNodes = palloc0(sizeof(int) * in.nTuples);
  
         /*
          * Form new leaf tuples and count up the total space needed.
          */
         totalLeafSizes = 0;
         for (i = 0; i < in.nTuples; i++)
         {
             if (state->leafTupDesc->natts > 1)
                 spgDeformLeafTuple(oldLeafs[i],
                                    state->leafTupDesc,
                                    leafDatums,
                                    leafIsnulls,
                                    isNulls);
  
             /*
              * Nulls tree can contain only null key values.
              */
             leafDatums[spgKeyColumn] = (Datum) 0;
             leafIsnulls[spgKeyColumn] = true;
  
             newLeafs[i] = spgFormLeafTuple(state, &oldLeafs[i]->heapPtr,
                                            leafDatums,
                                            leafIsnulls);
             totalLeafSizes += newLeafs[i]->size + sizeof(ItemIdData);
         }
     }
  
     /*
      * Check to see if the picksplit function failed to separate the values,
      * ie, it put them all into the same child node.  If so, select allTheSame
      * mode and create a random split instead.  See comments for
      * checkAllTheSame as to why we need to know if the new leaf tuples could
      * fit on one page.
      */
     allTheSame = checkAllTheSame(&in, &out,
                                  totalLeafSizes > SPGIST_PAGE_CAPACITY,
                                  &includeNew);
  
     /*
      * If checkAllTheSame decided we must exclude the new tuple, don't
      * consider it any further.
      */
     if (includeNew)
         maxToInclude = in.nTuples;
     else
     {
         maxToInclude = in.nTuples - 1;
         totalLeafSizes -= newLeafs[in.nTuples - 1]->size + sizeof(ItemIdData);
     }
  
     /*
      * Allocate per-node work arrays.  Since checkAllTheSame could replace
      * out.nNodes with a value larger than the number of tuples on the input
      * page, we can't allocate these arrays before here.
      */
     nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) * out.nNodes);
     leafSizes = (int *) palloc0(sizeof(int) * out.nNodes);
  
     /*
      * Form nodes of inner tuple and inner tuple itself
      */
     for (i = 0; i < out.nNodes; i++)
     {
         Datum       label = (Datum) 0;
         bool        labelisnull = (out.nodeLabels == NULL);
  
         if (!labelisnull)
             label = out.nodeLabels[i];
         nodes[i] = spgFormNodeTuple(state, label, labelisnull);
     }
     innerTuple = spgFormInnerTuple(state,
                                    out.hasPrefix, out.prefixDatum,
                                    out.nNodes, nodes);
     innerTuple->allTheSame = allTheSame;
  
     /*
      * Update nodes[] array to point into the newly formed innerTuple, so that
      * we can adjust their downlinks below.
      */
     SGITITERATE(innerTuple, i, node)
     {
         nodes[i] = node;
     }
  
     /*
      * Re-scan new leaf tuples and count up the space needed under each node.
      */
     for (i = 0; i < maxToInclude; i++)
     {
         n = out.mapTuplesToNodes[i];
         if (n < 0 || n >= out.nNodes)
             elog(ERROR, "inconsistent result of SPGiST picksplit function");
         leafSizes[n] += newLeafs[i]->size + sizeof(ItemIdData);
     }
  
     /*
      * To perform the split, we must insert a new inner tuple, which can't go
      * on a leaf page; and unless we are splitting the root page, we must then
      * update the parent tuple's downlink to point to the inner tuple.  If
      * there is room, we'll put the new inner tuple on the same page as the
      * parent tuple, otherwise we need another non-leaf buffer. But if the
      * parent page is the root, we can't add the new inner tuple there,
      * because the root page must have only one inner tuple.
      */
     xlrec.initInner = false;
     if (parent->buffer != InvalidBuffer &&
         !SpGistBlockIsRoot(parent->blkno) &&
         (SpGistPageGetFreeSpace(parent->page, 1) >=
          innerTuple->size + sizeof(ItemIdData)))
     {
         /* New inner tuple will fit on parent page */
         newInnerBuffer = parent->buffer;
     }
     else if (parent->buffer != InvalidBuffer)
     {
         /* Send tuple to page with next triple parity (see README) */
         newInnerBuffer = SpGistGetBuffer(index,
                                          GBUF_INNER_PARITY(parent->blkno + 1) |
                                          (isNulls ? GBUF_NULLS : 0),
                                          innerTuple->size + sizeof(ItemIdData),
                                          &xlrec.initInner);
     }
     else
     {
         /* Root page split ... inner tuple will go to root page */
         newInnerBuffer = InvalidBuffer;
     }
  
     /*
      * The new leaf tuples converted from the existing ones should require the
      * same or less space, and therefore should all fit onto one page
      * (although that's not necessarily the current page, since we can't
      * delete the old tuples but only replace them with placeholders).
      * However, the incoming new tuple might not also fit, in which case we
      * might need another picksplit cycle to reduce it some more.
      *
      * If there's not room to put everything back onto the current page, then
      * we decide on a per-node basis which tuples go to the new page. (We do
      * it like that because leaf tuple chains can't cross pages, so we must
      * place all leaf tuples belonging to the same parent node on the same
      * page.)
      *
      * If we are splitting the root page (turning it from a leaf page into an
      * inner page), then no leaf tuples can go back to the current page; they
      * must all go somewhere else.
      */
     if (!SpGistBlockIsRoot(current->blkno))
         currentFreeSpace = PageGetExactFreeSpace(current->page) + spaceToDelete;
     else
         currentFreeSpace = 0;   /* prevent assigning any tuples to current */
  
     xlrec.initDest = false;
  
     if (totalLeafSizes <= currentFreeSpace)
     {
         /* All the leaf tuples will fit on current page */
         newLeafBuffer = InvalidBuffer;
         /* mark new leaf tuple as included in insertions, if allowed */
         if (includeNew)
         {
             nToInsert++;
             insertedNew = true;
         }
         for (i = 0; i < nToInsert; i++)
             leafPageSelect[i] = 0;  /* signifies current page */
     }
     else if (in.nTuples == 1 && totalLeafSizes > SPGIST_PAGE_CAPACITY)
     {
         /*
          * We're trying to split up a long value by repeated suffixing, but
          * it's not going to fit yet.  Don't bother allocating a second leaf
          * buffer that we won't be able to use.
          */
         newLeafBuffer = InvalidBuffer;
         Assert(includeNew);
         Assert(nToInsert == 0);
     }
     else
     {
         /* We will need another leaf page */
         uint8      *nodePageSelect;
         int         curspace;
         int         newspace;
  
         newLeafBuffer = SpGistGetBuffer(index,
                                         GBUF_LEAF | (isNulls ? GBUF_NULLS : 0),
                                         Min(totalLeafSizes,
                                             SPGIST_PAGE_CAPACITY),
                                         &xlrec.initDest);
  
         /*
          * Attempt to assign node groups to the two pages.  We might fail to
          * do so, even if totalLeafSizes is less than the available space,
          * because we can't split a group across pages.
          */
         nodePageSelect = (uint8 *) palloc(sizeof(uint8) * out.nNodes);
  
         curspace = currentFreeSpace;
         newspace = PageGetExactFreeSpace(BufferGetPage(newLeafBuffer));
         for (i = 0; i < out.nNodes; i++)
         {
             if (leafSizes[i] <= curspace)
             {
                 nodePageSelect[i] = 0;  /* signifies current page */
                 curspace -= leafSizes[i];
             }
             else
             {
                 nodePageSelect[i] = 1;  /* signifies new leaf page */
                 newspace -= leafSizes[i];
             }
         }
         if (curspace >= 0 && newspace >= 0)
         {
             /* Successful assignment, so we can include the new leaf tuple */
             if (includeNew)
             {
                 nToInsert++;
                 insertedNew = true;
             }
         }
         else if (includeNew)
         {
             /* We must exclude the new leaf tuple from the split */
             int         nodeOfNewTuple = out.mapTuplesToNodes[in.nTuples - 1];
  
             leafSizes[nodeOfNewTuple] -=
                 newLeafs[in.nTuples - 1]->size + sizeof(ItemIdData);
  
             /* Repeat the node assignment process --- should succeed now */
             curspace = currentFreeSpace;
             newspace = PageGetExactFreeSpace(BufferGetPage(newLeafBuffer));
             for (i = 0; i < out.nNodes; i++)
             {
                 if (leafSizes[i] <= curspace)
                 {
                     nodePageSelect[i] = 0;  /* signifies current page */
                     curspace -= leafSizes[i];
                 }
                 else
                 {
                     nodePageSelect[i] = 1;  /* signifies new leaf page */
                     newspace -= leafSizes[i];
                 }
             }
             if (curspace < 0 || newspace < 0)
                 elog(ERROR, "failed to divide leaf tuple groups across pages");
         }
         else
         {
             /* oops, we already excluded new tuple ... should not get here */
             elog(ERROR, "failed to divide leaf tuple groups across pages");
         }
         /* Expand the per-node assignments to be shown per leaf tuple */
         for (i = 0; i < nToInsert; i++)
         {
             n = out.mapTuplesToNodes[i];
             leafPageSelect[i] = nodePageSelect[n];
         }
     }
  
     /* Start preparing WAL record */
     xlrec.nDelete = 0;
     xlrec.initSrc = isNew;
     xlrec.storesNulls = isNulls;
     xlrec.isRootSplit = SpGistBlockIsRoot(current->blkno);
  
     leafdata = leafptr = (char *) palloc(totalLeafSizes);
  
     /* Here we begin making the changes to the target pages */
     START_CRIT_SECTION();
  
     /*
      * Delete old leaf tuples from current buffer, except when we're splitting
      * the root; in that case there's no need because we'll re-init the page
      * below.  We do this first to make room for reinserting new leaf tuples.
      */
     if (!SpGistBlockIsRoot(current->blkno))
     {
         /*
          * Init buffer instead of deleting individual tuples, but only if
          * there aren't any other live tuples and only during build; otherwise
          * we need to set a redirection tuple for concurrent scans.
          */
         if (state->isBuild &&
             nToDelete + SpGistPageGetOpaque(current->page)->nPlaceholder ==
             PageGetMaxOffsetNumber(current->page))
         {
             SpGistInitBuffer(current->buffer,
                              SPGIST_LEAF | (isNulls ? SPGIST_NULLS : 0));
             xlrec.initSrc = true;
         }
         else if (isNew)
         {
             /* don't expose the freshly init'd buffer as a backup block */
             Assert(nToDelete == 0);
         }
         else
         {
             xlrec.nDelete = nToDelete;
  
             if (!state->isBuild)
             {
                 /*
                  * Need to create redirect tuple (it will point to new inner
                  * tuple) but right now the new tuple's location is not known
                  * yet.  So, set the redirection pointer to "impossible" value
                  * and remember its position to update tuple later.
                  */
                 if (nToDelete > 0)
                     redirectTuplePos = toDelete[0];
                 spgPageIndexMultiDelete(state, current->page,
                                         toDelete, nToDelete,
                                         SPGIST_REDIRECT,
                                         SPGIST_PLACEHOLDER,
                                         SPGIST_METAPAGE_BLKNO,
                                         FirstOffsetNumber);
             }
             else
             {
                 /*
                  * During index build there is not concurrent searches, so we
                  * don't need to create redirection tuple.
                  */
                 spgPageIndexMultiDelete(state, current->page,
                                         toDelete, nToDelete,
                                         SPGIST_PLACEHOLDER,
                                         SPGIST_PLACEHOLDER,
                                         InvalidBlockNumber,
                                         InvalidOffsetNumber);
             }
         }
     }
  
     /*
      * Put leaf tuples on proper pages, and update downlinks in innerTuple's
      * nodes.
      */
     startOffsets[0] = startOffsets[1] = InvalidOffsetNumber;
     for (i = 0; i < nToInsert; i++)
     {
         SpGistLeafTuple it = newLeafs[i];
         Buffer      leafBuffer;
         BlockNumber leafBlock;
         OffsetNumber newoffset;
  
         /* Which page is it going to? */
         leafBuffer = leafPageSelect[i] ? newLeafBuffer : current->buffer;
         leafBlock = BufferGetBlockNumber(leafBuffer);
  
         /* Link tuple into correct chain for its node */
         n = out.mapTuplesToNodes[i];
  
         if (ItemPointerIsValid(&nodes[n]->t_tid))
         {
             Assert(ItemPointerGetBlockNumber(&nodes[n]->t_tid) == leafBlock);
             SGLT_SET_NEXTOFFSET(it, ItemPointerGetOffsetNumber(&nodes[n]->t_tid));
         }
         else
             SGLT_SET_NEXTOFFSET(it, InvalidOffsetNumber);
  
         /* Insert it on page */
         newoffset = SpGistPageAddNewItem(state, BufferGetPage(leafBuffer),
                                          (Item) it, it->size,
                                          &startOffsets[leafPageSelect[i]],
                                          false);
         toInsert[i] = newoffset;
  
         /* ... and complete the chain linking */
         ItemPointerSet(&nodes[n]->t_tid, leafBlock, newoffset);
  
         /* Also copy leaf tuple into WAL data */
         memcpy(leafptr, newLeafs[i], newLeafs[i]->size);
         leafptr += newLeafs[i]->size;
     }
  
     /*
      * We're done modifying the other leaf buffer (if any), so mark it dirty.
      * current->buffer will be marked below, after we're entirely done
      * modifying it.
      */
     if (newLeafBuffer != InvalidBuffer)
     {
         MarkBufferDirty(newLeafBuffer);
     }
  
     /* Remember current buffer, since we're about to change "current" */
     saveCurrent = *current;
  
     /*
      * Store the new innerTuple
      */
     if (newInnerBuffer == parent->buffer && newInnerBuffer != InvalidBuffer)
     {
         /*
          * new inner tuple goes to parent page
          */
         Assert(current->buffer != parent->buffer);
  
         /* Repoint "current" at the new inner tuple */
         current->blkno = parent->blkno;
         current->buffer = parent->buffer;
         current->page = parent->page;
         xlrec.offnumInner = current->offnum =
             SpGistPageAddNewItem(state, current->page,
                                  (Item) innerTuple, innerTuple->size,
                                  NULL, false);
  
         /*
          * Update parent node link and mark parent page dirty
          */
         xlrec.innerIsParent = true;
         xlrec.offnumParent = parent->offnum;
         xlrec.nodeI = parent->node;
         saveNodeLink(index, parent, current->blkno, current->offnum);
  
         /*
          * Update redirection link (in old current buffer)
          */
         if (redirectTuplePos != InvalidOffsetNumber)
             setRedirectionTuple(&saveCurrent, redirectTuplePos,
                                 current->blkno, current->offnum);
  
         /* Done modifying old current buffer, mark it dirty */
         MarkBufferDirty(saveCurrent.buffer);
     }
     else if (parent->buffer != InvalidBuffer)
     {
         /*
          * new inner tuple will be stored on a new page
          */
         Assert(newInnerBuffer != InvalidBuffer);
  
         /* Repoint "current" at the new inner tuple */
         current->buffer = newInnerBuffer;
         current->blkno = BufferGetBlockNumber(current->buffer);
         current->page = BufferGetPage(current->buffer);
         xlrec.offnumInner = current->offnum =
             SpGistPageAddNewItem(state, current->page,
                                  (Item) innerTuple, innerTuple->size,
                                  NULL, false);
  
         /* Done modifying new current buffer, mark it dirty */
         MarkBufferDirty(current->buffer);
  
         /*
          * Update parent node link and mark parent page dirty
          */
         xlrec.innerIsParent = (parent->buffer == current->buffer);
         xlrec.offnumParent = parent->offnum;
         xlrec.nodeI = parent->node;
         saveNodeLink(index, parent, current->blkno, current->offnum);
  
         /*
          * Update redirection link (in old current buffer)
          */
         if (redirectTuplePos != InvalidOffsetNumber)
             setRedirectionTuple(&saveCurrent, redirectTuplePos,
                                 current->blkno, current->offnum);
  
         /* Done modifying old current buffer, mark it dirty */
         MarkBufferDirty(saveCurrent.buffer);
     }
     else
     {
         /*
          * Splitting root page, which was a leaf but now becomes inner page
          * (and so "current" continues to point at it)
          */
         Assert(SpGistBlockIsRoot(current->blkno));
         Assert(redirectTuplePos == InvalidOffsetNumber);
  
         SpGistInitBuffer(current->buffer, (isNulls ? SPGIST_NULLS : 0));
         xlrec.initInner = true;
         xlrec.innerIsParent = false;
  
         xlrec.offnumInner = current->offnum =
             PageAddItem(current->page, (Item) innerTuple, innerTuple->size,
                         InvalidOffsetNumber, false, false);
         if (current->offnum != FirstOffsetNumber)
             elog(ERROR, "failed to add item of size %u to SPGiST index page",
                  innerTuple->size);
  
         /* No parent link to update, nor redirection to do */
         xlrec.offnumParent = InvalidOffsetNumber;
         xlrec.nodeI = 0;
  
         /* Done modifying new current buffer, mark it dirty */
         MarkBufferDirty(current->buffer);
  
         /* saveCurrent doesn't represent a different buffer */
         saveCurrent.buffer = InvalidBuffer;
     }
  
     if (RelationNeedsWAL(index) && !state->isBuild)
     {
         XLogRecPtr  recptr;
         int         flags;
  
         XLogBeginInsert();
  
         xlrec.nInsert = nToInsert;
         XLogRegisterData((char *) &xlrec, SizeOfSpgxlogPickSplit);
  
         XLogRegisterData((char *) toDelete,
                          sizeof(OffsetNumber) * xlrec.nDelete);
         XLogRegisterData((char *) toInsert,
                          sizeof(OffsetNumber) * xlrec.nInsert);
         XLogRegisterData((char *) leafPageSelect,
                          sizeof(uint8) * xlrec.nInsert);
         XLogRegisterData((char *) innerTuple, innerTuple->size);
         XLogRegisterData(leafdata, leafptr - leafdata);
  
         /* Old leaf page */
         if (BufferIsValid(saveCurrent.buffer))
         {
             flags = REGBUF_STANDARD;
             if (xlrec.initSrc)
                 flags |= REGBUF_WILL_INIT;
             XLogRegisterBuffer(0, saveCurrent.buffer, flags);
         }
  
         /* New leaf page */
         if (BufferIsValid(newLeafBuffer))
         {
             flags = REGBUF_STANDARD;
             if (xlrec.initDest)
                 flags |= REGBUF_WILL_INIT;
             XLogRegisterBuffer(1, newLeafBuffer, flags);
         }
  
         /* Inner page */
         flags = REGBUF_STANDARD;
         if (xlrec.initInner)
             flags |= REGBUF_WILL_INIT;
         XLogRegisterBuffer(2, current->buffer, flags);
  
         /* Parent page, if different from inner page */
         if (parent->buffer != InvalidBuffer)
         {
             if (parent->buffer != current->buffer)
                 XLogRegisterBuffer(3, parent->buffer, REGBUF_STANDARD);
             else
                 Assert(xlrec.innerIsParent);
         }
  
         /* Issue the WAL record */
         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_PICKSPLIT);
  
         /* Update page LSNs on all affected pages */
         if (newLeafBuffer != InvalidBuffer)
         {
             Page        page = BufferGetPage(newLeafBuffer);
  
             PageSetLSN(page, recptr);
         }
  
         if (saveCurrent.buffer != InvalidBuffer)
         {
             Page        page = BufferGetPage(saveCurrent.buffer);
  
             PageSetLSN(page, recptr);
         }
  
         PageSetLSN(current->page, recptr);
  
         if (parent->buffer != InvalidBuffer)
         {
             PageSetLSN(parent->page, recptr);
         }
     }
  
     END_CRIT_SECTION();
  
     /* Update local free-space cache and unlock buffers */
     if (newLeafBuffer != InvalidBuffer)
     {
         SpGistSetLastUsedPage(index, newLeafBuffer);
         UnlockReleaseBuffer(newLeafBuffer);
     }
     if (saveCurrent.buffer != InvalidBuffer)
     {
         SpGistSetLastUsedPage(index, saveCurrent.buffer);
         UnlockReleaseBuffer(saveCurrent.buffer);
     }
  
     return insertedNew;
 }

Referenced by spgdoinsert().

◆ moveLeafs()

static void moveLeafs	(	Relation	index,
		SpGistState *	state,
		SPPageDesc *	current,
		SPPageDesc *	parent,
		SpGistLeafTuple	newLeafTuple,
		bool	isNulls
	)

static

Definition at line 387 of file spgdoinsert.c.

 {
     int         i,
                 nDelete,
                 nInsert,
                 size;
     Buffer      nbuf;
     Page        npage;
     OffsetNumber r = InvalidOffsetNumber,
                 startOffset = InvalidOffsetNumber;
     bool        replaceDead = false;
     OffsetNumber *toDelete;
     OffsetNumber *toInsert;
     BlockNumber nblkno;
     spgxlogMoveLeafs xlrec;
     char       *leafdata,
                *leafptr;
  
     /* This doesn't work on root page */
     Assert(parent->buffer != InvalidBuffer);
     Assert(parent->buffer != current->buffer);
  
     /* Locate the tuples to be moved, and count up the space needed */
     i = PageGetMaxOffsetNumber(current->page);
     toDelete = (OffsetNumber *) palloc(sizeof(OffsetNumber) * i);
     toInsert = (OffsetNumber *) palloc(sizeof(OffsetNumber) * (i + 1));
  
     size = newLeafTuple->size + sizeof(ItemIdData);
  
     nDelete = 0;
     i = current->offnum;
     while (i != InvalidOffsetNumber)
     {
         SpGistLeafTuple it;
  
         Assert(i >= FirstOffsetNumber &&
                i <= PageGetMaxOffsetNumber(current->page));
         it = (SpGistLeafTuple) PageGetItem(current->page,
                                            PageGetItemId(current->page, i));
  
         if (it->tupstate == SPGIST_LIVE)
         {
             toDelete[nDelete] = i;
             size += it->size + sizeof(ItemIdData);
             nDelete++;
         }
         else if (it->tupstate == SPGIST_DEAD)
         {
             /* We could see a DEAD tuple as first/only chain item */
             Assert(i == current->offnum);
             Assert(SGLT_GET_NEXTOFFSET(it) == InvalidOffsetNumber);
             /* We don't want to move it, so don't count it in size */
             toDelete[nDelete] = i;
             nDelete++;
             replaceDead = true;
         }
         else
             elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate);
  
         i = SGLT_GET_NEXTOFFSET(it);
     }
  
     /* Find a leaf page that will hold them */
     nbuf = SpGistGetBuffer(index, GBUF_LEAF | (isNulls ? GBUF_NULLS : 0),
                            size, &xlrec.newPage);
     npage = BufferGetPage(nbuf);
     nblkno = BufferGetBlockNumber(nbuf);
     Assert(nblkno != current->blkno);
  
     leafdata = leafptr = palloc(size);
  
     START_CRIT_SECTION();
  
     /* copy all the old tuples to new page, unless they're dead */
     nInsert = 0;
     if (!replaceDead)
     {
         for (i = 0; i < nDelete; i++)
         {
             SpGistLeafTuple it;
  
             it = (SpGistLeafTuple) PageGetItem(current->page,
                                                PageGetItemId(current->page, toDelete[i]));
             Assert(it->tupstate == SPGIST_LIVE);
  
             /*
              * Update chain link (notice the chain order gets reversed, but we
              * don't care).  We're modifying the tuple on the source page
              * here, but it's okay since we're about to delete it.
              */
             SGLT_SET_NEXTOFFSET(it, r);
  
             r = SpGistPageAddNewItem(state, npage, (Item) it, it->size,
                                      &startOffset, false);
  
             toInsert[nInsert] = r;
             nInsert++;
  
             /* save modified tuple into leafdata as well */
             memcpy(leafptr, it, it->size);
             leafptr += it->size;
         }
     }
  
     /* add the new tuple as well */
     SGLT_SET_NEXTOFFSET(newLeafTuple, r);
     r = SpGistPageAddNewItem(state, npage,
                              (Item) newLeafTuple, newLeafTuple->size,
                              &startOffset, false);
     toInsert[nInsert] = r;
     nInsert++;
     memcpy(leafptr, newLeafTuple, newLeafTuple->size);
     leafptr += newLeafTuple->size;
  
     /*
      * Now delete the old tuples, leaving a redirection pointer behind for the
      * first one, unless we're doing an index build; in which case there can't
      * be any concurrent scan so we need not provide a redirect.
      */
     spgPageIndexMultiDelete(state, current->page, toDelete, nDelete,
                             state->isBuild ? SPGIST_PLACEHOLDER : SPGIST_REDIRECT,
                             SPGIST_PLACEHOLDER,
                             nblkno, r);
  
     /* Update parent's downlink and mark parent page dirty */
     saveNodeLink(index, parent, nblkno, r);
  
     /* Mark the leaf pages too */
     MarkBufferDirty(current->buffer);
     MarkBufferDirty(nbuf);
  
     if (RelationNeedsWAL(index) && !state->isBuild)
     {
         XLogRecPtr  recptr;
  
         /* prepare WAL info */
         STORE_STATE(state, xlrec.stateSrc);
  
         xlrec.nMoves = nDelete;
         xlrec.replaceDead = replaceDead;
         xlrec.storesNulls = isNulls;
  
         xlrec.offnumParent = parent->offnum;
         xlrec.nodeI = parent->node;
  
         XLogBeginInsert();
         XLogRegisterData((char *) &xlrec, SizeOfSpgxlogMoveLeafs);
         XLogRegisterData((char *) toDelete,
                          sizeof(OffsetNumber) * nDelete);
         XLogRegisterData((char *) toInsert,
                          sizeof(OffsetNumber) * nInsert);
         XLogRegisterData((char *) leafdata, leafptr - leafdata);
  
         XLogRegisterBuffer(0, current->buffer, REGBUF_STANDARD);
         XLogRegisterBuffer(1, nbuf, REGBUF_STANDARD | (xlrec.newPage ? REGBUF_WILL_INIT : 0));
         XLogRegisterBuffer(2, parent->buffer, REGBUF_STANDARD);
  
         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_MOVE_LEAFS);
  
         PageSetLSN(current->page, recptr);
         PageSetLSN(npage, recptr);
         PageSetLSN(parent->page, recptr);
     }
  
     END_CRIT_SECTION();
  
     /* Update local free-space cache and release new buffer */
     SpGistSetLastUsedPage(index, nbuf);
     UnlockReleaseBuffer(nbuf);
 }

References Assert, SPPageDesc::blkno, SPPageDesc::buffer, BufferGetBlockNumber(), BufferGetPage(), elog, END_CRIT_SECTION, ERROR, FirstOffsetNumber, GBUF_LEAF, GBUF_NULLS, i, InvalidBuffer, InvalidOffsetNumber, MarkBufferDirty(), spgxlogMoveLeafs::newPage, spgxlogMoveLeafs::nMoves, SPPageDesc::node, spgxlogMoveLeafs::nodeI, SPPageDesc::offnum, spgxlogMoveLeafs::offnumParent, SPPageDesc::page, PageGetItem(), PageGetItemId(), PageGetMaxOffsetNumber(), PageSetLSN(), palloc(), REGBUF_STANDARD, REGBUF_WILL_INIT, RelationNeedsWAL, spgxlogMoveLeafs::replaceDead, saveNodeLink(), SGLT_GET_NEXTOFFSET, SGLT_SET_NEXTOFFSET, size, SpGistLeafTupleData::size, SizeOfSpgxlogMoveLeafs, SPGIST_DEAD, SPGIST_LIVE, SPGIST_PLACEHOLDER, SPGIST_REDIRECT, SpGistGetBuffer(), SpGistPageAddNewItem(), SpGistSetLastUsedPage(), spgPageIndexMultiDelete(), START_CRIT_SECTION, spgxlogMoveLeafs::stateSrc, STORE_STATE, spgxlogMoveLeafs::storesNulls, SpGistLeafTupleData::tupstate, UnlockReleaseBuffer(), XLOG_SPGIST_MOVE_LEAFS, XLogBeginInsert(), XLogInsert(), XLogRegisterBuffer(), and XLogRegisterData().

Referenced by spgdoinsert().

◆ saveNodeLink()

static void saveNodeLink	(	Relation	index,
		SPPageDesc *	parent,
		BlockNumber	blkno,
		OffsetNumber	offnum
	)

static

Definition at line 186 of file spgdoinsert.c.

 {
     SpGistInnerTuple innerTuple;
  
     innerTuple = (SpGistInnerTuple) PageGetItem(parent->page,
                                                 PageGetItemId(parent->page, parent->offnum));
  
     spgUpdateNodeLink(innerTuple, parent->node, blkno, offnum);
  
     MarkBufferDirty(parent->buffer);
 }

References SPPageDesc::buffer, MarkBufferDirty(), SPPageDesc::node, SPPageDesc::offnum, SPPageDesc::page, PageGetItem(), PageGetItemId(), and spgUpdateNodeLink().

Referenced by addLeafTuple(), doPickSplit(), moveLeafs(), and spgAddNodeAction().

◆ setRedirectionTuple()

static void setRedirectionTuple	(	SPPageDesc *	current,
		OffsetNumber	position,
		BlockNumber	blkno,
		OffsetNumber	offnum
	)

static

Definition at line 568 of file spgdoinsert.c.

 {
     SpGistDeadTuple dt;
  
     dt = (SpGistDeadTuple) PageGetItem(current->page,
                                        PageGetItemId(current->page, position));
     Assert(dt->tupstate == SPGIST_REDIRECT);
     Assert(ItemPointerGetBlockNumber(&dt->pointer) == SPGIST_METAPAGE_BLKNO);
     ItemPointerSet(&dt->pointer, blkno, offnum);
 }

References Assert, ItemPointerGetBlockNumber(), ItemPointerSet(), SPPageDesc::page, PageGetItem(), PageGetItemId(), SpGistDeadTupleData::pointer, SPGIST_METAPAGE_BLKNO, SPGIST_REDIRECT, and SpGistDeadTupleData::tupstate.

Referenced by doPickSplit().

◆ spgAddNodeAction()

static void spgAddNodeAction	(	Relation	index,
		SpGistState *	state,
		SpGistInnerTuple	innerTuple,
		SPPageDesc *	current,
		SPPageDesc *	parent,
		int	nodeN,
		Datum	nodeLabel
	)

static

Definition at line 1513 of file spgdoinsert.c.

 {
     SpGistInnerTuple newInnerTuple;
     spgxlogAddNode xlrec;
  
     /* Should not be applied to nulls */
     Assert(!SpGistPageStoresNulls(current->page));
  
     /* Construct new inner tuple with additional node */
     newInnerTuple = addNode(state, innerTuple, nodeLabel, nodeN);
  
     /* Prepare WAL record */
     STORE_STATE(state, xlrec.stateSrc);
     xlrec.offnum = current->offnum;
  
     /* we don't fill these unless we need to change the parent downlink */
     xlrec.parentBlk = -1;
     xlrec.offnumParent = InvalidOffsetNumber;
     xlrec.nodeI = 0;
  
     /* we don't fill these unless tuple has to be moved */
     xlrec.offnumNew = InvalidOffsetNumber;
     xlrec.newPage = false;
  
     if (PageGetExactFreeSpace(current->page) >=
         newInnerTuple->size - innerTuple->size)
     {
         /*
          * We can replace the inner tuple by new version in-place
          */
         START_CRIT_SECTION();
  
         PageIndexTupleDelete(current->page, current->offnum);
         if (PageAddItem(current->page,
                         (Item) newInnerTuple, newInnerTuple->size,
                         current->offnum, false, false) != current->offnum)
             elog(ERROR, "failed to add item of size %u to SPGiST index page",
                  newInnerTuple->size);
  
         MarkBufferDirty(current->buffer);
  
         if (RelationNeedsWAL(index) && !state->isBuild)
         {
             XLogRecPtr  recptr;
  
             XLogBeginInsert();
             XLogRegisterData((char *) &xlrec, sizeof(xlrec));
             XLogRegisterData((char *) newInnerTuple, newInnerTuple->size);
  
             XLogRegisterBuffer(0, current->buffer, REGBUF_STANDARD);
  
             recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_ADD_NODE);
  
             PageSetLSN(current->page, recptr);
         }
  
         END_CRIT_SECTION();
     }
     else
     {
         /*
          * move inner tuple to another page, and update parent
          */
         SpGistDeadTuple dt;
         SPPageDesc  saveCurrent;
  
         /*
          * It should not be possible to get here for the root page, since we
          * allow only one inner tuple on the root page, and spgFormInnerTuple
          * always checks that inner tuples don't exceed the size of a page.
          */
         if (SpGistBlockIsRoot(current->blkno))
             elog(ERROR, "cannot enlarge root tuple any more");
         Assert(parent->buffer != InvalidBuffer);
  
         saveCurrent = *current;
  
         xlrec.offnumParent = parent->offnum;
         xlrec.nodeI = parent->node;
  
         /*
          * obtain new buffer with the same parity as current, since it will be
          * a child of same parent tuple
          */
         current->buffer = SpGistGetBuffer(index,
                                           GBUF_INNER_PARITY(current->blkno),
                                           newInnerTuple->size + sizeof(ItemIdData),
                                           &xlrec.newPage);
         current->blkno = BufferGetBlockNumber(current->buffer);
         current->page = BufferGetPage(current->buffer);
  
         /*
          * Let's just make real sure new current isn't same as old.  Right now
          * that's impossible, but if SpGistGetBuffer ever got smart enough to
          * delete placeholder tuples before checking space, maybe it wouldn't
          * be impossible.  The case would appear to work except that WAL
          * replay would be subtly wrong, so I think a mere assert isn't enough
          * here.
          */
         if (current->blkno == saveCurrent.blkno)
             elog(ERROR, "SPGiST new buffer shouldn't be same as old buffer");
  
         /*
          * New current and parent buffer will both be modified; but note that
          * parent buffer could be same as either new or old current.
          */
         if (parent->buffer == saveCurrent.buffer)
             xlrec.parentBlk = 0;
         else if (parent->buffer == current->buffer)
             xlrec.parentBlk = 1;
         else
             xlrec.parentBlk = 2;
  
         START_CRIT_SECTION();
  
         /* insert new ... */
         xlrec.offnumNew = current->offnum =
             SpGistPageAddNewItem(state, current->page,
                                  (Item) newInnerTuple, newInnerTuple->size,
                                  NULL, false);
  
         MarkBufferDirty(current->buffer);
  
         /* update parent's downlink and mark parent page dirty */
         saveNodeLink(index, parent, current->blkno, current->offnum);
  
         /*
          * Replace old tuple with a placeholder or redirection tuple.  Unless
          * doing an index build, we have to insert a redirection tuple for
          * possible concurrent scans.  We can't just delete it in any case,
          * because that could change the offsets of other tuples on the page,
          * breaking downlinks from their parents.
          */
         if (state->isBuild)
             dt = spgFormDeadTuple(state, SPGIST_PLACEHOLDER,
                                   InvalidBlockNumber, InvalidOffsetNumber);
         else
             dt = spgFormDeadTuple(state, SPGIST_REDIRECT,
                                   current->blkno, current->offnum);
  
         PageIndexTupleDelete(saveCurrent.page, saveCurrent.offnum);
         if (PageAddItem(saveCurrent.page, (Item) dt, dt->size,
                         saveCurrent.offnum,
                         false, false) != saveCurrent.offnum)
             elog(ERROR, "failed to add item of size %u to SPGiST index page",
                  dt->size);
  
         if (state->isBuild)
             SpGistPageGetOpaque(saveCurrent.page)->nPlaceholder++;
         else
             SpGistPageGetOpaque(saveCurrent.page)->nRedirection++;
  
         MarkBufferDirty(saveCurrent.buffer);
  
         if (RelationNeedsWAL(index) && !state->isBuild)
         {
             XLogRecPtr  recptr;
             int         flags;
  
             XLogBeginInsert();
  
             /* orig page */
             XLogRegisterBuffer(0, saveCurrent.buffer, REGBUF_STANDARD);
             /* new page */
             flags = REGBUF_STANDARD;
             if (xlrec.newPage)
                 flags |= REGBUF_WILL_INIT;
             XLogRegisterBuffer(1, current->buffer, flags);
             /* parent page (if different from orig and new) */
             if (xlrec.parentBlk == 2)
                 XLogRegisterBuffer(2, parent->buffer, REGBUF_STANDARD);
  
             XLogRegisterData((char *) &xlrec, sizeof(xlrec));
             XLogRegisterData((char *) newInnerTuple, newInnerTuple->size);
  
             recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_ADD_NODE);
  
             /* we don't bother to check if any of these are redundant */
             PageSetLSN(current->page, recptr);
             PageSetLSN(parent->page, recptr);
             PageSetLSN(saveCurrent.page, recptr);
         }
  
         END_CRIT_SECTION();
  
         /* Release saveCurrent if it's not same as current or parent */
         if (saveCurrent.buffer != current->buffer &&
             saveCurrent.buffer != parent->buffer)
         {
             SpGistSetLastUsedPage(index, saveCurrent.buffer);
             UnlockReleaseBuffer(saveCurrent.buffer);
         }
     }
 }

References addNode(), Assert, SPPageDesc::blkno, SPPageDesc::buffer, BufferGetBlockNumber(), BufferGetPage(), elog, END_CRIT_SECTION, ERROR, GBUF_INNER_PARITY, InvalidBlockNumber, InvalidBuffer, InvalidOffsetNumber, MarkBufferDirty(), spgxlogAddNode::newPage, SPPageDesc::node, spgxlogAddNode::nodeI, SPPageDesc::offnum, spgxlogAddNode::offnum, spgxlogAddNode::offnumNew, spgxlogAddNode::offnumParent, SPPageDesc::page, PageAddItem, PageGetExactFreeSpace(), PageIndexTupleDelete(), PageSetLSN(), spgxlogAddNode::parentBlk, REGBUF_STANDARD, REGBUF_WILL_INIT, RelationNeedsWAL, saveNodeLink(), SpGistInnerTupleData::size, SpGistDeadTupleData::size, spgFormDeadTuple(), SPGIST_PLACEHOLDER, SPGIST_REDIRECT, SpGistBlockIsRoot, SpGistGetBuffer(), SpGistPageAddNewItem(), SpGistPageGetOpaque, SpGistPageStoresNulls, SpGistSetLastUsedPage(), START_CRIT_SECTION, spgxlogAddNode::stateSrc, STORE_STATE, UnlockReleaseBuffer(), XLOG_SPGIST_ADD_NODE, XLogBeginInsert(), XLogInsert(), XLogRegisterBuffer(), and XLogRegisterData().

Referenced by spgdoinsert().

◆ spgdoinsert()

bool spgdoinsert	(	Relation	index,
		SpGistState *	state,
		ItemPointer	heapPtr,
		Datum *	datums,
		bool *	isnulls
	)

Definition at line 1914 of file spgdoinsert.c.

 {
     bool        result = true;
     TupleDesc   leafDescriptor = state->leafTupDesc;
     bool        isnull = isnulls[spgKeyColumn];
     int         level = 0;
     Datum       leafDatums[INDEX_MAX_KEYS];
     int         leafSize;
     int         bestLeafSize;
     int         numNoProgressCycles = 0;
     SPPageDesc  current,
                 parent;
     FmgrInfo   *procinfo = NULL;
  
     /*
      * Look up FmgrInfo of the user-defined choose function once, to save
      * cycles in the loop below.
      */
     if (!isnull)
         procinfo = index_getprocinfo(index, 1, SPGIST_CHOOSE_PROC);
  
     /*
      * Prepare the leaf datum to insert.
      *
      * If an optional "compress" method is provided, then call it to form the
      * leaf key datum from the input datum.  Otherwise, store the input datum
      * as is.  Since we don't use index_form_tuple in this AM, we have to make
      * sure value to be inserted is not toasted; FormIndexDatum doesn't
      * guarantee that.  But we assume the "compress" method to return an
      * untoasted value.
      */
     if (!isnull)
     {
         if (OidIsValid(index_getprocid(index, 1, SPGIST_COMPRESS_PROC)))
         {
             FmgrInfo   *compressProcinfo = NULL;
  
             compressProcinfo = index_getprocinfo(index, 1, SPGIST_COMPRESS_PROC);
             leafDatums[spgKeyColumn] =
                 FunctionCall1Coll(compressProcinfo,
                                   index->rd_indcollation[spgKeyColumn],
                                   datums[spgKeyColumn]);
         }
         else
         {
             Assert(state->attLeafType.type == state->attType.type);
  
             if (state->attType.attlen == -1)
                 leafDatums[spgKeyColumn] =
                     PointerGetDatum(PG_DETOAST_DATUM(datums[spgKeyColumn]));
             else
                 leafDatums[spgKeyColumn] = datums[spgKeyColumn];
         }
     }
     else
         leafDatums[spgKeyColumn] = (Datum) 0;
  
     /* Likewise, ensure that any INCLUDE values are not toasted */
     for (int i = spgFirstIncludeColumn; i < leafDescriptor->natts; i++)
     {
         if (!isnulls[i])
         {
             if (TupleDescAttr(leafDescriptor, i)->attlen == -1)
                 leafDatums[i] = PointerGetDatum(PG_DETOAST_DATUM(datums[i]));
             else
                 leafDatums[i] = datums[i];
         }
         else
             leafDatums[i] = (Datum) 0;
     }
  
     /*
      * Compute space needed for a leaf tuple containing the given data.
      */
     leafSize = SpGistGetLeafTupleSize(leafDescriptor, leafDatums, isnulls);
     /* Account for an item pointer, too */
     leafSize += sizeof(ItemIdData);
  
     /*
      * If it isn't gonna fit, and the opclass can't reduce the datum size by
      * suffixing, bail out now rather than doing a lot of useless work.
      */
     if (leafSize > SPGIST_PAGE_CAPACITY &&
         (isnull || !state->config.longValuesOK))
         ereport(ERROR,
                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                  errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
                         leafSize - sizeof(ItemIdData),
                         SPGIST_PAGE_CAPACITY - sizeof(ItemIdData),
                         RelationGetRelationName(index)),
                  errhint("Values larger than a buffer page cannot be indexed.")));
     bestLeafSize = leafSize;
  
     /* Initialize "current" to the appropriate root page */
     current.blkno = isnull ? SPGIST_NULL_BLKNO : SPGIST_ROOT_BLKNO;
     current.buffer = InvalidBuffer;
     current.page = NULL;
     current.offnum = FirstOffsetNumber;
     current.node = -1;
  
     /* "parent" is invalid for the moment */
     parent.blkno = InvalidBlockNumber;
     parent.buffer = InvalidBuffer;
     parent.page = NULL;
     parent.offnum = InvalidOffsetNumber;
     parent.node = -1;
  
     /*
      * Before entering the loop, try to clear any pending interrupt condition.
      * If a query cancel is pending, we might as well accept it now not later;
      * while if a non-canceling condition is pending, servicing it here avoids
      * having to restart the insertion and redo all the work so far.
      */
     CHECK_FOR_INTERRUPTS();
  
     for (;;)
     {
         bool        isNew = false;
  
         /*
          * Bail out if query cancel is pending.  We must have this somewhere
          * in the loop since a broken opclass could produce an infinite
          * picksplit loop.  However, because we'll be holding buffer lock(s)
          * after the first iteration, ProcessInterrupts() wouldn't be able to
          * throw a cancel error here.  Hence, if we see that an interrupt is
          * pending, break out of the loop and deal with the situation below.
          * Set result = false because we must restart the insertion if the
          * interrupt isn't a query-cancel-or-die case.
          */
         if (INTERRUPTS_PENDING_CONDITION())
         {
             result = false;
             break;
         }
  
         if (current.blkno == InvalidBlockNumber)
         {
             /*
              * Create a leaf page.  If leafSize is too large to fit on a page,
              * we won't actually use the page yet, but it simplifies the API
              * for doPickSplit to always have a leaf page at hand; so just
              * quietly limit our request to a page size.
              */
             current.buffer =
                 SpGistGetBuffer(index,
                                 GBUF_LEAF | (isnull ? GBUF_NULLS : 0),
                                 Min(leafSize, SPGIST_PAGE_CAPACITY),
                                 &isNew);
             current.blkno = BufferGetBlockNumber(current.buffer);
         }
         else if (parent.buffer == InvalidBuffer)
         {
             /* we hold no parent-page lock, so no deadlock is possible */
             current.buffer = ReadBuffer(index, current.blkno);
             LockBuffer(current.buffer, BUFFER_LOCK_EXCLUSIVE);
         }
         else if (current.blkno != parent.blkno)
         {
             /* descend to a new child page */
             current.buffer = ReadBuffer(index, current.blkno);
  
             /*
              * Attempt to acquire lock on child page.  We must beware of
              * deadlock against another insertion process descending from that
              * page to our parent page (see README).  If we fail to get lock,
              * abandon the insertion and tell our caller to start over.
              *
              * XXX this could be improved, because failing to get lock on a
              * buffer is not proof of a deadlock situation; the lock might be
              * held by a reader, or even just background writer/checkpointer
              * process.  Perhaps it'd be worth retrying after sleeping a bit?
              */
             if (!ConditionalLockBuffer(current.buffer))
             {
                 ReleaseBuffer(current.buffer);
                 UnlockReleaseBuffer(parent.buffer);
                 return false;
             }
         }
         else
         {
             /* inner tuple can be stored on the same page as parent one */
             current.buffer = parent.buffer;
         }
         current.page = BufferGetPage(current.buffer);
  
         /* should not arrive at a page of the wrong type */
         if (isnull ? !SpGistPageStoresNulls(current.page) :
             SpGistPageStoresNulls(current.page))
             elog(ERROR, "SPGiST index page %u has wrong nulls flag",
                  current.blkno);
  
         if (SpGistPageIsLeaf(current.page))
         {
             SpGistLeafTuple leafTuple;
             int         nToSplit,
                         sizeToSplit;
  
             leafTuple = spgFormLeafTuple(state, heapPtr, leafDatums, isnulls);
             if (leafTuple->size + sizeof(ItemIdData) <=
                 SpGistPageGetFreeSpace(current.page, 1))
             {
                 /* it fits on page, so insert it and we're done */
                 addLeafTuple(index, state, leafTuple,
                              &current, &parent, isnull, isNew);
                 break;
             }
             else if ((sizeToSplit =
                       checkSplitConditions(index, state, &current,
                                            &nToSplit)) < SPGIST_PAGE_CAPACITY / 2 &&
                      nToSplit < 64 &&
                      leafTuple->size + sizeof(ItemIdData) + sizeToSplit <= SPGIST_PAGE_CAPACITY)
             {
                 /*
                  * the amount of data is pretty small, so just move the whole
                  * chain to another leaf page rather than splitting it.
                  */
                 Assert(!isNew);
                 moveLeafs(index, state, &current, &parent, leafTuple, isnull);
                 break;          /* we're done */
             }
             else
             {
                 /* picksplit */
                 if (doPickSplit(index, state, &current, &parent,
                                 leafTuple, level, isnull, isNew))
                     break;      /* doPickSplit installed new tuples */
  
                 /* leaf tuple will not be inserted yet */
                 pfree(leafTuple);
  
                 /*
                  * current now describes new inner tuple, go insert into it
                  */
                 Assert(!SpGistPageIsLeaf(current.page));
                 goto process_inner_tuple;
             }
         }
         else                    /* non-leaf page */
         {
             /*
              * Apply the opclass choose function to figure out how to insert
              * the given datum into the current inner tuple.
              */
             SpGistInnerTuple innerTuple;
             spgChooseIn in;
             spgChooseOut out;
  
             /*
              * spgAddNode and spgSplitTuple cases will loop back to here to
              * complete the insertion operation.  Just in case the choose
              * function is broken and produces add or split requests
              * repeatedly, check for query cancel (see comments above).
              */
     process_inner_tuple:
             if (INTERRUPTS_PENDING_CONDITION())
             {
                 result = false;
                 break;
             }
  
             innerTuple = (SpGistInnerTuple) PageGetItem(current.page,
                                                         PageGetItemId(current.page, current.offnum));
  
             in.datum = datums[spgKeyColumn];
             in.leafDatum = leafDatums[spgKeyColumn];
             in.level = level;
             in.allTheSame = innerTuple->allTheSame;
             in.hasPrefix = (innerTuple->prefixSize > 0);
             in.prefixDatum = SGITDATUM(innerTuple, state);
             in.nNodes = innerTuple->nNodes;
             in.nodeLabels = spgExtractNodeLabels(state, innerTuple);
  
             memset(&out, 0, sizeof(out));
  
             if (!isnull)
             {
                 /* use user-defined choose method */
                 FunctionCall2Coll(procinfo,
                                   index->rd_indcollation[0],
                                   PointerGetDatum(&in),
                                   PointerGetDatum(&out));
             }
             else
             {
                 /* force "match" action (to insert to random subnode) */
                 out.resultType = spgMatchNode;
             }
  
             if (innerTuple->allTheSame)
             {
                 /*
                  * It's not allowed to do an AddNode at an allTheSame tuple.
                  * Opclass must say "match", in which case we choose a random
                  * one of the nodes to descend into, or "split".
                  */
                 if (out.resultType == spgAddNode)
                     elog(ERROR, "cannot add a node to an allTheSame inner tuple");
                 else if (out.resultType == spgMatchNode)
                     out.result.matchNode.nodeN =
                         pg_prng_uint64_range(&pg_global_prng_state,
                                              0, innerTuple->nNodes - 1);
             }
  
             switch (out.resultType)
             {
                 case spgMatchNode:
                     /* Descend to N'th child node */
                     spgMatchNodeAction(index, state, innerTuple,
                                        &current, &parent,
                                        out.result.matchNode.nodeN);
                     /* Adjust level as per opclass request */
                     level += out.result.matchNode.levelAdd;
                     /* Replace leafDatum and recompute leafSize */
                     if (!isnull)
                     {
                         leafDatums[spgKeyColumn] = out.result.matchNode.restDatum;
                         leafSize = SpGistGetLeafTupleSize(leafDescriptor,
                                                           leafDatums, isnulls);
                         leafSize += sizeof(ItemIdData);
                     }
  
                     /*
                      * Check new tuple size; fail if it can't fit, unless the
                      * opclass says it can handle the situation by suffixing.
                      *
                      * However, the opclass can only shorten the leaf datum,
                      * which may not be enough to ever make the tuple fit,
                      * since INCLUDE columns might alone use more than a page.
                      * Depending on the opclass' behavior, that could lead to
                      * an infinite loop --- spgtextproc.c, for example, will
                      * just repeatedly generate an empty-string leaf datum
                      * once it runs out of data.  Actual bugs in opclasses
                      * might cause infinite looping, too.  To detect such a
                      * loop, check to see if we are making progress by
                      * reducing the leafSize in each pass.  This is a bit
                      * tricky though.  Because of alignment considerations,
                      * the total tuple size might not decrease on every pass.
                      * Also, there are edge cases where the choose method
                      * might seem to not make progress for a cycle or two.
                      * Somewhat arbitrarily, we allow up to 10 no-progress
                      * iterations before failing.  (This limit should be more
                      * than MAXALIGN, to accommodate opclasses that trim one
                      * byte from the leaf datum per pass.)
                      */
                     if (leafSize > SPGIST_PAGE_CAPACITY)
                     {
                         bool        ok = false;
  
                         if (state->config.longValuesOK && !isnull)
                         {
                             if (leafSize < bestLeafSize)
                             {
                                 ok = true;
                                 bestLeafSize = leafSize;
                                 numNoProgressCycles = 0;
                             }
                             else if (++numNoProgressCycles < 10)
                                 ok = true;
                         }
                         if (!ok)
                             ereport(ERROR,
                                     (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                                      errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
                                             leafSize - sizeof(ItemIdData),
                                             SPGIST_PAGE_CAPACITY - sizeof(ItemIdData),
                                             RelationGetRelationName(index)),
                                      errhint("Values larger than a buffer page cannot be indexed.")));
                     }
  
                     /*
                      * Loop around and attempt to insert the new leafDatum at
                      * "current" (which might reference an existing child
                      * tuple, or might be invalid to force us to find a new
                      * page for the tuple).
                      */
                     break;
                 case spgAddNode:
                     /* AddNode is not sensible if nodes don't have labels */
                     if (in.nodeLabels == NULL)
                         elog(ERROR, "cannot add a node to an inner tuple without node labels");
                     /* Add node to inner tuple, per request */
                     spgAddNodeAction(index, state, innerTuple,
                                      &current, &parent,
                                      out.result.addNode.nodeN,
                                      out.result.addNode.nodeLabel);
  
                     /*
                      * Retry insertion into the enlarged node.  We assume that
                      * we'll get a MatchNode result this time.
                      */
                     goto process_inner_tuple;
                     break;
                 case spgSplitTuple:
                     /* Split inner tuple, per request */
                     spgSplitNodeAction(index, state, innerTuple,
                                        &current, &out);
  
                     /* Retry insertion into the split node */
                     goto process_inner_tuple;
                     break;
                 default:
                     elog(ERROR, "unrecognized SPGiST choose result: %d",
                          (int) out.resultType);
                     break;
             }
         }
     }                           /* end loop */
  
     /*
      * Release any buffers we're still holding.  Beware of possibility that
      * current and parent reference same buffer.
      */
     if (current.buffer != InvalidBuffer)
     {
         SpGistSetLastUsedPage(index, current.buffer);
         UnlockReleaseBuffer(current.buffer);
     }
     if (parent.buffer != InvalidBuffer &&
         parent.buffer != current.buffer)
     {
         SpGistSetLastUsedPage(index, parent.buffer);
         UnlockReleaseBuffer(parent.buffer);
     }
  
     /*
      * We do not support being called while some outer function is holding a
      * buffer lock (or any other reason to postpone query cancels).  If that
      * were the case, telling the caller to retry would create an infinite
      * loop.
      */
     Assert(INTERRUPTS_CAN_BE_PROCESSED());
  
     /*
      * Finally, check for interrupts again.  If there was a query cancel,
      * ProcessInterrupts() will be able to throw the error here.  If it was
      * some other kind of interrupt that can just be cleared, return false to
      * tell our caller to retry.
      */
     CHECK_FOR_INTERRUPTS();
  
     return result;
 }

References addLeafTuple(), spgChooseOut::addNode, spgChooseIn::allTheSame, SpGistInnerTupleData::allTheSame, Assert, SPPageDesc::blkno, SPPageDesc::buffer, BUFFER_LOCK_EXCLUSIVE, BufferGetBlockNumber(), BufferGetPage(), CHECK_FOR_INTERRUPTS, checkSplitConditions(), ConditionalLockBuffer(), spgChooseIn::datum, doPickSplit(), elog, ereport, errcode(), errhint(), errmsg(), ERROR, FirstOffsetNumber, FunctionCall1Coll(), FunctionCall2Coll(), GBUF_LEAF, GBUF_NULLS, spgChooseIn::hasPrefix, i, index_getprocid(), index_getprocinfo(), INDEX_MAX_KEYS, INTERRUPTS_CAN_BE_PROCESSED, INTERRUPTS_PENDING_CONDITION, InvalidBlockNumber, InvalidBuffer, InvalidOffsetNumber, spgChooseIn::leafDatum, spgChooseIn::level, LockBuffer(), spgChooseOut::matchNode, Min, moveLeafs(), TupleDescData::natts, spgChooseIn::nNodes, SpGistInnerTupleData::nNodes, SPPageDesc::node, spgChooseIn::nodeLabels, SPPageDesc::offnum, OidIsValid, SPPageDesc::page, PageGetItem(), PageGetItemId(), pfree(), PG_DETOAST_DATUM, pg_global_prng_state, pg_prng_uint64_range(), PointerGetDatum(), spgChooseIn::prefixDatum, SpGistInnerTupleData::prefixSize, ReadBuffer(), RelationGetRelationName, ReleaseBuffer(), spgChooseOut::result, spgChooseOut::resultType, SGITDATUM, SpGistLeafTupleData::size, spgAddNode, spgAddNodeAction(), spgExtractNodeLabels(), spgFirstIncludeColumn, spgFormLeafTuple(), SPGIST_CHOOSE_PROC, SPGIST_COMPRESS_PROC, SPGIST_NULL_BLKNO, SPGIST_PAGE_CAPACITY, SPGIST_ROOT_BLKNO, SpGistGetBuffer(), SpGistGetLeafTupleSize(), SpGistPageGetFreeSpace, SpGistPageIsLeaf, SpGistPageStoresNulls, SpGistSetLastUsedPage(), spgKeyColumn, spgMatchNode, spgMatchNodeAction(), spgSplitNodeAction(), spgSplitTuple, TupleDescAttr, and UnlockReleaseBuffer().

Referenced by spginsert(), and spgistBuildCallback().

◆ spgMatchNodeAction()

static void spgMatchNodeAction	(	Relation	index,
		SpGistState *	state,
		SpGistInnerTuple	innerTuple,
		SPPageDesc *	current,
		SPPageDesc *	parent,
		int	nodeN
	)

static

Definition at line 1459 of file spgdoinsert.c.

 {
     int         i;
     SpGistNodeTuple node;
  
     /* Release previous parent buffer if any */
     if (parent->buffer != InvalidBuffer &&
         parent->buffer != current->buffer)
     {
         SpGistSetLastUsedPage(index, parent->buffer);
         UnlockReleaseBuffer(parent->buffer);
     }
  
     /* Repoint parent to specified node of current inner tuple */
     parent->blkno = current->blkno;
     parent->buffer = current->buffer;
     parent->page = current->page;
     parent->offnum = current->offnum;
     parent->node = nodeN;
  
     /* Locate that node */
     SGITITERATE(innerTuple, i, node)
     {
         if (i == nodeN)
             break;
     }
  
     if (i != nodeN)
         elog(ERROR, "failed to find requested node %d in SPGiST inner tuple",
              nodeN);
  
     /* Point current to the downlink location, if any */
     if (ItemPointerIsValid(&node->t_tid))
     {
         current->blkno = ItemPointerGetBlockNumber(&node->t_tid);
         current->offnum = ItemPointerGetOffsetNumber(&node->t_tid);
     }
     else
     {
         /* Downlink is empty, so we'll need to find a new page */
         current->blkno = InvalidBlockNumber;
         current->offnum = InvalidOffsetNumber;
     }
  
     current->buffer = InvalidBuffer;
     current->page = NULL;
 }

References SPPageDesc::blkno, SPPageDesc::buffer, elog, ERROR, i, InvalidBlockNumber, InvalidBuffer, InvalidOffsetNumber, ItemPointerGetBlockNumber(), ItemPointerGetOffsetNumber(), ItemPointerIsValid(), SPPageDesc::node, SPPageDesc::offnum, SPPageDesc::page, SGITITERATE, SpGistSetLastUsedPage(), IndexTupleData::t_tid, and UnlockReleaseBuffer().

Referenced by spgdoinsert().

◆ spgPageIndexMultiDelete()

void spgPageIndexMultiDelete	(	SpGistState *	state,
		Page	page,
		OffsetNumber *	itemnos,
		int	nitems,
		int	firststate,
		int	reststate,
		BlockNumber	blkno,
		OffsetNumber	offnum
	)

Definition at line 131 of file spgdoinsert.c.

 {
     OffsetNumber firstItem;
     OffsetNumber sortednos[MaxIndexTuplesPerPage];
     SpGistDeadTuple tuple = NULL;
     int         i;
  
     if (nitems == 0)
         return;                 /* nothing to do */
  
     /*
      * For efficiency we want to use PageIndexMultiDelete, which requires the
      * targets to be listed in sorted order, so we have to sort the itemnos
      * array.  (This also greatly simplifies the math for reinserting the
      * replacement tuples.)  However, we must not scribble on the caller's
      * array, so we have to make a copy.
      */
     memcpy(sortednos, itemnos, sizeof(OffsetNumber) * nitems);
     if (nitems > 1)
         qsort(sortednos, nitems, sizeof(OffsetNumber), cmpOffsetNumbers);
  
     PageIndexMultiDelete(page, sortednos, nitems);
  
     firstItem = itemnos[0];
  
     for (i = 0; i < nitems; i++)
     {
         OffsetNumber itemno = sortednos[i];
         int         tupstate;
  
         tupstate = (itemno == firstItem) ? firststate : reststate;
         if (tuple == NULL || tuple->tupstate != tupstate)
             tuple = spgFormDeadTuple(state, tupstate, blkno, offnum);
  
         if (PageAddItem(page, (Item) tuple, tuple->size,
                         itemno, false, false) != itemno)
             elog(ERROR, "failed to add item of size %u to SPGiST index page",
                  tuple->size);
  
         if (tupstate == SPGIST_REDIRECT)
             SpGistPageGetOpaque(page)->nRedirection++;
         else if (tupstate == SPGIST_PLACEHOLDER)
             SpGistPageGetOpaque(page)->nPlaceholder++;
     }
 }

References cmpOffsetNumbers(), elog, ERROR, i, MaxIndexTuplesPerPage, nitems, PageAddItem, PageIndexMultiDelete(), qsort, SpGistDeadTupleData::size, spgFormDeadTuple(), SPGIST_PLACEHOLDER, SPGIST_REDIRECT, SpGistPageGetOpaque, and SpGistDeadTupleData::tupstate.

Referenced by doPickSplit(), moveLeafs(), spgRedoMoveLeafs(), spgRedoPickSplit(), spgRedoVacuumLeaf(), and vacuumLeafPage().

◆ spgSplitNodeAction()

static void spgSplitNodeAction	(	Relation	index,
		SpGistState *	state,
		SpGistInnerTuple	innerTuple,
		SPPageDesc *	current,
		spgChooseOut *	out
	)

static

Definition at line 1715 of file spgdoinsert.c.

 {
     SpGistInnerTuple prefixTuple,
                 postfixTuple;
     SpGistNodeTuple node,
                *nodes;
     BlockNumber postfixBlkno;
     OffsetNumber postfixOffset;
     int         i;
     spgxlogSplitTuple xlrec;
     Buffer      newBuffer = InvalidBuffer;
  
     /* Should not be applied to nulls */
     Assert(!SpGistPageStoresNulls(current->page));
  
     /* Check opclass gave us sane values */
     if (out->result.splitTuple.prefixNNodes <= 0 ||
         out->result.splitTuple.prefixNNodes > SGITMAXNNODES)
         elog(ERROR, "invalid number of prefix nodes: %d",
              out->result.splitTuple.prefixNNodes);
     if (out->result.splitTuple.childNodeN < 0 ||
         out->result.splitTuple.childNodeN >=
         out->result.splitTuple.prefixNNodes)
         elog(ERROR, "invalid child node number: %d",
              out->result.splitTuple.childNodeN);
  
     /*
      * Construct new prefix tuple with requested number of nodes.  We'll fill
      * in the childNodeN'th node's downlink below.
      */
     nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) *
                                        out->result.splitTuple.prefixNNodes);
  
     for (i = 0; i < out->result.splitTuple.prefixNNodes; i++)
     {
         Datum       label = (Datum) 0;
         bool        labelisnull;
  
         labelisnull = (out->result.splitTuple.prefixNodeLabels == NULL);
         if (!labelisnull)
             label = out->result.splitTuple.prefixNodeLabels[i];
         nodes[i] = spgFormNodeTuple(state, label, labelisnull);
     }
  
     prefixTuple = spgFormInnerTuple(state,
                                     out->result.splitTuple.prefixHasPrefix,
                                     out->result.splitTuple.prefixPrefixDatum,
                                     out->result.splitTuple.prefixNNodes,
                                     nodes);
  
     /* it must fit in the space that innerTuple now occupies */
     if (prefixTuple->size > innerTuple->size)
         elog(ERROR, "SPGiST inner-tuple split must not produce longer prefix");
  
     /*
      * Construct new postfix tuple, containing all nodes of innerTuple with
      * same node datums, but with the prefix specified by the picksplit
      * function.
      */
     nodes = palloc(sizeof(SpGistNodeTuple) * innerTuple->nNodes);
     SGITITERATE(innerTuple, i, node)
     {
         nodes[i] = node;
     }
  
     postfixTuple = spgFormInnerTuple(state,
                                      out->result.splitTuple.postfixHasPrefix,
                                      out->result.splitTuple.postfixPrefixDatum,
                                      innerTuple->nNodes, nodes);
  
     /* Postfix tuple is allTheSame if original tuple was */
     postfixTuple->allTheSame = innerTuple->allTheSame;
  
     /* prep data for WAL record */
     xlrec.newPage = false;
  
     /*
      * If we can't fit both tuples on the current page, get a new page for the
      * postfix tuple.  In particular, can't split to the root page.
      *
      * For the space calculation, note that prefixTuple replaces innerTuple
      * but postfixTuple will be a new entry.
      */
     if (SpGistBlockIsRoot(current->blkno) ||
         SpGistPageGetFreeSpace(current->page, 1) + innerTuple->size <
         prefixTuple->size + postfixTuple->size + sizeof(ItemIdData))
     {
         /*
          * Choose page with next triple parity, because postfix tuple is a
          * child of prefix one
          */
         newBuffer = SpGistGetBuffer(index,
                                     GBUF_INNER_PARITY(current->blkno + 1),
                                     postfixTuple->size + sizeof(ItemIdData),
                                     &xlrec.newPage);
     }
  
     START_CRIT_SECTION();
  
     /*
      * Replace old tuple by prefix tuple
      */
     PageIndexTupleDelete(current->page, current->offnum);
     xlrec.offnumPrefix = PageAddItem(current->page,
                                      (Item) prefixTuple, prefixTuple->size,
                                      current->offnum, false, false);
     if (xlrec.offnumPrefix != current->offnum)
         elog(ERROR, "failed to add item of size %u to SPGiST index page",
              prefixTuple->size);
  
     /*
      * put postfix tuple into appropriate page
      */
     if (newBuffer == InvalidBuffer)
     {
         postfixBlkno = current->blkno;
         xlrec.offnumPostfix = postfixOffset =
             SpGistPageAddNewItem(state, current->page,
                                  (Item) postfixTuple, postfixTuple->size,
                                  NULL, false);
         xlrec.postfixBlkSame = true;
     }
     else
     {
         postfixBlkno = BufferGetBlockNumber(newBuffer);
         xlrec.offnumPostfix = postfixOffset =
             SpGistPageAddNewItem(state, BufferGetPage(newBuffer),
                                  (Item) postfixTuple, postfixTuple->size,
                                  NULL, false);
         MarkBufferDirty(newBuffer);
         xlrec.postfixBlkSame = false;
     }
  
     /*
      * And set downlink pointer in the prefix tuple to point to postfix tuple.
      * (We can't avoid this step by doing the above two steps in opposite
      * order, because there might not be enough space on the page to insert
      * the postfix tuple first.)  We have to update the local copy of the
      * prefixTuple too, because that's what will be written to WAL.
      */
     spgUpdateNodeLink(prefixTuple, out->result.splitTuple.childNodeN,
                       postfixBlkno, postfixOffset);
     prefixTuple = (SpGistInnerTuple) PageGetItem(current->page,
                                                  PageGetItemId(current->page, current->offnum));
     spgUpdateNodeLink(prefixTuple, out->result.splitTuple.childNodeN,
                       postfixBlkno, postfixOffset);
  
     MarkBufferDirty(current->buffer);
  
     if (RelationNeedsWAL(index) && !state->isBuild)
     {
         XLogRecPtr  recptr;
  
         XLogBeginInsert();
         XLogRegisterData((char *) &xlrec, sizeof(xlrec));
         XLogRegisterData((char *) prefixTuple, prefixTuple->size);
         XLogRegisterData((char *) postfixTuple, postfixTuple->size);
  
         XLogRegisterBuffer(0, current->buffer, REGBUF_STANDARD);
         if (newBuffer != InvalidBuffer)
         {
             int         flags;
  
             flags = REGBUF_STANDARD;
             if (xlrec.newPage)
                 flags |= REGBUF_WILL_INIT;
             XLogRegisterBuffer(1, newBuffer, flags);
         }
  
         recptr = XLogInsert(RM_SPGIST_ID, XLOG_SPGIST_SPLIT_TUPLE);
  
         PageSetLSN(current->page, recptr);
  
         if (newBuffer != InvalidBuffer)
         {
             PageSetLSN(BufferGetPage(newBuffer), recptr);
         }
     }
  
     END_CRIT_SECTION();
  
     /* Update local free-space cache and release buffer */
     if (newBuffer != InvalidBuffer)
     {
         SpGistSetLastUsedPage(index, newBuffer);
         UnlockReleaseBuffer(newBuffer);
     }
 }

References SpGistInnerTupleData::allTheSame, Assert, SPPageDesc::blkno, SPPageDesc::buffer, BufferGetBlockNumber(), BufferGetPage(), elog, END_CRIT_SECTION, ERROR, GBUF_INNER_PARITY, i, InvalidBuffer, label, MarkBufferDirty(), spgxlogSplitTuple::newPage, SpGistInnerTupleData::nNodes, SPPageDesc::offnum, spgxlogSplitTuple::offnumPostfix, spgxlogSplitTuple::offnumPrefix, SPPageDesc::page, PageAddItem, PageGetItem(), PageGetItemId(), PageIndexTupleDelete(), PageSetLSN(), palloc(), spgxlogSplitTuple::postfixBlkSame, REGBUF_STANDARD, REGBUF_WILL_INIT, RelationNeedsWAL, spgChooseOut::result, SGITITERATE, SGITMAXNNODES, SpGistInnerTupleData::size, spgFormInnerTuple(), spgFormNodeTuple(), SpGistBlockIsRoot, SpGistGetBuffer(), SpGistPageAddNewItem(), SpGistPageGetFreeSpace, SpGistPageStoresNulls, SpGistSetLastUsedPage(), spgUpdateNodeLink(), spgChooseOut::splitTuple, START_CRIT_SECTION, UnlockReleaseBuffer(), XLOG_SPGIST_SPLIT_TUPLE, XLogBeginInsert(), XLogInsert(), XLogRegisterBuffer(), and XLogRegisterData().

Referenced by spgdoinsert().

◆ spgUpdateNodeLink()

void spgUpdateNodeLink	(	SpGistInnerTuple	tup,
		int	nodeN,
		BlockNumber	blkno,
		OffsetNumber	offset
	)

Definition at line 52 of file spgdoinsert.c.

 {
     int         i;
     SpGistNodeTuple node;
  
     SGITITERATE(tup, i, node)
     {
         if (i == nodeN)
         {
             ItemPointerSet(&node->t_tid, blkno, offset);
             return;
         }
     }
  
     elog(ERROR, "failed to find requested node %d in SPGiST inner tuple",
          nodeN);
 }

References elog, ERROR, i, ItemPointerSet(), SGITITERATE, and IndexTupleData::t_tid.

Referenced by saveNodeLink(), spgRedoAddLeaf(), spgRedoAddNode(), spgRedoMoveLeafs(), spgRedoPickSplit(), and spgSplitNodeAction().

Data Structures

Typedefs

Functions

Typedef Documentation

◆ SPPageDesc

Function Documentation

◆ addLeafTuple()

◆ addNode()

◆ checkAllTheSame()

◆ checkSplitConditions()

◆ cmpOffsetNumbers()

◆ doPickSplit()

◆ moveLeafs()

◆ saveNodeLink()

◆ setRedirectionTuple()

◆ spgAddNodeAction()

◆ spgdoinsert()

◆ spgMatchNodeAction()

◆ spgPageIndexMultiDelete()

◆ spgSplitNodeAction()

◆ spgUpdateNodeLink()