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