/*-------------------------------------------------------------------------

 *

 * spgscan.c

 *    routines for scanning SP-GiST indexes

 *

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 * IDENTIFICATION

 *      src/backend/access/spgist/spgscan.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "access/genam.h"

#include "access/relscan.h"

#include "access/spgist_private.h"

#include "executor/instrument_node.h"

#include "miscadmin.h"

#include "pgstat.h"

#include "storage/bufmgr.h"

#include "utils/datum.h"

#include "utils/float.h"

#include "utils/lsyscache.h"

#include "utils/memutils.h"

#include "utils/rel.h"

typedef void (*storeRes_func) (SpGistScanOpaque so, ItemPointer heapPtr,

                 Datum leafValue, bool isNull,

                 SpGistLeafTuple leafTuple, bool recheck,

                 bool recheckDistances, double *distances);

/*

 * Pairing heap comparison function for the SpGistSearchItem queue.

 * KNN-searches currently only support NULLS LAST.  So, preserve this logic

 * here.

 */

static int

pairingheap_SpGistSearchItem_cmp(const pairingheap_node *a,

                 const pairingheap_node *b, void *arg)

{

  const SpGistSearchItem *sa = (const SpGistSearchItem *) a;

  const SpGistSearchItem *sb = (const SpGistSearchItem *) b;

  SpGistScanOpaque so = (SpGistScanOpaque) arg;

  int     i;

  if (sa->isNull)

  {

    if (!sb->isNull)

      return -1;

  }

  else if (sb->isNull)

  {

    return 1;

  }

  else

  {

    /* Order according to distance comparison */

    for (i = 0; i < so->numberOfNonNullOrderBys; i++)

    {

      if (isnan(sa->distances[i]) && isnan(sb->distances[i]))

        continue;   /* NaN == NaN */

      if (isnan(sa->distances[i]))

        return -1;   /* NaN > number */

      if (isnan(sb->distances[i]))

        return 1;   /* number < NaN */

      if (sa->distances[i] != sb->distances[i])

        return (sa->distances[i] < sb->distances[i]) ? 1 : -1;

    }

  }

  /* Leaf items go before inner pages, to ensure a depth-first search */

  if (sa->isLeaf && !sb->isLeaf)

    return 1;

  if (!sa->isLeaf && sb->isLeaf)

    return -1;

  return 0;

}

static void

spgFreeSearchItem(SpGistScanOpaque so, SpGistSearchItem *item)

{

  /* value is of type attType if isLeaf, else of type attLeafType */

  /* (no, that is not backwards; yes, it's confusing) */

  if (!(item->isLeaf ? so->state.attType.attbyval :

      so->state.attLeafType.attbyval) &&

    DatumGetPointer(item->value) != NULL)

    pfree(DatumGetPointer(item->value));

  if (item->leafTuple)

    pfree(item->leafTuple);

  if (item->traversalValue)

    pfree(item->traversalValue);

  pfree(item);

}

/*

 * Add SpGistSearchItem to queue

 *

 * Called in queue context

 */

static void

spgAddSearchItemToQueue(SpGistScanOpaque so, SpGistSearchItem *item)

{

  pairingheap_add(so->scanQueue, &item->phNode);

}

static SpGistSearchItem *

spgAllocSearchItem(SpGistScanOpaque so, bool isnull, double *distances)

{

  /* allocate distance array only for non-NULL items */

  SpGistSearchItem *item =

    palloc(SizeOfSpGistSearchItem(isnull ? 0 : so->numberOfNonNullOrderBys));

  item->isNull = isnull;

  if (!isnull && so->numberOfNonNullOrderBys > 0)

    memcpy(item->distances, distances,

         sizeof(item->distances[0]) * so->numberOfNonNullOrderBys);

  return item;

}

static void

spgAddStartItem(SpGistScanOpaque so, bool isnull)

