text: /*-------------------------------------------------------------------------
   *
   * spgscan.c
   *	  routines for scanning SP-GiST indexes
   *
   *
   * Portions Copyright (c) 1996-2014, 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/relscan.h"
  #include "access/spgist_private.h"
  #include "miscadmin.h"
  #include "storage/bufmgr.h"
  #include "utils/datum.h"
  #include "utils/memutils.h"
  #include "utils/rel.h"
  
  
  typedef void (*storeRes_func) (SpGistScanOpaque so, ItemPointer heapPtr,
  								 Datum leafValue, bool isnull, bool recheck);
  
  typedef struct ScanStackEntry
  {
  	Datum		reconstructedValue;		/* value reconstructed from parent */
  	int			level;			/* level of items on this page */
  	ItemPointerData ptr;		/* block and offset to scan from */
  } ScanStackEntry;
  
  
  /* Free a ScanStackEntry */
  static void
  freeScanStackEntry(SpGistScanOpaque so, ScanStackEntry *stackEntry)
  {
  	if (!so->state.attType.attbyval &&
  		DatumGetPointer(stackEntry->reconstructedValue) != NULL)
  		pfree(DatumGetPointer(stackEntry->reconstructedValue));
  	pfree(stackEntry);
  }
  
  /* Free the entire stack */
  static void
  freeScanStack(SpGistScanOpaque so)
  {
  	ListCell   *lc;
  
  	foreach(lc, so->scanStack)
  	{
  		freeScanStackEntry(so, (ScanStackEntry *) lfirst(lc));
  	}
  	list_free(so->scanStack);
  	so->scanStack = NIL;
  }
  
  /*
   * Initialize scanStack to search the root page, resetting
   * any previously active scan
   */
  static void
  resetSpGistScanOpaque(SpGistScanOpaque so)
  {
  	ScanStackEntry *startEntry;
  
  	freeScanStack(so);
  
  	if (so->searchNulls)
  	{
  		/* Stack a work item to scan the null index entries */
  		startEntry = (ScanStackEntry *) palloc0(sizeof(ScanStackEntry));
  		ItemPointerSet(&startEntry->ptr, SPGIST_NULL_BLKNO, FirstOffsetNumber);
  		so->scanStack = lappend(so->scanStack, startEntry);
  	}
  
  	if (so->searchNonNulls)
  	{
  		/* Stack a work item to scan the non-null index entries */
  		startEntry = (ScanStackEntry *) palloc0(sizeof(ScanStackEntry));
  		ItemPointerSet(&startEntry->ptr, SPGIST_ROOT_BLKNO, FirstOffsetNumber);
  		so->scanStack = lappend(so->scanStack, startEntry);
  	}
  
  	if (so->want_itup)
  	{
  		/* Must pfree IndexTuples to avoid memory leak */
  		int			i;
  
  		for (i = 0; i < so->nPtrs; i++)
  			pfree(so->indexTups[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;
  
  	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;
  	}
  }
  
  Datum
  spgbeginscan(PG_FUNCTION_ARGS)
  {
  	Relation	rel = (Relation) PG_GETARG_POINTER(0);
  	int			keysz = PG_GETARG_INT32(1);
  
  	/* ScanKey			scankey = (ScanKey) PG_GETARG_POINTER(2); */
  	IndexScanDesc scan;
  	SpGistScanOpaque so;
  
  	scan = RelationGetIndexScan(rel, keysz, 0);
  
  	so = (SpGistScanOpaque) palloc0(sizeof(SpGistScanOpaqueData));
  	if (keysz > 0)
  		so->keyData = (ScanKey) palloc(sizeof(ScanKeyData) * keysz);
  	else
  		so->keyData = NULL;
  	initSpGistState(&so->state, scan->indexRelation);
  	so->tempCxt = AllocSetContextCreate(CurrentMemoryContext,
  										"SP-GiST search temporary context",
  										ALLOCSET_DEFAULT_MINSIZE,
  										ALLOCSET_DEFAULT_INITSIZE,
  										ALLOCSET_DEFAULT_MAXSIZE);
  
  	/* Set up indexTupDesc and xs_itupdesc in case it's an index-only scan */
  	so->indexTupDesc = scan->xs_itupdesc = RelationGetDescr(rel);
  
  	scan->opaque = so;
  
  	PG_RETURN_POINTER(scan);
  }
  
  Datum
  spgrescan(PG_FUNCTION_ARGS)
  {
  	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  	SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
  	ScanKey		scankey = (ScanKey) PG_GETARG_POINTER(1);
  
  	/* copy scankeys into local storage */
  	if (scankey && scan->numberOfKeys > 0)
  	{
  		memmove(scan->keyData, scankey,
  				scan->numberOfKeys * sizeof(ScanKeyData));
  	}
  
  	/* preprocess scankeys, set up the representation in *so */
  	spgPrepareScanKeys(scan);
  
  	/* set up starting stack entries */
  	resetSpGistScanOpaque(so);
  
  	PG_RETURN_VOID();
  }
  
  Datum
  spgendscan(PG_FUNCTION_ARGS)
  {
  	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  	SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
  
  	MemoryContextDelete(so->tempCxt);
  
  	PG_RETURN_VOID();
  }
  
  Datum
  spgmarkpos(PG_FUNCTION_ARGS)
  {
  	elog(ERROR, "SPGiST does not support mark/restore");
  	PG_RETURN_VOID();
  }
  
  Datum
  spgrestrpos(PG_FUNCTION_ARGS)
  {
  	elog(ERROR, "SPGiST does not support mark/restore");
  	PG_RETURN_VOID();
  }
  
  /*
   * Test whether a leaf tuple satisfies all the scan keys
   *
   * *leafValue is set to the reconstructed datum, if provided
   * *recheck is set true if any of the operators are lossy
   */
  static bool
  spgLeafTest(Relation index, SpGistScanOpaque so,
  			SpGistLeafTuple leafTuple, bool isnull,
  			int level, Datum reconstructedValue,
  			Datum *leafValue, bool *recheck)
  {
  	bool		result;
  	Datum		leafDatum;
  	spgLeafConsistentIn in;
  	spgLeafConsistentOut out;
  	FmgrInfo   *procinfo;
  	MemoryContext oldCtx;
  
  	if (isnull)
  	{
  		/* Should not have arrived on a nulls page unless nulls are wanted */
  		Assert(so->searchNulls);
  		*leafValue = (Datum) 0;
  		*recheck = false;
  		return true;
  	}
  
  	leafDatum = SGLTDATUM(leafTuple, &so->state);
  
  	/* use temp context for calling leaf_consistent */
  	oldCtx = MemoryContextSwitchTo(so->tempCxt);
  
  	in.scankeys = so->keyData;
  	in.nkeys = so->numberOfKeys;
  	in.reconstructedValue = reconstructedValue;
  	in.level = level;
  	in.returnData = so->want_itup;
  	in.leafDatum = leafDatum;
  
  	out.leafValue = (Datum) 0;
  	out.recheck = false;
  
  	procinfo = index_getprocinfo(index, 1, SPGIST_LEAF_CONSISTENT_PROC);
  	result = DatumGetBool(FunctionCall2Coll(procinfo,
  											index->rd_indcollation[0],
  											PointerGetDatum(&in),
  											PointerGetDatum(&out)));
  
  	*leafValue = out.leafValue;
  	*recheck = out.recheck;
  
  	MemoryContextSwitchTo(oldCtx);
  
  	return result;
  }
  
  /*
   * 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)
  	{
  		ScanStackEntry *stackEntry;
  		BlockNumber blkno;
  		OffsetNumber offset;
  		Page		page;
  		bool		isnull;
  
  		/* Pull next to-do item from the list */
  		if (so->scanStack == NIL)
  			break;				/* there are no more pages to scan */
  
  		stackEntry = (ScanStackEntry *) linitial(so->scanStack);
  		so->scanStack = list_delete_first(so->scanStack);
  
  redirect:
  		/* Check for interrupts, just in case of infinite loop */
  		CHECK_FOR_INTERRUPTS();
  
  		blkno = ItemPointerGetBlockNumber(&stackEntry->ptr);
  		offset = ItemPointerGetOffsetNumber(&stackEntry->ptr);
  
  		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))
  		{
  			SpGistLeafTuple leafTuple;
  			OffsetNumber max = PageGetMaxOffsetNumber(page);
  			Datum		leafValue = (Datum) 0;
  			bool		recheck = false;
  
  			if (SpGistBlockIsRoot(blkno))
  			{
  				/* When root is a leaf, examine all its tuples */
  				for (offset = FirstOffsetNumber; offset <= max; offset++)
  				{
  					leafTuple = (SpGistLeafTuple)
  						PageGetItem(page, PageGetItemId(page, offset));
  					if (leafTuple->tupstate != SPGIST_LIVE)
  					{
  						/* all tuples on root should be live */
  						elog(ERROR, "unexpected SPGiST tuple state: %d",
  							 leafTuple->tupstate);
  					}
  
  					Assert(ItemPointerIsValid(&leafTuple->heapPtr));
  					if (spgLeafTest(index, so,
  									leafTuple, isnull,
  									stackEntry->level,
  									stackEntry->reconstructedValue,
  									&leafValue,
  									&recheck))
  					{
  						storeRes(so, &leafTuple->heapPtr,
  								 leafValue, isnull, recheck);
  						reportedSome = true;
  					}
  				}
  			}
  			else
  			{
  				/* Normal case: just examine the chain we arrived at */
  				while (offset != InvalidOffsetNumber)
  				{
  					Assert(offset >= FirstOffsetNumber && offset <= max);
  					leafTuple = (SpGistLeafTuple)
  						PageGetItem(page, PageGetItemId(page, offset));
  					if (leafTuple->tupstate != SPGIST_LIVE)
  					{
  						if (leafTuple->tupstate == SPGIST_REDIRECT)
  						{
  							/* redirection tuple should be first in chain */
  							Assert(offset == ItemPointerGetOffsetNumber(&stackEntry->ptr));
  							/* transfer attention to redirect point */
  							stackEntry->ptr = ((SpGistDeadTuple) leafTuple)->pointer;
  							Assert(ItemPointerGetBlockNumber(&stackEntry->ptr) != SPGIST_METAPAGE_BLKNO);
  							goto redirect;
  						}
  						if (leafTuple->tupstate == SPGIST_DEAD)
  						{
  							/* dead tuple should be first in chain */
  							Assert(offset == ItemPointerGetOffsetNumber(&stackEntry->ptr));
  							/* No live entries on this page */
  							Assert(leafTuple->nextOffset == InvalidOffsetNumber);
  							break;
  						}
  						/* We should not arrive at a placeholder */
  						elog(ERROR, "unexpected SPGiST tuple state: %d",
  							 leafTuple->tupstate);
  					}
  
  					Assert(ItemPointerIsValid(&leafTuple->heapPtr));
  					if (spgLeafTest(index, so,
  									leafTuple, isnull,
  									stackEntry->level,
  									stackEntry->reconstructedValue,
  									&leafValue,
  									&recheck))
  					{
  						storeRes(so, &leafTuple->heapPtr,
  								 leafValue, isnull, recheck);
  						reportedSome = true;
  					}
  
  					offset = leafTuple->nextOffset;
  				}
  			}
  		}
  		else	/* page is inner */
  		{
  			SpGistInnerTuple innerTuple;
  			spgInnerConsistentIn in;
  			spgInnerConsistentOut out;
  			FmgrInfo   *procinfo;
  			SpGistNodeTuple *nodes;
  			SpGistNodeTuple node;
  			int			i;
  			MemoryContext oldCtx;
  
  			innerTuple = (SpGistInnerTuple) PageGetItem(page,
  												PageGetItemId(page, offset));
  
  			if (innerTuple->tupstate != SPGIST_LIVE)
  			{
  				if (innerTuple->tupstate == SPGIST_REDIRECT)
  				{
  					/* transfer attention to redirect point */
  					stackEntry->ptr = ((SpGistDeadTuple) innerTuple)->pointer;
  					Assert(ItemPointerGetBlockNumber(&stackEntry->ptr) != SPGIST_METAPAGE_BLKNO);
  					goto redirect;
  				}
  				elog(ERROR, "unexpected SPGiST tuple state: %d",
  					 innerTuple->tupstate);
  			}
  
  			/* use temp context for calling inner_consistent */
  			oldCtx = MemoryContextSwitchTo(so->tempCxt);
  
  			in.scankeys = so->keyData;
  			in.nkeys = so->numberOfKeys;
  			in.reconstructedValue = stackEntry->reconstructedValue;
  			in.level = stackEntry->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);
  
  			/* collect node pointers */
  			nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) * in.nNodes);
  			SGITITERATE(innerTuple, i, node)
  			{
  				nodes[i] = node;
  			}
  
  			memset(&out, 0, sizeof(out));
  
  			if (!isnull)
  			{
  				/* use user-defined inner consistent method */
  				procinfo = index_getprocinfo(index, 1, SPGIST_INNER_CONSISTENT_PROC);
  				FunctionCall2Coll(procinfo,
  								  index->rd_indcollation[0],
  								  PointerGetDatum(&in),
  								  PointerGetDatum(&out));
  			}
  			else
  			{
  				/* force all children to be visited */
  				out.nNodes = in.nNodes;
  				out.nodeNumbers = (int *) palloc(sizeof(int) * in.nNodes);
  				for (i = 0; i < in.nNodes; i++)
  					out.nodeNumbers[i] = i;
  			}
  
  			MemoryContextSwitchTo(oldCtx);
  
  			/* If allTheSame, they should all or none of 'em match */
  			if (innerTuple->allTheSame)
  				if (out.nNodes != 0 && out.nNodes != in.nNodes)
  					elog(ERROR, "inconsistent inner_consistent results for allTheSame inner tuple");
  
  			for (i = 0; i < out.nNodes; i++)
  			{
  				int			nodeN = out.nodeNumbers[i];
  
  				Assert(nodeN >= 0 && nodeN < in.nNodes);
  				if (ItemPointerIsValid(&nodes[nodeN]->t_tid))
  				{
  					ScanStackEntry *newEntry;
  
  					/* Create new work item for this node */
  					newEntry = palloc(sizeof(ScanStackEntry));
  					newEntry->ptr = nodes[nodeN]->t_tid;
  					if (out.levelAdds)
  						newEntry->level = stackEntry->level + out.levelAdds[i];
  					else
  						newEntry->level = stackEntry->level;
  					/* Must copy value out of temp context */
  					if (out.reconstructedValues)
  						newEntry->reconstructedValue =
  							datumCopy(out.reconstructedValues[i],
  									  so->state.attType.attbyval,
  									  so->state.attType.attlen);
  					else
  						newEntry->reconstructedValue = (Datum) 0;
  
  					so->scanStack = lcons(newEntry, so->scanStack);
  				}
  			}
  		}
  
  		/* done with this scan stack entry */
  		freeScanStackEntry(so, stackEntry);
  		/* clear temp context before proceeding to the next one */
  		MemoryContextReset(so->tempCxt);
  	}
  
  	if (buffer != InvalidBuffer)
  		UnlockReleaseBuffer(buffer);
  }
  
  /* storeRes subroutine for getbitmap case */
  static void
  storeBitmap(SpGistScanOpaque so, ItemPointer heapPtr,
  			Datum leafValue, bool isnull, bool recheck)
  {
  	tbm_add_tuples(so->tbm, heapPtr, 1, recheck);
  	so->ntids++;
  }
  
  Datum
  spggetbitmap(PG_FUNCTION_ARGS)
  {
  	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  	TIDBitmap  *tbm = (TIDBitmap *) PG_GETARG_POINTER(1);
  	SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
  
  	/* Copy want_itup to *so so we don't need to pass it around separately */
  	so->want_itup = false;
  
  	so->tbm = tbm;
  	so->ntids = 0;
  
  	spgWalk(scan->indexRelation, so, true, storeBitmap);
  
  	PG_RETURN_INT64(so->ntids);
  }
  
  /* storeRes subroutine for gettuple case */
  static void
  storeGettuple(SpGistScanOpaque so, ItemPointer heapPtr,
  			  Datum leafValue, bool isnull, bool recheck)
  {
  	Assert(so->nPtrs < MaxIndexTuplesPerPage);
  	so->heapPtrs[so->nPtrs] = *heapPtr;
  	so->recheck[so->nPtrs] = recheck;
  	if (so->want_itup)
  	{
  		/*
  		 * Reconstruct desired IndexTuple.  We have to copy the datum out of
  		 * the temp context anyway, so we may as well create the tuple here.
  		 */
  		so->indexTups[so->nPtrs] = index_form_tuple(so->indexTupDesc,
  													&leafValue,
  													&isnull);
  	}
  	so->nPtrs++;
  }
  
  Datum
  spggettuple(PG_FUNCTION_ARGS)
  {
  	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  	ScanDirection dir = (ScanDirection) PG_GETARG_INT32(1);
  	SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
  
  	if (dir != ForwardScanDirection)
  		elog(ERROR, "SP-GiST only supports forward scan direction");
  
  	/* Copy want_itup to *so so we don't need to pass it around separately */
  	so->want_itup = scan->xs_want_itup;
  
  	for (;;)
  	{
  		if (so->iPtr < so->nPtrs)
  		{
  			/* continuing to return tuples from a leaf page */
  			scan->xs_ctup.t_self = so->heapPtrs[so->iPtr];
  			scan->xs_recheck = so->recheck[so->iPtr];
  			scan->xs_itup = so->indexTups[so->iPtr];
  			so->iPtr++;
  			PG_RETURN_BOOL(true);
  		}
  
  		if (so->want_itup)
  		{
  			/* Must pfree IndexTuples to avoid memory leak */
  			int			i;
  
  			for (i = 0; i < so->nPtrs; i++)
  				pfree(so->indexTups[i]);
  		}
  		so->iPtr = so->nPtrs = 0;
  
  		spgWalk(scan->indexRelation, so, false, storeGettuple);
  
  		if (so->nPtrs == 0)
  			break;				/* must have completed scan */
  	}
  
  	PG_RETURN_BOOL(false);
  }
  
  Datum
  spgcanreturn(PG_FUNCTION_ARGS)
  {
  	Relation	index = (Relation) PG_GETARG_POINTER(0);
  	SpGistCache *cache;
  
  	/* We can do it if the opclass config function says so */
  	cache = spgGetCache(index);
  
  	PG_RETURN_BOOL(cache->config.canReturnData);
  }
  