LCOV - code coverage report
Current view: top level - src/backend/access/spgist - spgutils.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13beta1 Lines: 340 377 90.2 %
Date: 2020-06-03 10:06:28 Functions: 22 22 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-2020, 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/transam.h"
      23             : #include "access/xact.h"
      24             : #include "catalog/pg_amop.h"
      25             : #include "commands/vacuum.h"
      26             : #include "storage/bufmgr.h"
      27             : #include "storage/indexfsm.h"
      28             : #include "storage/lmgr.h"
      29             : #include "utils/builtins.h"
      30             : #include "utils/catcache.h"
      31             : #include "utils/index_selfuncs.h"
      32             : #include "utils/lsyscache.h"
      33             : #include "utils/syscache.h"
      34             : 
      35             : 
      36             : /*
      37             :  * SP-GiST handler function: return IndexAmRoutine with access method parameters
      38             :  * and callbacks.
      39             :  */
      40             : Datum
      41         652 : spghandler(PG_FUNCTION_ARGS)
      42             : {
      43         652 :     IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);
      44             : 
      45         652 :     amroutine->amstrategies = 0;
      46         652 :     amroutine->amsupport = SPGISTNProc;
      47         652 :     amroutine->amoptsprocnum = SPGIST_OPTIONS_PROC;
      48         652 :     amroutine->amcanorder = false;
      49         652 :     amroutine->amcanorderbyop = true;
      50         652 :     amroutine->amcanbackward = false;
      51         652 :     amroutine->amcanunique = false;
      52         652 :     amroutine->amcanmulticol = false;
      53         652 :     amroutine->amoptionalkey = true;
      54         652 :     amroutine->amsearcharray = false;
      55         652 :     amroutine->amsearchnulls = true;
      56         652 :     amroutine->amstorage = false;
      57         652 :     amroutine->amclusterable = false;
      58         652 :     amroutine->ampredlocks = false;
      59         652 :     amroutine->amcanparallel = false;
      60         652 :     amroutine->amcaninclude = false;
      61         652 :     amroutine->amusemaintenanceworkmem = false;
      62         652 :     amroutine->amparallelvacuumoptions =
      63             :         VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_COND_CLEANUP;
      64         652 :     amroutine->amkeytype = InvalidOid;
      65             : 
      66         652 :     amroutine->ambuild = spgbuild;
      67         652 :     amroutine->ambuildempty = spgbuildempty;
      68         652 :     amroutine->aminsert = spginsert;
      69         652 :     amroutine->ambulkdelete = spgbulkdelete;
      70         652 :     amroutine->amvacuumcleanup = spgvacuumcleanup;
      71         652 :     amroutine->amcanreturn = spgcanreturn;
      72         652 :     amroutine->amcostestimate = spgcostestimate;
      73         652 :     amroutine->amoptions = spgoptions;
      74         652 :     amroutine->amproperty = spgproperty;
      75         652 :     amroutine->ambuildphasename = NULL;
      76         652 :     amroutine->amvalidate = spgvalidate;
      77         652 :     amroutine->ambeginscan = spgbeginscan;
      78         652 :     amroutine->amrescan = spgrescan;
      79         652 :     amroutine->amgettuple = spggettuple;
      80         652 :     amroutine->amgetbitmap = spggetbitmap;
      81         652 :     amroutine->amendscan = spgendscan;
      82         652 :     amroutine->ammarkpos = NULL;
      83         652 :     amroutine->amrestrpos = NULL;
      84         652 :     amroutine->amestimateparallelscan = NULL;
      85         652 :     amroutine->aminitparallelscan = NULL;
      86         652 :     amroutine->amparallelrescan = NULL;
      87             : 
      88         652 :     PG_RETURN_POINTER(amroutine);
      89             : }
      90             : 
      91             : /* Fill in a SpGistTypeDesc struct with info about the specified data type */
      92             : static void
      93         558 : fillTypeDesc(SpGistTypeDesc *desc, Oid type)
      94             : {
      95         558 :     desc->type = type;
      96         558 :     get_typlenbyval(type, &desc->attlen, &desc->attbyval);
      97         558 : }
      98             : 
      99             : /*
     100             :  * Fetch local cache of AM-specific info about the index, initializing it
     101             :  * if necessary
     102             :  */
     103             : SpGistCache *
     104     1612328 : spgGetCache(Relation index)
     105             : {
     106             :     SpGistCache *cache;
     107             : 
     108     1612328 :     if (index->rd_amcache == NULL)
     109             :     {
     110             :         Oid         atttype;
     111             :         spgConfigIn in;
     112             :         FmgrInfo   *procinfo;
     113             :         Buffer      metabuffer;
     114             :         SpGistMetaPageData *metadata;
     115             : 
     116         186 :         cache = MemoryContextAllocZero(index->rd_indexcxt,
     117             :                                        sizeof(SpGistCache));
     118             : 
     119             :         /* SPGiST doesn't support multi-column indexes */
     120             :         Assert(index->rd_att->natts == 1);
     121             : 
     122             :         /*
     123             :          * Get the actual data type of the indexed column from the index
     124             :          * tupdesc.  We pass this to the opclass config function so that
     125             :          * polymorphic opclasses are possible.
     126             :          */
     127         186 :         atttype = TupleDescAttr(index->rd_att, 0)->atttypid;
     128             : 
     129             :         /* Call the config function to get config info for the opclass */
     130         186 :         in.attType = atttype;
     131             : 
     132         186 :         procinfo = index_getprocinfo(index, 1, SPGIST_CONFIG_PROC);
     133         372 :         FunctionCall2Coll(procinfo,
     134         186 :                           index->rd_indcollation[0],
     135             :                           PointerGetDatum(&in),
     136         186 :                           PointerGetDatum(&cache->config));
     137             : 
     138             :         /* Get the information we need about each relevant datatype */
     139         186 :         fillTypeDesc(&cache->attType, atttype);
     140             : 
     141         186 :         if (OidIsValid(cache->config.leafType) &&
     142          14 :             cache->config.leafType != atttype)
     143             :         {
     144           0 :             if (!OidIsValid(index_getprocid(index, 1, SPGIST_COMPRESS_PROC)))
     145           0 :                 ereport(ERROR,
     146             :                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     147             :                          errmsg("compress method must be defined when leaf type is different from input type")));
     148             : 
     149           0 :             fillTypeDesc(&cache->attLeafType, cache->config.leafType);
     150             :         }
     151             :         else
     152             :         {
     153         186 :             cache->attLeafType = cache->attType;
     154             :         }
     155             : 
     156         186 :         fillTypeDesc(&cache->attPrefixType, cache->config.prefixType);
     157         186 :         fillTypeDesc(&cache->attLabelType, cache->config.labelType);
     158             : 
     159             :         /* Last, get the lastUsedPages data from the metapage */
     160         186 :         metabuffer = ReadBuffer(index, SPGIST_METAPAGE_BLKNO);
     161         186 :         LockBuffer(metabuffer, BUFFER_LOCK_SHARE);
     162             : 
     163         186 :         metadata = SpGistPageGetMeta(BufferGetPage(metabuffer));
     164             : 
     165         186 :         if (metadata->magicNumber != SPGIST_MAGIC_NUMBER)
     166           0 :             elog(ERROR, "index \"%s\" is not an SP-GiST index",
     167             :                  RelationGetRelationName(index));
     168             : 
     169         186 :         cache->lastUsedPages = metadata->lastUsedPages;
     170             : 
     171         186 :         UnlockReleaseBuffer(metabuffer);
     172             : 
     173         186 :         index->rd_amcache = (void *) cache;
     174             :     }
     175             :     else
     176             :     {
     177             :         /* assume it's up to date */
     178     1612142 :         cache = (SpGistCache *) index->rd_amcache;
     179             :     }
     180             : 
     181     1612328 :     return cache;
     182             : }
     183             : 
     184             : /* Initialize SpGistState for working with the given index */
     185             : void
     186      157912 : initSpGistState(SpGistState *state, Relation index)
     187             : {
     188             :     SpGistCache *cache;
     189             : 
     190             :     /* Get cached static information about index */
     191      157912 :     cache = spgGetCache(index);
     192             : 
     193      157912 :     state->config = cache->config;
     194      157912 :     state->attType = cache->attType;
     195      157912 :     state->attLeafType = cache->attLeafType;
     196      157912 :     state->attPrefixType = cache->attPrefixType;
     197      157912 :     state->attLabelType = cache->attLabelType;
     198             : 
     199             :     /* Make workspace for constructing dead tuples */
     200      157912 :     state->deadTupleStorage = palloc0(SGDTSIZE);
     201             : 
     202             :     /* Set XID to use in redirection tuples */
     203      157912 :     state->myXid = GetTopTransactionIdIfAny();
     204             : 
     205             :     /* Assume we're not in an index build (spgbuild will override) */
     206      157912 :     state->isBuild = false;
     207      157912 : }
     208             : 
     209             : /*
     210             :  * Allocate a new page (either by recycling, or by extending the index file).
     211             :  *
     212             :  * The returned buffer is already pinned and exclusive-locked.
     213             :  * Caller is responsible for initializing the page by calling SpGistInitBuffer.
     214             :  */
     215             : Buffer
     216        3780 : SpGistNewBuffer(Relation index)
     217             : {
     218             :     Buffer      buffer;
     219             :     bool        needLock;
     220             : 
     221             :     /* First, try to get a page from FSM */
     222             :     for (;;)
     223           0 :     {
     224        3780 :         BlockNumber blkno = GetFreeIndexPage(index);
     225             : 
     226        3780 :         if (blkno == InvalidBlockNumber)
     227        3776 :             break;              /* nothing known to FSM */
     228             : 
     229             :         /*
     230             :          * The fixed pages shouldn't ever be listed in FSM, but just in case
     231             :          * one is, ignore it.
     232             :          */
     233           4 :         if (SpGistBlockIsFixed(blkno))
     234           0 :             continue;
     235             : 
     236           4 :         buffer = ReadBuffer(index, blkno);
     237             : 
     238             :         /*
     239             :          * We have to guard against the possibility that someone else already
     240             :          * recycled this page; the buffer may be locked if so.
     241             :          */
     242           4 :         if (ConditionalLockBuffer(buffer))
     243             :         {
     244           4 :             Page        page = BufferGetPage(buffer);
     245             : 
     246           4 :             if (PageIsNew(page))
     247           0 :                 return buffer;  /* OK to use, if never initialized */
     248             : 
     249           4 :             if (SpGistPageIsDeleted(page) || PageIsEmpty(page))
     250           4 :                 return buffer;  /* OK to use */
     251             : 
     252           0 :             LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
     253             :         }
     254             : 
     255             :         /* Can't use it, so release buffer and try again */
     256           0 :         ReleaseBuffer(buffer);
     257             :     }
     258             : 
     259             :     /* Must extend the file */
     260        3776 :     needLock = !RELATION_IS_LOCAL(index);
     261        3776 :     if (needLock)
     262        1520 :         LockRelationForExtension(index, ExclusiveLock);
     263             : 
     264        3776 :     buffer = ReadBuffer(index, P_NEW);
     265        3776 :     LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
     266             : 
     267        3776 :     if (needLock)
     268        1520 :         UnlockRelationForExtension(index, ExclusiveLock);
     269             : 
     270        3776 :     return buffer;
     271             : }
     272             : 
     273             : /*
     274             :  * Update index metapage's lastUsedPages info from local cache, if possible
     275             :  *
     276             :  * Updating meta page isn't critical for index working, so
     277             :  * 1 use ConditionalLockBuffer to improve concurrency
     278             :  * 2 don't WAL-log metabuffer changes to decrease WAL traffic
     279             :  */
     280             : void
     281      157332 : SpGistUpdateMetaPage(Relation index)
     282             : {
     283      157332 :     SpGistCache *cache = (SpGistCache *) index->rd_amcache;
     284             : 
     285      157332 :     if (cache != NULL)
     286             :     {
     287             :         Buffer      metabuffer;
     288             : 
     289      157332 :         metabuffer = ReadBuffer(index, SPGIST_METAPAGE_BLKNO);
     290             : 
     291      157332 :         if (ConditionalLockBuffer(metabuffer))
     292             :         {
     293      157332 :             Page        metapage = BufferGetPage(metabuffer);
     294      157332 :             SpGistMetaPageData *metadata = SpGistPageGetMeta(metapage);
     295             : 
     296      157332 :             metadata->lastUsedPages = cache->lastUsedPages;
     297             : 
     298             :             /*
     299             :              * Set pd_lower just past the end of the metadata.  This is
     300             :              * essential, because without doing so, metadata will be lost if
     301             :              * xlog.c compresses the page.  (We must do this here because
     302             :              * pre-v11 versions of PG did not set the metapage's pd_lower
     303             :              * correctly, so a pg_upgraded index might contain the wrong
     304             :              * value.)
     305             :              */
     306      157332 :             ((PageHeader) metapage)->pd_lower =
     307      157332 :                 ((char *) metadata + sizeof(SpGistMetaPageData)) - (char *) metapage;
     308             : 
     309      157332 :             MarkBufferDirty(metabuffer);
     310      157332 :             UnlockReleaseBuffer(metabuffer);
     311             :         }
     312             :         else
     313             :         {
     314           0 :             ReleaseBuffer(metabuffer);
     315             :         }
     316             :     }
     317      157332 : }
     318             : 
     319             : /* Macro to select proper element of lastUsedPages cache depending on flags */
     320             : /* Masking flags with SPGIST_CACHED_PAGES is just for paranoia's sake */
     321             : #define GET_LUP(c, f)  (&(c)->lastUsedPages.cachedPage[((unsigned int) (f)) % SPGIST_CACHED_PAGES])
     322             : 
     323             : /*
     324             :  * Allocate and initialize a new buffer of the type and parity specified by
     325             :  * flags.  The returned buffer is already pinned and exclusive-locked.
     326             :  *
     327             :  * When requesting an inner page, if we get one with the wrong parity,
     328             :  * we just release the buffer and try again.  We will get a different page
     329             :  * because GetFreeIndexPage will have marked the page used in FSM.  The page
     330             :  * is entered in our local lastUsedPages cache, so there's some hope of
     331             :  * making use of it later in this session, but otherwise we rely on VACUUM
     332             :  * to eventually re-enter the page in FSM, making it available for recycling.
     333             :  * Note that such a page does not get marked dirty here, so unless it's used
     334             :  * fairly soon, the buffer will just get discarded and the page will remain
     335             :  * as it was on disk.
     336             :  *
     337             :  * When we return a buffer to the caller, the page is *not* entered into
     338             :  * the lastUsedPages cache; we expect the caller will do so after it's taken
     339             :  * whatever space it will use.  This is because after the caller has used up
     340             :  * some space, the page might have less space than whatever was cached already
     341             :  * so we'd rather not trash the old cache entry.
     342             :  */
     343             : static Buffer
     344        3510 : allocNewBuffer(Relation index, int flags)
     345             : {
     346        3510 :     SpGistCache *cache = spgGetCache(index);
     347        3510 :     uint16      pageflags = 0;
     348             : 
     349        3510 :     if (GBUF_REQ_LEAF(flags))
     350        3446 :         pageflags |= SPGIST_LEAF;
     351        3510 :     if (GBUF_REQ_NULLS(flags))
     352           0 :         pageflags |= SPGIST_NULLS;
     353             : 
     354             :     for (;;)
     355          48 :     {
     356             :         Buffer      buffer;
     357             : 
     358        3558 :         buffer = SpGistNewBuffer(index);
     359        3558 :         SpGistInitBuffer(buffer, pageflags);
     360             : 
     361        3558 :         if (pageflags & SPGIST_LEAF)
     362             :         {
     363             :             /* Leaf pages have no parity concerns, so just use it */
     364        3446 :             return buffer;
     365             :         }
     366             :         else
     367             :         {
     368         112 :             BlockNumber blkno = BufferGetBlockNumber(buffer);
     369         112 :             int         blkFlags = GBUF_INNER_PARITY(blkno);
     370             : 
     371         112 :             if ((flags & GBUF_PARITY_MASK) == blkFlags)
     372             :             {
     373             :                 /* Page has right parity, use it */
     374          64 :                 return buffer;
     375             :             }
     376             :             else
     377             :             {
     378             :                 /* Page has wrong parity, record it in cache and try again */
     379          48 :                 if (pageflags & SPGIST_NULLS)
     380           0 :                     blkFlags |= GBUF_NULLS;
     381          48 :                 cache->lastUsedPages.cachedPage[blkFlags].blkno = blkno;
     382          48 :                 cache->lastUsedPages.cachedPage[blkFlags].freeSpace =
     383          48 :                     PageGetExactFreeSpace(BufferGetPage(buffer));
     384          48 :                 UnlockReleaseBuffer(buffer);
     385             :             }
     386             :         }
     387             :     }
     388             : }
     389             : 
     390             : /*
     391             :  * Get a buffer of the type and parity specified by flags, having at least
     392             :  * as much free space as indicated by needSpace.  We use the lastUsedPages
     393             :  * cache to assign the same buffer previously requested when possible.
     394             :  * The returned buffer is already pinned and exclusive-locked.
     395             :  *
     396             :  * *isNew is set true if the page was initialized here, false if it was
     397             :  * already valid.
     398             :  */
     399             : Buffer
     400        6342 : SpGistGetBuffer(Relation index, int flags, int needSpace, bool *isNew)
     401             : {
     402        6342 :     SpGistCache *cache = spgGetCache(index);
     403             :     SpGistLastUsedPage *lup;
     404             : 
     405             :     /* Bail out if even an empty page wouldn't meet the demand */
     406        6342 :     if (needSpace > SPGIST_PAGE_CAPACITY)
     407           0 :         elog(ERROR, "desired SPGiST tuple size is too big");
     408             : 
     409             :     /*
     410             :      * If possible, increase the space request to include relation's
     411             :      * fillfactor.  This ensures that when we add unrelated tuples to a page,
     412             :      * we try to keep 100-fillfactor% available for adding tuples that are
     413             :      * related to the ones already on it.  But fillfactor mustn't cause an
     414             :      * error for requests that would otherwise be legal.
     415             :      */
     416        6342 :     needSpace += SpGistGetTargetPageFreeSpace(index);
     417        6342 :     needSpace = Min(needSpace, SPGIST_PAGE_CAPACITY);
     418             : 
     419             :     /* Get the cache entry for this flags setting */
     420        6342 :     lup = GET_LUP(cache, flags);
     421             : 
     422             :     /* If we have nothing cached, just turn it over to allocNewBuffer */
     423        6342 :     if (lup->blkno == InvalidBlockNumber)
     424             :     {
     425         108 :         *isNew = true;
     426         108 :         return allocNewBuffer(index, flags);
     427             :     }
     428             : 
     429             :     /* fixed pages should never be in cache */
     430             :     Assert(!SpGistBlockIsFixed(lup->blkno));
     431             : 
     432             :     /* If cached freeSpace isn't enough, don't bother looking at the page */
     433        6234 :     if (lup->freeSpace >= needSpace)
     434             :     {
     435             :         Buffer      buffer;
     436             :         Page        page;
     437             : 
     438        2832 :         buffer = ReadBuffer(index, lup->blkno);
     439             : 
     440        2832 :         if (!ConditionalLockBuffer(buffer))
     441             :         {
     442             :             /*
     443             :              * buffer is locked by another process, so return a new buffer
     444             :              */
     445           0 :             ReleaseBuffer(buffer);
     446           0 :             *isNew = true;
     447           0 :             return allocNewBuffer(index, flags);
     448             :         }
     449             : 
     450        2832 :         page = BufferGetPage(buffer);
     451             : 
     452        2832 :         if (PageIsNew(page) || SpGistPageIsDeleted(page) || PageIsEmpty(page))
     453             :         {
     454             :             /* OK to initialize the page */
     455          96 :             uint16      pageflags = 0;
     456             : 
     457          96 :             if (GBUF_REQ_LEAF(flags))
     458          92 :                 pageflags |= SPGIST_LEAF;
     459          96 :             if (GBUF_REQ_NULLS(flags))
     460           0 :                 pageflags |= SPGIST_NULLS;
     461          96 :             SpGistInitBuffer(buffer, pageflags);
     462          96 :             lup->freeSpace = PageGetExactFreeSpace(page) - needSpace;
     463          96 :             *isNew = true;
     464          96 :             return buffer;
     465             :         }
     466             : 
     467             :         /*
     468             :          * Check that page is of right type and has enough space.  We must
     469             :          * recheck this since our cache isn't necessarily up to date.
     470             :          */
     471        5472 :         if ((GBUF_REQ_LEAF(flags) ? SpGistPageIsLeaf(page) : !SpGistPageIsLeaf(page)) &&
     472        2736 :             (GBUF_REQ_NULLS(flags) ? SpGistPageStoresNulls(page) : !SpGistPageStoresNulls(page)))
     473             :         {
     474        2736 :             int         freeSpace = PageGetExactFreeSpace(page);
     475             : 
     476        2736 :             if (freeSpace >= needSpace)
     477             :             {
     478             :                 /* Success, update freespace info and return the buffer */
     479        2736 :                 lup->freeSpace = freeSpace - needSpace;
     480        2736 :                 *isNew = false;
     481        2736 :                 return buffer;
     482             :             }
     483             :         }
     484             : 
     485             :         /*
     486             :          * fallback to allocation of new buffer
     487             :          */
     488           0 :         UnlockReleaseBuffer(buffer);
     489             :     }
     490             : 
     491             :     /* No success with cache, so return a new buffer */
     492        3402 :     *isNew = true;
     493        3402 :     return allocNewBuffer(index, flags);
     494             : }
     495             : 
     496             : /*
     497             :  * Update lastUsedPages cache when done modifying a page.
     498             :  *
     499             :  * We update the appropriate cache entry if it already contained this page
     500             :  * (its freeSpace is likely obsolete), or if this page has more space than
     501             :  * whatever we had cached.
     502             :  */
     503             : void
     504     1443400 : SpGistSetLastUsedPage(Relation index, Buffer buffer)
     505             : {
     506     1443400 :     SpGistCache *cache = spgGetCache(index);
     507             :     SpGistLastUsedPage *lup;
     508             :     int         freeSpace;
     509     1443400 :     Page        page = BufferGetPage(buffer);
     510     1443400 :     BlockNumber blkno = BufferGetBlockNumber(buffer);
     511             :     int         flags;
     512             : 
     513             :     /* Never enter fixed pages (root pages) in cache, though */
     514     1443400 :     if (SpGistBlockIsFixed(blkno))
     515      483354 :         return;
     516             : 
     517      960046 :     if (SpGistPageIsLeaf(page))
     518      482660 :         flags = GBUF_LEAF;
     519             :     else
     520      477386 :         flags = GBUF_INNER_PARITY(blkno);
     521      960046 :     if (SpGistPageStoresNulls(page))
     522           0 :         flags |= GBUF_NULLS;
     523             : 
     524      960046 :     lup = GET_LUP(cache, flags);
     525             : 
     526      960046 :     freeSpace = PageGetExactFreeSpace(page);
     527      960046 :     if (lup->blkno == InvalidBlockNumber || lup->blkno == blkno ||
     528      248048 :         lup->freeSpace < freeSpace)
     529             :     {
     530      717074 :         lup->blkno = blkno;
     531      717074 :         lup->freeSpace = freeSpace;
     532             :     }
     533             : }
     534             : 
     535             : /*
     536             :  * Initialize an SPGiST page to empty, with specified flags
     537             :  */
     538             : void
     539        4120 : SpGistInitPage(Page page, uint16 f)
     540             : {
     541             :     SpGistPageOpaque opaque;
     542             : 
     543        4120 :     PageInit(page, BLCKSZ, MAXALIGN(sizeof(SpGistPageOpaqueData)));
     544        4120 :     opaque = SpGistPageGetOpaque(page);
     545        4120 :     memset(opaque, 0, sizeof(SpGistPageOpaqueData));
     546        4120 :     opaque->flags = f;
     547        4120 :     opaque->spgist_page_id = SPGIST_PAGE_ID;
     548        4120 : }
     549             : 
     550             : /*
     551             :  * Initialize a buffer's page to empty, with specified flags
     552             :  */
     553             : void
     554        4046 : SpGistInitBuffer(Buffer b, uint16 f)
     555             : {
     556             :     Assert(BufferGetPageSize(b) == BLCKSZ);
     557        4046 :     SpGistInitPage(BufferGetPage(b), f);
     558        4046 : }
     559             : 
     560             : /*
     561             :  * Initialize metadata page
     562             :  */
     563             : void
     564          74 : SpGistInitMetapage(Page page)
     565             : {
     566             :     SpGistMetaPageData *metadata;
     567             :     int         i;
     568             : 
     569          74 :     SpGistInitPage(page, SPGIST_META);
     570          74 :     metadata = SpGistPageGetMeta(page);
     571          74 :     memset(metadata, 0, sizeof(SpGistMetaPageData));
     572          74 :     metadata->magicNumber = SPGIST_MAGIC_NUMBER;
     573             : 
     574             :     /* initialize last-used-page cache to empty */
     575         666 :     for (i = 0; i < SPGIST_CACHED_PAGES; i++)
     576         592 :         metadata->lastUsedPages.cachedPage[i].blkno = InvalidBlockNumber;
     577             : 
     578             :     /*
     579             :      * Set pd_lower just past the end of the metadata.  This is essential,
     580             :      * because without doing so, metadata will be lost if xlog.c compresses
     581             :      * the page.
     582             :      */
     583          74 :     ((PageHeader) page)->pd_lower =
     584          74 :         ((char *) metadata + sizeof(SpGistMetaPageData)) - (char *) page;
     585          74 : }
     586             : 
     587             : /*
     588             :  * reloptions processing for SPGiST
     589             :  */
     590             : bytea *
     591          76 : spgoptions(Datum reloptions, bool validate)
     592             : {
     593             :     static const relopt_parse_elt tab[] = {
     594             :         {"fillfactor", RELOPT_TYPE_INT, offsetof(SpGistOptions, fillfactor)},
     595             :     };
     596             : 
     597          76 :     return (bytea *) build_reloptions(reloptions, validate,
     598             :                                       RELOPT_KIND_SPGIST,
     599             :                                       sizeof(SpGistOptions),
     600             :                                       tab, lengthof(tab));
     601             : 
     602             : }
     603             : 
     604             : /*
     605             :  * Get the space needed to store a non-null datum of the indicated type.
     606             :  * Note the result is already rounded up to a MAXALIGN boundary.
     607             :  * Also, we follow the SPGiST convention that pass-by-val types are
     608             :  * just stored in their Datum representation (compare memcpyDatum).
     609             :  */
     610             : unsigned int
     611    13985348 : SpGistGetTypeSize(SpGistTypeDesc *att, Datum datum)
     612             : {
     613             :     unsigned int size;
     614             : 
     615    13985348 :     if (att->attbyval)
     616        3704 :         size = sizeof(Datum);
     617    13981644 :     else if (att->attlen > 0)
     618    12755064 :         size = att->attlen;
     619             :     else
     620     1226580 :         size = VARSIZE_ANY(datum);
     621             : 
     622    13985348 :     return MAXALIGN(size);
     623             : }
     624             : 
     625             : /*
     626             :  * Copy the given non-null datum to *target
     627             :  */
     628             : static void
     629      934104 : memcpyDatum(void *target, SpGistTypeDesc *att, Datum datum)
     630             : {
     631             :     unsigned int size;
     632             : 
     633      934104 :     if (att->attbyval)
     634             :     {
     635        3704 :         memcpy(target, &datum, sizeof(Datum));
     636             :     }
     637             :     else
     638             :     {
     639      930400 :         size = (att->attlen > 0) ? att->attlen : VARSIZE_ANY(datum);
     640      930400 :         memcpy(target, DatumGetPointer(datum), size);
     641             :     }
     642      934104 : }
     643             : 
     644             : /*
     645             :  * Construct a leaf tuple containing the given heap TID and datum value
     646             :  */
     647             : SpGistLeafTuple
     648      926824 : spgFormLeafTuple(SpGistState *state, ItemPointer heapPtr,
     649             :                  Datum datum, bool isnull)
     650             : {
     651             :     SpGistLeafTuple tup;
     652             :     unsigned int size;
     653             : 
     654             :     /* compute space needed (note result is already maxaligned) */
     655      926824 :     size = SGLTHDRSZ;
     656      926824 :     if (!isnull)
     657      926776 :         size += SpGistGetTypeSize(&state->attLeafType, datum);
     658             : 
     659             :     /*
     660             :      * Ensure that we can replace the tuple with a dead tuple later.  This
     661             :      * test is unnecessary when !isnull, but let's be safe.
     662             :      */
     663      926824 :     if (size < SGDTSIZE)
     664           0 :         size = SGDTSIZE;
     665             : 
     666             :     /* OK, form the tuple */
     667      926824 :     tup = (SpGistLeafTuple) palloc0(size);
     668             : 
     669      926824 :     tup->size = size;
     670      926824 :     tup->nextOffset = InvalidOffsetNumber;
     671      926824 :     tup->heapPtr = *heapPtr;
     672      926824 :     if (!isnull)
     673      926776 :         memcpyDatum(SGLTDATAPTR(tup), &state->attLeafType, datum);
     674             : 
     675      926824 :     return tup;
     676             : }
     677             : 
     678             : /*
     679             :  * Construct a node (to go into an inner tuple) containing the given label
     680             :  *
     681             :  * Note that the node's downlink is just set invalid here.  Caller will fill
     682             :  * it in later.
     683             :  */
     684             : SpGistNodeTuple
     685       25516 : spgFormNodeTuple(SpGistState *state, Datum label, bool isnull)
     686             : {
     687             :     SpGistNodeTuple tup;
     688             :     unsigned int size;
     689       25516 :     unsigned short infomask = 0;
     690             : 
     691             :     /* compute space needed (note result is already maxaligned) */
     692       25516 :     size = SGNTHDRSZ;
     693       25516 :     if (!isnull)
     694        3244 :         size += SpGistGetTypeSize(&state->attLabelType, label);
     695             : 
     696             :     /*
     697             :      * Here we make sure that the size will fit in the field reserved for it
     698             :      * in t_info.
     699             :      */
     700       25516 :     if ((size & INDEX_SIZE_MASK) != size)
     701           0 :         ereport(ERROR,
     702             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
     703             :                  errmsg("index row requires %zu bytes, maximum size is %zu",
     704             :                         (Size) size, (Size) INDEX_SIZE_MASK)));
     705             : 
     706       25516 :     tup = (SpGistNodeTuple) palloc0(size);
     707             : 
     708       25516 :     if (isnull)
     709       22272 :         infomask |= INDEX_NULL_MASK;
     710             :     /* we don't bother setting the INDEX_VAR_MASK bit */
     711       25516 :     infomask |= size;
     712       25516 :     tup->t_info = infomask;
     713             : 
     714             :     /* The TID field will be filled in later */
     715       25516 :     ItemPointerSetInvalid(&tup->t_tid);
     716             : 
     717       25516 :     if (!isnull)
     718        3244 :         memcpyDatum(SGNTDATAPTR(tup), &state->attLabelType, label);
     719             : 
     720       25516 :     return tup;
     721             : }
     722             : 
     723             : /*
     724             :  * Construct an inner tuple containing the given prefix and node array
     725             :  */
     726             : SpGistInnerTuple
     727        5214 : spgFormInnerTuple(SpGistState *state, bool hasPrefix, Datum prefix,
     728             :                   int nNodes, SpGistNodeTuple *nodes)
     729             : {
     730             :     SpGistInnerTuple tup;
     731             :     unsigned int size;
     732             :     unsigned int prefixSize;
     733             :     int         i;
     734             :     char       *ptr;
     735             : 
     736             :     /* Compute size needed */
     737        5214 :     if (hasPrefix)
     738        4084 :         prefixSize = SpGistGetTypeSize(&state->attPrefixType, prefix);
     739             :     else
     740        1130 :         prefixSize = 0;
     741             : 
     742        5214 :     size = SGITHDRSZ + prefixSize;
     743             : 
     744             :     /* Note: we rely on node tuple sizes to be maxaligned already */
     745       35978 :     for (i = 0; i < nNodes; i++)
     746       30764 :         size += IndexTupleSize(nodes[i]);
     747             : 
     748             :     /*
     749             :      * Ensure that we can replace the tuple with a dead tuple later.  This
     750             :      * test is unnecessary given current tuple layouts, but let's be safe.
     751             :      */
     752        5214 :     if (size < SGDTSIZE)
     753           0 :         size = SGDTSIZE;
     754             : 
     755             :     /*
     756             :      * Inner tuple should be small enough to fit on a page
     757             :      */
     758        5214 :     if (size > SPGIST_PAGE_CAPACITY - sizeof(ItemIdData))
     759           0 :         ereport(ERROR,
     760             :                 (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
     761             :                  errmsg("SP-GiST inner tuple size %zu exceeds maximum %zu",
     762             :                         (Size) size,
     763             :                         SPGIST_PAGE_CAPACITY - sizeof(ItemIdData)),
     764             :                  errhint("Values larger than a buffer page cannot be indexed.")));
     765             : 
     766             :     /*
     767             :      * Check for overflow of header fields --- probably can't fail if the
     768             :      * above succeeded, but let's be paranoid
     769             :      */
     770        5214 :     if (size > SGITMAXSIZE ||
     771        5214 :         prefixSize > SGITMAXPREFIXSIZE ||
     772             :         nNodes > SGITMAXNNODES)
     773           0 :         elog(ERROR, "SPGiST inner tuple header field is too small");
     774             : 
     775             :     /* OK, form the tuple */
     776        5214 :     tup = (SpGistInnerTuple) palloc0(size);
     777             : 
     778        5214 :     tup->nNodes = nNodes;
     779        5214 :     tup->prefixSize = prefixSize;
     780        5214 :     tup->size = size;
     781             : 
     782        5214 :     if (hasPrefix)
     783        4084 :         memcpyDatum(SGITDATAPTR(tup), &state->attPrefixType, prefix);
     784             : 
     785        5214 :     ptr = (char *) SGITNODEPTR(tup);
     786             : 
     787       35978 :     for (i = 0; i < nNodes; i++)
     788             :     {
     789       30764 :         SpGistNodeTuple node = nodes[i];
     790             : 
     791       30764 :         memcpy(ptr, node, IndexTupleSize(node));
     792       30764 :         ptr += IndexTupleSize(node);
     793             :     }
     794             : 
     795        5214 :     return tup;
     796             : }
     797             : 
     798             : /*
     799             :  * Construct a "dead" tuple to replace a tuple being deleted.
     800             :  *
     801             :  * The state can be SPGIST_REDIRECT, SPGIST_DEAD, or SPGIST_PLACEHOLDER.
     802             :  * For a REDIRECT tuple, a pointer (blkno+offset) must be supplied, and
     803             :  * the xid field is filled in automatically.
     804             :  *
     805             :  * This is called in critical sections, so we don't use palloc; the tuple
     806             :  * is built in preallocated storage.  It should be copied before another
     807             :  * call with different parameters can occur.
     808             :  */
     809             : SpGistDeadTuple
     810        7864 : spgFormDeadTuple(SpGistState *state, int tupstate,
     811             :                  BlockNumber blkno, OffsetNumber offnum)
     812             : {
     813        7864 :     SpGistDeadTuple tuple = (SpGistDeadTuple) state->deadTupleStorage;
     814             : 
     815        7864 :     tuple->tupstate = tupstate;
     816        7864 :     tuple->size = SGDTSIZE;
     817        7864 :     tuple->nextOffset = InvalidOffsetNumber;
     818             : 
     819        7864 :     if (tupstate == SPGIST_REDIRECT)
     820             :     {
     821        1254 :         ItemPointerSet(&tuple->pointer, blkno, offnum);
     822             :         Assert(TransactionIdIsValid(state->myXid));
     823        1254 :         tuple->xid = state->myXid;
     824             :     }
     825             :     else
     826             :     {
     827        6610 :         ItemPointerSetInvalid(&tuple->pointer);
     828        6610 :         tuple->xid = InvalidTransactionId;
     829             :     }
     830             : 
     831        7864 :     return tuple;
     832             : }
     833             : 
     834             : /*
     835             :  * Extract the label datums of the nodes within innerTuple
     836             :  *
     837             :  * Returns NULL if label datums are NULLs
     838             :  */
     839             : Datum *
     840    12586262 : spgExtractNodeLabels(SpGistState *state, SpGistInnerTuple innerTuple)
     841             : {
     842             :     Datum      *nodeLabels;
     843             :     int         i;
     844             :     SpGistNodeTuple node;
     845             : 
     846             :     /* Either all the labels must be NULL, or none. */
     847    12586262 :     node = SGITNODEPTR(innerTuple);
     848    12586262 :     if (IndexTupleHasNulls(node))
     849             :     {
     850    67576468 :         SGITITERATE(innerTuple, i, node)
     851             :         {
     852    55131396 :             if (!IndexTupleHasNulls(node))
     853           0 :                 elog(ERROR, "some but not all node labels are null in SPGiST inner tuple");
     854             :         }
     855             :         /* They're all null, so just return NULL */
     856    12445072 :         return NULL;
     857             :     }
     858             :     else
     859             :     {
     860      141190 :         nodeLabels = (Datum *) palloc(sizeof(Datum) * innerTuple->nNodes);
     861     1526450 :         SGITITERATE(innerTuple, i, node)
     862             :         {
     863     1385260 :             if (IndexTupleHasNulls(node))
     864           0 :                 elog(ERROR, "some but not all node labels are null in SPGiST inner tuple");
     865     1385260 :             nodeLabels[i] = SGNTDATUM(node, state);
     866             :         }
     867      141190 :         return nodeLabels;
     868             :     }
     869             : }
     870             : 
     871             : /*
     872             :  * Add a new item to the page, replacing a PLACEHOLDER item if possible.
     873             :  * Return the location it's inserted at, or InvalidOffsetNumber on failure.
     874             :  *
     875             :  * If startOffset isn't NULL, we start searching for placeholders at
     876             :  * *startOffset, and update that to the next place to search.  This is just
     877             :  * an optimization for repeated insertions.
     878             :  *
     879             :  * If errorOK is false, we throw error when there's not enough room,
     880             :  * rather than returning InvalidOffsetNumber.
     881             :  */
     882             : OffsetNumber
     883      971082 : SpGistPageAddNewItem(SpGistState *state, Page page, Item item, Size size,
     884             :                      OffsetNumber *startOffset, bool errorOK)
     885             : {
     886      971082 :     SpGistPageOpaque opaque = SpGistPageGetOpaque(page);
     887             :     OffsetNumber i,
     888             :                 maxoff,
     889             :                 offnum;
     890             : 
     891      971082 :     if (opaque->nPlaceholder > 0 &&
     892      273394 :         PageGetExactFreeSpace(page) + SGDTSIZE >= MAXALIGN(size))
     893             :     {
     894             :         /* Try to replace a placeholder */
     895      273394 :         maxoff = PageGetMaxOffsetNumber(page);
     896      273394 :         offnum = InvalidOffsetNumber;
     897             : 
     898             :         for (;;)
     899             :         {
     900      273394 :             if (startOffset && *startOffset != InvalidOffsetNumber)
     901       76478 :                 i = *startOffset;
     902             :             else
     903      196916 :                 i = FirstOffsetNumber;
     904    18807256 :             for (; i <= maxoff; i++)
     905             :             {
     906    18807256 :                 SpGistDeadTuple it = (SpGistDeadTuple) PageGetItem(page,
     907             :                                                                    PageGetItemId(page, i));
     908             : 
     909    18807256 :                 if (it->tupstate == SPGIST_PLACEHOLDER)
     910             :                 {
     911      273394 :                     offnum = i;
     912      273394 :                     break;
     913             :                 }
     914             :             }
     915             : 
     916             :             /* Done if we found a placeholder */
     917      273394 :             if (offnum != InvalidOffsetNumber)
     918      273394 :                 break;
     919             : 
     920           0 :             if (startOffset && *startOffset != InvalidOffsetNumber)
     921             :             {
     922             :                 /* Hint was no good, re-search from beginning */
     923           0 :                 *startOffset = InvalidOffsetNumber;
     924           0 :                 continue;
     925             :             }
     926             : 
     927             :             /* Hmm, no placeholder found? */
     928           0 :             opaque->nPlaceholder = 0;
     929           0 :             break;
     930             :         }
     931             : 
     932      273394 :         if (offnum != InvalidOffsetNumber)
     933             :         {
     934             :             /* Replace the placeholder tuple */
     935      273394 :             PageIndexTupleDelete(page, offnum);
     936             : 
     937      273394 :             offnum = PageAddItem(page, item, size, offnum, false, false);
     938             : 
     939             :             /*
     940             :              * We should not have failed given the size check at the top of
     941             :              * the function, but test anyway.  If we did fail, we must PANIC
     942             :              * because we've already deleted the placeholder tuple, and
     943             :              * there's no other way to keep the damage from getting to disk.
     944             :              */
     945      273394 :             if (offnum != InvalidOffsetNumber)
     946             :             {
     947             :                 Assert(opaque->nPlaceholder > 0);
     948      273394 :                 opaque->nPlaceholder--;
     949      273394 :                 if (startOffset)
     950       78200 :                     *startOffset = offnum + 1;
     951             :             }
     952             :             else
     953           0 :                 elog(PANIC, "failed to add item of size %u to SPGiST index page",
     954             :                      (int) size);
     955             : 
     956      273394 :             return offnum;
     957             :         }
     958             :     }
     959             : 
     960             :     /* No luck in replacing a placeholder, so just add it to the page */
     961      697688 :     offnum = PageAddItem(page, item, size,
     962             :                          InvalidOffsetNumber, false, false);
     963             : 
     964      697688 :     if (offnum == InvalidOffsetNumber && !errorOK)
     965           0 :         elog(ERROR, "failed to add item of size %u to SPGiST index page",
     966             :              (int) size);
     967             : 
     968      697688 :     return offnum;
     969             : }
     970             : 
     971             : /*
     972             :  *  spgproperty() -- Check boolean properties of indexes.
     973             :  *
     974             :  * This is optional for most AMs, but is required for SP-GiST because the core
     975             :  * property code doesn't support AMPROP_DISTANCE_ORDERABLE.
     976             :  */
     977             : bool
     978         124 : spgproperty(Oid index_oid, int attno,
     979             :             IndexAMProperty prop, const char *propname,
     980             :             bool *res, bool *isnull)
     981             : {
     982             :     Oid         opclass,
     983             :                 opfamily,
     984             :                 opcintype;
     985             :     CatCList   *catlist;
     986             :     int         i;
     987             : 
     988             :     /* Only answer column-level inquiries */
     989         124 :     if (attno == 0)
     990          44 :         return false;
     991             : 
     992          80 :     switch (prop)
     993             :     {
     994           8 :         case AMPROP_DISTANCE_ORDERABLE:
     995           8 :             break;
     996          72 :         default:
     997          72 :             return false;
     998             :     }
     999             : 
    1000             :     /*
    1001             :      * Currently, SP-GiST distance-ordered scans require that there be a
    1002             :      * distance operator in the opclass with the default types. So we assume
    1003             :      * that if such a operator exists, then there's a reason for it.
    1004             :      */
    1005             : 
    1006             :     /* First we need to know the column's opclass. */
    1007           8 :     opclass = get_index_column_opclass(index_oid, attno);
    1008           8 :     if (!OidIsValid(opclass))
    1009             :     {
    1010           0 :         *isnull = true;
    1011           0 :         return true;
    1012             :     }
    1013             : 
    1014             :     /* Now look up the opclass family and input datatype. */
    1015           8 :     if (!get_opclass_opfamily_and_input_type(opclass, &opfamily, &opcintype))
    1016             :     {
    1017           0 :         *isnull = true;
    1018           0 :         return true;
    1019             :     }
    1020             : 
    1021             :     /* And now we can check whether the operator is provided. */
    1022           8 :     catlist = SearchSysCacheList1(AMOPSTRATEGY,
    1023             :                                   ObjectIdGetDatum(opfamily));
    1024             : 
    1025           8 :     *res = false;
    1026             : 
    1027          68 :     for (i = 0; i < catlist->n_members; i++)
    1028             :     {
    1029          64 :         HeapTuple   amoptup = &catlist->members[i]->tuple;
    1030          64 :         Form_pg_amop amopform = (Form_pg_amop) GETSTRUCT(amoptup);
    1031             : 
    1032          64 :         if (amopform->amoppurpose == AMOP_ORDER &&
    1033           4 :             (amopform->amoplefttype == opcintype ||
    1034           4 :              amopform->amoprighttype == opcintype) &&
    1035           4 :             opfamily_can_sort_type(amopform->amopsortfamily,
    1036             :                                    get_op_rettype(amopform->amopopr)))
    1037             :         {
    1038           4 :             *res = true;
    1039           4 :             break;
    1040             :         }
    1041             :     }
    1042             : 
    1043           8 :     ReleaseSysCacheList(catlist);
    1044             : 
    1045           8 :     *isnull = false;
    1046             : 
    1047           8 :     return true;
    1048             : }

Generated by: LCOV version 1.13