{

  SpGistSearchItem *startEntry =

    spgAllocSearchItem(so, isnull, so->zeroDistances);

  ItemPointerSet(&startEntry->heapPtr,

           isnull ? SPGIST_NULL_BLKNO : SPGIST_ROOT_BLKNO,

           FirstOffsetNumber);

  startEntry->isLeaf = false;

  startEntry->level = 0;

  startEntry->value = (Datum) 0;

  startEntry->leafTuple = NULL;

  startEntry->traversalValue = NULL;

  startEntry->recheck = false;

  startEntry->recheckDistances = false;

  spgAddSearchItemToQueue(so, startEntry);

}

/*

 * Initialize queue to search the root page, resetting

 * any previously active scan

 */

static void

resetSpGistScanOpaque(SpGistScanOpaque so)

{

  MemoryContext oldCtx;

  MemoryContextReset(so->traversalCxt);

  oldCtx = MemoryContextSwitchTo(so->traversalCxt);

  /* initialize queue only for distance-ordered scans */

  so->scanQueue = pairingheap_allocate(pairingheap_SpGistSearchItem_cmp, so);

  if (so->searchNulls)

    /* Add a work item to scan the null index entries */

    spgAddStartItem(so, true);

  if (so->searchNonNulls)

    /* Add a work item to scan the non-null index entries */

    spgAddStartItem(so, false);

  MemoryContextSwitchTo(oldCtx);

  if (so->numberOfOrderBys > 0)

  {

    /* Must pfree distances to avoid memory leak */

    int     i;

    for (i = 0; i < so->nPtrs; i++)

      if (so->distances[i])

        pfree(so->distances[i]);

  }

  if (so->want_itup)

  {

    /* Must pfree reconstructed tuples to avoid memory leak */

    int     i;

    for (i = 0; i < so->nPtrs; i++)

      pfree(so->reconTups[i]);

  }

  so->iPtr = so->nPtrs = 0;

}

/*

 * Prepare scan keys in SpGistScanOpaque from caller-given scan keys

 *

 * Sets searchNulls, searchNonNulls, numberOfKeys, keyData fields of *so.

 *

 * The point here is to eliminate null-related considerations from what the

 * opclass consistent functions need to deal with.  We assume all SPGiST-

 * indexable operators are strict, so any null RHS value makes the scan

 * condition unsatisfiable.  We also pull out any IS NULL/IS NOT NULL

 * conditions; their effect is reflected into searchNulls/searchNonNulls.

 */

static void

spgPrepareScanKeys(IndexScanDesc scan)

{

  SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;

  bool    qual_ok;

  bool    haveIsNull;

  bool    haveNotNull;

  int     nkeys;

  int     i;

  so->numberOfOrderBys = scan->numberOfOrderBys;

  so->orderByData = scan->orderByData;

  if (so->numberOfOrderBys <= 0)

    so->numberOfNonNullOrderBys = 0;

  else

  {

    int     j = 0;

    /*

     * Remove all NULL keys, but remember their offsets in the original

     * array.

     */

    for (i = 0; i < scan->numberOfOrderBys; i++)

    {

      ScanKey   skey = &so->orderByData[i];

      if (skey->sk_flags & SK_ISNULL)

        so->nonNullOrderByOffsets[i] = -1;

      else

      {

        if (i != j)

          so->orderByData[j] = *skey;

        so->nonNullOrderByOffsets[i] = j++;

      }

    }

    so->numberOfNonNullOrderBys = j;

  }

  if (scan->numberOfKeys <= 0)

  {

    /* If no quals, whole-index scan is required */

    so->searchNulls = true;

    so->searchNonNulls = true;

    so->numberOfKeys = 0;

    return;

  }

  /* Examine the given quals */

  qual_ok = true;

  haveIsNull = haveNotNull = false;

  nkeys = 0;

  for (i = 0; i < scan->numberOfKeys; i++)

  {

    ScanKey   skey = &scan->keyData[i];

    if (skey->sk_flags & SK_SEARCHNULL)

      haveIsNull = true;

    else if (skey->sk_flags & SK_SEARCHNOTNULL)

      haveNotNull = true;

    else if (skey->sk_flags & SK_ISNULL)

    {

      /* ordinary qual with null argument - unsatisfiable */

      qual_ok = false;

      break;

    }

    else

    {

      /* ordinary qual, propagate into so->keyData */

      so->keyData[nkeys++] = *skey;

      /* this effectively creates a not-null requirement */

      haveNotNull = true;

    }

  }

  /* IS NULL in combination with something else is unsatisfiable */

  if (haveIsNull && haveNotNull)

    qual_ok = false;

  /* Emit results */

  if (qual_ok)

  {

    so->searchNulls = haveIsNull;

    so->searchNonNulls = haveNotNull;

    so->numberOfKeys = nkeys;

  }

  else

  {

    so->searchNulls = false;

    so->searchNonNulls = false;

    so->numberOfKeys = 0;

  }

}

