1/*-------------------------------------------------------------------------
2 *
3 * spgscan.c
4 * routines for scanning SP-GiST indexes
5 *
6 *
7 * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
8 * Portions Copyright (c) 1994, Regents of the University of California
9 *
10 * IDENTIFICATION
11 * src/backend/access/spgist/spgscan.c
12 *
13 *-------------------------------------------------------------------------
14 */
15
16#include "postgres.h"
17
18#include "access/genam.h"
19#include "access/relscan.h"
20#include "access/spgist_private.h"
21#include "miscadmin.h"
22#include "storage/bufmgr.h"
23#include "utils/datum.h"
24#include "utils/float.h"
25#include "utils/lsyscache.h"
26#include "utils/memutils.h"
27#include "utils/rel.h"
28
29typedef void (*storeRes_func) (SpGistScanOpaque so, ItemPointer heapPtr,
30 Datum leafValue, bool isNull, bool recheck,
31 bool recheckDistances, double *distances);
32
33/*
34 * Pairing heap comparison function for the SpGistSearchItem queue.
35 * KNN-searches currently only support NULLS LAST. So, preserve this logic
36 * here.
37 */
38static int
39pairingheap_SpGistSearchItem_cmp(const pairingheap_node *a,
40 const pairingheap_node *b, void *arg)
41{
42 const SpGistSearchItem *sa = (const SpGistSearchItem *) a;
43 const SpGistSearchItem *sb = (const SpGistSearchItem *) b;
44 SpGistScanOpaque so = (SpGistScanOpaque) arg;
45 int i;
46
47 if (sa->isNull)
48 {
49 if (!sb->isNull)
50 return -1;
51 }
52 else if (sb->isNull)
53 {
54 return 1;
55 }
56 else
57 {
58 /* Order according to distance comparison */
59 for (i = 0; i < so->numberOfNonNullOrderBys; i++)
60 {
61 if (isnan(sa->distances[i]) && isnan(sb->distances[i]))
62 continue; /* NaN == NaN */
63 if (isnan(sa->distances[i]))
64 return -1; /* NaN > number */
65 if (isnan(sb->distances[i]))
66 return 1; /* number < NaN */
67 if (sa->distances[i] != sb->distances[i])
68 return (sa->distances[i] < sb->distances[i]) ? 1 : -1;
69 }
70 }
71
72 /* Leaf items go before inner pages, to ensure a depth-first search */
73 if (sa->isLeaf && !sb->isLeaf)
74 return 1;
75 if (!sa->isLeaf && sb->isLeaf)
76 return -1;
77
78 return 0;
79}
80
81static void
82spgFreeSearchItem(SpGistScanOpaque so, SpGistSearchItem *item)
83{
84 if (!so->state.attLeafType.attbyval &&
85 DatumGetPointer(item->value) != NULL)
86 pfree(DatumGetPointer(item->value));
87
88 if (item->traversalValue)
89 pfree(item->traversalValue);
90
91 pfree(item);
92}
93
94/*
95 * Add SpGistSearchItem to queue
96 *
97 * Called in queue context
98 */
99static void
100spgAddSearchItemToQueue(SpGistScanOpaque so, SpGistSearchItem *item)
101{
102 pairingheap_add(so->scanQueue, &item->phNode);
103}
104
105static SpGistSearchItem *
106spgAllocSearchItem(SpGistScanOpaque so, bool isnull, double *distances)
107{
108 /* allocate distance array only for non-NULL items */
109 SpGistSearchItem *item =
110 palloc(SizeOfSpGistSearchItem(isnull ? 0 : so->numberOfNonNullOrderBys));
111
112 item->isNull = isnull;
113
114 if (!isnull && so->numberOfNonNullOrderBys > 0)
115 memcpy(item->distances, distances,
116 sizeof(item->distances[0]) * so->numberOfNonNullOrderBys);
117
118 return item;
119}
120
121static void
122spgAddStartItem(SpGistScanOpaque so, bool isnull)
123{
124 SpGistSearchItem *startEntry =
125 spgAllocSearchItem(so, isnull, so->zeroDistances);
126
127 ItemPointerSet(&startEntry->heapPtr,
128 isnull ? SPGIST_NULL_BLKNO : SPGIST_ROOT_BLKNO,
129 FirstOffsetNumber);
130 startEntry->isLeaf = false;
131 startEntry->level = 0;
132 startEntry->value = (Datum) 0;
133 startEntry->traversalValue = NULL;
134 startEntry->recheck = false;
135 startEntry->recheckDistances = false;
136
137 spgAddSearchItemToQueue(so, startEntry);
138}
139
140/*
141 * Initialize queue to search the root page, resetting
142 * any previously active scan
143 */
144static void
145resetSpGistScanOpaque(SpGistScanOpaque so)
146{
147 MemoryContext oldCtx;
148
149 /*
150 * clear traversal context before proceeding to the next scan; this must
151 * not happen before the freeScanStack above, else we get double-free
152 * crashes.
153 */
154 MemoryContextReset(so->traversalCxt);
155
156 oldCtx = MemoryContextSwitchTo(so->traversalCxt);
157
158 /* initialize queue only for distance-ordered scans */
159 so->scanQueue = pairingheap_allocate(pairingheap_SpGistSearchItem_cmp, so);
160
161 if (so->searchNulls)
162 /* Add a work item to scan the null index entries */
163 spgAddStartItem(so, true);
164
165 if (so->searchNonNulls)
166 /* Add a work item to scan the non-null index entries */
167 spgAddStartItem(so, false);
168
169 MemoryContextSwitchTo(oldCtx);
170
171 if (so->numberOfOrderBys > 0)
172 {
173 /* Must pfree distances to avoid memory leak */
174 int i;
175
176 for (i = 0; i < so->nPtrs; i++)
177 if (so->distances[i])
178 pfree(so->distances[i]);
179 }
180
181 if (so->want_itup)
182 {
183 /* Must pfree reconstructed tuples to avoid memory leak */
184 int i;
185
186 for (i = 0; i < so->nPtrs; i++)
187 pfree(so->reconTups[i]);
188 }
189 so->iPtr = so->nPtrs = 0;
190}
191
192/*
193 * Prepare scan keys in SpGistScanOpaque from caller-given scan keys
194 *
195 * Sets searchNulls, searchNonNulls, numberOfKeys, keyData fields of *so.
196 *
197 * The point here is to eliminate null-related considerations from what the
198 * opclass consistent functions need to deal with. We assume all SPGiST-
199 * indexable operators are strict, so any null RHS value makes the scan
200 * condition unsatisfiable. We also pull out any IS NULL/IS NOT NULL
201 * conditions; their effect is reflected into searchNulls/searchNonNulls.
202 */
203static void
204spgPrepareScanKeys(IndexScanDesc scan)
205{
206 SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
207 bool qual_ok;
208 bool haveIsNull;
209 bool haveNotNull;
210 int nkeys;
211 int i;
212
213 so->numberOfOrderBys = scan->numberOfOrderBys;
214 so->orderByData = scan->orderByData;
215
216 if (so->numberOfOrderBys <= 0)
217 so->numberOfNonNullOrderBys = 0;
218 else
219 {
220 int j = 0;
221
222 /*
223 * Remove all NULL keys, but remember their offsets in the original
224 * array.
225 */
226 for (i = 0; i < scan->numberOfOrderBys; i++)
227 {
228 ScanKey skey = &so->orderByData[i];
229
230 if (skey->sk_flags & SK_ISNULL)
231 so->nonNullOrderByOffsets[i] = -1;
232 else
233 {
234 if (i != j)
235 so->orderByData[j] = *skey;
236
237 so->nonNullOrderByOffsets[i] = j++;
238 }
239 }
240
241 so->numberOfNonNullOrderBys = j;
242 }
243
244 if (scan->numberOfKeys <= 0)
245 {
246 /* If no quals, whole-index scan is required */
247 so->searchNulls = true;
248 so->searchNonNulls = true;
249 so->numberOfKeys = 0;
250 return;
251 }
252
253 /* Examine the given quals */
254 qual_ok = true;
255 haveIsNull = haveNotNull = false;
256 nkeys = 0;
257 for (i = 0; i < scan->numberOfKeys; i++)
258 {
259 ScanKey skey = &scan->keyData[i];
260
261 if (skey->sk_flags & SK_SEARCHNULL)
262 haveIsNull = true;
263 else if (skey->sk_flags & SK_SEARCHNOTNULL)
264 haveNotNull = true;
265 else if (skey->sk_flags & SK_ISNULL)
266 {
267 /* ordinary qual with null argument - unsatisfiable */
268 qual_ok = false;
269 break;
270 }
271 else
272 {
273 /* ordinary qual, propagate into so->keyData */
274 so->keyData[nkeys++] = *skey;
275 /* this effectively creates a not-null requirement */
276 haveNotNull = true;
277 }
278 }
279
280 /* IS NULL in combination with something else is unsatisfiable */
281 if (haveIsNull && haveNotNull)
282 qual_ok = false;
283
284 /* Emit results */
285 if (qual_ok)
286 {
287 so->searchNulls = haveIsNull;
288 so->searchNonNulls = haveNotNull;
289 so->numberOfKeys = nkeys;
290 }
291 else
292 {
293 so->searchNulls = false;
294 so->searchNonNulls = false;
295 so->numberOfKeys = 0;
296 }
297}
298
299IndexScanDesc
300spgbeginscan(Relation rel, int keysz, int orderbysz)
301{
302 IndexScanDesc scan;
303 SpGistScanOpaque so;
304 int i;
305
306 scan = RelationGetIndexScan(rel, keysz, orderbysz);
307
308 so = (SpGistScanOpaque) palloc0(sizeof(SpGistScanOpaqueData));
309 if (keysz > 0)
310 so->keyData = (ScanKey) palloc(sizeof(ScanKeyData) * keysz);
311 else
312 so->keyData = NULL;
313 initSpGistState(&so->state, scan->indexRelation);
314
315 so->tempCxt = AllocSetContextCreate(CurrentMemoryContext,
316 "SP-GiST search temporary context",
317 ALLOCSET_DEFAULT_SIZES);
318 so->traversalCxt = AllocSetContextCreate(CurrentMemoryContext,
319 "SP-GiST traversal-value context",
320 ALLOCSET_DEFAULT_SIZES);
321
322 /* Set up indexTupDesc and xs_hitupdesc in case it's an index-only scan */
323 so->indexTupDesc = scan->xs_hitupdesc = RelationGetDescr(rel);
324
325 /* Allocate various arrays needed for order-by scans */
326 if (scan->numberOfOrderBys > 0)
327 {
328 /* This will be filled in spgrescan, but allocate the space here */
329 so->orderByTypes = (Oid *)
330 palloc(sizeof(Oid) * scan->numberOfOrderBys);
331 so->nonNullOrderByOffsets = (int *)
332 palloc(sizeof(int) * scan->numberOfOrderBys);
333
334 /* These arrays have constant contents, so we can fill them now */
335 so->zeroDistances = (double *)
336 palloc(sizeof(double) * scan->numberOfOrderBys);
337 so->infDistances = (double *)
338 palloc(sizeof(double) * scan->numberOfOrderBys);
339
340 for (i = 0; i < scan->numberOfOrderBys; i++)
341 {
342 so->zeroDistances[i] = 0.0;
343 so->infDistances[i] = get_float8_infinity();
344 }
345
346 scan->xs_orderbyvals = (Datum *)
347 palloc0(sizeof(Datum) * scan->numberOfOrderBys);
348 scan->xs_orderbynulls = (bool *)
349 palloc(sizeof(bool) * scan->numberOfOrderBys);
350 memset(scan->xs_orderbynulls, true,
351 sizeof(bool) * scan->numberOfOrderBys);
352 }
353
354 fmgr_info_copy(&so->innerConsistentFn,
355 index_getprocinfo(rel, 1, SPGIST_INNER_CONSISTENT_PROC),
356 CurrentMemoryContext);
357
358 fmgr_info_copy(&so->leafConsistentFn,
359 index_getprocinfo(rel, 1, SPGIST_LEAF_CONSISTENT_PROC),
360 CurrentMemoryContext);
361
362 so->indexCollation = rel->rd_indcollation[0];
363
364 scan->opaque = so;
365
366 return scan;
367}
368
369void
370spgrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,
371 ScanKey orderbys, int norderbys)
372{
373 SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
374
375 /* copy scankeys into local storage */
376 if (scankey && scan->numberOfKeys > 0)
377 memmove(scan->keyData, scankey,
378 scan->numberOfKeys * sizeof(ScanKeyData));
379
380 /* initialize order-by data if needed */
381 if (orderbys && scan->numberOfOrderBys > 0)
382 {
383 int i;
384
385 memmove(scan->orderByData, orderbys,
386 scan->numberOfOrderBys * sizeof(ScanKeyData));
387
388 for (i = 0; i < scan->numberOfOrderBys; i++)
389 {
390 ScanKey skey = &scan->orderByData[i];
391
392 /*
393 * Look up the datatype returned by the original ordering
394 * operator. SP-GiST always uses a float8 for the distance
395 * function, but the ordering operator could be anything else.
396 *
397 * XXX: The distance function is only allowed to be lossy if the
398 * ordering operator's result type is float4 or float8. Otherwise
399 * we don't know how to return the distance to the executor. But
400 * we cannot check that here, as we won't know if the distance
401 * function is lossy until it returns *recheck = true for the
402 * first time.
403 */
404 so->orderByTypes[i] = get_func_rettype(skey->sk_func.fn_oid);
405 }
406 }
407
408 /* preprocess scankeys, set up the representation in *so */
409 spgPrepareScanKeys(scan);
410
411 /* set up starting queue entries */
412 resetSpGistScanOpaque(so);
413}
414
415void
416spgendscan(IndexScanDesc scan)
417{
418 SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
419
420 MemoryContextDelete(so->tempCxt);
421 MemoryContextDelete(so->traversalCxt);
422
423 if (so->keyData)
424 pfree(so->keyData);
425
426 if (so->state.deadTupleStorage)
427 pfree(so->state.deadTupleStorage);
428
429 if (scan->numberOfOrderBys > 0)
430 {
431 pfree(so->orderByTypes);
432 pfree(so->nonNullOrderByOffsets);
433 pfree(so->zeroDistances);
434 pfree(so->infDistances);
435 pfree(scan->xs_orderbyvals);
436 pfree(scan->xs_orderbynulls);
437 }
438
439 pfree(so);
440}
441
442/*
443 * Leaf SpGistSearchItem constructor, called in queue context
444 */
445static SpGistSearchItem *
446spgNewHeapItem(SpGistScanOpaque so, int level, ItemPointer heapPtr,
447 Datum leafValue, bool recheck, bool recheckDistances,
448 bool isnull, double *distances)
449{
450 SpGistSearchItem *item = spgAllocSearchItem(so, isnull, distances);
451
452 item->level = level;
453 item->heapPtr = *heapPtr;
454 /* copy value to queue cxt out of tmp cxt */
455 item->value = isnull ? (Datum) 0 :
456 datumCopy(leafValue, so->state.attLeafType.attbyval,
457 so->state.attLeafType.attlen);
458 item->traversalValue = NULL;
459 item->isLeaf = true;
460 item->recheck = recheck;
461 item->recheckDistances = recheckDistances;
462
463 return item;
464}
465
466/*
467 * Test whether a leaf tuple satisfies all the scan keys
468 *
469 * *reportedSome is set to true if:
470 * the scan is not ordered AND the item satisfies the scankeys
471 */
472static bool
473spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item,
474 SpGistLeafTuple leafTuple, bool isnull,
475 bool *reportedSome, storeRes_func storeRes)
476{
477 Datum leafValue;
478 double *distances;
479 bool result;
480 bool recheck;
481 bool recheckDistances;
482
483 if (isnull)
484 {
485 /* Should not have arrived on a nulls page unless nulls are wanted */
486 Assert(so->searchNulls);
487 leafValue = (Datum) 0;
488 distances = NULL;
489 recheck = false;
490 recheckDistances = false;
491 result = true;
492 }
493 else
494 {
495 spgLeafConsistentIn in;
496 spgLeafConsistentOut out;
497
498 /* use temp context for calling leaf_consistent */
499 MemoryContext oldCxt = MemoryContextSwitchTo(so->tempCxt);
500
501 in.scankeys = so->keyData;
502 in.nkeys = so->numberOfKeys;
503 in.orderbys = so->orderByData;
504 in.norderbys = so->numberOfNonNullOrderBys;
505 in.reconstructedValue = item->value;
506 in.traversalValue = item->traversalValue;
507 in.level = item->level;
508 in.returnData = so->want_itup;
509 in.leafDatum = SGLTDATUM(leafTuple, &so->state);
510
511 out.leafValue = (Datum) 0;
512 out.recheck = false;
513 out.distances = NULL;
514 out.recheckDistances = false;
515
516 result = DatumGetBool(FunctionCall2Coll(&so->leafConsistentFn,
517 so->indexCollation,
518 PointerGetDatum(&in),
519 PointerGetDatum(&out)));
520 recheck = out.recheck;
521 recheckDistances = out.recheckDistances;
522 leafValue = out.leafValue;
523 distances = out.distances;
524
525 MemoryContextSwitchTo(oldCxt);
526 }
527
528 if (result)
529 {
530 /* item passes the scankeys */
531 if (so->numberOfNonNullOrderBys > 0)
532 {
533 /* the scan is ordered -> add the item to the queue */
534 MemoryContext oldCxt = MemoryContextSwitchTo(so->traversalCxt);
535 SpGistSearchItem *heapItem = spgNewHeapItem(so, item->level,
536 &leafTuple->heapPtr,
537 leafValue,
538 recheck,
539 recheckDistances,
540 isnull,
541 distances);
542
543 spgAddSearchItemToQueue(so, heapItem);
544
545 MemoryContextSwitchTo(oldCxt);
546 }
547 else
548 {
549 /* non-ordered scan, so report the item right away */
550 Assert(!recheckDistances);
551 storeRes(so, &leafTuple->heapPtr, leafValue, isnull,
552 recheck, false, NULL);
553 *reportedSome = true;
554 }
555 }
556
557 return result;
558}
559
560/* A bundle initializer for inner_consistent methods */
561static void
562spgInitInnerConsistentIn(spgInnerConsistentIn *in,
563 SpGistScanOpaque so,
564 SpGistSearchItem *item,
565 SpGistInnerTuple innerTuple)
566{
567 in->scankeys = so->keyData;
568 in->orderbys = so->orderByData;
569 in->nkeys = so->numberOfKeys;
570 in->norderbys = so->numberOfNonNullOrderBys;
571 in->reconstructedValue = item->value;
572 in->traversalMemoryContext = so->traversalCxt;
573 in->traversalValue = item->traversalValue;
574 in->level = item->level;
575 in->returnData = so->want_itup;
576 in->allTheSame = innerTuple->allTheSame;
577 in->hasPrefix = (innerTuple->prefixSize > 0);
578 in->prefixDatum = SGITDATUM(innerTuple, &so->state);
579 in->nNodes = innerTuple->nNodes;
580 in->nodeLabels = spgExtractNodeLabels(&so->state, innerTuple);
581}
582
583static SpGistSearchItem *
584spgMakeInnerItem(SpGistScanOpaque so,
585 SpGistSearchItem *parentItem,
586 SpGistNodeTuple tuple,
587 spgInnerConsistentOut *out, int i, bool isnull,
588 double *distances)
589{
590 SpGistSearchItem *item = spgAllocSearchItem(so, isnull, distances);
591
592 item->heapPtr = tuple->t_tid;
593 item->level = out->levelAdds ? parentItem->level + out->levelAdds[i]
594 : parentItem->level;
595
596 /* Must copy value out of temp context */
597 item->value = out->reconstructedValues
598 ? datumCopy(out->reconstructedValues[i],
599 so->state.attLeafType.attbyval,
600 so->state.attLeafType.attlen)
601 : (Datum) 0;
602
603 /*
604 * Elements of out.traversalValues should be allocated in
605 * in.traversalMemoryContext, which is actually a long lived context of
606 * index scan.
607 */
608 item->traversalValue =
609 out->traversalValues ? out->traversalValues[i] : NULL;
610
611 item->isLeaf = false;
612 item->recheck = false;
613 item->recheckDistances = false;
614
615 return item;
616}
617
618static void
619spgInnerTest(SpGistScanOpaque so, SpGistSearchItem *item,
620 SpGistInnerTuple innerTuple, bool isnull)
621{
622 MemoryContext oldCxt = MemoryContextSwitchTo(so->tempCxt);
623 spgInnerConsistentOut out;
624 int nNodes = innerTuple->nNodes;
625 int i;
626
627 memset(&out, 0, sizeof(out));
628
629 if (!isnull)
630 {
631 spgInnerConsistentIn in;
632
633 spgInitInnerConsistentIn(&in, so, item, innerTuple);
634
635 /* use user-defined inner consistent method */
636 FunctionCall2Coll(&so->innerConsistentFn,
637 so->indexCollation,
638 PointerGetDatum(&in),
639 PointerGetDatum(&out));
640 }
641 else
642 {
643 /* force all children to be visited */
644 out.nNodes = nNodes;
645 out.nodeNumbers = (int *) palloc(sizeof(int) * nNodes);
646 for (i = 0; i < nNodes; i++)
647 out.nodeNumbers[i] = i;
648 }
649
650 /* If allTheSame, they should all or none of them match */
651 if (innerTuple->allTheSame && out.nNodes != 0 && out.nNodes != nNodes)
652 elog(ERROR, "inconsistent inner_consistent results for allTheSame inner tuple");
653
654 if (out.nNodes)
655 {
656 /* collect node pointers */
657 SpGistNodeTuple node;
658 SpGistNodeTuple *nodes = (SpGistNodeTuple *) palloc(
659 sizeof(SpGistNodeTuple) * nNodes);
660
661 SGITITERATE(innerTuple, i, node)
662 {
663 nodes[i] = node;
664 }
665
666 MemoryContextSwitchTo(so->traversalCxt);
667
668 for (i = 0; i < out.nNodes; i++)
669 {
670 int nodeN = out.nodeNumbers[i];
671 SpGistSearchItem *innerItem;
672 double *distances;
673
674 Assert(nodeN >= 0 && nodeN < nNodes);
675
676 node = nodes[nodeN];
677
678 if (!ItemPointerIsValid(&node->t_tid))
679 continue;
680
681 /*
682 * Use infinity distances if innerConsistent() failed to return
683 * them or if is a NULL item (their distances are really unused).
684 */
685 distances = out.distances ? out.distances[i] : so->infDistances;
686
687 innerItem = spgMakeInnerItem(so, item, node, &out, i, isnull,
688 distances);
689
690 spgAddSearchItemToQueue(so, innerItem);
691 }
692 }
693
694 MemoryContextSwitchTo(oldCxt);
695}
696
697/* Returns a next item in an (ordered) scan or null if the index is exhausted */
698static SpGistSearchItem *
699spgGetNextQueueItem(SpGistScanOpaque so)
700{
701 if (pairingheap_is_empty(so->scanQueue))
702 return NULL; /* Done when both heaps are empty */
703
704 /* Return item; caller is responsible to pfree it */
705 return (SpGistSearchItem *) pairingheap_remove_first(so->scanQueue);
706}
707
708enum SpGistSpecialOffsetNumbers
709{
710 SpGistBreakOffsetNumber = InvalidOffsetNumber,
711 SpGistRedirectOffsetNumber = MaxOffsetNumber + 1,
712 SpGistErrorOffsetNumber = MaxOffsetNumber + 2
713};
714
715static OffsetNumber
716spgTestLeafTuple(SpGistScanOpaque so,
717 SpGistSearchItem *item,
718 Page page, OffsetNumber offset,
719 bool isnull, bool isroot,
720 bool *reportedSome,
721 storeRes_func storeRes)
722{
723 SpGistLeafTuple leafTuple = (SpGistLeafTuple)
724 PageGetItem(page, PageGetItemId(page, offset));
725
726 if (leafTuple->tupstate != SPGIST_LIVE)
727 {
728 if (!isroot) /* all tuples on root should be live */
729 {
730 if (leafTuple->tupstate == SPGIST_REDIRECT)
731 {
732 /* redirection tuple should be first in chain */
733 Assert(offset == ItemPointerGetOffsetNumber(&item->heapPtr));
734 /* transfer attention to redirect point */
735 item->heapPtr = ((SpGistDeadTuple) leafTuple)->pointer;
736 Assert(ItemPointerGetBlockNumber(&item->heapPtr) != SPGIST_METAPAGE_BLKNO);
737 return SpGistRedirectOffsetNumber;
738 }
739
740 if (leafTuple->tupstate == SPGIST_DEAD)
741 {
742 /* dead tuple should be first in chain */
743 Assert(offset == ItemPointerGetOffsetNumber(&item->heapPtr));
744 /* No live entries on this page */
745 Assert(leafTuple->nextOffset == InvalidOffsetNumber);
746 return SpGistBreakOffsetNumber;
747 }
748 }
749
750 /* We should not arrive at a placeholder */
751 elog(ERROR, "unexpected SPGiST tuple state: %d", leafTuple->tupstate);
752 return SpGistErrorOffsetNumber;
753 }
754
755 Assert(ItemPointerIsValid(&leafTuple->heapPtr));
756
757 spgLeafTest(so, item, leafTuple, isnull, reportedSome, storeRes);
758
759 return leafTuple->nextOffset;
760}
761
762/*
763 * Walk the tree and report all tuples passing the scan quals to the storeRes
764 * subroutine.
765 *
766 * If scanWholeIndex is true, we'll do just that. If not, we'll stop at the
767 * next page boundary once we have reported at least one tuple.
768 */
769static void
770spgWalk(Relation index, SpGistScanOpaque so, bool scanWholeIndex,
771 storeRes_func storeRes, Snapshot snapshot)
772{
773 Buffer buffer = InvalidBuffer;
774 bool reportedSome = false;
775
776 while (scanWholeIndex || !reportedSome)
777 {
778 SpGistSearchItem *item = spgGetNextQueueItem(so);
779
780 if (item == NULL)
781 break; /* No more items in queue -> done */
782
783redirect:
784 /* Check for interrupts, just in case of infinite loop */
785 CHECK_FOR_INTERRUPTS();
786
787 if (item->isLeaf)
788 {
789 /* We store heap items in the queue only in case of ordered search */
790 Assert(so->numberOfNonNullOrderBys > 0);
791 storeRes(so, &item->heapPtr, item->value, item->isNull,
792 item->recheck, item->recheckDistances, item->distances);
793 reportedSome = true;
794 }
795 else
796 {
797 BlockNumber blkno = ItemPointerGetBlockNumber(&item->heapPtr);
798 OffsetNumber offset = ItemPointerGetOffsetNumber(&item->heapPtr);
799 Page page;
800 bool isnull;
801
802 if (buffer == InvalidBuffer)
803 {
804 buffer = ReadBuffer(index, blkno);
805 LockBuffer(buffer, BUFFER_LOCK_SHARE);
806 }
807 else if (blkno != BufferGetBlockNumber(buffer))
808 {
809 UnlockReleaseBuffer(buffer);
810 buffer = ReadBuffer(index, blkno);
811 LockBuffer(buffer, BUFFER_LOCK_SHARE);
812 }
813
814 /* else new pointer points to the same page, no work needed */
815
816 page = BufferGetPage(buffer);
817 TestForOldSnapshot(snapshot, index, page);
818
819 isnull = SpGistPageStoresNulls(page) ? true : false;
820
821 if (SpGistPageIsLeaf(page))
822 {
823 /* Page is a leaf - that is, all it's tuples are heap items */
824 OffsetNumber max = PageGetMaxOffsetNumber(page);
825
826 if (SpGistBlockIsRoot(blkno))
827 {
828 /* When root is a leaf, examine all its tuples */
829 for (offset = FirstOffsetNumber; offset <= max; offset++)
830 (void) spgTestLeafTuple(so, item, page, offset,
831 isnull, true,
832 &reportedSome, storeRes);
833 }
834 else
835 {
836 /* Normal case: just examine the chain we arrived at */
837 while (offset != InvalidOffsetNumber)
838 {
839 Assert(offset >= FirstOffsetNumber && offset <= max);
840 offset = spgTestLeafTuple(so, item, page, offset,
841 isnull, false,
842 &reportedSome, storeRes);
843 if (offset == SpGistRedirectOffsetNumber)
844 goto redirect;
845 }
846 }
847 }
848 else /* page is inner */
849 {
850 SpGistInnerTuple innerTuple = (SpGistInnerTuple)
851 PageGetItem(page, PageGetItemId(page, offset));
852
853 if (innerTuple->tupstate != SPGIST_LIVE)
854 {
855 if (innerTuple->tupstate == SPGIST_REDIRECT)
856 {
857 /* transfer attention to redirect point */
858 item->heapPtr = ((SpGistDeadTuple) innerTuple)->pointer;
859 Assert(ItemPointerGetBlockNumber(&item->heapPtr) !=
860 SPGIST_METAPAGE_BLKNO);
861 goto redirect;
862 }
863 elog(ERROR, "unexpected SPGiST tuple state: %d",
864 innerTuple->tupstate);
865 }
866
867 spgInnerTest(so, item, innerTuple, isnull);
868 }
869 }
870
871 /* done with this scan item */
872 spgFreeSearchItem(so, item);
873 /* clear temp context before proceeding to the next one */
874 MemoryContextReset(so->tempCxt);
875 }
876
877 if (buffer != InvalidBuffer)
878 UnlockReleaseBuffer(buffer);
879}
880
881
882/* storeRes subroutine for getbitmap case */
883static void
884storeBitmap(SpGistScanOpaque so, ItemPointer heapPtr,
885 Datum leafValue, bool isnull, bool recheck, bool recheckDistances,
886 double *distances)
887{
888 Assert(!recheckDistances && !distances);
889 tbm_add_tuples(so->tbm, heapPtr, 1, recheck);
890 so->ntids++;
891}
892
893int64
894spggetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
895{
896 SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
897
898 /* Copy want_itup to *so so we don't need to pass it around separately */
899 so->want_itup = false;
900
901 so->tbm = tbm;
902 so->ntids = 0;
903
904 spgWalk(scan->indexRelation, so, true, storeBitmap, scan->xs_snapshot);
905
906 return so->ntids;
907}
908
909/* storeRes subroutine for gettuple case */
910static void
911storeGettuple(SpGistScanOpaque so, ItemPointer heapPtr,
912 Datum leafValue, bool isnull, bool recheck, bool recheckDistances,
913 double *nonNullDistances)
914{
915 Assert(so->nPtrs < MaxIndexTuplesPerPage);
916 so->heapPtrs[so->nPtrs] = *heapPtr;
917 so->recheck[so->nPtrs] = recheck;
918 so->recheckDistances[so->nPtrs] = recheckDistances;
919
920 if (so->numberOfOrderBys > 0)
921 {
922 if (isnull || so->numberOfNonNullOrderBys <= 0)
923 so->distances[so->nPtrs] = NULL;
924 else
925 {
926 IndexOrderByDistance *distances =
927 palloc(sizeof(distances[0]) * so->numberOfOrderBys);
928 int i;
929
930 for (i = 0; i < so->numberOfOrderBys; i++)
931 {
932 int offset = so->nonNullOrderByOffsets[i];
933
934 if (offset >= 0)
935 {
936 /* Copy non-NULL distance value */
937 distances[i].value = nonNullDistances[offset];
938 distances[i].isnull = false;
939 }
940 else
941 {
942 /* Set distance's NULL flag. */
943 distances[i].value = 0.0;
944 distances[i].isnull = true;
945 }
946 }
947
948 so->distances[so->nPtrs] = distances;
949 }
950 }
951
952 if (so->want_itup)
953 {
954 /*
955 * Reconstruct index data. We have to copy the datum out of the temp
956 * context anyway, so we may as well create the tuple here.
957 */
958 so->reconTups[so->nPtrs] = heap_form_tuple(so->indexTupDesc,
959 &leafValue,
960 &isnull);
961 }
962 so->nPtrs++;
963}
964
965bool
966spggettuple(IndexScanDesc scan, ScanDirection dir)
967{
968 SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
969
970 if (dir != ForwardScanDirection)
971 elog(ERROR, "SP-GiST only supports forward scan direction");
972
973 /* Copy want_itup to *so so we don't need to pass it around separately */
974 so->want_itup = scan->xs_want_itup;
975
976 for (;;)
977 {
978 if (so->iPtr < so->nPtrs)
979 {
980 /* continuing to return reported tuples */
981 scan->xs_heaptid = so->heapPtrs[so->iPtr];
982 scan->xs_recheck = so->recheck[so->iPtr];
983 scan->xs_hitup = so->reconTups[so->iPtr];
984
985 if (so->numberOfOrderBys > 0)
986 index_store_float8_orderby_distances(scan, so->orderByTypes,
987 so->distances[so->iPtr],
988 so->recheckDistances[so->iPtr]);
989 so->iPtr++;
990 return true;
991 }
992
993 if (so->numberOfOrderBys > 0)
994 {
995 /* Must pfree distances to avoid memory leak */
996 int i;
997
998 for (i = 0; i < so->nPtrs; i++)
999 if (so->distances[i])
1000 pfree(so->distances[i]);
1001 }
1002
1003 if (so->want_itup)
1004 {
1005 /* Must pfree reconstructed tuples to avoid memory leak */
1006 int i;
1007
1008 for (i = 0; i < so->nPtrs; i++)
1009 pfree(so->reconTups[i]);
1010 }
1011 so->iPtr = so->nPtrs = 0;
1012
1013 spgWalk(scan->indexRelation, so, false, storeGettuple,
1014 scan->xs_snapshot);
1015
1016 if (so->nPtrs == 0)
1017 break; /* must have completed scan */
1018 }
1019
1020 return false;
1021}
1022
1023bool
1024spgcanreturn(Relation index, int attno)
1025{
1026 SpGistCache *cache;
1027
1028 /* We can do it if the opclass config function says so */
1029 cache = spgGetCache(index);
1030
1031 return cache->config.canReturnData;
1032}
1033