gist.c

PostgreSQL7.4.6 for Linux

/*-------------------------------------------------------------------------
 *
 * gist.c
 *	  interface routines for the postgres GiST index access method.
 *
 *
 * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  $Header: /cvsroot/pgsql/src/backend/access/gist/gist.c,v 1.105 2003/08/04 02:39:56 momjian 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 "miscadmin.h"


#undef GIST_PAGEADDITEM

#define ATTSIZE( datum, TupDesc, i, isnull ) \
	( \
		( isnull ) ? 0 : \
			att_addlength(0, (TupDesc)->attrs[(i)-1]->attlen, (datum)) \
	)

/* result's status */
#define INSERTED	0x01
#define SPLITED		0x02

/* group flags ( in gistSplit ) */
#define LEFT_ADDED	0x01
#define RIGHT_ADDED 0x02
#define BOTH_ADDED	( LEFT_ADDED | RIGHT_ADDED )

/*
 * This defines only for shorter code, used in gistgetadjusted
 * and gistadjsubkey only
 */
#define FILLITEM(evp, isnullkey, okey, okeyb, rkey, rkeyb)	 do { \
		if ( isnullkey ) {										  \
				gistentryinit((evp), rkey, r, (Page) NULL ,		  \
						(OffsetNumber) 0, rkeyb, FALSE);		  \
		} else {												  \
				gistentryinit((evp), okey, r, (Page) NULL,		  \
						(OffsetNumber) 0, okeyb, FALSE);		  \
		}														  \
} while(0)

#define FILLEV(isnull1, key1, key1b, isnull2, key2, key2b) do { \
	FILLITEM(*ev0p, isnull1, key1, key1b, key2, key2b);		\
	FILLITEM(*ev1p, isnull2, key2, key2b, key1, key1b);		\
} while(0);

/* Working state for gistbuild and its callback */
typedef struct
{
	GISTSTATE	giststate;
	int			numindexattrs;
	double		indtuples;
} GISTBuildState;


/* non-export function prototypes */
static void gistbuildCallback(Relation index,
				  HeapTuple htup,
				  Datum *attdata,
				  char *nulls,
				  bool tupleIsAlive,
				  void *state);
static void gistdoinsert(Relation r,
			 IndexTuple itup,
			 InsertIndexResult *res,
			 GISTSTATE *GISTstate);
static int gistlayerinsert(Relation r, BlockNumber blkno,
				IndexTuple **itup,
				int *len,
				InsertIndexResult *res,
				GISTSTATE *giststate);
static OffsetNumber gistwritebuffer(Relation r,
				Page page,
				IndexTuple *itup,
				int len,
				OffsetNumber off);
static int gistnospace(Page page,
			IndexTuple *itvec, int len);
static IndexTuple *gistreadbuffer(Buffer buffer, int *len);
static IndexTuple *gistjoinvector(
			   IndexTuple *itvec, int *len,
			   IndexTuple *additvec, int addlen);
static IndexTuple gistunion(Relation r, IndexTuple *itvec,
		  int len, GISTSTATE *giststate);

static IndexTuple gistgetadjusted(Relation r,
				IndexTuple oldtup,
				IndexTuple addtup,
				GISTSTATE *giststate);
static int gistfindgroup(GISTSTATE *giststate,
			  GISTENTRY *valvec, GIST_SPLITVEC *spl);
static void gistadjsubkey(Relation r,
			  IndexTuple *itup, int *len,
			  GIST_SPLITVEC *v,
			  GISTSTATE *giststate);
static IndexTuple gistFormTuple(GISTSTATE *giststate,
			Relation r, Datum attdata[], int datumsize[], bool isnull[]);
static IndexTuple *gistSplit(Relation r,
		  Buffer buffer,
		  IndexTuple *itup,
		  int *len,
		  GISTSTATE *giststate,
		  InsertIndexResult *res);
static void gistnewroot(Relation r,
			IndexTuple *itup, int len);
static void GISTInitBuffer(Buffer b, uint32 f);
static OffsetNumber gistchoose(Relation r, Page p,
		   IndexTuple it,
		   GISTSTATE *giststate);
static void gistdelete(Relation r, ItemPointer tid);

#ifdef GIST_PAGEADDITEM
static IndexTuple gist_tuple_replacekey(Relation r,
					  GISTENTRY entry, IndexTuple t);
#endif
static void gistcentryinit(GISTSTATE *giststate, int nkey,
			   GISTENTRY *e, Datum k,
			   Relation r, Page pg,
			   OffsetNumber o, int b, bool l, bool isNull);
static void gistDeCompressAtt(GISTSTATE *giststate, Relation r,
				  IndexTuple tuple, Page p, OffsetNumber o,
				  GISTENTRY attdata[], bool decompvec[], bool isnull[]);
static void gistFreeAtt(Relation r, GISTENTRY attdata[], bool decompvec[]);
static void gistpenalty(GISTSTATE *giststate, int attno,
			GISTENTRY *key1, bool isNull1,
			GISTENTRY *key2, bool isNull2,
			float *penalty);

