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