rtree.c

关系型数据库 Postgresql 6.5.2

/*-------------------------------------------------------------------------
 *
 * rtree.c
 *	  interface routines for the postgres rtree indexed access method.
 *
 * Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  $Header: /usr/local/cvsroot/pgsql/src/backend/access/rtree/rtree.c,v 1.32.2.1 1999/08/02 05:24:44 scrappy Exp $
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/rtree.h"
#include "catalog/index.h"
#include "executor/executor.h"
#include "utils/geo_decls.h"


typedef struct SPLITVEC
{
	OffsetNumber *spl_left;
	int			spl_nleft;
	char	   *spl_ldatum;
	OffsetNumber *spl_right;
	int			spl_nright;
	char	   *spl_rdatum;
} SPLITVEC;

typedef struct RTSTATE
{
	FmgrInfo	unionFn;		/* union function */
	FmgrInfo	sizeFn;			/* size function */
	FmgrInfo	interFn;		/* intersection function */
} RTSTATE;

/* non-export function prototypes */
static InsertIndexResult rtdoinsert(Relation r, IndexTuple itup,
		   RTSTATE *rtstate);
static void rttighten(Relation r, RTSTACK *stk, char *datum, int att_size,
		  RTSTATE *rtstate);
static InsertIndexResult dosplit(Relation r, Buffer buffer, RTSTACK *stack,
		IndexTuple itup, RTSTATE *rtstate);
static void rtintinsert(Relation r, RTSTACK *stk, IndexTuple ltup,
			IndexTuple rtup, RTSTATE *rtstate);
static void rtnewroot(Relation r, IndexTuple lt, IndexTuple rt);
static void picksplit(Relation r, Page page, SPLITVEC *v, IndexTuple itup,
		  RTSTATE *rtstate);
static void RTInitBuffer(Buffer b, uint32 f);
static OffsetNumber choose(Relation r, Page p, IndexTuple it,
	   RTSTATE *rtstate);
static int	nospace(Page p, IndexTuple it);
static void initRtstate(RTSTATE *rtstate, Relation index);


