gist.c

关系型数据库 Postgresql 6.5.2

static BlockNumber
gistChooseSubtree(Relation r, IndexTuple itup,	/* itup has compressed
												 * entry */
				  int level,
				  GISTSTATE *giststate,
				  GISTSTACK **retstack /* out */ ,
				  Buffer *leafbuf /* out */ )
{
	Buffer		buffer;
	BlockNumber blk;
	GISTSTACK  *stack;
	Page		page;
	GISTPageOpaque opaque;
	IndexTuple	which;

	blk = GISTP_ROOT;
	buffer = InvalidBuffer;
	stack = (GISTSTACK *) 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 = (GISTPageOpaque) PageGetSpecialPointer(page);
		if (!(opaque->flags & F_LEAF))
		{
			GISTSTACK  *n;
			ItemId		iid;

			n = (GISTSTACK *) palloc(sizeof(GISTSTACK));
			n->gs_parent = stack;
			n->gs_blk = blk;
			n->gs_child = gistchoose(r, page, itup, giststate);
			stack = n;

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

	*retstack = stack;
	*leafbuf = buffer;

	return blk;
}


static void
gistAdjustKeys(Relation r,
			   GISTSTACK *stk,
			   BlockNumber blk,
			   char *datum,		/* datum is uncompressed */
			   int att_size,
			   GISTSTATE *giststate)
{
	char	   *oldud;
	Page		p;
	Buffer		b;
	bool		result;
	bytea	   *evec;
	GISTENTRY	centry,
			   *ev0p,
			   *ev1p;
	int			size,
				datumsize;
	IndexTuple	tid;

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

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

	oldud = (char *) PageGetItem(p, PageGetItemId(p, stk->gs_child));
	tid = (IndexTuple) oldud;
	size = IndexTupleSize((IndexTuple) oldud) - sizeof(IndexTupleData);
	oldud += sizeof(IndexTupleData);

	evec = (bytea *) palloc(2 * sizeof(GISTENTRY) + VARHDRSZ);
	VARSIZE(evec) = 2 * sizeof(GISTENTRY) + VARHDRSZ;

	/* insert decompressed oldud into entry vector */
	gistdentryinit(giststate, &((GISTENTRY *) VARDATA(evec))[0],
				   oldud, r, p, stk->gs_child,
				   size, FALSE);
	ev0p = &((GISTENTRY *) VARDATA(evec))[0];

	/* insert datum entry into entry vector */
	gistentryinit(((GISTENTRY *) VARDATA(evec))[1], datum,
		(Relation) NULL, (Page) NULL, (OffsetNumber) 0, att_size, FALSE);
	ev1p = &((GISTENTRY *) VARDATA(evec))[1];

	/* form union of decompressed entries */
	datum = (*fmgr_faddr(&giststate->unionFn)) (evec, &datumsize);

	/* did union leave decompressed version of oldud unchanged? */
	(*fmgr_faddr(&giststate->equalFn)) (ev0p->pred, datum, &result);
	if (!result)
	{
		TupleDesc	td = RelationGetDescr(r);

		/* compress datum for storage on page */
		gistcentryinit(giststate, &centry, datum, ev0p->rel, ev0p->page,
					   ev0p->offset, datumsize, FALSE);
		if (td->attrs[0]->attlen >= 0)
		{
			memmove(oldud, centry.pred, att_size);
			gistAdjustKeys(r, stk->gs_parent, stk->gs_blk, datum, att_size,
						   giststate);
		}
		else if (VARSIZE(centry.pred) == VARSIZE(oldud))
		{
			memmove(oldud, centry.pred, VARSIZE(centry.pred));
			gistAdjustKeys(r, stk->gs_parent, stk->gs_blk, datum, att_size,
						   giststate);
		}
		else
		{

			/*
			 * * new datum is not the same size as the old. * We have to
			 * delete the old entry and insert the new * one.  Note that
			 * this may cause a split here!
			 */
			IndexTuple	newtup;
			ItemPointerData oldtid;
			char	   *isnull;
			TupleDesc	tupDesc;
			InsertIndexResult res;

			/* delete old tuple */
			ItemPointerSet(&oldtid, stk->gs_blk, stk->gs_child);
			gistdelete(r, (ItemPointer) &oldtid);

			/* generate and insert new tuple */
			tupDesc = r->rd_att;
			isnull = (char *) palloc(r->rd_rel->relnatts);
			MemSet(isnull, ' ', r->rd_rel->relnatts);
			newtup = (IndexTuple) index_formtuple(tupDesc,
										 (Datum *) &centry.pred, isnull);
			pfree(isnull);
			/* set pointer in new tuple to point to current child */
			ItemPointerSet(&oldtid, blk, 1);
			newtup->t_tid = oldtid;

			/* inserting the new entry also adjust keys above */
			res = gistentryinsert(r, stk, newtup, giststate);

			/* in stack, set info to point to new tuple */
			stk->gs_blk = ItemPointerGetBlockNumber(&(res->pointerData));
			stk->gs_child = ItemPointerGetOffsetNumber(&(res->pointerData));

			pfree(res);
		}
		WriteBuffer(b);

		if (centry.pred != datum)
			pfree(datum);
	}
	else
		ReleaseBuffer(b);
	pfree(evec);
}

/*
 *	gistSplit -- split a page in the tree.
 *
 */
static InsertIndexResult
gistSplit(Relation r,
		  Buffer buffer,
		  GISTSTACK *stack,
		  IndexTuple itup,		/* contains compressed entry */
		  GISTSTATE *giststate)
{
	Page		p;
	Buffer		leftbuf,
				rightbuf;
	Page		left,
				right;
	ItemId		itemid;
	IndexTuple	item;
	IndexTuple	ltup,
				rtup,
				newtup;
	OffsetNumber maxoff;
	OffsetNumber i;
	OffsetNumber leftoff,
				rightoff;
	BlockNumber lbknum,
				rbknum;
	BlockNumber bufblock;
	GISTPageOpaque opaque;
	int			blank;
	InsertIndexResult res;
	char	   *isnull;
	GIST_SPLITVEC v;
	TupleDesc	tupDesc;
	bytea	   *entryvec;
	bool	   *decompvec;
	IndexTuple	item_1;
	GISTENTRY	tmpdentry,
				tmpentry;

	isnull = (char *) palloc(r->rd_rel->relnatts);
	for (blank = 0; blank < r->rd_rel->relnatts; blank++)
		isnull[blank] = ' ';
	p = (Page) BufferGetPage(buffer);
	opaque = (GISTPageOpaque) 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) == GISTP_ROOT)
	{
		leftbuf = ReadBuffer(r, P_NEW);
		GISTInitBuffer(leftbuf, opaque->flags);
		lbknum = BufferGetBlockNumber(leftbuf);
		left = (Page) BufferGetPage(leftbuf);
	}
	else
	{
		leftbuf = buffer;
		IncrBufferRefCount(buffer);
		lbknum = BufferGetBlockNumber(buffer);
		left = (Page) PageGetTempPage(p, sizeof(GISTPageOpaqueData));
	}

	rightbuf = ReadBuffer(r, P_NEW);
	GISTInitBuffer(rightbuf, opaque->flags);
	rbknum = BufferGetBlockNumber(rightbuf);
	right = (Page) BufferGetPage(rightbuf);

	/* generate the item array */
	maxoff = PageGetMaxOffsetNumber(p);
	entryvec = (bytea *) palloc(VARHDRSZ + (maxoff + 2) * sizeof(GISTENTRY));
	decompvec = (bool *) palloc(VARHDRSZ + (maxoff + 2) * sizeof(bool));
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		item_1 = (IndexTuple) PageGetItem(p, PageGetItemId(p, i));
		gistdentryinit(giststate, &((GISTENTRY *) VARDATA(entryvec))[i],
					   (((char *) item_1) + sizeof(IndexTupleData)),
					   r, p, i,
				 IndexTupleSize(item_1) - sizeof(IndexTupleData), FALSE);
		if ((char *) (((GISTENTRY *) VARDATA(entryvec))[i].pred)
			== (((char *) item_1) + sizeof(IndexTupleData)))
			decompvec[i] = FALSE;
		else
			decompvec[i] = TRUE;
	}

	/* add the new datum as the last entry */
	gistdentryinit(giststate, &(((GISTENTRY *) VARDATA(entryvec))[maxoff + 1]),
				   (((char *) itup) + sizeof(IndexTupleData)),
				   (Relation) NULL, (Page) NULL,
				   (OffsetNumber) 0, tmpentry.bytes, FALSE);
	if ((char *) (((GISTENTRY *) VARDATA(entryvec))[maxoff + 1]).pred !=
		(((char *) itup) + sizeof(IndexTupleData)))
		decompvec[maxoff + 1] = TRUE;
	else
		decompvec[maxoff + 1] = FALSE;

	VARSIZE(entryvec) = (maxoff + 2) * sizeof(GISTENTRY) + VARHDRSZ;

	/* now let the user-defined picksplit function set up the split vector */
	(*fmgr_faddr(&giststate->picksplitFn)) (entryvec, &v);

	/* compress ldatum and rdatum */
	gistcentryinit(giststate, &tmpentry, v.spl_ldatum, (Relation) NULL,
				   (Page) NULL, (OffsetNumber) 0,
				   ((GISTENTRY *) VARDATA(entryvec))[i].bytes, FALSE);
	if (v.spl_ldatum != tmpentry.pred)
		pfree(v.spl_ldatum);
	v.spl_ldatum = tmpentry.pred;

	gistcentryinit(giststate, &tmpentry, v.spl_rdatum, (Relation) NULL,
				   (Page) NULL, (OffsetNumber) 0,
				   ((GISTENTRY *) VARDATA(entryvec))[i].bytes, FALSE);
	if (v.spl_rdatum != tmpentry.pred)
		pfree(v.spl_rdatum);
	v.spl_rdatum = tmpentry.pred;

	/* clean up the entry vector: its preds need to be deleted, too */
	for (i = FirstOffsetNumber; i <= maxoff + 1; i = OffsetNumberNext(i))
		if (decompvec[i])
			pfree(((GISTENTRY *) VARDATA(entryvec))[i].pred);
	pfree(entryvec);
	pfree(decompvec);

	leftoff = rightoff = FirstOffsetNumber;
	maxoff = PageGetMaxOffsetNumber(p);
	for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
	{
		itemid = PageGetItemId(p, i);
		item = (IndexTuple) PageGetItem(p, itemid);

		if (i == *(v.spl_left))
		{
			gistPageAddItem(giststate, r, left, (Item) item,
							IndexTupleSize(item),
							leftoff, LP_USED, &tmpdentry, &newtup);
			leftoff = OffsetNumberNext(leftoff);
			v.spl_left++;		/* advance in left split vector */
			/* be tidy */
			if (tmpdentry.pred != (((char *) item) + sizeof(IndexTupleData)))
				pfree(tmpdentry.pred);
			if ((IndexTuple) item != newtup)
				pfree(newtup);
		}
		else
		{
			gistPageAddItem(giststate, r, right, (Item) item,
							IndexTupleSize(item),
							rightoff, LP_USED, &tmpdentry, &newtup);
			rightoff = OffsetNumberNext(rightoff);
			v.spl_right++;		/* advance in right split vector */
			/* be tidy */
			if (tmpdentry.pred != (((char *) item) + sizeof(IndexTupleData)))
				pfree(tmpdentry.pred);
			if (item != newtup)
				pfree(newtup);
		}
	}

	/* build an InsertIndexResult for this insertion */
	res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));

	/* now insert the new index tuple */
	if (*(v.spl_left) != FirstOffsetNumber)
	{
		gistPageAddItem(giststate, r, left, (Item) itup,
						IndexTupleSize(itup),
						leftoff, LP_USED, &tmpdentry, &newtup);
		leftoff = OffsetNumberNext(leftoff);
		ItemPointerSet(&(res->pointerData), lbknum, leftoff);
		/* be tidy */
		if (tmpdentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
			pfree(tmpdentry.pred);
		if (itup != newtup)
			pfree(newtup);
	}
	else
	{
		gistPageAddItem(giststate, r, right, (Item) itup,
						IndexTupleSize(itup),
						rightoff, LP_USED, &tmpdentry, &newtup);
		rightoff = OffsetNumberNext(rightoff);
		ItemPointerSet(&(res->pointerData), rbknum, rightoff);
		/* be tidy */
		if (tmpdentry.pred != (((char *) itup) + sizeof(IndexTupleData)))
			pfree(tmpdentry.pred);
		if (itup != newtup)
			pfree(newtup);
	}

	if ((bufblock = BufferGetBlockNumber(buffer)) != GISTP_ROOT)
		PageRestoreTempPage(left, p);
	WriteBuffer(leftbuf);
	WriteBuffer(rightbuf);

	/*
	 * Okay, the page is split.  We have three things left to do:
	 *
	 * 1)  Adjust any active scans on this index to cope with changes we
	 * introduced in its structure by splitting this page.
	 *
	 * 2)  "Tighten" the bounding box of the pointer to the left page in the
	 * parent node in the tree, if any.  Since we moved a bunch of stuff
	 * off the left page, we expect it to get smaller.	This happens in
	 * the internal insertion routine.
	 *
	 * 3)  Insert a pointer to the right page in the parent.  This may cause
	 * the parent to split.  If it does, we need to repeat steps one and
	 * two for each split node in the tree.
	 */

	/* adjust active scans */
	gistadjscans(r, GISTOP_SPLIT, bufblock, FirstOffsetNumber);

	tupDesc = r->rd_att;

	ltup = (IndexTuple) index_formtuple(tupDesc,
									  (Datum *) &(v.spl_ldatum), isnull);
	rtup = (IndexTuple) index_formtuple(tupDesc,
									  (Datum *) &(v.spl_rdatum), isnull);
	pfree(isnull);

	/* set pointers to new child pages in the internal index tuples */
	ItemPointerSet(&(ltup->t_tid), lbknum, 1);
	ItemPointerSet(&(rtup->t_tid), rbknum, 1);

	gistintinsert(r, stack, ltup, rtup, giststate);

	pfree(ltup);
	pfree(rtup);

	return res;
}

/*
** After a split, we need to overwrite the old entry's key in the parent,
** and install install an entry for the new key into the parent.
*/
static void
gistintinsert(Relation r,
			  GISTSTACK *stk,
			  IndexTuple ltup,	/* new version of entry for old page */
			  IndexTuple rtup,	/* entry for new page */
			  GISTSTATE *giststate)
{
	ItemPointerData ltid;

	if (stk == (GISTSTACK *) NULL)
	{
		gistnewroot(giststate, r, ltup, rtup);
		return;
	}

	/* remove old left pointer, insert the 2 new entries */
	ItemPointerSet(&ltid, stk->gs_blk, stk->gs_child);
	gistdelete(r, (ItemPointer) &ltid);
	gistentryinserttwo(r, stk, ltup, rtup, giststate);
}


/*
** Insert two entries onto one page, handling a split for either one!
*/
static void
gistentryinserttwo(Relation r, GISTSTACK *stk, IndexTuple ltup,
				   IndexTuple rtup, GISTSTATE *giststate)
{
	Buffer		b;
	Page		p;