gistsplit.c

postgresql8.3.4源码,开源数据库

/*-------------------------------------------------------------------------
 *
 * gistsplit.c
 *	  Split page algorithm
 *
 *
 * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  $PostgreSQL: pgsql/src/backend/access/gist/gistsplit.c,v 1.5 2008/01/01 19:45:46 momjian Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/gist_private.h"

typedef struct
{
	Datum	   *attr;
	int			len;
	OffsetNumber *entries;
	bool	   *isnull;
	bool	   *equiv;
} GistSplitUnion;


/*
 * Forms unions of subkeys after page split, but
 * uses only tuples aren't in groups of equalent tuples
 */
static void
gistunionsubkeyvec(GISTSTATE *giststate, IndexTuple *itvec,
				   GistSplitUnion *gsvp, int startkey)
{
	IndexTuple *cleanedItVec;
	int			i,
				cleanedLen = 0;

	cleanedItVec = (IndexTuple *) palloc(sizeof(IndexTuple) * gsvp->len);

	for (i = 0; i < gsvp->len; i++)
	{
		if (gsvp->equiv && gsvp->equiv[gsvp->entries[i]])
			continue;

		cleanedItVec[cleanedLen++] = itvec[gsvp->entries[i] - 1];
	}

	gistMakeUnionItVec(giststate, cleanedItVec, cleanedLen, startkey,
					   gsvp->attr, gsvp->isnull);

	pfree(cleanedItVec);
}

/*
 * unions subkeys for after user picksplit over attno-1 column
 */
static void
gistunionsubkey(GISTSTATE *giststate, IndexTuple *itvec, GistSplitVector *spl, int attno)
{
	GistSplitUnion gsvp;

	gsvp.equiv = spl->spl_equiv;

	gsvp.attr = spl->spl_lattr;
	gsvp.len = spl->splitVector.spl_nleft;
	gsvp.entries = spl->splitVector.spl_left;
	gsvp.isnull = spl->spl_lisnull;

	gistunionsubkeyvec(giststate, itvec, &gsvp, attno);

	gsvp.attr = spl->spl_rattr;
	gsvp.len = spl->splitVector.spl_nright;
	gsvp.entries = spl->splitVector.spl_right;
	gsvp.isnull = spl->spl_risnull;

	gistunionsubkeyvec(giststate, itvec, &gsvp, attno);
}

/*
 * find group in vector with equivalent value
 */
static int
gistfindgroup(Relation r, GISTSTATE *giststate, GISTENTRY *valvec, GistSplitVector *spl, int attno)
{
	int			i;
	GISTENTRY	entry;
	int			len = 0;

	/*
	 * attno key is always not null (see gistSplitByKey), so we may not check
	 * for nulls
	 */
	gistentryinit(entry, spl->splitVector.spl_rdatum, r, NULL, (OffsetNumber) 0, FALSE);
	for (i = 0; i < spl->splitVector.spl_nleft; i++)
	{
		float		penalty = gistpenalty(giststate, attno, &entry, false,
							   &valvec[spl->splitVector.spl_left[i]], false);

		if (penalty == 0.0)
		{
			spl->spl_equiv[spl->splitVector.spl_left[i]] = true;
			len++;
		}
	}

	gistentryinit(entry, spl->splitVector.spl_ldatum, r, NULL, (OffsetNumber) 0, FALSE);
	for (i = 0; i < spl->splitVector.spl_nright; i++)
	{
		float		penalty = gistpenalty(giststate, attno, &entry, false,
							  &valvec[spl->splitVector.spl_right[i]], false);

		if (penalty == 0.0)
		{
			spl->spl_equiv[spl->splitVector.spl_right[i]] = true;
			len++;
		}
	}

	return len;
}

static void
cleanupOffsets(OffsetNumber *a, int *len, bool *equiv, int *LenEquiv)
{
	int			curlen,
				i;
	OffsetNumber *curwpos;

	curlen = *len;
	curwpos = a;
	for (i = 0; i < *len; i++)
	{
		if (equiv[a[i]] == FALSE)
		{
			*curwpos = a[i];
			curwpos++;
		}
		else
		{
			/* corner case: we shouldn't make void array */
			if (curlen == 1)
			{
				equiv[a[i]] = FALSE;	/* mark item as non-equivalent */
				i--;			/* redo the same */
				*LenEquiv -= 1;
				continue;
			}
			else
				curlen--;
		}
	}

	*len = curlen;
}