IndexScanDesc

spgbeginscan(Relation rel, int keysz, int orderbysz)

{

  IndexScanDesc scan;

  SpGistScanOpaque so;

  int     i;

  scan = RelationGetIndexScan(rel, keysz, orderbysz);

  so = palloc0_object(SpGistScanOpaqueData);

  if (keysz > 0)

    so->keyData = palloc_array(ScanKeyData, keysz);

  else

    so->keyData = NULL;

  initSpGistState(&so->state, scan->indexRelation);

  so->tempCxt = AllocSetContextCreate(CurrentMemoryContext,

                    "SP-GiST search temporary context",

                    ALLOCSET_DEFAULT_SIZES);

  so->traversalCxt = AllocSetContextCreate(CurrentMemoryContext,

                       "SP-GiST traversal-value context",

                       ALLOCSET_DEFAULT_SIZES);

  /*

   * Set up reconTupDesc and xs_hitupdesc in case it's an index-only scan,

   * making sure that the key column is shown as being of type attType.

   * (It's rather annoying to do this work when it might be wasted, but for

   * most opclasses we can re-use the index reldesc instead of making one.)

   */

  so->reconTupDesc = scan->xs_hitupdesc =

    getSpGistTupleDesc(rel, &so->state.attType);

  /* Allocate various arrays needed for order-by scans */

  if (scan->numberOfOrderBys > 0)

  {

    /* This will be filled in spgrescan, but allocate the space here */

    so->orderByTypes = palloc_array(Oid, scan->numberOfOrderBys);

    so->nonNullOrderByOffsets = palloc_array(int, scan->numberOfOrderBys);

    /* These arrays have constant contents, so we can fill them now */

    so->zeroDistances = palloc_array(double, scan->numberOfOrderBys);

    so->infDistances = palloc_array(double, scan->numberOfOrderBys);

    for (i = 0; i < scan->numberOfOrderBys; i++)

    {

      so->zeroDistances[i] = 0.0;

      so->infDistances[i] = get_float8_infinity();

    }

    scan->xs_orderbyvals = palloc0_array(Datum, scan->numberOfOrderBys);

    scan->xs_orderbynulls = palloc_array(bool, scan->numberOfOrderBys);

    memset(scan->xs_orderbynulls, true,

         sizeof(bool) * scan->numberOfOrderBys);

  }

  fmgr_info_copy(&so->innerConsistentFn,

           index_getprocinfo(rel, 1, SPGIST_INNER_CONSISTENT_PROC),

           CurrentMemoryContext);

  fmgr_info_copy(&so->leafConsistentFn,

           index_getprocinfo(rel, 1, SPGIST_LEAF_CONSISTENT_PROC),

           CurrentMemoryContext);

  so->indexCollation = rel->rd_indcollation[0];

  scan->opaque = so;

  return scan;

}

void

spgrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,

      ScanKey orderbys, int norderbys)

