LCOV - code coverage report
Current view: top level - src/backend/access/spgist - spgutils.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18devel Lines: 434 473 91.8 %
Date: 2025-04-01 14:15:22 Functions: 26 26 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * spgutils.c
       4             :  *    various support functions for SP-GiST
       5             :  *
       6             :  *
       7             :  * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
       8             :  * Portions Copyright (c) 1994, Regents of the University of California
       9             :  *
      10             :  * IDENTIFICATION
      11             :  *          src/backend/access/spgist/spgutils.c
      12             :  *
      13             :  *-------------------------------------------------------------------------
      14             :  */
      15             : 
      16             : #include "postgres.h"
      17             : 
      18             : #include "access/amvalidate.h"
      19             : #include "access/htup_details.h"
      20             : #include "access/reloptions.h"
      21             : #include "access/spgist_private.h"
      22             : #include "access/toast_compression.h"
      23             : #include "access/transam.h"
      24             : #include "access/xact.h"
      25             : #include "catalog/pg_amop.h"
      26             : #include "commands/vacuum.h"
      27             : #include "nodes/nodeFuncs.h"
      28             : #include "parser/parse_coerce.h"
      29             : #include "storage/bufmgr.h"
      30             : #include "storage/indexfsm.h"
      31             : #include "utils/catcache.h"
      32             : #include "utils/fmgrprotos.h"
      33             : #include "utils/index_selfuncs.h"
      34             : #include "utils/lsyscache.h"
      35             : #include "utils/rel.h"
      36             : #include "utils/syscache.h"
      37             : 
      38             : 
      39             : /*
      40             :  * SP-GiST handler function: return IndexAmRoutine with access method parameters
      41             :  * and callbacks.
      42             :  */
      43             : Datum
      44        1324 : spghandler(PG_FUNCTION_ARGS)
      45             : {
      46        1324 :     IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);
      47             : 
      48        1324 :     amroutine->amstrategies = 0;
      49        1324 :     amroutine->amsupport = SPGISTNProc;
      50        1324 :     amroutine->amoptsprocnum = SPGIST_OPTIONS_PROC;
      51        1324 :     amroutine->amcanorder = false;
      52        1324 :     amroutine->amcanorderbyop = true;
      53        1324 :     amroutine->amcanhash = false;
      54        1324 :     amroutine->amconsistentequality = false;
      55        1324 :     amroutine->amconsistentordering = false;
      56        1324 :     amroutine->amcanbackward = false;
      57        1324 :     amroutine->amcanunique = false;
      58        1324 :     amroutine->amcanmulticol = false;
      59        1324 :     amroutine->amoptionalkey = true;
      60        1324 :     amroutine->amsearcharray = false;
      61        1324 :     amroutine->amsearchnulls = true;
      62        1324 :     amroutine->amstorage = true;
      63        1324 :     amroutine->amclusterable = false;
      64        1324 :     amroutine->ampredlocks = false;
      65        1324 :     amroutine->amcanparallel = false;
      66        1324 :     amroutine->amcanbuildparallel = false;
      67        1324 :     amroutine->amcaninclude = true;
      68        1324 :     amroutine->amusemaintenanceworkmem = false;
      69        1324 :     amroutine->amsummarizing = false;
      70        1324 :     amroutine->amparallelvacuumoptions =
      71             :         VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_COND_CLEANUP;
      72        1324 :     amroutine->amkeytype = InvalidOid;
      73             : 
      74        1324 :     amroutine->ambuild = spgbuild;
      75        1324 :     amroutine->ambuildempty = spgbuildempty;
      76        1324 :     amroutine->aminsert = spginsert;
      77        1324 :     amroutine->aminsertcleanup = NULL;
      78        1324 :     amroutine->ambulkdelete = spgbulkdelete;
      79        1324 :     amroutine->amvacuumcleanup = spgvacuumcleanup;
      80        1324 :     amroutine->amcanreturn = spgcanreturn;
      81        1324 :     amroutine->amcostestimate = spgcostestimate;
      82        1324 :     amroutine->amgettreeheight = NULL;
      83        1324 :     amroutine->amoptions = spgoptions;
      84        1324 :     amroutine->amproperty = spgproperty;
      85        1324 :     amroutine->ambuildphasename = NULL;
      86        1324 :     amroutine->amvalidate = spgvalidate;
      87        1324 :     amroutine->amadjustmembers = spgadjustmembers;
      88        1324 :     amroutine->ambeginscan = spgbeginscan;
      89        1324 :     amroutine->amrescan = spgrescan;
      90        1324 :     amroutine->amgettuple = spggettuple;
      91        1324 :     amroutine->amgetbitmap = spggetbitmap;
      92        1324 :     amroutine->amendscan = spgendscan;
      93        1324 :     amroutine->ammarkpos = NULL;
      94        1324 :     amroutine->amrestrpos = NULL;
      95        1324 :     amroutine->amestimateparallelscan = NULL;
      96        1324 :     amroutine->aminitparallelscan = NULL;
      97        1324 :     amroutine->amparallelrescan = NULL;
      98        1324 :     amroutine->amtranslatestrategy = NULL;
      99        1324 :     amroutine->amtranslatecmptype = NULL;
     100             : 
     101        1324 :     PG_RETURN_POINTER(amroutine);
     102             : }
     103             : 
     104             : /*
     105             :  * GetIndexInputType
     106             :  *      Determine the nominal input data type for an index column
     107             :  *
     108             :  * We define the "nominal" input type as the associated opclass's opcintype,
     109             :  * or if that is a polymorphic type, the base type of the heap column or
     110             :  * expression that is the index's input.  The reason for preferring the
     111             :  * opcintype is that non-polymorphic opclasses probably don't want to hear
     112             :  * about binary-compatible input types.  For instance, if a text opclass
     113             :  * is being used with a varchar heap column, we want to report "text" not
     114             :  * "varchar".  Likewise, opclasses don't want to hear about domain types,
     115             :  * so if we do consult the actual input type, we make sure to flatten domains.
     116             :  *
     117             :  * At some point maybe this should go somewhere else, but it's not clear
     118             :  * if any other index AMs have a use for it.
     119             :  */
     120             : static Oid
     121         412 : GetIndexInputType(Relation index, AttrNumber indexcol)
     122             : {
     123             :     Oid         opcintype;
     124             :     AttrNumber  heapcol;
     125             :     List       *indexprs;
     126             :     ListCell   *indexpr_item;
     127             : 
     128             :     Assert(index->rd_index != NULL);
     129             :     Assert(indexcol > 0 && indexcol <= index->rd_index->indnkeyatts);
     130         412 :     opcintype = index->rd_opcintype[indexcol - 1];
     131         412 :     if (!IsPolymorphicType(opcintype))
     132         308 :         return opcintype;
     133         104 :     heapcol = index->rd_index->indkey.values[indexcol - 1];
     134         104 :     if (heapcol != 0)           /* Simple index column? */
     135          92 :         return getBaseType(get_atttype(index->rd_index->indrelid, heapcol));
     136             : 
     137             :     /*
     138             :      * If the index expressions are already cached, skip calling
     139             :      * RelationGetIndexExpressions, as it will make a copy which is overkill.
     140             :      * We're not going to modify the trees, and we're not going to do anything
     141             :      * that would invalidate the relcache entry before we're done.
     142             :      */
     143          12 :     if (index->rd_indexprs)
     144           0 :         indexprs = index->rd_indexprs;
     145             :     else
     146          12 :         indexprs = RelationGetIndexExpressions(index);
     147          12 :     indexpr_item = list_head(indexprs);
     148          12 :     for (int i = 1; i <= index->rd_index->indnkeyatts; i++)
     149             :     {
     150          12 :         if (index->rd_index->indkey.values[i - 1] == 0)
     151             :         {
     152             :             /* expression column */
     153          12 :             if (indexpr_item == NULL)
     154           0 :                 elog(ERROR, "wrong number of index expressions");
     155          12 :             if (i == indexcol)
     156          12 :                 return getBaseType(exprType((Node *) lfirst(indexpr_item)));
     157           0 :             indexpr_item = lnext(indexprs, indexpr_item);
     158             :         }
     159             :     }
     160           0 :     elog(ERROR, "wrong number of index expressions");
     161             :     return InvalidOid;          /* keep compiler quiet */
     162             : }
     163             : 
     164             : /* Fill in a SpGistTypeDesc struct with info about the specified data type */
     165             : static void
     166        1266 : fillTypeDesc(SpGistTypeDesc *desc, Oid type)
     167             : {
     168             :     HeapTuple   tp;
     169             :     Form_pg_type typtup;
     170             : 
     171        1266 :     desc->type = type;
     172        1266 :     tp = SearchSysCache1(TYPEOID, ObjectIdGetDatum(type));
     173        1266 :     if (!HeapTupleIsValid(tp))
     174           0 :         elog(ERROR, "cache lookup failed for type %u", type);
     175        1266 :     typtup = (Form_pg_type) GETSTRUCT(tp);
     176        1266 :     desc->attlen = typtup->typlen;
     177        1266 :     desc->attbyval = typtup->typbyval;
     178        1266 :     desc->attalign = typtup->typalign;
     179        1266 :     desc->attstorage = typtup->typstorage;
     180        1266 :     ReleaseSysCache(tp);
     181        1266 : }
     182             : 
     183             : /*
     184             :  * Fetch local cache of AM-specific info about the index, initializing it
     185             :  * if necessary
     186             :  */
     187             : SpGistCache *
     188     2653872 : spgGetCache(Relation index)
     189             : {
     190             :     SpGistCache *cache;
     191             : 
     192     2653872 :     if (index->rd_amcache == NULL)
     193             :     {
     194             :         Oid         atttype;
     195             :         spgConfigIn in;
     196             :         FmgrInfo   *procinfo;
     197             : 
     198         412 :         cache = MemoryContextAllocZero(index->rd_indexcxt,
     199             :                                        sizeof(SpGistCache));
     200             : 
     201             :         /* SPGiST must have one key column and can also have INCLUDE columns */
     202             :         Assert(IndexRelationGetNumberOfKeyAttributes(index) == 1);
     203             :         Assert(IndexRelationGetNumberOfAttributes(index) <= INDEX_MAX_KEYS);
     204             : 
     205             :         /*
     206             :          * Get the actual (well, nominal) data type of the key column.  We
     207             :          * pass this to the opclass config function so that polymorphic
     208             :          * opclasses are possible.
     209             :          */
     210         412 :         atttype = GetIndexInputType(index, spgKeyColumn + 1);
     211             : 
     212             :         /* Call the config function to get config info for the opclass */
     213         412 :         in.attType = atttype;
     214             : 
     215         412 :         procinfo = index_getprocinfo(index, 1, SPGIST_CONFIG_PROC);
     216         412 :         FunctionCall2Coll(procinfo,
     217         412 :                           index->rd_indcollation[spgKeyColumn],
     218             :                           PointerGetDatum(&in),
     219         412 :                           PointerGetDatum(&cache->config));
     220             : 
     221             :         /*
     222             :          * If leafType isn't specified, use the declared index column type,
     223             :          * which index.c will have derived from the opclass's opcintype.
     224             :          * (Although we now make spgvalidate.c warn if these aren't the same,
     225             :          * old user-defined opclasses may not set the STORAGE parameter
     226             :          * correctly, so believe leafType if it's given.)
     227             :          */
     228         412 :         if (!OidIsValid(cache->config.leafType))
     229             :         {
     230         382 :             cache->config.leafType =
     231         382 :                 TupleDescAttr(RelationGetDescr(index), spgKeyColumn)->atttypid;
     232             : 
     233             :             /*
     234             :              * If index column type is binary-coercible to atttype (for
     235             :              * example, it's a domain over atttype), treat it as plain atttype
     236             :              * to avoid thinking we need to compress.
     237             :              */
     238         396 :             if (cache->config.leafType != atttype &&
     239          14 :                 IsBinaryCoercible(cache->config.leafType, atttype))
     240          14 :                 cache->config.leafType = atttype;
     241             :         }
     242             : 
     243             :         /* Get the information we need about each relevant datatype */
     244         412 :         fillTypeDesc(&cache->attType, atttype);
     245             : 
     246         412 :         if (cache->config.leafType != atttype)
     247             :         {
     248          30 :             if (!OidIsValid(index_getprocid(index, 1, SPGIST_COMPRESS_PROC)))
     249           0 :                 ereport(ERROR,
     250             :                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     251             :                          errmsg("compress method must be defined when leaf type is different from input type")));
     252             : 
     253          30 :             fillTypeDesc(&cache->attLeafType, cache->config.leafType);
     254             :         }
     255             :         else
     256             :         {
     257             :             /* Save lookups in this common case */
     258         382 :             cache->attLeafType = cache->attType;
     259             :         }
     260             : 
     261         412 :         fillTypeDesc(&cache->attPrefixType, cache->config.prefixType);
     262         412 :         fillTypeDesc(&cache->attLabelType, cache->config.labelType);
     263             : 
     264             :         /*
     265             :          * Finally, if it's a real index (not a partitioned one), get the
     266             :          * lastUsedPages data from the metapage
     267             :          */
     268         412 :         if (index->rd_rel->relkind != RELKIND_PARTITIONED_INDEX)
     269             :         {
     270             :             Buffer      metabuffer;
     271             :             SpGistMetaPageData *metadata;
     272             : 
     273         406 :             metabuffer = ReadBuffer(index, SPGIST_METAPAGE_BLKNO);
     274         406 :             LockBuffer(metabuffer, BUFFER_LOCK_SHARE);
     275             : 
     276         406 :             metadata = SpGistPageGetMeta(BufferGetPage(metabuffer));
     277             : 
     278         406 :             if (metadata->magicNumber != SPGIST_MAGIC_NUMBER)
     279           0 :                 elog(ERROR, "index \"%s\" is not an SP-GiST index",
     280             :                      RelationGetRelationName(index));
     281             : 
     282         406 :             cache->lastUsedPages = metadata->lastUsedPages;
     283             : 
     284         406 :             UnlockReleaseBuffer(metabuffer);
     285             :         }
     286             : 
     287         412 :         index->rd_amcache = cache;
     288             :     }
     289             :     else
     290             :     {
     291             :         /* assume it's up to date */
     292     2653460 :         cache = (SpGistCache *) index->rd_amcache;
     293             :     }
     294             : 
     295     2653872 :     return cache;
     296             : }
     297             : 
     298             : /*
     299             :  * Compute a tuple descriptor for leaf tuples or index-only-scan result tuples.
     300             :  *
     301             :  * We can use the relcache's tupdesc as-is in many cases, and it's always
     302             :  * OK so far as any INCLUDE columns are concerned.  However, the entry for
     303             :  * the key column has to match leafType in the first case or attType in the
     304             :  * second case.  While the relcache's tupdesc *should* show leafType, this
     305             :  * might not hold for legacy user-defined opclasses, since before v14 they
     306             :  * were not allowed to declare their true storage type in CREATE OPCLASS.
     307             :  * Also, attType can be different from what is in the relcache.
     308             :  *
     309             :  * This function gives back either a pointer to the relcache's tupdesc
     310             :  * if that is suitable, or a palloc'd copy that's been adjusted to match
     311             :  * the specified key column type.  We can avoid doing any catalog lookups
     312             :  * here by insisting that the caller pass an SpGistTypeDesc not just an OID.
     313             :  */
     314             : TupleDesc
     315      244896 : getSpGistTupleDesc(Relation index, SpGistTypeDesc *keyType)
     316             : {
     317             :     TupleDesc   outTupDesc;
     318             :     Form_pg_attribute att;
     319             : 
     320      489792 :     if (keyType->type ==
     321      244896 :         TupleDescAttr(RelationGetDescr(index), spgKeyColumn)->atttypid)
     322      244766 :         outTupDesc = RelationGetDescr(index);
     323             :     else
     324             :     {
     325         130 :         outTupDesc = CreateTupleDescCopy(RelationGetDescr(index));
     326         130 :         att = TupleDescAttr(outTupDesc, spgKeyColumn);
     327             :         /* It's sufficient to update the type-dependent fields of the column */
     328         130 :         att->atttypid = keyType->type;
     329         130 :         att->atttypmod = -1;
     330         130 :         att->attlen = keyType->attlen;
     331         130 :         att->attbyval = keyType->attbyval;
     332         130 :         att->attalign = keyType->attalign;
     333         130 :         att->attstorage = keyType->attstorage;
     334             :         /* We shouldn't need to bother with making these valid: */
     335         130 :         att->attcompression = InvalidCompressionMethod;
     336         130 :         att->attcollation = InvalidOid;
     337             :         /* In case we changed typlen, we'd better reset following offsets */
     338         146 :         for (int i = spgFirstIncludeColumn; i < outTupDesc->natts; i++)
     339          16 :             TupleDescCompactAttr(outTupDesc, i)->attcacheoff = -1;
     340             : 
     341         130 :         populate_compact_attribute(outTupDesc, spgKeyColumn);
     342             :     }
     343      244896 :     return outTupDesc;
     344             : }
     345             : 
     346             : /* Initialize SpGistState for working with the given index */
     347             : void
     348      243992 : initSpGistState(SpGistState *state, Relation index)
     349             : {
     350             :     SpGistCache *cache;
     351             : 
     352      243992 :     state->index = index;
     353             : 
     354             :     /* Get cached static information about index */
     355      243992 :     cache = spgGetCache(index);
     356             : 
     357      243992 :     state->config = cache->config;
     358      243992 :     state->attType = cache->attType;
     359      243992 :     state->attLeafType = cache->attLeafType;
     360      243992 :     state->attPrefixType = cache->attPrefixType;
     361      243992 :     state->attLabelType = cache->attLabelType;
     362             : 
     363             :     /* Ensure we have a valid descriptor for leaf tuples */
     364      243992 :     state->leafTupDesc = getSpGistTupleDesc(state->index, &state->attLeafType);
     365             : 
     366             :     /* Make workspace for constructing dead tuples */
     367      243992 :     state->deadTupleStorage = palloc0(SGDTSIZE);
     368             : 
     369             :     /*
     370             :      * Set horizon XID to use in redirection tuples.  Use our own XID if we
     371             :      * have one, else use InvalidTransactionId.  The latter case can happen in
     372             :      * VACUUM or REINDEX CONCURRENTLY, and in neither case would it be okay to
     373             :      * force an XID to be assigned.  VACUUM won't create any redirection
     374             :      * tuples anyway, but REINDEX CONCURRENTLY can.  Fortunately, REINDEX
     375             :      * CONCURRENTLY doesn't mark the index valid until the end, so there could
     376             :      * never be any concurrent scans "in flight" to a redirection tuple it has
     377             :      * inserted.  And it locks out VACUUM until the end, too.  So it's okay
     378             :      * for VACUUM to immediately expire a redirection tuple that contains an
     379             :      * invalid xid.
     380             :      */
     381      243992 :     state->redirectXid = GetTopTransactionIdIfAny();
     382             : 
     383             :     /* Assume we're not in an index build (spgbuild will override) */
     384      243992 :     state->isBuild = false;
     385      243992 : }
     386             : 
     387             : /*
     388             :  * Allocate a new page (either by recycling, or by extending the index file).
     389             :  *
     390             :  * The returned buffer is already pinned and exclusive-locked.
     391             :  * Caller is responsible for initializing the page by calling SpGistInitBuffer.
     392             :  */
     393             : Buffer
     394        6542 : SpGistNewBuffer(Relation index)
     395             : {
     396             :     Buffer      buffer;
     397             : 
     398             :     /* First, try to get a page from FSM */
     399             :     for (;;)
     400           0 :     {
     401        6542 :         BlockNumber blkno = GetFreeIndexPage(index);
     402             : 
     403        6542 :         if (blkno == InvalidBlockNumber)
     404        6534 :             break;              /* nothing known to FSM */
     405             : 
     406             :         /*
     407             :          * The fixed pages shouldn't ever be listed in FSM, but just in case
     408             :          * one is, ignore it.
     409             :          */
     410           8 :         if (SpGistBlockIsFixed(blkno))
     411           0 :             continue;
     412             : 
     413           8 :         buffer = ReadBuffer(index, blkno);
     414             : 
     415             :         /*
     416             :          * We have to guard against the possibility that someone else already
     417             :          * recycled this page; the buffer may be locked if so.
     418             :          */
     419           8 :         if (ConditionalLockBuffer(buffer))
     420             :         {
     421           8 :             Page        page = BufferGetPage(buffer);
     422             : 
     423           8 :             if (PageIsNew(page))
     424           2 :                 return buffer;  /* OK to use, if never initialized */
     425             : 
     426           6 :             if (SpGistPageIsDeleted(page) || PageIsEmpty(page))
     427           6 :                 return buffer;  /* OK to use */
     428             : 
     429           0 :             LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
     430             :         }
     431             : 
     432             :         /* Can't use it, so release buffer and try again */
     433           0 :         ReleaseBuffer(buffer);
     434             :     }
     435             : 
     436        6534 :     buffer = ExtendBufferedRel(BMR_REL(index), MAIN_FORKNUM, NULL,
     437             :                                EB_LOCK_FIRST);
     438             : 
     439        6534 :     return buffer;
     440             : }
     441             : 
     442             : /*
     443             :  * Update index metapage's lastUsedPages info from local cache, if possible
     444             :  *
     445             :  * Updating meta page isn't critical for index working, so
     446             :  * 1 use ConditionalLockBuffer to improve concurrency
     447             :  * 2 don't WAL-log metabuffer changes to decrease WAL traffic
     448             :  */
     449             : void
     450      242920 : SpGistUpdateMetaPage(Relation index)
     451             : {
     452      242920 :     SpGistCache *cache = (SpGistCache *) index->rd_amcache;
     453             : 
     454      242920 :     if (cache != NULL)
     455             :     {
     456             :         Buffer      metabuffer;
     457             : 
     458      242920 :         metabuffer = ReadBuffer(index, SPGIST_METAPAGE_BLKNO);
     459             : 
     460      242920 :         if (ConditionalLockBuffer(metabuffer))
     461             :         {
     462      242912 :             Page        metapage = BufferGetPage(metabuffer);
     463      242912 :             SpGistMetaPageData *metadata = SpGistPageGetMeta(metapage);
     464             : 
     465      242912 :             metadata->lastUsedPages = cache->lastUsedPages;
     466             : 
     467             :             /*
     468             :              * Set pd_lower just past the end of the metadata.  This is
     469             :              * essential, because without doing so, metadata will be lost if
     470             :              * xlog.c compresses the page.  (We must do this here because
     471             :              * pre-v11 versions of PG did not set the metapage's pd_lower
     472             :              * correctly, so a pg_upgraded index might contain the wrong
     473             :              * value.)
     474             :              */
     475      242912 :             ((PageHeader) metapage)->pd_lower =
     476      242912 :                 ((char *) metadata + sizeof(SpGistMetaPageData)) - (char *) metapage;
     477             : 
     478      242912 :             MarkBufferDirty(metabuffer);
     479      242912 :             UnlockReleaseBuffer(metabuffer);
     480             :         }
     481             :         else
     482             :         {
     483           8 :             ReleaseBuffer(metabuffer);
     484             :         }
     485             :     }
     486      242920 : }
     487             : 
     488             : /* Macro to select proper element of lastUsedPages cache depending on flags */
     489             : /* Masking flags with SPGIST_CACHED_PAGES is just for paranoia's sake */
     490             : #define GET_LUP(c, f)  (&(c)->lastUsedPages.cachedPage[((unsigned int) (f)) % SPGIST_CACHED_PAGES])
     491             : 
     492             : /*
     493             :  * Allocate and initialize a new buffer of the type and parity specified by
     494             :  * flags.  The returned buffer is already pinned and exclusive-locked.
     495             :  *
     496             :  * When requesting an inner page, if we get one with the wrong parity,
     497             :  * we just release the buffer and try again.  We will get a different page
     498             :  * because GetFreeIndexPage will have marked the page used in FSM.  The page
     499             :  * is entered in our local lastUsedPages cache, so there's some hope of
     500             :  * making use of it later in this session, but otherwise we rely on VACUUM
     501             :  * to eventually re-enter the page in FSM, making it available for recycling.
     502             :  * Note that such a page does not get marked dirty here, so unless it's used
     503             :  * fairly soon, the buffer will just get discarded and the page will remain
     504             :  * as it was on disk.
     505             :  *
     506             :  * When we return a buffer to the caller, the page is *not* entered into
     507             :  * the lastUsedPages cache; we expect the caller will do so after it's taken
     508             :  * whatever space it will use.  This is because after the caller has used up
     509             :  * some space, the page might have less space than whatever was cached already
     510             :  * so we'd rather not trash the old cache entry.
     511             :  */
     512             : static Buffer
     513        5832 : allocNewBuffer(Relation index, int flags)
     514             : {
     515        5832 :     SpGistCache *cache = spgGetCache(index);
     516        5832 :     uint16      pageflags = 0;
     517             : 
     518        5832 :     if (GBUF_REQ_LEAF(flags))
     519        5726 :         pageflags |= SPGIST_LEAF;
     520        5832 :     if (GBUF_REQ_NULLS(flags))
     521           0 :         pageflags |= SPGIST_NULLS;
     522             : 
     523             :     for (;;)
     524          86 :     {
     525             :         Buffer      buffer;
     526             : 
     527        5918 :         buffer = SpGistNewBuffer(index);
     528        5918 :         SpGistInitBuffer(buffer, pageflags);
     529             : 
     530        5918 :         if (pageflags & SPGIST_LEAF)
     531             :         {
     532             :             /* Leaf pages have no parity concerns, so just use it */
     533        5726 :             return buffer;
     534             :         }
     535             :         else
     536             :         {
     537         192 :             BlockNumber blkno = BufferGetBlockNumber(buffer);
     538         192 :             int         blkFlags = GBUF_INNER_PARITY(blkno);
     539             : 
     540         192 :             if ((flags & GBUF_PARITY_MASK) == blkFlags)
     541             :             {
     542             :                 /* Page has right parity, use it */
     543         106 :                 return buffer;
     544             :             }
     545             :             else
     546             :             {
     547             :                 /* Page has wrong parity, record it in cache and try again */
     548          86 :                 if (pageflags & SPGIST_NULLS)
     549           0 :                     blkFlags |= GBUF_NULLS;
     550          86 :                 cache->lastUsedPages.cachedPage[blkFlags].blkno = blkno;
     551          86 :                 cache->lastUsedPages.cachedPage[blkFlags].freeSpace =
     552          86 :                     PageGetExactFreeSpace(BufferGetPage(buffer));
     553          86 :                 UnlockReleaseBuffer(buffer);
     554             :             }
     555             :         }
     556             :     }
     557             : }
     558             : 
     559             : /*
     560             :  * Get a buffer of the type and parity specified by flags, having at least
     561             :  * as much free space as indicated by needSpace.  We use the lastUsedPages
     562             :  * cache to assign the same buffer previously requested when possible.
     563             :  * The returned buffer is already pinned and exclusive-locked.
     564             :  *
     565             :  * *isNew is set true if the page was initialized here, false if it was
     566             :  * already valid.
     567             :  */
     568             : Buffer
     569       10906 : SpGistGetBuffer(Relation index, int flags, int needSpace, bool *isNew)
     570             : {
     571       10906 :     SpGistCache *cache = spgGetCache(index);
     572             :     SpGistLastUsedPage *lup;
     573             : 
     574             :     /* Bail out if even an empty page wouldn't meet the demand */
     575       10906 :     if (needSpace > SPGIST_PAGE_CAPACITY)
     576           0 :         elog(ERROR, "desired SPGiST tuple size is too big");
     577             : 
     578             :     /*
     579             :      * If possible, increase the space request to include relation's
     580             :      * fillfactor.  This ensures that when we add unrelated tuples to a page,
     581             :      * we try to keep 100-fillfactor% available for adding tuples that are
     582             :      * related to the ones already on it.  But fillfactor mustn't cause an
     583             :      * error for requests that would otherwise be legal.
     584             :      */
     585       10906 :     needSpace += SpGistGetTargetPageFreeSpace(index);
     586       10906 :     needSpace = Min(needSpace, SPGIST_PAGE_CAPACITY);
     587             : 
     588             :     /* Get the cache entry for this flags setting */
     589       10906 :     lup = GET_LUP(cache, flags);
     590             : 
     591             :     /* If we have nothing cached, just turn it over to allocNewBuffer */
     592       10906 :     if (lup->blkno == InvalidBlockNumber)
     593             :     {
     594         182 :         *isNew = true;
     595         182 :         return allocNewBuffer(index, flags);
     596             :     }
     597             : 
     598             :     /* fixed pages should never be in cache */
     599             :     Assert(!SpGistBlockIsFixed(lup->blkno));
     600             : 
     601             :     /* If cached freeSpace isn't enough, don't bother looking at the page */
     602       10724 :     if (lup->freeSpace >= needSpace)
     603             :     {
     604             :         Buffer      buffer;
     605             :         Page        page;
     606             : 
     607        5076 :         buffer = ReadBuffer(index, lup->blkno);
     608             : 
     609        5076 :         if (!ConditionalLockBuffer(buffer))
     610             :         {
     611             :             /*
     612             :              * buffer is locked by another process, so return a new buffer
     613             :              */
     614           2 :             ReleaseBuffer(buffer);
     615           2 :             *isNew = true;
     616           2 :             return allocNewBuffer(index, flags);
     617             :         }
     618             : 
     619        5074 :         page = BufferGetPage(buffer);
     620             : 
     621        5074 :         if (PageIsNew(page) || SpGistPageIsDeleted(page) || PageIsEmpty(page))
     622             :         {
     623             :             /* OK to initialize the page */
     624         188 :             uint16      pageflags = 0;
     625             : 
     626         188 :             if (GBUF_REQ_LEAF(flags))
     627         182 :                 pageflags |= SPGIST_LEAF;
     628         188 :             if (GBUF_REQ_NULLS(flags))
     629           0 :                 pageflags |= SPGIST_NULLS;
     630         188 :             SpGistInitBuffer(buffer, pageflags);
     631         188 :             lup->freeSpace = PageGetExactFreeSpace(page) - needSpace;
     632         188 :             *isNew = true;
     633         188 :             return buffer;
     634             :         }
     635             : 
     636             :         /*
     637             :          * Check that page is of right type and has enough space.  We must
     638             :          * recheck this since our cache isn't necessarily up to date.
     639             :          */
     640        9772 :         if ((GBUF_REQ_LEAF(flags) ? SpGistPageIsLeaf(page) : !SpGistPageIsLeaf(page)) &&
     641        4886 :             (GBUF_REQ_NULLS(flags) ? SpGistPageStoresNulls(page) : !SpGistPageStoresNulls(page)))
     642             :         {
     643        4886 :             int         freeSpace = PageGetExactFreeSpace(page);
     644             : 
     645        4886 :             if (freeSpace >= needSpace)
     646             :             {
     647             :                 /* Success, update freespace info and return the buffer */
     648        4886 :                 lup->freeSpace = freeSpace - needSpace;
     649        4886 :                 *isNew = false;
     650        4886 :                 return buffer;
     651             :             }
     652             :         }
     653             : 
     654             :         /*
     655             :          * fallback to allocation of new buffer
     656             :          */
     657           0 :         UnlockReleaseBuffer(buffer);
     658             :     }
     659             : 
     660             :     /* No success with cache, so return a new buffer */
     661        5648 :     *isNew = true;
     662        5648 :     return allocNewBuffer(index, flags);
     663             : }
     664             : 
     665             : /*
     666             :  * Update lastUsedPages cache when done modifying a page.
     667             :  *
     668             :  * We update the appropriate cache entry if it already contained this page
     669             :  * (its freeSpace is likely obsolete), or if this page has more space than
     670             :  * whatever we had cached.
     671             :  */
     672             : void
     673     2391316 : SpGistSetLastUsedPage(Relation index, Buffer buffer)
     674             : {
     675     2391316 :     SpGistCache *cache = spgGetCache(index);
     676             :     SpGistLastUsedPage *lup;
     677             :     int         freeSpace;
     678     2391316 :     Page        page = BufferGetPage(buffer);
     679     2391316 :     BlockNumber blkno = BufferGetBlockNumber(buffer);
     680             :     int         flags;
     681             : 
     682             :     /* Never enter fixed pages (root pages) in cache, though */
     683     2391316 :     if (SpGistBlockIsFixed(blkno))
     684      805560 :         return;
     685             : 
     686     1585756 :     if (SpGistPageIsLeaf(page))
     687      800854 :         flags = GBUF_LEAF;
     688             :     else
     689      784902 :         flags = GBUF_INNER_PARITY(blkno);
     690     1585756 :     if (SpGistPageStoresNulls(page))
     691           0 :         flags |= GBUF_NULLS;
     692             : 
     693     1585756 :     lup = GET_LUP(cache, flags);
     694             : 
     695     1585756 :     freeSpace = PageGetExactFreeSpace(page);
     696     1585756 :     if (lup->blkno == InvalidBlockNumber || lup->blkno == blkno ||
     697      420076 :         lup->freeSpace < freeSpace)
     698             :     {
     699     1174328 :         lup->blkno = blkno;
     700     1174328 :         lup->freeSpace = freeSpace;
     701             :     }
     702             : }
     703             : 
     704             : /*
     705             :  * Initialize an SPGiST page to empty, with specified flags
     706             :  */
     707             : void
     708        7626 : SpGistInitPage(Page page, uint16 f)
     709             : {
     710             :     SpGistPageOpaque opaque;
     711             : 
     712        7626 :     PageInit(page, BLCKSZ, sizeof(SpGistPageOpaqueData));
     713        7626 :     opaque = SpGistPageGetOpaque(page);
     714        7626 :     opaque->flags = f;
     715        7626 :     opaque->spgist_page_id = SPGIST_PAGE_ID;
     716        7626 : }
     717             : 
     718             : /*
     719             :  * Initialize a buffer's page to empty, with specified flags
     720             :  */
     721             : void
     722        7394 : SpGistInitBuffer(Buffer b, uint16 f)
     723             : {
     724             :     Assert(BufferGetPageSize(b) == BLCKSZ);
     725        7394 :     SpGistInitPage(BufferGetPage(b), f);
     726        7394 : }
     727             : 
     728             : /*
     729             :  * Initialize metadata page
     730             :  */
     731             : void
     732         216 : SpGistInitMetapage(Page page)
     733             : {
     734             :     SpGistMetaPageData *metadata;
     735             :     int         i;
     736             : 
     737         216 :     SpGistInitPage(page, SPGIST_META);
     738         216 :     metadata = SpGistPageGetMeta(page);
     739         216 :     memset(metadata, 0, sizeof(SpGistMetaPageData));
     740         216 :     metadata->magicNumber = SPGIST_MAGIC_NUMBER;
     741             : 
     742             :     /* initialize last-used-page cache to empty */
     743        1944 :     for (i = 0; i < SPGIST_CACHED_PAGES; i++)
     744        1728 :         metadata->lastUsedPages.cachedPage[i].blkno = InvalidBlockNumber;
     745             : 
     746             :     /*
     747             :      * Set pd_lower just past the end of the metadata.  This is essential,
     748             :      * because without doing so, metadata will be lost if xlog.c compresses
     749             :      * the page.
     750             :      */
     751         216 :     ((PageHeader) page)->pd_lower =
     752         216 :         ((char *) metadata + sizeof(SpGistMetaPageData)) - (char *) page;
     753         216 : }
     754             : 
     755             : /*
     756             :  * reloptions processing for SPGiST
     757             :  */
     758             : bytea *
     759         128 : spgoptions(Datum reloptions, bool validate)
     760             : {
     761             :     static const relopt_parse_elt tab[] = {
     762             :         {"fillfactor", RELOPT_TYPE_INT, offsetof(SpGistOptions, fillfactor)},
     763             :     };
     764             : 
     765         128 :     return (bytea *) build_reloptions(reloptions, validate,
     766             :                                       RELOPT_KIND_SPGIST,
     767             :                                       sizeof(SpGistOptions),
     768             :                                       tab, lengthof(tab));
     769             : }
     770             : 
     771             : /*
     772             :  * Get the space needed to store a non-null datum of the indicated type
     773             :  * in an inner tuple (that is, as a prefix or node label).
     774             :  * Note the result is already rounded up to a MAXALIGN boundary.
     775             :  * Here we follow the convention that pass-by-val types are just stored
     776             :  * in their Datum representation (compare memcpyInnerDatum).
     777             :  */
     778             : unsigned int
     779       12330 : SpGistGetInnerTypeSize(SpGistTypeDesc *att, Datum datum)
     780             : {
     781             :     unsigned int size;
     782             : 
     783       12330 :     if (att->attbyval)
     784        6484 :         size = sizeof(Datum);
     785        5846 :     else if (att->attlen > 0)
     786        3962 :         size = att->attlen;
     787             :     else
     788        1884 :         size = VARSIZE_ANY(datum);
     789             : 
     790       12330 :     return MAXALIGN(size);
     791             : }
     792             : 
     793             : /*
     794             :  * Copy the given non-null datum to *target, in the inner-tuple case
     795             :  */
     796             : static void
     797       12330 : memcpyInnerDatum(void *target, SpGistTypeDesc *att, Datum datum)
     798             : {
     799             :     unsigned int size;
     800             : 
     801       12330 :     if (att->attbyval)
     802             :     {
     803        6484 :         memcpy(target, &datum, sizeof(Datum));
     804             :     }
     805             :     else
     806             :     {
     807        5846 :         size = (att->attlen > 0) ? att->attlen : VARSIZE_ANY(datum);
     808        5846 :         memcpy(target, DatumGetPointer(datum), size);
     809             :     }
     810       12330 : }
     811             : 
     812             : /*
     813             :  * Compute space required for a leaf tuple holding the given data.
     814             :  *
     815             :  * This must match the size-calculation portion of spgFormLeafTuple.
     816             :  */
     817             : Size
     818    19455914 : SpGistGetLeafTupleSize(TupleDesc tupleDescriptor,
     819             :                        const Datum *datums, const bool *isnulls)
     820             : {
     821             :     Size        size;
     822             :     Size        data_size;
     823    19455914 :     bool        needs_null_mask = false;
     824    19455914 :     int         natts = tupleDescriptor->natts;
     825             : 
     826             :     /*
     827             :      * Decide whether we need a nulls bitmask.
     828             :      *
     829             :      * If there is only a key attribute (natts == 1), never use a bitmask, for
     830             :      * compatibility with the pre-v14 layout of leaf tuples.  Otherwise, we
     831             :      * need one if any attribute is null.
     832             :      */
     833    19455914 :     if (natts > 1)
     834             :     {
     835      993914 :         for (int i = 0; i < natts; i++)
     836             :         {
     837      680744 :             if (isnulls[i])
     838             :             {
     839       18318 :                 needs_null_mask = true;
     840       18318 :                 break;
     841             :             }
     842             :         }
     843             :     }
     844             : 
     845             :     /*
     846             :      * Calculate size of the data part; same as for heap tuples.
     847             :      */
     848    19455914 :     data_size = heap_compute_data_size(tupleDescriptor, datums, isnulls);
     849             : 
     850             :     /*
     851             :      * Compute total size.
     852             :      */
     853    19455914 :     size = SGLTHDRSZ(needs_null_mask);
     854    19455914 :     size += data_size;
     855    19455914 :     size = MAXALIGN(size);
     856             : 
     857             :     /*
     858             :      * Ensure that we can replace the tuple with a dead tuple later. This test
     859             :      * is unnecessary when there are any non-null attributes, but be safe.
     860             :      */
     861    19455914 :     if (size < SGDTSIZE)
     862           0 :         size = SGDTSIZE;
     863             : 
     864    19455914 :     return size;
     865             : }
     866             : 
     867             : /*
     868             :  * Construct a leaf tuple containing the given heap TID and datum values
     869             :  */
     870             : SpGistLeafTuple
     871     1527250 : spgFormLeafTuple(SpGistState *state, ItemPointer heapPtr,
     872             :                  const Datum *datums, const bool *isnulls)
     873             : {
     874             :     SpGistLeafTuple tup;
     875     1527250 :     TupleDesc   tupleDescriptor = state->leafTupDesc;
     876             :     Size        size;
     877             :     Size        hoff;
     878             :     Size        data_size;
     879     1527250 :     bool        needs_null_mask = false;
     880     1527250 :     int         natts = tupleDescriptor->natts;
     881             :     char       *tp;             /* ptr to tuple data */
     882     1527250 :     uint16      tupmask = 0;    /* unused heap_fill_tuple output */
     883             : 
     884             :     /*
     885             :      * Decide whether we need a nulls bitmask.
     886             :      *
     887             :      * If there is only a key attribute (natts == 1), never use a bitmask, for
     888             :      * compatibility with the pre-v14 layout of leaf tuples.  Otherwise, we
     889             :      * need one if any attribute is null.
     890             :      */
     891     1527250 :     if (natts > 1)
     892             :     {
     893      420754 :         for (int i = 0; i < natts; i++)
     894             :         {
     895      292402 :             if (isnulls[i])
     896             :             {
     897       12064 :                 needs_null_mask = true;
     898       12064 :                 break;
     899             :             }
     900             :         }
     901             :     }
     902             : 
     903             :     /*
     904             :      * Calculate size of the data part; same as for heap tuples.
     905             :      */
     906     1527250 :     data_size = heap_compute_data_size(tupleDescriptor, datums, isnulls);
     907             : 
     908             :     /*
     909             :      * Compute total size.
     910             :      */
     911     1527250 :     hoff = SGLTHDRSZ(needs_null_mask);
     912     1527250 :     size = hoff + data_size;
     913     1527250 :     size = MAXALIGN(size);
     914             : 
     915             :     /*
     916             :      * Ensure that we can replace the tuple with a dead tuple later. This test
     917             :      * is unnecessary when there are any non-null attributes, but be safe.
     918             :      */
     919     1527250 :     if (size < SGDTSIZE)
     920           0 :         size = SGDTSIZE;
     921             : 
     922             :     /* OK, form the tuple */
     923     1527250 :     tup = (SpGistLeafTuple) palloc0(size);
     924             : 
     925     1527250 :     tup->size = size;
     926     1527250 :     SGLT_SET_NEXTOFFSET(tup, InvalidOffsetNumber);
     927     1527250 :     tup->heapPtr = *heapPtr;
     928             : 
     929     1527250 :     tp = (char *) tup + hoff;
     930             : 
     931     1527250 :     if (needs_null_mask)
     932             :     {
     933             :         bits8      *bp;         /* ptr to null bitmap in tuple */
     934             : 
     935             :         /* Set nullmask presence bit in SpGistLeafTuple header */
     936       12064 :         SGLT_SET_HASNULLMASK(tup, true);
     937             :         /* Fill the data area and null mask */
     938       12064 :         bp = (bits8 *) ((char *) tup + sizeof(SpGistLeafTupleData));
     939       12064 :         heap_fill_tuple(tupleDescriptor, datums, isnulls, tp, data_size,
     940             :                         &tupmask, bp);
     941             :     }
     942     1515186 :     else if (natts > 1 || !isnulls[spgKeyColumn])
     943             :     {
     944             :         /* Fill data area only */
     945     1515114 :         heap_fill_tuple(tupleDescriptor, datums, isnulls, tp, data_size,
     946             :                         &tupmask, (bits8 *) NULL);
     947             :     }
     948             :     /* otherwise we have no data, nor a bitmap, to fill */
     949             : 
     950     1527250 :     return tup;
     951             : }
     952             : 
     953             : /*
     954             :  * Construct a node (to go into an inner tuple) containing the given label
     955             :  *
     956             :  * Note that the node's downlink is just set invalid here.  Caller will fill
     957             :  * it in later.
     958             :  */
     959             : SpGistNodeTuple
     960       40554 : spgFormNodeTuple(SpGistState *state, Datum label, bool isnull)
     961             : {
     962             :     SpGistNodeTuple tup;
     963             :     unsigned int size;
     964       40554 :     unsigned short infomask = 0;
     965             : 
     966             :     /* compute space needed (note result is already maxaligned) */
     967       40554 :     size = SGNTHDRSZ;
     968       40554 :     if (!isnull)
     969        5794 :         size += SpGistGetInnerTypeSize(&state->attLabelType, label);
     970             : 
     971             :     /*
     972             :      * Here we make sure that the size will fit in the field reserved for it
     973             :      * in t_info.
     974             :      */
     975       40554 :     if ((size & INDEX_SIZE_MASK) != size)
     976           0 :         ereport(ERROR,
     977             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
     978             :                  errmsg("index row requires %zu bytes, maximum size is %zu",
     979             :                         (Size) size, (Size) INDEX_SIZE_MASK)));
     980             : 
     981       40554 :     tup = (SpGistNodeTuple) palloc0(size);
     982             : 
     983       40554 :     if (isnull)
     984       34760 :         infomask |= INDEX_NULL_MASK;
     985             :     /* we don't bother setting the INDEX_VAR_MASK bit */
     986       40554 :     infomask |= size;
     987       40554 :     tup->t_info = infomask;
     988             : 
     989             :     /* The TID field will be filled in later */
     990       40554 :     ItemPointerSetInvalid(&tup->t_tid);
     991             : 
     992       40554 :     if (!isnull)
     993        5794 :         memcpyInnerDatum(SGNTDATAPTR(tup), &state->attLabelType, label);
     994             : 
     995       40554 :     return tup;
     996             : }
     997             : 
     998             : /*
     999             :  * Construct an inner tuple containing the given prefix and node array
    1000             :  */
    1001             : SpGistInnerTuple
    1002        8530 : spgFormInnerTuple(SpGistState *state, bool hasPrefix, Datum prefix,
    1003             :                   int nNodes, SpGistNodeTuple *nodes)
    1004             : {
    1005             :     SpGistInnerTuple tup;
    1006             :     unsigned int size;
    1007             :     unsigned int prefixSize;
    1008             :     int         i;
    1009             :     char       *ptr;
    1010             : 
    1011             :     /* Compute size needed */
    1012        8530 :     if (hasPrefix)
    1013        6536 :         prefixSize = SpGistGetInnerTypeSize(&state->attPrefixType, prefix);
    1014             :     else
    1015        1994 :         prefixSize = 0;
    1016             : 
    1017        8530 :     size = SGITHDRSZ + prefixSize;
    1018             : 
    1019             :     /* Note: we rely on node tuple sizes to be maxaligned already */
    1020       59374 :     for (i = 0; i < nNodes; i++)
    1021       50844 :         size += IndexTupleSize(nodes[i]);
    1022             : 
    1023             :     /*
    1024             :      * Ensure that we can replace the tuple with a dead tuple later.  This
    1025             :      * test is unnecessary given current tuple layouts, but let's be safe.
    1026             :      */
    1027        8530 :     if (size < SGDTSIZE)
    1028           0 :         size = SGDTSIZE;
    1029             : 
    1030             :     /*
    1031             :      * Inner tuple should be small enough to fit on a page
    1032             :      */
    1033        8530 :     if (size > SPGIST_PAGE_CAPACITY - sizeof(ItemIdData))
    1034           0 :         ereport(ERROR,
    1035             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
    1036             :                  errmsg("SP-GiST inner tuple size %zu exceeds maximum %zu",
    1037             :                         (Size) size,
    1038             :                         SPGIST_PAGE_CAPACITY - sizeof(ItemIdData)),
    1039             :                  errhint("Values larger than a buffer page cannot be indexed.")));
    1040             : 
    1041             :     /*
    1042             :      * Check for overflow of header fields --- probably can't fail if the
    1043             :      * above succeeded, but let's be paranoid
    1044             :      */
    1045        8530 :     if (size > SGITMAXSIZE ||
    1046        8530 :         prefixSize > SGITMAXPREFIXSIZE ||
    1047             :         nNodes > SGITMAXNNODES)
    1048           0 :         elog(ERROR, "SPGiST inner tuple header field is too small");
    1049             : 
    1050             :     /* OK, form the tuple */
    1051        8530 :     tup = (SpGistInnerTuple) palloc0(size);
    1052             : 
    1053        8530 :     tup->nNodes = nNodes;
    1054        8530 :     tup->prefixSize = prefixSize;
    1055        8530 :     tup->size = size;
    1056             : 
    1057        8530 :     if (hasPrefix)
    1058        6536 :         memcpyInnerDatum(SGITDATAPTR(tup), &state->attPrefixType, prefix);
    1059             : 
    1060        8530 :     ptr = (char *) SGITNODEPTR(tup);
    1061             : 
    1062       59374 :     for (i = 0; i < nNodes; i++)
    1063             :     {
    1064       50844 :         SpGistNodeTuple node = nodes[i];
    1065             : 
    1066       50844 :         memcpy(ptr, node, IndexTupleSize(node));
    1067       50844 :         ptr += IndexTupleSize(node);
    1068             :     }
    1069             : 
    1070        8530 :     return tup;
    1071             : }
    1072             : 
    1073             : /*
    1074             :  * Construct a "dead" tuple to replace a tuple being deleted.
    1075             :  *
    1076             :  * The state can be SPGIST_REDIRECT, SPGIST_DEAD, or SPGIST_PLACEHOLDER.
    1077             :  * For a REDIRECT tuple, a pointer (blkno+offset) must be supplied, and
    1078             :  * the xid field is filled in automatically.
    1079             :  *
    1080             :  * This is called in critical sections, so we don't use palloc; the tuple
    1081             :  * is built in preallocated storage.  It should be copied before another
    1082             :  * call with different parameters can occur.
    1083             :  */
    1084             : SpGistDeadTuple
    1085       13540 : spgFormDeadTuple(SpGistState *state, int tupstate,
    1086             :                  BlockNumber blkno, OffsetNumber offnum)
    1087             : {
    1088       13540 :     SpGistDeadTuple tuple = (SpGistDeadTuple) state->deadTupleStorage;
    1089             : 
    1090       13540 :     tuple->tupstate = tupstate;
    1091       13540 :     tuple->size = SGDTSIZE;
    1092       13540 :     SGLT_SET_NEXTOFFSET(tuple, InvalidOffsetNumber);
    1093             : 
    1094       13540 :     if (tupstate == SPGIST_REDIRECT)
    1095             :     {
    1096        2384 :         ItemPointerSet(&tuple->pointer, blkno, offnum);
    1097        2384 :         tuple->xid = state->redirectXid;
    1098             :     }
    1099             :     else
    1100             :     {
    1101       11156 :         ItemPointerSetInvalid(&tuple->pointer);
    1102       11156 :         tuple->xid = InvalidTransactionId;
    1103             :     }
    1104             : 
    1105       13540 :     return tuple;
    1106             : }
    1107             : 
    1108             : /*
    1109             :  * Convert an SPGiST leaf tuple into Datum/isnull arrays.
    1110             :  *
    1111             :  * The caller must allocate sufficient storage for the output arrays.
    1112             :  * (INDEX_MAX_KEYS entries should be enough.)
    1113             :  */
    1114             : void
    1115       60848 : spgDeformLeafTuple(SpGistLeafTuple tup, TupleDesc tupleDescriptor,
    1116             :                    Datum *datums, bool *isnulls, bool keyColumnIsNull)
    1117             : {
    1118       60848 :     bool        hasNullsMask = SGLT_GET_HASNULLMASK(tup);
    1119             :     char       *tp;             /* ptr to tuple data */
    1120             :     bits8      *bp;             /* ptr to null bitmap in tuple */
    1121             : 
    1122       60848 :     if (keyColumnIsNull && tupleDescriptor->natts == 1)
    1123             :     {
    1124             :         /*
    1125             :          * Trivial case: there is only the key attribute and we're in a nulls
    1126             :          * tree.  The hasNullsMask bit in the tuple header should not be set
    1127             :          * (and thus we can't use index_deform_tuple_internal), but
    1128             :          * nonetheless the result is NULL.
    1129             :          *
    1130             :          * Note: currently this is dead code, because noplace calls this when
    1131             :          * there is only the key attribute.  But we should cover the case.
    1132             :          */
    1133             :         Assert(!hasNullsMask);
    1134             : 
    1135           0 :         datums[spgKeyColumn] = (Datum) 0;
    1136           0 :         isnulls[spgKeyColumn] = true;
    1137           0 :         return;
    1138             :     }
    1139             : 
    1140       60848 :     tp = (char *) tup + SGLTHDRSZ(hasNullsMask);
    1141       60848 :     bp = (bits8 *) ((char *) tup + sizeof(SpGistLeafTupleData));
    1142             : 
    1143       60848 :     index_deform_tuple_internal(tupleDescriptor,
    1144             :                                 datums, isnulls,
    1145             :                                 tp, bp, hasNullsMask);
    1146             : 
    1147             :     /*
    1148             :      * Key column isnull value from the tuple should be consistent with
    1149             :      * keyColumnIsNull flag from the caller.
    1150             :      */
    1151             :     Assert(keyColumnIsNull == isnulls[spgKeyColumn]);
    1152             : }
    1153             : 
    1154             : /*
    1155             :  * Extract the label datums of the nodes within innerTuple
    1156             :  *
    1157             :  * Returns NULL if label datums are NULLs
    1158             :  */
    1159             : Datum *
    1160    18676896 : spgExtractNodeLabels(SpGistState *state, SpGistInnerTuple innerTuple)
    1161             : {
    1162             :     Datum      *nodeLabels;
    1163             :     int         i;
    1164             :     SpGistNodeTuple node;
    1165             : 
    1166             :     /* Either all the labels must be NULL, or none. */
    1167    18676896 :     node = SGITNODEPTR(innerTuple);
    1168    18676896 :     if (IndexTupleHasNulls(node))
    1169             :     {
    1170   100677996 :         SGITITERATE(innerTuple, i, node)
    1171             :         {
    1172    82234966 :             if (!IndexTupleHasNulls(node))
    1173           0 :                 elog(ERROR, "some but not all node labels are null in SPGiST inner tuple");
    1174             :         }
    1175             :         /* They're all null, so just return NULL */
    1176    18443030 :         return NULL;
    1177             :     }
    1178             :     else
    1179             :     {
    1180      233866 :         nodeLabels = (Datum *) palloc(sizeof(Datum) * innerTuple->nNodes);
    1181     2683416 :         SGITITERATE(innerTuple, i, node)
    1182             :         {
    1183     2449550 :             if (IndexTupleHasNulls(node))
    1184           0 :                 elog(ERROR, "some but not all node labels are null in SPGiST inner tuple");
    1185     2449550 :             nodeLabels[i] = SGNTDATUM(node, state);
    1186             :         }
    1187      233866 :         return nodeLabels;
    1188             :     }
    1189             : }
    1190             : 
    1191             : /*
    1192             :  * Add a new item to the page, replacing a PLACEHOLDER item if possible.
    1193             :  * Return the location it's inserted at, or InvalidOffsetNumber on failure.
    1194             :  *
    1195             :  * If startOffset isn't NULL, we start searching for placeholders at
    1196             :  * *startOffset, and update that to the next place to search.  This is just
    1197             :  * an optimization for repeated insertions.
    1198             :  *
    1199             :  * If errorOK is false, we throw error when there's not enough room,
    1200             :  * rather than returning InvalidOffsetNumber.
    1201             :  */
    1202             : OffsetNumber
    1203     1618528 : SpGistPageAddNewItem(SpGistState *state, Page page, Item item, Size size,
    1204             :                      OffsetNumber *startOffset, bool errorOK)
    1205             : {
    1206     1618528 :     SpGistPageOpaque opaque = SpGistPageGetOpaque(page);
    1207             :     OffsetNumber i,
    1208             :                 maxoff,
    1209             :                 offnum;
    1210             : 
    1211     1618528 :     if (opaque->nPlaceholder > 0 &&
    1212      461796 :         PageGetExactFreeSpace(page) + SGDTSIZE >= MAXALIGN(size))
    1213             :     {
    1214             :         /* Try to replace a placeholder */
    1215      461796 :         maxoff = PageGetMaxOffsetNumber(page);
    1216      461796 :         offnum = InvalidOffsetNumber;
    1217             : 
    1218             :         for (;;)
    1219             :         {
    1220      461796 :             if (startOffset && *startOffset != InvalidOffsetNumber)
    1221      114926 :                 i = *startOffset;
    1222             :             else
    1223      346870 :                 i = FirstOffsetNumber;
    1224    31587354 :             for (; i <= maxoff; i++)
    1225             :             {
    1226    31587354 :                 SpGistDeadTuple it = (SpGistDeadTuple) PageGetItem(page,
    1227    31587354 :                                                                    PageGetItemId(page, i));
    1228             : 
    1229    31587354 :                 if (it->tupstate == SPGIST_PLACEHOLDER)
    1230             :                 {
    1231      461796 :                     offnum = i;
    1232      461796 :                     break;
    1233             :                 }
    1234             :             }
    1235             : 
    1236             :             /* Done if we found a placeholder */
    1237      461796 :             if (offnum != InvalidOffsetNumber)
    1238      461796 :                 break;
    1239             : 
    1240           0 :             if (startOffset && *startOffset != InvalidOffsetNumber)
    1241             :             {
    1242             :                 /* Hint was no good, re-search from beginning */
    1243           0 :                 *startOffset = InvalidOffsetNumber;
    1244           0 :                 continue;
    1245             :             }
    1246             : 
    1247             :             /* Hmm, no placeholder found? */
    1248           0 :             opaque->nPlaceholder = 0;
    1249           0 :             break;
    1250             :         }
    1251             : 
    1252      461796 :         if (offnum != InvalidOffsetNumber)
    1253             :         {
    1254             :             /* Replace the placeholder tuple */
    1255      461796 :             PageIndexTupleDelete(page, offnum);
    1256             : 
    1257      461796 :             offnum = PageAddItem(page, item, size, offnum, false, false);
    1258             : 
    1259             :             /*
    1260             :              * We should not have failed given the size check at the top of
    1261             :              * the function, but test anyway.  If we did fail, we must PANIC
    1262             :              * because we've already deleted the placeholder tuple, and
    1263             :              * there's no other way to keep the damage from getting to disk.
    1264             :              */
    1265      461796 :             if (offnum != InvalidOffsetNumber)
    1266             :             {
    1267             :                 Assert(opaque->nPlaceholder > 0);
    1268      461796 :                 opaque->nPlaceholder--;
    1269      461796 :                 if (startOffset)
    1270      117560 :                     *startOffset = offnum + 1;
    1271             :             }
    1272             :             else
    1273           0 :                 elog(PANIC, "failed to add item of size %zu to SPGiST index page",
    1274             :                      size);
    1275             : 
    1276      461796 :             return offnum;
    1277             :         }
    1278             :     }
    1279             : 
    1280             :     /* No luck in replacing a placeholder, so just add it to the page */
    1281     1156732 :     offnum = PageAddItem(page, item, size,
    1282             :                          InvalidOffsetNumber, false, false);
    1283             : 
    1284     1156732 :     if (offnum == InvalidOffsetNumber && !errorOK)
    1285           0 :         elog(ERROR, "failed to add item of size %zu to SPGiST index page",
    1286             :              size);
    1287             : 
    1288     1156732 :     return offnum;
    1289             : }
    1290             : 
    1291             : /*
    1292             :  *  spgproperty() -- Check boolean properties of indexes.
    1293             :  *
    1294             :  * This is optional for most AMs, but is required for SP-GiST because the core
    1295             :  * property code doesn't support AMPROP_DISTANCE_ORDERABLE.
    1296             :  */
    1297             : bool
    1298         186 : spgproperty(Oid index_oid, int attno,
    1299             :             IndexAMProperty prop, const char *propname,
    1300             :             bool *res, bool *isnull)
    1301             : {
    1302             :     Oid         opclass,
    1303             :                 opfamily,
    1304             :                 opcintype;
    1305             :     CatCList   *catlist;
    1306             :     int         i;
    1307             : 
    1308             :     /* Only answer column-level inquiries */
    1309         186 :     if (attno == 0)
    1310          66 :         return false;
    1311             : 
    1312         120 :     switch (prop)
    1313             :     {
    1314          12 :         case AMPROP_DISTANCE_ORDERABLE:
    1315          12 :             break;
    1316         108 :         default:
    1317         108 :             return false;
    1318             :     }
    1319             : 
    1320             :     /*
    1321             :      * Currently, SP-GiST distance-ordered scans require that there be a
    1322             :      * distance operator in the opclass with the default types. So we assume
    1323             :      * that if such an operator exists, then there's a reason for it.
    1324             :      */
    1325             : 
    1326             :     /* First we need to know the column's opclass. */
    1327          12 :     opclass = get_index_column_opclass(index_oid, attno);
    1328          12 :     if (!OidIsValid(opclass))
    1329             :     {
    1330           0 :         *isnull = true;
    1331           0 :         return true;
    1332             :     }
    1333             : 
    1334             :     /* Now look up the opclass family and input datatype. */
    1335          12 :     if (!get_opclass_opfamily_and_input_type(opclass, &opfamily, &opcintype))
    1336             :     {
    1337           0 :         *isnull = true;
    1338           0 :         return true;
    1339             :     }
    1340             : 
    1341             :     /* And now we can check whether the operator is provided. */
    1342          12 :     catlist = SearchSysCacheList1(AMOPSTRATEGY,
    1343             :                                   ObjectIdGetDatum(opfamily));
    1344             : 
    1345          12 :     *res = false;
    1346             : 
    1347         102 :     for (i = 0; i < catlist->n_members; i++)
    1348             :     {
    1349          96 :         HeapTuple   amoptup = &catlist->members[i]->tuple;
    1350          96 :         Form_pg_amop amopform = (Form_pg_amop) GETSTRUCT(amoptup);
    1351             : 
    1352          96 :         if (amopform->amoppurpose == AMOP_ORDER &&
    1353           6 :             (amopform->amoplefttype == opcintype ||
    1354           6 :              amopform->amoprighttype == opcintype) &&
    1355           6 :             opfamily_can_sort_type(amopform->amopsortfamily,
    1356             :                                    get_op_rettype(amopform->amopopr)))
    1357             :         {
    1358           6 :             *res = true;
    1359           6 :             break;
    1360             :         }
    1361             :     }
    1362             : 
    1363          12 :     ReleaseSysCacheList(catlist);
    1364             : 
    1365          12 :     *isnull = false;
    1366             : 
    1367          12 :     return true;
    1368             : }

Generated by: LCOV version 1.14