gist.c

关系型数据库 Postgresql 6.5.2

/*-------------------------------------------------------------------------
 *
 * gist.c
 *	  interface routines for the postgres GiST index access method.
 *
 *
 *
 * IDENTIFICATION
 *	  $Header: /usr/local/cvsroot/pgsql/src/backend/access/gist/gist.c,v 1.38.2.1 1999/08/02 05:24:28 scrappy Exp $
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/genam.h"
#include "access/gist.h"
#include "access/gistscan.h"
#include "access/heapam.h"
#include "catalog/index.h"
#include "catalog/pg_index.h"
#include "executor/executor.h"
#include "utils/syscache.h"


/* non-export function prototypes */
static InsertIndexResult gistdoinsert(Relation r, IndexTuple itup,
			 GISTSTATE *GISTstate);
static InsertIndexResult gistentryinsert(Relation r, GISTSTACK *stk,
				IndexTuple tup,
				GISTSTATE *giststate);
static void gistentryinserttwo(Relation r, GISTSTACK *stk, IndexTuple ltup,
				   IndexTuple rtup, GISTSTATE *giststate);
static void gistAdjustKeys(Relation r, GISTSTACK *stk, BlockNumber blk,
			   char *datum, int att_size, GISTSTATE *giststate);
static void gistintinsert(Relation r, GISTSTACK *stk, IndexTuple ltup,
			  IndexTuple rtup, GISTSTATE *giststate);
static InsertIndexResult gistSplit(Relation r, Buffer buffer,
		  GISTSTACK *stack, IndexTuple itup,
		  GISTSTATE *giststate);
static void gistnewroot(GISTSTATE *giststate, Relation r, IndexTuple lt,
			IndexTuple rt);
static void GISTInitBuffer(Buffer b, uint32 f);
static BlockNumber gistChooseSubtree(Relation r, IndexTuple itup, int level,
				  GISTSTATE *giststate,
				  GISTSTACK **retstack, Buffer *leafbuf);
static OffsetNumber gistchoose(Relation r, Page p, IndexTuple it,
		   GISTSTATE *giststate);
static int	gistnospace(Page p, IndexTuple it);
void		gistdelete(Relation r, ItemPointer tid);
static IndexTuple gist_tuple_replacekey(Relation r, GISTENTRY entry, IndexTuple t);
static void gistcentryinit(GISTSTATE *giststate, GISTENTRY *e, char *pr,
			   Relation r, Page pg, OffsetNumber o, int b, bool l);
#ifdef GISTDEBUG
static char *int_range_out(INTRANGE *r);
#endif