{

  SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;

  /* copy scankeys into local storage */

  if (scankey && scan->numberOfKeys > 0)

    memcpy(scan->keyData, scankey, scan->numberOfKeys * sizeof(ScanKeyData));

  /* initialize order-by data if needed */

  if (orderbys && scan->numberOfOrderBys > 0)

  {

    int     i;

    memcpy(scan->orderByData, orderbys, scan->numberOfOrderBys * sizeof(ScanKeyData));

    for (i = 0; i < scan->numberOfOrderBys; i++)

    {

      ScanKey   skey = &scan->orderByData[i];

      /*

       * Look up the datatype returned by the original ordering

       * operator. SP-GiST always uses a float8 for the distance

       * function, but the ordering operator could be anything else.

       *

       * XXX: The distance function is only allowed to be lossy if the

       * ordering operator's result type is float4 or float8.  Otherwise

       * we don't know how to return the distance to the executor.  But

       * we cannot check that here, as we won't know if the distance

       * function is lossy until it returns *recheck = true for the

       * first time.

       */

      so->orderByTypes[i] = get_func_rettype(skey->sk_func.fn_oid);

    }

  }

  /* preprocess scankeys, set up the representation in *so */

  spgPrepareScanKeys(scan);

  /* set up starting queue entries */

  resetSpGistScanOpaque(so);

  /* count an indexscan for stats */

  pgstat_count_index_scan(scan->indexRelation);

  if (scan->instrument)

    scan->instrument->nsearches++;

}

void

spgendscan(IndexScanDesc scan)

{

  SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;

  MemoryContextDelete(so->tempCxt);

  MemoryContextDelete(so->traversalCxt);

  if (so->keyData)

    pfree(so->keyData);

  if (so->state.leafTupDesc &&

    so->state.leafTupDesc != RelationGetDescr(so->state.index))

    FreeTupleDesc(so->state.leafTupDesc);

  if (so->state.deadTupleStorage)

    pfree(so->state.deadTupleStorage);

  if (scan->numberOfOrderBys > 0)

  {

    pfree(so->orderByTypes);

    pfree(so->nonNullOrderByOffsets);

    pfree(so->zeroDistances);

    pfree(so->infDistances);

    pfree(scan->xs_orderbyvals);

    pfree(scan->xs_orderbynulls);

  }

  pfree(so);

}

/*

 * Leaf SpGistSearchItem constructor, called in queue context

 */

static SpGistSearchItem *

spgNewHeapItem(SpGistScanOpaque so, int level, SpGistLeafTuple leafTuple,

         Datum leafValue, bool recheck, bool recheckDistances,

         bool isnull, double *distances)

{

  SpGistSearchItem *item = spgAllocSearchItem(so, isnull, distances);

  item->level = level;

  item->heapPtr = leafTuple->heapPtr;

  /*

   * If we need the reconstructed value, copy it to queue cxt out of tmp

   * cxt.  Caution: the leaf_consistent method may not have supplied a value

   * if we didn't ask it to, and mildly-broken methods might supply one of

   * the wrong type.  The correct leafValue type is attType not leafType.

   */

  if (so->want_itup)

  {

    item->value = isnull ? (Datum) 0 :

      datumCopy(leafValue, so->state.attType.attbyval,

            so->state.attType.attlen);

    /*

     * If we're going to need to reconstruct INCLUDE attributes, store the

     * whole leaf tuple so we can get the INCLUDE attributes out of it.

     */

    if (so->state.leafTupDesc->natts > 1)

    {

      item->leafTuple = palloc(leafTuple->size);

      memcpy(item->leafTuple, leafTuple, leafTuple->size);

    }

    else

      item->leafTuple = NULL;

  }

  else

  {

    item->value = (Datum) 0;

    item->leafTuple = NULL;

  }

  item->traversalValue = NULL;

  item->isLeaf = true;

  item->recheck = recheck;

  item->recheckDistances = recheckDistances;

  return item;

}

/*

 * Test whether a leaf tuple satisfies all the scan keys

 *

 * *reportedSome is set to true if:

 *    the scan is not ordered AND the item satisfies the scankeys

 */

static bool

spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item,

      SpGistLeafTuple leafTuple, bool isnull,

      bool *reportedSome, storeRes_func storeRes)