void
rtbuild(Relation heap,
		Relation index,
		int natts,
		AttrNumber *attnum,
		IndexStrategy istrat,
		uint16 pcount,
		Datum *params,
		FuncIndexInfo *finfo,
		PredInfo *predInfo)
{
	HeapScanDesc scan;
	AttrNumber	i;
	HeapTuple	htup;
	IndexTuple	itup;
	TupleDesc	hd,
				id;
	InsertIndexResult res;
	Datum	   *d;
	bool	   *nulls;
	Buffer		buffer = InvalidBuffer;
	int			nb,
				nh,
				ni;

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

#endif
	Oid			hrelid,
				irelid;
	Node	   *pred,
			   *oldPred;
	RTSTATE		rtState;

	initRtstate(&rtState, index);

	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);
		RTInitBuffer(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
	{
		econtext = NULL;
		tupleTable = NULL;
		slot = NULL;
	}
#endif	 /* OMIT_PARTIAL_INDEX */

	/* count 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' : ' ');
		}

		/* 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
		 * rtdoinsert directly.  Normal access method calls dispatch
		 * through rtinsert, 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 = rtdoinsert(index, itup, &rtState);
		pfree(itup);
		pfree(res);
	}

	/* 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 counted 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
	 * CommandCounterIncrement(), 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);
}

/*
 *	rtinsert -- wrapper for rtree tuple insertion.
 *
 *	  This is the public interface routine for tuple insertion in rtrees.
 *	  It doesn't do any work; just locks the relation and passes the buck.
 */
InsertIndexResult
rtinsert(Relation r, Datum *datum, char *nulls, ItemPointer ht_ctid, Relation heapRel)
{
	InsertIndexResult res;
	IndexTuple	itup;
	RTSTATE		rtState;

	/* generate an index tuple */
	itup = index_formtuple(RelationGetDescr(r), datum, nulls);
	itup->t_tid = *ht_ctid;
	initRtstate(&rtState, r);

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

	res = rtdoinsert(r, itup, &rtState);

	return res;
}

static InsertIndexResult
rtdoinsert(Relation r, IndexTuple itup, RTSTATE *rtstate)
{
	Page		page;
	Buffer		buffer;
	BlockNumber blk;
	IndexTuple	which;
	OffsetNumber l;
	RTSTACK    *stack;
	InsertIndexResult res;
	RTreePageOpaque opaque;
	char	   *datum;

	blk = P_ROOT;
	buffer = InvalidBuffer;
	stack = (RTSTACK *) NULL;

	do
	{
		/* let go of current buffer before getting next */
		if (buffer != InvalidBuffer)
			ReleaseBuffer(buffer);

		/* get next buffer */
		buffer = ReadBuffer(r, blk);
		page = (Page) BufferGetPage(buffer);

		opaque = (RTreePageOpaque) PageGetSpecialPointer(page);
		if (!(opaque->flags & F_LEAF))
		{
			RTSTACK    *n;
			ItemId		iid;

			n = (RTSTACK *) palloc(sizeof(RTSTACK));
			n->rts_parent = stack;
			n->rts_blk = blk;
			n->rts_child = choose(r, page, itup, rtstate);
			stack = n;

			iid = PageGetItemId(page, n->rts_child);
			which = (IndexTuple) PageGetItem(page, iid);
			blk = ItemPointerGetBlockNumber(&(which->t_tid));
		}
	} while (!(opaque->flags & F_LEAF));

	if (nospace(page, itup))
	{
		/* need to do a split */
		res = dosplit(r, buffer, stack, itup, rtstate);
		freestack(stack);
		WriteBuffer(buffer);	/* don't forget to release buffer! */
		return res;
	}

	/* add the item and write the buffer */
	if (PageIsEmpty(page))
	{
		l = PageAddItem(page, (Item) itup, IndexTupleSize(itup),
						FirstOffsetNumber,
						LP_USED);
	}
	else
	{
		l = PageAddItem(page, (Item) itup, IndexTupleSize(itup),
						OffsetNumberNext(PageGetMaxOffsetNumber(page)),
						LP_USED);
	}

	WriteBuffer(buffer);

	datum = (((char *) itup) + sizeof(IndexTupleData));

	/* now expand the page boundary in the parent to include the new child */
	rttighten(r, stack, datum,
			  (IndexTupleSize(itup) - sizeof(IndexTupleData)), rtstate);
	freestack(stack);

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

	return res;
}

static void
rttighten(Relation r,
		  RTSTACK *stk,
		  char *datum,
		  int att_size,
		  RTSTATE *rtstate)
{
	char	   *oldud;
	char	   *tdatum;
	Page		p;
	float		old_size,
				newd_size;
	Buffer		b;

	if (stk == (RTSTACK *) NULL)
		return;

	b = ReadBuffer(r, stk->rts_blk);
	p = BufferGetPage(b);

	oldud = (char *) PageGetItem(p, PageGetItemId(p, stk->rts_child));
	oldud += sizeof(IndexTupleData);

	(*fmgr_faddr(&rtstate->sizeFn)) (oldud, &old_size);
	datum = (char *) (*fmgr_faddr(&rtstate->unionFn)) (oldud, datum);

	(*fmgr_faddr(&rtstate->sizeFn)) (datum, &newd_size);

	if (newd_size != old_size)
	{
		TupleDesc	td = RelationGetDescr(r);

		if (td->attrs[0]->attlen < 0)
		{

			/*
			 * This is an internal page, so 'oldud' had better be a union
			 * (constant-length) key, too.	(See comment below.)
			 */
			Assert(VARSIZE(datum) == VARSIZE(oldud));
			memmove(oldud, datum, VARSIZE(datum));
		}
		else
			memmove(oldud, datum, att_size);
		WriteBuffer(b);

		/*
		 * The user may be defining an index on variable-sized data (like
		 * polygons).  If so, we need to get a constant-sized datum for
		 * insertion on the internal page.	We do this by calling the
		 * union proc, which is guaranteed to return a rectangle.
		 */

		tdatum = (char *) (*fmgr_faddr(&rtstate->unionFn)) (datum, datum);
		rttighten(r, stk->rts_parent, tdatum, att_size, rtstate);
		pfree(tdatum);
	}
	else
		ReleaseBuffer(b);
	pfree(datum);
}

/*
 *	dosplit -- split a page in the tree.
 *
 *	  This is the quadratic-cost split algorithm Guttman describes in
 *	  his paper.  The reason we chose it is that you can implement this
 *	  with less information about the data types on which you're operating.
 */
static InsertIndexResult
dosplit(Relation r,
		Buffer buffer,
		RTSTACK *stack,
		IndexTuple itup,
		RTSTATE *rtstate)
{
	Page		p;
	Buffer		leftbuf,
				rightbuf;
	Page		left,
				right;
	ItemId		itemid;
	IndexTuple	item;
	IndexTuple	ltup,
				rtup;
	OffsetNumber maxoff;
	OffsetNumber i;
	OffsetNumber leftoff,
				rightoff;
	BlockNumber lbknum,
				rbknum;
	BlockNumber bufblock;
	RTreePageOpaque opaque;
	int			blank;
	InsertIndexResult res;
	char	   *isnull;
	SPLITVEC	v;
	TupleDesc	tupDesc;

	isnull = (char *) palloc(r->rd_rel->relnatts);
	for (blank = 0; blank < r->rd_rel->relnatts; blank++)
		isnull[blank] = ' ';
	p = (Page) BufferGetPage(buffer);
	opaque = (RTreePageOpaque) PageGetSpecialPointer(p);

	/*
	 * The root of the tree is the first block in the relation.  If we're
	 * about to split the root, we need to do some hocus-pocus to enforce
	 * this guarantee.
	 */

	if (BufferGetBlockNumber(buffer) == P_ROOT)
	{
		leftbuf = ReadBuffer(r, P_NEW);
		RTInitBuffer(leftbuf, opaque->flags);
		lbknum = BufferGetBlockNumber(leftbuf);
		left = (Page) BufferGetPage(leftbuf);
	}
	else
	{
		leftbuf = buffer;
		IncrBufferRefCount(buffer);
		lbknum = BufferGetBlockNumber(buffer);
		left = (Page) PageGetTempPage(p, sizeof(RTreePageOpaqueData));
	}