gist.c

PostgreSQL7.4.6 for Linux

static int
gistfindgroup(GISTSTATE *giststate, GISTENTRY *valvec, GIST_SPLITVEC *spl)
{
	int			i,
				j,
				len;
	int			curid = 1;
	bool		result;

	/*
	 * first key is always not null (see gistinsert), so we may not check
	 * for nulls
	 */

	for (i = 0; i < spl->spl_nleft; i++)
	{
		if (spl->spl_idgrp[spl->spl_left[i]])
			continue;
		len = 0;
		/* find all equal value in right part */
		for (j = 0; j < spl->spl_nright; j++)
		{
			if (spl->spl_idgrp[spl->spl_right[j]])
				continue;
			FunctionCall3(&giststate->equalFn[0],
						  valvec[spl->spl_left[i]].key,
						  valvec[spl->spl_right[j]].key,
						  PointerGetDatum(&result));
			if (result)
			{
				spl->spl_idgrp[spl->spl_right[j]] = curid;
				len++;
			}
		}
		/* find all other equal value in left part */
		if (len)
		{
			/* add current val to list of equial values */
			spl->spl_idgrp[spl->spl_left[i]] = curid;
			/* searching .. */
			for (j = i + 1; j < spl->spl_nleft; j++)
			{
				if (spl->spl_idgrp[spl->spl_left[j]])
					continue;
				FunctionCall3(&giststate->equalFn[0],
							  valvec[spl->spl_left[i]].key,
							  valvec[spl->spl_left[j]].key,
							  PointerGetDatum(&result));
				if (result)
				{
					spl->spl_idgrp[spl->spl_left[j]] = curid;
					len++;
				}
			}
			spl->spl_ngrp[curid] = len + 1;
			curid++;
		}
	}

	return curid;
}

/*
 * Insert equivalent tuples to left or right page
 * with minimize penalty
 */