static void
placeOne(Relation r, GISTSTATE *giststate, GistSplitVector *v, IndexTuple itup, OffsetNumber off, int attno)
{
	GISTENTRY	identry[INDEX_MAX_KEYS];
	bool		isnull[INDEX_MAX_KEYS];
	bool		toLeft = true;

	gistDeCompressAtt(giststate, r, itup, NULL, (OffsetNumber) 0, identry, isnull);

	for (; attno < giststate->tupdesc->natts; attno++)
	{
		float		lpenalty,
					rpenalty;
		GISTENTRY	entry;

		gistentryinit(entry, v->spl_lattr[attno], r, NULL, 0, FALSE);
		lpenalty = gistpenalty(giststate, attno, &entry, v->spl_lisnull[attno], identry + attno, isnull[attno]);
		gistentryinit(entry, v->spl_rattr[attno], r, NULL, 0, FALSE);
		rpenalty = gistpenalty(giststate, attno, &entry, v->spl_risnull[attno], identry + attno, isnull[attno]);

		if (lpenalty != rpenalty)
		{
			if (lpenalty > rpenalty)
				toLeft = false;
			break;
		}
	}

	if (toLeft)
		v->splitVector.spl_left[v->splitVector.spl_nleft++] = off;
	else
		v->splitVector.spl_right[v->splitVector.spl_nright++] = off;
}

#define SWAPVAR( s, d, t ) \
do {	\
	(t) = (s); \
	(s) = (d); \
	(d) = (t); \
} while(0)

/*
 * adjust left and right unions according to splits by previous
 * split by firsts columns. This function is called only in case
 * when pickSplit doesn't support subspplit.
 */

static void
supportSecondarySplit(Relation r, GISTSTATE *giststate, int attno, GIST_SPLITVEC *sv, Datum oldL, Datum oldR)
{
	bool		leaveOnLeft = true,
				tmpBool;
	GISTENTRY	entryL,
				entryR,
				entrySL,
				entrySR;

	gistentryinit(entryL, oldL, r, NULL, 0, FALSE);
	gistentryinit(entryR, oldR, r, NULL, 0, FALSE);
	gistentryinit(entrySL, sv->spl_ldatum, r, NULL, 0, FALSE);
	gistentryinit(entrySR, sv->spl_rdatum, r, NULL, 0, FALSE);

	if (sv->spl_ldatum_exists && sv->spl_rdatum_exists)
	{
		float		penalty1,
					penalty2;

		penalty1 = gistpenalty(giststate, attno, &entryL, false, &entrySL, false) +
			gistpenalty(giststate, attno, &entryR, false, &entrySR, false);
		penalty2 = gistpenalty(giststate, attno, &entryL, false, &entrySR, false) +
			gistpenalty(giststate, attno, &entryR, false, &entrySL, false);

		if (penalty1 > penalty2)
			leaveOnLeft = false;

	}
	else
	{
		GISTENTRY  *entry1 = (sv->spl_ldatum_exists) ? &entryL : &entryR;
		float		penalty1,
					penalty2;

		/*
		 * there is only one previously defined union, so we just choose swap
		 * or not by lowest penalty
		 */

		penalty1 = gistpenalty(giststate, attno, entry1, false, &entrySL, false);
		penalty2 = gistpenalty(giststate, attno, entry1, false, &entrySR, false);

		if (penalty1 < penalty2)
			leaveOnLeft = (sv->spl_ldatum_exists) ? true : false;
		else
			leaveOnLeft = (sv->spl_rdatum_exists) ? true : false;
	}

	if (leaveOnLeft == false)
	{
		/*
		 * swap left and right
		 */
		OffsetNumber *off,
					noff;
		Datum		datum;

		SWAPVAR(sv->spl_left, sv->spl_right, off);
		SWAPVAR(sv->spl_nleft, sv->spl_nright, noff);
		SWAPVAR(sv->spl_ldatum, sv->spl_rdatum, datum);
		gistentryinit(entrySL, sv->spl_ldatum, r, NULL, 0, FALSE);
		gistentryinit(entrySR, sv->spl_rdatum, r, NULL, 0, FALSE);
	}

	if (sv->spl_ldatum_exists)
		gistMakeUnionKey(giststate, attno, &entryL, false, &entrySL, false,
						 &sv->spl_ldatum, &tmpBool);

	if (sv->spl_rdatum_exists)
		gistMakeUnionKey(giststate, attno, &entryR, false, &entrySR, false,
						 &sv->spl_rdatum, &tmpBool);

	sv->spl_ldatum_exists = sv->spl_rdatum_exists = false;
}

/*
 * Calls user picksplit method for attno columns to split vector to
 * two vectors. May use attno+n columns data to
 * get better split.
 * Returns TRUE and v->spl_equiv = NULL if left and right unions of attno columns are the same,
 * so caller may find better split
 * Returns TRUE and v->spl_equiv != NULL if there is tuples which may be freely moved
 */
static bool
gistUserPicksplit(Relation r, GistEntryVector *entryvec, int attno, GistSplitVector *v,
				  IndexTuple *itup, int len, GISTSTATE *giststate)
{
	GIST_SPLITVEC *sv = &v->splitVector;

	/*
	 * now let the user-defined picksplit function set up the split vector; in
	 * entryvec have no null value!!
	 */