{

  Datum   leafValue;

  double     *distances;

  bool    result;

  bool    recheck;

  bool    recheckDistances;

  if (isnull)

  {

    /* Should not have arrived on a nulls page unless nulls are wanted */

    Assert(so->searchNulls);

    leafValue = (Datum) 0;

    distances = NULL;

    recheck = false;

    recheckDistances = false;

    result = true;

  }

  else

  {

    spgLeafConsistentIn in;

    spgLeafConsistentOut out;

    /* use temp context for calling leaf_consistent */

    MemoryContext oldCxt = MemoryContextSwitchTo(so->tempCxt);

    in.scankeys = so->keyData;

    in.nkeys = so->numberOfKeys;

    in.orderbys = so->orderByData;

    in.norderbys = so->numberOfNonNullOrderBys;

    Assert(!item->isLeaf);  /* else reconstructedValue would be wrong type */

    in.reconstructedValue = item->value;

    in.traversalValue = item->traversalValue;

    in.level = item->level;

    in.returnData = so->want_itup;

    in.leafDatum = SGLTDATUM(leafTuple, &so->state);

    out.leafValue = (Datum) 0;

    out.recheck = false;

    out.distances = NULL;

    out.recheckDistances = false;

    result = DatumGetBool(FunctionCall2Coll(&so->leafConsistentFn,

                        so->indexCollation,

                        PointerGetDatum(&in),

                        PointerGetDatum(&out)));

    recheck = out.recheck;

    recheckDistances = out.recheckDistances;

    leafValue = out.leafValue;

    distances = out.distances;

    MemoryContextSwitchTo(oldCxt);

  }

  if (result)

  {

    /* item passes the scankeys */

    if (so->numberOfNonNullOrderBys > 0)

    {

      /* the scan is ordered -> add the item to the queue */

      MemoryContext oldCxt = MemoryContextSwitchTo(so->traversalCxt);

      SpGistSearchItem *heapItem = spgNewHeapItem(so, item->level,

                            leafTuple,

                            leafValue,

                            recheck,

                            recheckDistances,

                            isnull,

                            distances);

      spgAddSearchItemToQueue(so, heapItem);

      MemoryContextSwitchTo(oldCxt);

    }

    else

    {

      /* non-ordered scan, so report the item right away */

      Assert(!recheckDistances);

      storeRes(so, &leafTuple->heapPtr, leafValue, isnull,

           leafTuple, recheck, false, NULL);

      *reportedSome = true;

    }

  }

  return result;

}

/* A bundle initializer for inner_consistent methods */

static void

spgInitInnerConsistentIn(spgInnerConsistentIn *in,

             SpGistScanOpaque so,

             SpGistSearchItem *item,

             SpGistInnerTuple innerTuple)

{

  in->scankeys = so->keyData;

  in->orderbys = so->orderByData;

  in->nkeys = so->numberOfKeys;

  in->norderbys = so->numberOfNonNullOrderBys;

  Assert(!item->isLeaf);    /* else reconstructedValue would be wrong type */

  in->reconstructedValue = item->value;

  in->traversalMemoryContext = so->traversalCxt;

  in->traversalValue = item->traversalValue;

  in->level = item->level;

  in->returnData = so->want_itup;

  in->allTheSame = innerTuple->allTheSame;

  in->hasPrefix = (innerTuple->prefixSize > 0);

  in->prefixDatum = SGITDATUM(innerTuple, &so->state);

  in->nNodes = innerTuple->nNodes;

  in->nodeLabels = spgExtractNodeLabels(&so->state, innerTuple);

}

static SpGistSearchItem *

spgMakeInnerItem(SpGistScanOpaque so,

         SpGistSearchItem *parentItem,

         SpGistNodeTuple tuple,

         spgInnerConsistentOut *out, int i, bool isnull,

         double *distances)

{

  SpGistSearchItem *item = spgAllocSearchItem(so, isnull, distances);

  item->heapPtr = tuple->t_tid;

  item->level = out->levelAdds ? parentItem->level + out->levelAdds[i]

    : parentItem->level;

  /* Must copy value out of temp context */

  /* (recall that reconstructed values are of type leafType) */

  item->value = out->reconstructedValues

    ? datumCopy(out->reconstructedValues[i],

          so->state.attLeafType.attbyval,

          so->state.attLeafType.attlen)

    : (Datum) 0;

  item->leafTuple = NULL;

  /*

   * Elements of out.traversalValues should be allocated in

   * in.traversalMemoryContext, which is actually a long lived context of

   * index scan.

   */

  item->traversalValue =

    out->traversalValues ? out->traversalValues[i] : NULL;

  item->isLeaf = false;

  item->recheck = false;

  item->recheckDistances = false;

  return item;

}

