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

Generated by: LCOV version 1.14