static void
gistadjsubkey(Relation r,
			  IndexTuple *itup, /* contains compressed entry */
			  int *len,
			  GIST_SPLITVEC *v,
			  GISTSTATE *giststate
)
{
	int			curlen;
	OffsetNumber *curwpos;
	bool		decfree[INDEX_MAX_KEYS];
	GISTENTRY	entry,
				identry[INDEX_MAX_KEYS],
			   *ev0p,
			   *ev1p;
	float		lpenalty,
				rpenalty;
	bytea	   *evec;
	char	   *storage;
	int			datumsize;
	bool		isnull[INDEX_MAX_KEYS];
	int			i,
				j;
	Datum		datum;

	/* clear vectors */
	curlen = v->spl_nleft;
	curwpos = v->spl_left;
	for (i = 0; i < v->spl_nleft; i++)
		if (v->spl_idgrp[v->spl_left[i]] == 0)
		{
			*curwpos = v->spl_left[i];
			curwpos++;
		}
		else
			curlen--;
	v->spl_nleft = curlen;

	curlen = v->spl_nright;
	curwpos = v->spl_right;
	for (i = 0; i < v->spl_nright; i++)
		if (v->spl_idgrp[v->spl_right[i]] == 0)
		{
			*curwpos = v->spl_right[i];
			curwpos++;
		}
		else
			curlen--;
	v->spl_nright = curlen;

	/* workaround for 64-bit: ensure GISTENTRY array is maxaligned */
	storage = (char *) palloc(2 * sizeof(GISTENTRY) + MAXALIGN(VARHDRSZ));
	evec = (bytea *) (storage + MAXALIGN(VARHDRSZ) - VARHDRSZ);
	VARATT_SIZEP(evec) = 2 * sizeof(GISTENTRY) + VARHDRSZ;
	ev0p = &((GISTENTRY *) VARDATA(evec))[0];
	ev1p = &((GISTENTRY *) VARDATA(evec))[1];

	/* add equivalent tuple */
	for (i = 0; i < *len; i++)
	{
		if (v->spl_idgrp[i + 1] == 0)	/* already inserted */
			continue;
		gistDeCompressAtt(giststate, r, itup[i], (Page) NULL, (OffsetNumber) 0,
						  identry, decfree, isnull);

		v->spl_ngrp[v->spl_idgrp[i + 1]]--;
		if (v->spl_ngrp[v->spl_idgrp[i + 1]] == 0 &&
		(v->spl_grpflag[v->spl_idgrp[i + 1]] & BOTH_ADDED) != BOTH_ADDED)
		{

			/* force last in group */
			rpenalty = 1.0;
			lpenalty = (v->spl_grpflag[v->spl_idgrp[i + 1]] & LEFT_ADDED) ? 2.0 : 0.0;
		}
		else
		{
			/* where? */
			for (j = 1; j < r->rd_att->natts; j++)
			{
				gistentryinit(entry, v->spl_lattr[j], r, (Page) NULL,
						   (OffsetNumber) 0, v->spl_lattrsize[j], FALSE);
				gistpenalty(giststate, j, &entry, v->spl_lisnull[j],
							&identry[j], isnull[j], &lpenalty);

				gistentryinit(entry, v->spl_rattr[j], r, (Page) NULL,
						   (OffsetNumber) 0, v->spl_rattrsize[j], FALSE);
				gistpenalty(giststate, j, &entry, v->spl_risnull[j],
							&identry[j], isnull[j], &rpenalty);

				if (lpenalty != rpenalty)
					break;
			}
		}
		/* add */
		if (lpenalty < rpenalty)
		{
			v->spl_grpflag[v->spl_idgrp[i + 1]] |= LEFT_ADDED;
			v->spl_left[v->spl_nleft] = i + 1;
			v->spl_nleft++;
			for (j = 1; j < r->rd_att->natts; j++)
			{
				if (isnull[j] && v->spl_lisnull[j])
				{
					v->spl_lattr[j] = (Datum) 0;
					v->spl_lattrsize[j] = 0;
				}
				else
				{
					FILLEV(
						   v->spl_lisnull[j], v->spl_lattr[j], v->spl_lattrsize[j],
						   isnull[j], identry[j].key, identry[j].bytes
						);

					datum = FunctionCall2(&giststate->unionFn[j],
										  PointerGetDatum(evec),
										  PointerGetDatum(&datumsize));

					if ((!isAttByVal(giststate, j)) && !v->spl_lisnull[j])
						pfree(DatumGetPointer(v->spl_lattr[j]));
					v->spl_lattr[j] = datum;
					v->spl_lattrsize[j] = datumsize;
					v->spl_lisnull[j] = false;
				}
			}
		}
		else
		{
			v->spl_grpflag[v->spl_idgrp[i + 1]] |= RIGHT_ADDED;
			v->spl_right[v->spl_nright] = i + 1;
			v->spl_nright++;
			for (j = 1; j < r->rd_att->natts; j++)
			{
				if (isnull[j] && v->spl_risnull[j])
				{
					v->spl_rattr[j] = (Datum) 0;
					v->spl_rattrsize[j] = 0;
				}
				else
				{
					FILLEV(
						   v->spl_risnull[j], v->spl_rattr[j], v->spl_rattrsize[j],
						   isnull[j], identry[j].key, identry[j].bytes
						);

					datum = FunctionCall2(&giststate->unionFn[j],
										  PointerGetDatum(evec),
										  PointerGetDatum(&datumsize));

					if ((!isAttByVal(giststate, j)) && !v->spl_risnull[j])
						pfree(DatumGetPointer(v->spl_rattr[j]));

					v->spl_rattr[j] = datum;
					v->spl_rattrsize[j] = datumsize;
					v->spl_risnull[j] = false;
				}
			}

		}
		gistFreeAtt(r, identry, decfree);
	}
	pfree(storage);				/* pfree(evec); */
}

/*
 *	gistSplit -- split a page in the tree.
 */