#undef GISTDEBUG

#ifdef GISTDEBUG
static void gist_dumptree(Relation r, int level, BlockNumber blk, OffsetNumber coff);
#endif

/*
 * routine to build an index.  Basically calls insert over and over
 */
Datum
gistbuild(PG_FUNCTION_ARGS)
{
	Relation	heap = (Relation) PG_GETARG_POINTER(0);
	Relation	index = (Relation) PG_GETARG_POINTER(1);
	IndexInfo  *indexInfo = (IndexInfo *) PG_GETARG_POINTER(2);
	double		reltuples;
	GISTBuildState buildstate;
	Buffer		buffer;

	/* no locking is needed */

	initGISTstate(&buildstate.giststate, index);

	/*
	 * We expect to be called exactly once for any index relation. If
	 * that's not the case, big trouble's what we have.
	 */
	if (RelationGetNumberOfBlocks(index) != 0)
		elog(ERROR, "index \"%s\" already contains data",
			 RelationGetRelationName(index));

	/* initialize the root page */
	buffer = ReadBuffer(index, P_NEW);
	GISTInitBuffer(buffer, F_LEAF);
	WriteBuffer(buffer);

	/* build the index */
	buildstate.numindexattrs = indexInfo->ii_NumIndexAttrs;
	buildstate.indtuples = 0;

	/* do the heap scan */
	reltuples = IndexBuildHeapScan(heap, index, indexInfo,
								gistbuildCallback, (void *) &buildstate);

	/* okay, all heap tuples are indexed */

	/*
	 * Since we just counted the tuples in the heap, we update its stats
	 * in pg_class to guarantee that the planner takes advantage of the
	 * index we just created.  But, only update statistics during normal
	 * index definitions, not for indices on system catalogs created
	 * during bootstrap processing.  We must close the relations before
	 * updating statistics to guarantee that the relcache entries are
	 * flushed when we increment the command counter in UpdateStats(). But
	 * we do not release any locks on the relations; those will be held
	 * until end of transaction.
	 */
	if (IsNormalProcessingMode())
	{
		Oid			hrelid = RelationGetRelid(heap);
		Oid			irelid = RelationGetRelid(index);

		heap_close(heap, NoLock);
		index_close(index);
		UpdateStats(hrelid, reltuples);
		UpdateStats(irelid, buildstate.indtuples);
	}

	freeGISTstate(&buildstate.giststate);
#ifdef GISTDEBUG
	gist_dumptree(index, 0, GISTP_ROOT, 0);
#endif

	PG_RETURN_VOID();
}

/*
 * Per-tuple callback from IndexBuildHeapScan
 */