static void

spgInnerTest(SpGistScanOpaque so, SpGistSearchItem *item,

       SpGistInnerTuple innerTuple, bool isnull)

{

  MemoryContext oldCxt = MemoryContextSwitchTo(so->tempCxt);

  spgInnerConsistentOut out;

  int     nNodes = innerTuple->nNodes;

  int     i;

  memset(&out, 0, sizeof(out));

  if (!isnull)

  {

    spgInnerConsistentIn in;

    spgInitInnerConsistentIn(&in, so, item, innerTuple);

    /* use user-defined inner consistent method */

    FunctionCall2Coll(&so->innerConsistentFn,

              so->indexCollation,

              PointerGetDatum(&in),

              PointerGetDatum(&out));

  }

  else

  {

    /* force all children to be visited */

    out.nNodes = nNodes;

    out.nodeNumbers = palloc_array(int, nNodes);

    for (i = 0; i < nNodes; i++)

      out.nodeNumbers[i] = i;

  }

  /* If allTheSame, they should all or none of them match */

  if (innerTuple->allTheSame && out.nNodes != 0 && out.nNodes != nNodes)

    elog(ERROR, "inconsistent inner_consistent results for allTheSame inner tuple");

  if (out.nNodes)

  {

    /* collect node pointers */

    SpGistNodeTuple node;

    SpGistNodeTuple *nodes = palloc_array(SpGistNodeTuple, nNodes);

    SGITITERATE(innerTuple, i, node)

    {

      nodes[i] = node;

    }

    MemoryContextSwitchTo(so->traversalCxt);

    for (i = 0; i < out.nNodes; i++)

    {

      int     nodeN = out.nodeNumbers[i];

      SpGistSearchItem *innerItem;

      double     *distances;

      Assert(nodeN >= 0 && nodeN < nNodes);

      node = nodes[nodeN];

      if (!ItemPointerIsValid(&node->t_tid))

        continue;

      /*

       * Use infinity distances if innerConsistentFn() failed to return

       * them or if is a NULL item (their distances are really unused).

       */

      distances = out.distances ? out.distances[i] : so->infDistances;

      innerItem = spgMakeInnerItem(so, item, node, &out, i, isnull,

                     distances);

      spgAddSearchItemToQueue(so, innerItem);

    }

  }

  MemoryContextSwitchTo(oldCxt);

}

/* Returns a next item in an (ordered) scan or null if the index is exhausted */

static SpGistSearchItem *

spgGetNextQueueItem(SpGistScanOpaque so)

{

  if (pairingheap_is_empty(so->scanQueue))

    return NULL;     /* Done when both heaps are empty */

  /* Return item; caller is responsible to pfree it */

  return (SpGistSearchItem *) pairingheap_remove_first(so->scanQueue);

}

enum SpGistSpecialOffsetNumbers

{

  SpGistBreakOffsetNumber = InvalidOffsetNumber,

  SpGistRedirectOffsetNumber = MaxOffsetNumber + 1,

  SpGistErrorOffsetNumber = MaxOffsetNumber + 2,

};

static OffsetNumber

spgTestLeafTuple(SpGistScanOpaque so,

         SpGistSearchItem *item,

         Page page, OffsetNumber offset,

         bool isnull, bool isroot,

         bool *reportedSome,

         storeRes_func storeRes)