/*
** routine to build an index.  Basically calls insert over and over
*/
void
gistbuild(Relation heap,
		  Relation index,
		  int natts,
		  AttrNumber *attnum,
		  IndexStrategy istrat,
		  uint16 pint,
		  Datum *params,
		  FuncIndexInfo *finfo,
		  PredInfo *predInfo)
{
	HeapScanDesc scan;
	AttrNumber	i;
	HeapTuple	htup;
	IndexTuple	itup;
	TupleDesc	hd,
				id;
	InsertIndexResult res;
	Datum	   *d;
	bool	   *nulls;
	int			nb,
				nh,
				ni;

#ifndef OMIT_PARTIAL_INDEX
	ExprContext *econtext;
	TupleTable	tupleTable;
	TupleTableSlot *slot;

#endif
	Oid			hrelid,
				irelid;
	Node	   *pred,
			   *oldPred;
	GISTSTATE	giststate;
	GISTENTRY	tmpcentry;
	Buffer		buffer = InvalidBuffer;
	bool	   *compvec;

	/* no locking is needed */

	setheapoverride(true);		/* so we can see the new pg_index tuple */
	initGISTstate(&giststate, index);
	setheapoverride(false);

	pred = predInfo->pred;
	oldPred = predInfo->oldPred;

	/*
	 * We expect to be called exactly once for any index relation. If
	 * that's not the case, big trouble's what we have.
	 */

	if (oldPred == NULL && (nb = RelationGetNumberOfBlocks(index)) != 0)
		elog(ERROR, "%s already contains data", index->rd_rel->relname.data);

	/* initialize the root page (if this is a new index) */
	if (oldPred == NULL)
	{
		buffer = ReadBuffer(index, P_NEW);
		GISTInitBuffer(buffer, F_LEAF);
		WriteBuffer(buffer);
	}

	/* init the tuple descriptors and get set for a heap scan */
	hd = RelationGetDescr(heap);
	id = RelationGetDescr(index);
	d = (Datum *) palloc(natts * sizeof(*d));
	nulls = (bool *) palloc(natts * sizeof(*nulls));

	/*
	 * If this is a predicate (partial) index, we will need to evaluate
	 * the predicate using ExecQual, which requires the current tuple to
	 * be in a slot of a TupleTable.  In addition, ExecQual must have an
	 * ExprContext referring to that slot.	Here, we initialize dummy
	 * TupleTable and ExprContext objects for this purpose. --Nels, Feb
	 * '92
	 */
#ifndef OMIT_PARTIAL_INDEX
	if (pred != NULL || oldPred != NULL)
	{
		tupleTable = ExecCreateTupleTable(1);
		slot = ExecAllocTableSlot(tupleTable);
		econtext = makeNode(ExprContext);
		FillDummyExprContext(econtext, slot, hd, buffer);
	}
	else
/* shut the compiler up */
	{
		tupleTable = NULL;
		slot = NULL;
		econtext = NULL;
	}
#endif	 /* OMIT_PARTIAL_INDEX */
	/* int the tuples as we insert them */
	nh = ni = 0;

	scan = heap_beginscan(heap, 0, SnapshotNow, 0, (ScanKey) NULL);

	while (HeapTupleIsValid(htup = heap_getnext(scan, 0)))
	{
		nh++;

		/*
		 * If oldPred != NULL, this is an EXTEND INDEX command, so skip
		 * this tuple if it was already in the existing partial index
		 */
		if (oldPred != NULL)
		{
#ifndef OMIT_PARTIAL_INDEX
			/* SetSlotContents(slot, htup); */
			slot->val = htup;
			if (ExecQual((List *) oldPred, econtext) == true)
			{
				ni++;
				continue;
			}
#endif	 /* OMIT_PARTIAL_INDEX */
		}

		/*
		 * Skip this tuple if it doesn't satisfy the partial-index
		 * predicate
		 */
		if (pred != NULL)
		{
#ifndef OMIT_PARTIAL_INDEX
			/* SetSlotContents(slot, htup); */
			slot->val = htup;
			if (ExecQual((List *) pred, econtext) == false)
				continue;
#endif	 /* OMIT_PARTIAL_INDEX */
		}

		ni++;

		/*
		 * For the current heap tuple, extract all the attributes we use
		 * in this index, and note which are null.
		 */

		for (i = 1; i <= natts; i++)
		{
			int			attoff;
			bool		attnull;

			/*
			 * Offsets are from the start of the tuple, and are
			 * zero-based; indices are one-based.  The next call returns i
			 * - 1.  That's data hiding for you.
			 */

			attoff = AttrNumberGetAttrOffset(i);

			/*
			 * d[attoff] = HeapTupleGetAttributeValue(htup, buffer,
			 */
			d[attoff] = GetIndexValue(htup,
									  hd,
									  attoff,
									  attnum,
									  finfo,
									  &attnull);
			nulls[attoff] = (attnull ? 'n' : ' ');
		}

		/* immediately compress keys to normalize */
		compvec = (bool *) palloc(sizeof(bool) * natts);
		for (i = 0; i < natts; i++)
		{
			gistcentryinit(&giststate, &tmpcentry, (char *) d[i],
						   (Relation) NULL, (Page) NULL, (OffsetNumber) 0,
						   -1 /* size is currently bogus */ , TRUE);
			if (d[i] != (Datum) tmpcentry.pred && !(giststate.keytypbyval))
				compvec[i] = TRUE;
			else
				compvec[i] = FALSE;
			d[i] = (Datum) tmpcentry.pred;
		}

		/* form an index tuple and point it at the heap tuple */
		itup = index_formtuple(id, &d[0], nulls);
		itup->t_tid = htup->t_self;

		/*
		 * Since we already have the index relation locked, we call
		 * gistdoinsert directly.  Normal access method calls dispatch
		 * through gistinsert, which locks the relation for write.	This
		 * is the right thing to do if you're inserting single tups, but
		 * not when you're initializing the whole index at once.
		 */

		res = gistdoinsert(index, itup, &giststate);
		for (i = 0; i < natts; i++)
			if (compvec[i] == TRUE)
				pfree((char *) d[i]);
		pfree(itup);
		pfree(res);
		pfree(compvec);
	}

	/* okay, all heap tuples are indexed */
	heap_endscan(scan);

	if (pred != NULL || oldPred != NULL)
	{
#ifndef OMIT_PARTIAL_INDEX
		ExecDestroyTupleTable(tupleTable, true);
		pfree(econtext);
#endif	 /* OMIT_PARTIAL_INDEX */
	}

	/*
	 * Since we just inted the tuples in the heap, we update its stats in
	 * pg_relation to guarantee that the planner takes advantage of the
	 * index we just created.  UpdateStats() does a
	 * CommandinterIncrement(), which flushes changed entries from the
	 * system relcache.  The act of constructing an index changes these
	 * heap and index tuples in the system catalogs, so they need to be
	 * flushed.  We close them to guarantee that they will be.
	 */

	hrelid = RelationGetRelid(heap);
	irelid = RelationGetRelid(index);
	heap_close(heap);
	index_close(index);

	UpdateStats(hrelid, nh, true);
	UpdateStats(irelid, ni, false);

	if (oldPred != NULL)
	{
		if (ni == nh)
			pred = NULL;
		UpdateIndexPredicate(irelid, oldPred, pred);
	}

	/* be tidy */
	pfree(nulls);
	pfree(d);
}