static IndexTuple *
gistSplit(Relation r,
		  Buffer buffer,
		  IndexTuple *itup,		/* contains compressed entry */
		  int *len,
		  GISTSTATE *giststate,
		  InsertIndexResult *res)
{
	Page		p;
	Buffer		leftbuf,
				rightbuf;
	Page		left,
				right;
	IndexTuple *lvectup,
			   *rvectup,
			   *newtup;
	BlockNumber lbknum,
				rbknum;
	GISTPageOpaque opaque;
	GIST_SPLITVEC v;
	bytea	   *entryvec;
	char	   *storage;
	bool	   *decompvec;
	int			i,
				j,
				nlen;
	int			MaxGrpId = 1;
	Datum		datum;
	bool		IsNull;

	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 */
	/* workaround for 64-bit: ensure GISTENTRY array is maxaligned */
	storage = palloc(MAXALIGN(VARHDRSZ) + (*len + 1) * sizeof(GISTENTRY));
	entryvec = (bytea *) (storage + MAXALIGN(VARHDRSZ) - VARHDRSZ);
	decompvec = (bool *) palloc((*len + 1) * sizeof(bool));
	VARATT_SIZEP(entryvec) = (*len + 1) * sizeof(GISTENTRY) + VARHDRSZ;
	for (i = 1; i <= *len; i++)
	{
		datum = index_getattr(itup[i - 1], 1, giststate->tupdesc, &IsNull);
		gistdentryinit(giststate, 0, &((GISTENTRY *) VARDATA(entryvec))[i],
					   datum, r, p, i,
		   ATTSIZE(datum, giststate->tupdesc, 1, IsNull), FALSE, IsNull);
		if ((!isAttByVal(giststate, 0)) && ((GISTENTRY *) VARDATA(entryvec))[i].key != datum)
			decompvec[i] = TRUE;
		else
			decompvec[i] = FALSE;
	}

	/*
	 * now let the user-defined picksplit function set up the split
	 * vector; in entryvec have no null value!!
	 */
	FunctionCall2(&giststate->picksplitFn[0],
				  PointerGetDatum(entryvec),
				  PointerGetDatum(&v));

	/* compatibility with old code */
	if (v.spl_left[v.spl_nleft - 1] == InvalidOffsetNumber)
		v.spl_left[v.spl_nleft - 1] = (OffsetNumber) *len;
	if (v.spl_right[v.spl_nright - 1] == InvalidOffsetNumber)
		v.spl_right[v.spl_nright - 1] = (OffsetNumber) *len;

	v.spl_lattr[0] = v.spl_ldatum;
	v.spl_rattr[0] = v.spl_rdatum;
	v.spl_lisnull[0] = false;
	v.spl_risnull[0] = false;

	/*
	 * if index is multikey, then we must to try get smaller bounding box
	 * for subkey(s)
	 */
	if (r->rd_att->natts > 1)
	{
		v.spl_idgrp = (int *) palloc0(sizeof(int) * (*len + 1));
		v.spl_grpflag = (char *) palloc0(sizeof(char) * (*len + 1));
		v.spl_ngrp = (int *) palloc(sizeof(int) * (*len + 1));

		MaxGrpId = gistfindgroup(giststate, (GISTENTRY *) VARDATA(entryvec), &v);

		/* form union of sub keys for each page (l,p) */
		gistunionsubkey(r, giststate, itup, &v);

		/*
		 * if possible, we insert equivalent tuples with control by
		 * penalty for a subkey(s)
		 */
		if (MaxGrpId > 1)
			gistadjsubkey(r, itup, len, &v, giststate);

		pfree(v.spl_idgrp);
		pfree(v.spl_grpflag);
		pfree(v.spl_ngrp);
	}

	/* clean up the entry vector: its keys need to be deleted, too */
	for (i = 1; i <= *len; i++)
		if (decompvec[i])
			pfree(DatumGetPointer(((GISTENTRY *) VARDATA(entryvec))[i].key));
	pfree(storage);				/* pfree(entryvec); */
	pfree(decompvec);

	/* form left and right vector */
	lvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * v.spl_nleft);
	rvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * v.spl_nright);

	for (i = 0; i < v.spl_nleft; i++)
		lvectup[i] = itup[v.spl_left[i] - 1];

	for (i = 0; i < v.spl_nright; i++)
		rvectup[i] = itup[v.spl_right[i] - 1];


	/* write on disk (may be need another split) */
	if (gistnospace(right, rvectup, v.spl_nright))
	{
		nlen = v.spl_nright;
		newtup = gistSplit(r, rightbuf, rvectup, &nlen, giststate,
			  (res && rvectup[nlen - 1] == itup[*len - 1]) ? res : NULL);
		ReleaseBuffer(rightbuf);
		for (j = 1; j < r->rd_att->natts; j++)
			if ((!isAttByVal(giststate, j)) && !v.spl_risnull[j])
				pfree(DatumGetPointer(v.spl_rattr[j]));
	}
	else
	{
		OffsetNumber l;

		l = gistwritebuffer(r, right, rvectup, v.spl_nright, FirstOffsetNumber);
		WriteBuffer(rightbuf);

		if (res)
			ItemPointerSet(&((*res)->pointerData), rbknum, l);

		nlen = 1;
		newtup = (IndexTuple *) palloc(sizeof(IndexTuple) * 1);
		newtup[0] = gistFormTuple(giststate, r, v.spl_rattr, v.spl_rattrsize, v.spl_risnull);
		ItemPointerSet(&(newtup[0]->t_tid), rbknum, 1);
	}


	if (gistnospace(left, lvectup, v.spl_nleft))
	{
		int			llen = v.spl_nleft;
		IndexTuple *lntup;

		lntup = gistSplit(r, leftbuf, lvectup, &llen, giststate,
			  (res && lvectup[llen - 1] == itup[*len - 1]) ? res : NULL);
		ReleaseBuffer(leftbuf);

		for (j = 1; j < r->rd_att->natts; j++)
			if ((!isAttByVal(giststate, j)) && !v.spl_lisnull[j])
				pfree(DatumGetPointer(v.spl_lattr[j]));

		newtup = gistjoinvector(newtup, &nlen, lntup, llen);
		pfree(lntup);
	}
	else
	{
		OffsetNumber l;

		l = gistwritebuffer(r, left, lvectup, v.spl_nleft, FirstOffsetNumber);
		if (BufferGetBlockNumber(buffer) != GISTP_ROOT)
			PageRestoreTempPage(left, p);

		WriteBuffer(leftbuf);

		if (res)
			ItemPointerSet(&((*res)->pointerData), lbknum, l);

		nlen += 1;
		newtup = (IndexTuple *) repalloc((void *) newtup, sizeof(IndexTuple) * nlen);
		newtup[nlen - 1] = gistFormTuple(giststate, r, v.spl_lattr, v.spl_lattrsize, v.spl_lisnull);
		ItemPointerSet(&(newtup[nlen - 1]->t_tid), lbknum, 1);
	}

	/* !!! pfree */
	pfree(rvectup);
	pfree(lvectup);
	pfree(v.spl_left);
	pfree(v.spl_right);

	*len = nlen;
	return newtup;
}

static void
gistnewroot(Relation r, IndexTuple *itup, int len)
{
	Buffer		b;
	Page		p;

	b = ReadBuffer(r, GISTP_ROOT);
	GISTInitBuffer(b, 0);
	p = BufferGetPage(b);

	gistwritebuffer(r, p, itup, len, FirstOffsetNumber);
	WriteBuffer(b);
}

static void
GISTInitBuffer(Buffer b, uint32 f)
{
	GISTPageOpaque opaque;
	Page		page;
	Size		pageSize;

	pageSize = BufferGetPageSize(b);

	page = BufferGetPage(b);

	PageInit(page, pageSize, sizeof(GISTPageOpaqueData));

	opaque = (GISTPageOpaque) PageGetSpecialPointer(page);
	opaque->flags = f;
}


/*
** find entry with lowest penalty
*/
static OffsetNumber
gistchoose(Relation r, Page p, IndexTuple it,	/* it has compressed entry */
		   GISTSTATE *giststate)
{
	OffsetNumber maxoff;
	OffsetNumber i;
	Datum		datum;
	float		usize;
	OffsetNumber which;
	float		sum_grow,
				which_grow[INDEX_MAX_KEYS];