{

  SpGistLeafTuple leafTuple = (SpGistLeafTuple)

    PageGetItem(page, PageGetItemId(page, offset));

  if (leafTuple->tupstate != SPGIST_LIVE)

  {

    if (!isroot)     /* all tuples on root should be live */

    {

      if (leafTuple->tupstate == SPGIST_REDIRECT)

      {

        /* redirection tuple should be first in chain */

        Assert(offset == ItemPointerGetOffsetNumber(&item->heapPtr));

        /* transfer attention to redirect point */

        item->heapPtr = ((SpGistDeadTuple) leafTuple)->pointer;

        Assert(ItemPointerGetBlockNumber(&item->heapPtr) != SPGIST_METAPAGE_BLKNO);

        return SpGistRedirectOffsetNumber;

      }

      if (leafTuple->tupstate == SPGIST_DEAD)

      {

        /* dead tuple should be first in chain */

        Assert(offset == ItemPointerGetOffsetNumber(&item->heapPtr));

        /* No live entries on this page */

        Assert(SGLT_GET_NEXTOFFSET(leafTuple) == InvalidOffsetNumber);

        return SpGistBreakOffsetNumber;

      }

    }

    /* We should not arrive at a placeholder */

    elog(ERROR, "unexpected SPGiST tuple state: %d", leafTuple->tupstate);

    return SpGistErrorOffsetNumber;

  }

  Assert(ItemPointerIsValid(&leafTuple->heapPtr));

  spgLeafTest(so, item, leafTuple, isnull, reportedSome, storeRes);

  return SGLT_GET_NEXTOFFSET(leafTuple);

}

/*

 * Walk the tree and report all tuples passing the scan quals to the storeRes

 * subroutine.

 *

 * If scanWholeIndex is true, we'll do just that.  If not, we'll stop at the

 * next page boundary once we have reported at least one tuple.

 */

static void

spgWalk(Relation index, SpGistScanOpaque so, bool scanWholeIndex,

    storeRes_func storeRes)

{

  Buffer    buffer = InvalidBuffer;

  bool    reportedSome = false;

  while (scanWholeIndex || !reportedSome)

  {

    SpGistSearchItem *item = spgGetNextQueueItem(so);

    if (item == NULL)

      break;       /* No more items in queue -> done */

redirect:

    /* Check for interrupts, just in case of infinite loop */

    CHECK_FOR_INTERRUPTS();

    if (item->isLeaf)

    {

      /* We store heap items in the queue only in case of ordered search */

      Assert(so->numberOfNonNullOrderBys > 0);

      storeRes(so, &item->heapPtr, item->value, item->isNull,

           item->leafTuple, item->recheck,

           item->recheckDistances, item->distances);

      reportedSome = true;

    }

    else

    {

      BlockNumber blkno = ItemPointerGetBlockNumber(&item->heapPtr);

      OffsetNumber offset = ItemPointerGetOffsetNumber(&item->heapPtr);

      Page    page;

      bool    isnull;

      if (buffer == InvalidBuffer)

      {

        buffer = ReadBuffer(index, blkno);

        LockBuffer(buffer, BUFFER_LOCK_SHARE);

      }

      else if (blkno != BufferGetBlockNumber(buffer))

      {

        UnlockReleaseBuffer(buffer);

        buffer = ReadBuffer(index, blkno);

        LockBuffer(buffer, BUFFER_LOCK_SHARE);

      }

      /* else new pointer points to the same page, no work needed */

      page = BufferGetPage(buffer);

      isnull = SpGistPageStoresNulls(page) ? true : false;

      if (SpGistPageIsLeaf(page))

      {

        /* Page is a leaf - that is, all its tuples are heap items */

        OffsetNumber max = PageGetMaxOffsetNumber(page);

        if (SpGistBlockIsRoot(blkno))

        {

          /* When root is a leaf, examine all its tuples */

          for (offset = FirstOffsetNumber; offset <= max; offset++)

            (void) spgTestLeafTuple(so, item, page, offset,

                        isnull, true,

                        &reportedSome, storeRes);

        }

        else

        {

          /* Normal case: just examine the chain we arrived at */

          while (offset != InvalidOffsetNumber)

          {

            Assert(offset >= FirstOffsetNumber && offset <= max);

            offset = spgTestLeafTuple(so, item, page, offset,

                          isnull, false,

                          &reportedSome, storeRes);

            if (offset == SpGistRedirectOffsetNumber)

              goto redirect;

          }

        }

      }

      else        /* page is inner */

      {

        SpGistInnerTuple innerTuple = (SpGistInnerTuple)

          PageGetItem(page, PageGetItemId(page, offset));

        if (innerTuple->tupstate != SPGIST_LIVE)

        {