/*
 *	gistinsert -- wrapper for GiST tuple insertion.
 *
 *	  This is the public interface routine for tuple insertion in GiSTs.
 *	  It doesn't do any work; just locks the relation and passes the buck.
 */
InsertIndexResult
gistinsert(Relation r, Datum *datum, char *nulls, ItemPointer ht_ctid, Relation heapRel)
{
	InsertIndexResult res;
	IndexTuple	itup;
	GISTSTATE	giststate;
	GISTENTRY	tmpentry;
	int			i;
	bool	   *compvec;

	initGISTstate(&giststate, r);

	/* immediately compress keys to normalize */
	compvec = (bool *) palloc(sizeof(bool) * r->rd_att->natts);
	for (i = 0; i < r->rd_att->natts; i++)
	{
		gistcentryinit(&giststate, &tmpentry, (char *) datum[i],
					   (Relation) NULL, (Page) NULL, (OffsetNumber) 0,
					   -1 /* size is currently bogus */ , TRUE);
		if (datum[i] != (Datum) tmpentry.pred && !(giststate.keytypbyval))
			compvec[i] = TRUE;
		else
			compvec[i] = FALSE;
		datum[i] = (Datum) tmpentry.pred;
	}
	itup = index_formtuple(RelationGetDescr(r), datum, nulls);
	itup->t_tid = *ht_ctid;

	/*
	 * Notes in ExecUtils:ExecOpenIndices()
	 *
	 * RelationSetLockForWrite(r);
	 */

	res = gistdoinsert(r, itup, &giststate);
	for (i = 0; i < r->rd_att->natts; i++)
		if (compvec[i] == TRUE)
			pfree((char *) datum[i]);
	pfree(itup);
	pfree(compvec);

	return res;
}

/*
** Take a compressed entry, and install it on a page.  Since we now know
** where the entry will live, we decompress it and recompress it using
** that knowledge (some compression routines may want to fish around
** on the page, for example, or do something special for leaf nodes.)
*/
static OffsetNumber
gistPageAddItem(GISTSTATE *giststate,
				Relation r,
				Page page,
				Item item,
				Size size,
				OffsetNumber offsetNumber,
				ItemIdFlags flags,
				GISTENTRY *dentry,
				IndexTuple *newtup)
{
	GISTENTRY	tmpcentry;
	IndexTuple	itup = (IndexTuple) item;

	/*
	 * recompress the item given that we now know the exact page and
	 * offset for insertion
	 */
	gistdentryinit(giststate, dentry,
				   (((char *) itup) + sizeof(IndexTupleData)),
			  (Relation) 0, (Page) 0, (OffsetNumber) InvalidOffsetNumber,
				   IndexTupleSize(itup) - sizeof(IndexTupleData), FALSE);
	gistcentryinit(giststate, &tmpcentry, dentry->pred, r, page,
				   offsetNumber, dentry->bytes, FALSE);
	*newtup = gist_tuple_replacekey(r, *dentry, itup);
	/* be tidy */
	if (tmpcentry.pred != dentry->pred
		&& tmpcentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
		pfree(tmpcentry.pred);

	return (PageAddItem(page, (Item) *newtup, IndexTupleSize(*newtup),
						offsetNumber, flags));
}


static InsertIndexResult
gistdoinsert(Relation r,
			 IndexTuple itup,	/* itup contains compressed entry */
			 GISTSTATE *giststate)
{
	GISTENTRY	tmpdentry;
	InsertIndexResult res;
	OffsetNumber l;
	GISTSTACK  *stack;
	Buffer		buffer;
	BlockNumber blk;
	Page		page;
	OffsetNumber off;
	IndexTuple	newtup;

	/* 3rd arg is ignored for now */
	blk = gistChooseSubtree(r, itup, 0, giststate, &stack, &buffer);
	page = (Page) BufferGetPage(buffer);

	if (gistnospace(page, itup))
	{
		/* need to do a split */
		res = gistSplit(r, buffer, stack, itup, giststate);
		gistfreestack(stack);
		WriteBuffer(buffer);	/* don't forget to release buffer! */
		return res;
	}

	if (PageIsEmpty(page))
		off = FirstOffsetNumber;
	else
		off = OffsetNumberNext(PageGetMaxOffsetNumber(page));

	/* add the item and write the buffer */
	l = gistPageAddItem(giststate, r, page, (Item) itup, IndexTupleSize(itup),
						off, LP_USED, &tmpdentry, &newtup);
	WriteBuffer(buffer);

	/* now expand the page boundary in the parent to include the new child */
	gistAdjustKeys(r, stack, blk, tmpdentry.pred, tmpdentry.bytes, giststate);
	gistfreestack(stack);

	/* be tidy */
	if (itup != newtup)
		pfree(newtup);
	if (tmpdentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
		pfree(tmpdentry.pred);

	/* build and return an InsertIndexResult for this insertion */
	res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
	ItemPointerSet(&(res->pointerData), blk, l);

	return res;
}