static void
gistbuildCallback(Relation index,
				  HeapTuple htup,
				  Datum *attdata,
				  char *nulls,
				  bool tupleIsAlive,
				  void *state)
{
	GISTBuildState *buildstate = (GISTBuildState *) state;
	IndexTuple	itup;
	bool		compvec[INDEX_MAX_KEYS];
	GISTENTRY	tmpcentry;
	int			i;

	/* GiST cannot index tuples with leading NULLs */
	if (nulls[0] == 'n')
		return;

	/* immediately compress keys to normalize */
	for (i = 0; i < buildstate->numindexattrs; i++)
	{
		if (nulls[i] == 'n')
		{
			attdata[i] = (Datum) 0;
			compvec[i] = FALSE;
		}
		else
		{
			gistcentryinit(&buildstate->giststate, i, &tmpcentry, attdata[i],
						   (Relation) NULL, (Page) NULL, (OffsetNumber) 0,
						 -1 /* size is currently bogus */ , TRUE, FALSE);
			if (attdata[i] != tmpcentry.key &&
				!(isAttByVal(&buildstate->giststate, i)))
				compvec[i] = TRUE;
			else
				compvec[i] = FALSE;
			attdata[i] = tmpcentry.key;
		}
	}

	/* form an index tuple and point it at the heap tuple */
	itup = index_formtuple(buildstate->giststate.tupdesc, attdata, 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.
	 */
	gistdoinsert(index, itup, NULL, &buildstate->giststate);

	buildstate->indtuples += 1;

	for (i = 0; i < buildstate->numindexattrs; i++)
		if (compvec[i])
			pfree(DatumGetPointer(attdata[i]));

	pfree(itup);
}

/*
 *	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.
 */
Datum
gistinsert(PG_FUNCTION_ARGS)
{
	Relation	r = (Relation) PG_GETARG_POINTER(0);
	Datum	   *datum = (Datum *) PG_GETARG_POINTER(1);
	char	   *nulls = (char *) PG_GETARG_POINTER(2);
	ItemPointer ht_ctid = (ItemPointer) PG_GETARG_POINTER(3);

#ifdef NOT_USED
	Relation	heapRel = (Relation) PG_GETARG_POINTER(4);
	bool		checkUnique = PG_GETARG_BOOL(5);
#endif
	InsertIndexResult res;
	IndexTuple	itup;
	GISTSTATE	giststate;
	GISTENTRY	tmpentry;
	int			i;
	bool		compvec[INDEX_MAX_KEYS];

	/*
	 * Since GIST is not marked "amconcurrent" in pg_am, caller should
	 * have acquired exclusive lock on index relation.	We need no locking
	 * here.
	 */

	/* GiST cannot index tuples with leading NULLs */
	if (nulls[0] == 'n')
	{
		res = NULL;
		PG_RETURN_POINTER(res);
	}

	initGISTstate(&giststate, r);

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

	res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
	gistdoinsert(r, itup, &res, &giststate);

	for (i = 0; i < r->rd_att->natts; i++)
		if (compvec[i] == TRUE)
			pfree(DatumGetPointer(datum[i]));
	pfree(itup);
	freeGISTstate(&giststate);

	PG_RETURN_POINTER(res);
}

#ifdef GIST_PAGEADDITEM
/*
 * 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;
	OffsetNumber retval;
	Datum		datum;
	bool		IsNull;

	/*
	 * recompress the item given that we now know the exact page and
	 * offset for insertion
	 */
	datum = index_getattr(itup, 1, r->rd_att, &IsNull);
	gistdentryinit(giststate, 0, dentry, datum,
				   (Relation) 0, (Page) 0,
				   (OffsetNumber) InvalidOffsetNumber,
				   ATTSIZE(datum, r, 1, IsNull),
				   FALSE, IsNull);
	gistcentryinit(giststate, 0, &tmpcentry, dentry->key, r, page,
				   offsetNumber, dentry->bytes, FALSE);
	*newtup = gist_tuple_replacekey(r, tmpcentry, itup);
	retval = PageAddItem(page, (Item) *newtup, IndexTupleSize(*newtup),
						 offsetNumber, flags);
	if (retval == InvalidOffsetNumber)
		elog(ERROR, "failed to add index item to \"%s\"",
			 RelationGetRelationName(r));
	/* be tidy */
	if (DatumGetPointer(tmpcentry.key) != NULL &&
		tmpcentry.key != dentry->key &&
		tmpcentry.key != datum)
		pfree(DatumGetPointer(tmpcentry.key));
	return (retval);
}
#endif

static void
gistdoinsert(Relation r,
			 IndexTuple itup,
			 InsertIndexResult *res,
			 GISTSTATE *giststate)
{
	IndexTuple *instup;
	int			i,
				ret,
				len = 1;

	instup = (IndexTuple *) palloc(sizeof(IndexTuple));
	instup[0] = (IndexTuple) palloc(IndexTupleSize(itup));
	memcpy(instup[0], itup, IndexTupleSize(itup));

	ret = gistlayerinsert(r, GISTP_ROOT, &instup, &len, res, giststate);
	if (ret & SPLITED)
		gistnewroot(r, instup, len);

	for (i = 0; i < len; i++)
		pfree(instup[i]);
	pfree(instup);
}

static int
gistlayerinsert(Relation r, BlockNumber blkno,
				IndexTuple **itup,		/* in - out, has compressed entry */
				int *len,		/* in - out */
				InsertIndexResult *res, /* out */
				GISTSTATE *giststate)
{
	Buffer		buffer;
	Page		page;
	OffsetNumber child;
	int			ret;
	GISTPageOpaque opaque;

	buffer = ReadBuffer(r, blkno);
	page = (Page) BufferGetPage(buffer);
	opaque = (GISTPageOpaque) PageGetSpecialPointer(page);

	if (!(opaque->flags & F_LEAF))
	{
		/* internal page, so we must walk on tree */
		/* len IS equal 1 */
		ItemId		iid;
		BlockNumber nblkno;
		ItemPointerData oldtid;
		IndexTuple	oldtup;

		child = gistchoose(r, page, *(*itup), giststate);
		iid = PageGetItemId(page, child);
		oldtup = (IndexTuple) PageGetItem(page, iid);
		nblkno = ItemPointerGetBlockNumber(&(oldtup->t_tid));

		/*
		 * After this call: 1. if child page was splited, then itup
		 * contains keys for each page 2. if  child page wasn't splited,
		 * then itup contains additional for adjustment of current key
		 */
		ret = gistlayerinsert(r, nblkno, itup, len, res, giststate);

		/* nothing inserted in child */
		if (!(ret & INSERTED))
		{
			ReleaseBuffer(buffer);
			return 0x00;
		}

		/* child does not splited */
		if (!(ret & SPLITED))
		{
			IndexTuple	newtup = gistgetadjusted(r, oldtup, (*itup)[0], giststate);

			if (!newtup)
			{
				/* not need to update key */
				ReleaseBuffer(buffer);
				return 0x00;
			}

			pfree((*itup)[0]);	/* !!! */
			(*itup)[0] = newtup;