Restructure index AM interface for index building and index tuple deletion,

per previous discussion on pghackers.  Most of the duplicate code in
different AMs' ambuild routines has been moved out to a common routine
in index.c; this means that all index types now do the right things about
inserting recently-dead tuples, etc.  (I also removed support for EXTEND
INDEX in the ambuild routines, since that's about to go away anyway, and
it cluttered the code a lot.)  The retail indextuple deletion routines have
been replaced by a "bulk delete" routine in which the indexscan is inside
the access method.  I haven't pushed this change as far as it should go yet,
but it should allow considerable simplification of the internal bookkeeping
for deletions.  Also, add flag columns to pg_am to eliminate various
hardcoded tests on AM OIDs, and remove unused pg_am columns.

Fix rtree and gist index types to not attempt to store NULLs; before this,
gist usually crashed, while rtree managed not to crash but computed wacko
bounding boxes for NULL entries (which might have had something to do with
the performance problems we've heard about occasionally).

Add AtEOXact routines to hash, rtree, and gist, all of which have static
state that needs to be reset after an error.  We discovered this need long
ago for btree, but missed the other guys.

Oh, one more thing: concurrent VACUUM is now the default.

1 files changed, 167 insertions, 176 deletions

diff --git a/src/backend/access/rtree/rtree.c b/src/backend/access/rtree/rtree.c
index a8c6a13ea3c..21831ef5d61 100644
--- a/src/backend/access/rtree/rtree.c
+++ b/src/backend/access/rtree/rtree.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/rtree/Attic/rtree.c,v 1.62 2001/05/07 00:43:16 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/rtree/Attic/rtree.c,v 1.63 2001/07/15 22:48:16 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -62,7 +62,20 @@ typedef struct RTSTATE
 	FmgrInfo	interFn;		/* intersection function */
 } RTSTATE;
 
+/* Working state for rtbuild and its callback */
+typedef struct
+{
+	RTSTATE		rtState;
+	double		indtuples;
+} RTBuildState;
+
 /* non-export function prototypes */
+static void rtbuildCallback(Relation index,
+							HeapTuple htup,
+							Datum *attdata,
+							char *nulls,
+							bool tupleIsAlive,
+							void *state);
 static InsertIndexResult rtdoinsert(Relation r, IndexTuple itup,
 		   RTSTATE *rtstate);
 static void rttighten(Relation r, RTSTACK *stk, Datum datum, int att_size,
@@ -81,165 +94,44 @@ static int	nospace(Page p, IndexTuple it);
 static void initRtstate(RTSTATE *rtstate, Relation index);
 
 
+/*
+ * routine to build an index.  Basically calls insert over and over
+ */
 Datum
 rtbuild(PG_FUNCTION_ARGS)
 {
 	Relation	heap = (Relation) PG_GETARG_POINTER(0);
 	Relation	index = (Relation) PG_GETARG_POINTER(1);
 	IndexInfo  *indexInfo = (IndexInfo *) PG_GETARG_POINTER(2);
-	Node	   *oldPred = (Node *) PG_GETARG_POINTER(3);
-
-#ifdef NOT_USED
-	IndexStrategy istrat = (IndexStrategy) PG_GETARG_POINTER(4);
-
-#endif
-	HeapScanDesc hscan;
-	HeapTuple	htup;
-	IndexTuple	itup;
-	TupleDesc	htupdesc,
-				itupdesc;
-	Datum		attdata[INDEX_MAX_KEYS];
-	char		nulls[INDEX_MAX_KEYS];
-	double		nhtups,
-				nitups;
-	Node	   *pred = indexInfo->ii_Predicate;
-
-#ifndef OMIT_PARTIAL_INDEX
-	TupleTable	tupleTable;
-	TupleTableSlot *slot;
+	double		reltuples;
+	RTBuildState buildstate;
+	Buffer		buffer;
 
-#endif
-	ExprContext *econtext;
-	InsertIndexResult res = NULL;
-	Buffer		buffer = InvalidBuffer;
-	RTSTATE		rtState;
+	/* no locking is needed */
 
-	initRtstate(&rtState, index);
+	initRtstate(&buildstate.rtState, 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 (oldPred == NULL && RelationGetNumberOfBlocks(index) != 0)
-		elog(ERROR, "%s already contains data", RelationGetRelationName(index));
-
-	/* initialize the root page (if this is a new index) */
-	if (oldPred == NULL)
-	{
-		buffer = ReadBuffer(index, P_NEW);
-		RTInitBuffer(buffer, F_LEAF);
-		WriteBuffer(buffer);
-	}
-
-	/* get tuple descriptors for heap and index relations */
-	htupdesc = RelationGetDescr(heap);
-	itupdesc = RelationGetDescr(index);
-
-	/*
-	 * 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
-	 *
-	 * We construct the ExprContext anyway since we need a per-tuple
-	 * temporary memory context for function evaluation -- tgl July 00
-	 */
-#ifndef OMIT_PARTIAL_INDEX
-	if (pred != NULL || oldPred != NULL)
-	{
-		tupleTable = ExecCreateTupleTable(1);
-		slot = ExecAllocTableSlot(tupleTable);
-		ExecSetSlotDescriptor(slot, htupdesc, false);
-	}
-	else
-	{
-		tupleTable = NULL;
-		slot = NULL;
-	}
-	econtext = MakeExprContext(slot, TransactionCommandContext);
-#else
-	econtext = MakeExprContext(NULL, TransactionCommandContext);
-#endif	 /* OMIT_PARTIAL_INDEX */
-
-	/* count the tuples as we insert them */
-	nhtups = nitups = 0.0;
-
-	/* start a heap scan */
-	hscan = heap_beginscan(heap, 0, SnapshotNow, 0, (ScanKey) NULL);
-
-	while (HeapTupleIsValid(htup = heap_getnext(hscan, 0)))
-	{
-		MemoryContextReset(econtext->ecxt_per_tuple_memory);
+	if (RelationGetNumberOfBlocks(index) != 0)
+		elog(ERROR, "%s already contains data",
+			 RelationGetRelationName(index));
 
-		nhtups += 1.0;
-
-#ifndef OMIT_PARTIAL_INDEX
-
-		/*
-		 * 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)
-		{
-			slot->val = htup;
-			if (ExecQual((List *) oldPred, econtext, false))
-			{
-				nitups += 1.0;
-				continue;
-			}
-		}
-
-		/*
-		 * Skip this tuple if it doesn't satisfy the partial-index
-		 * predicate
-		 */
-		if (pred != NULL)
-		{
-			slot->val = htup;
-			if (!ExecQual((List *) pred, econtext, false))
-				continue;
-		}
-#endif	 /* OMIT_PARTIAL_INDEX */
-
-		nitups += 1.0;
-
-		/*
-		 * For the current heap tuple, extract all the attributes we use
-		 * in this index, and note which are null.
-		 */
-		FormIndexDatum(indexInfo,
-					   htup,
-					   htupdesc,
-					   econtext->ecxt_per_tuple_memory,
-					   attdata,
-					   nulls);
-
-		/* form an index tuple and point it at the heap tuple */
-		itup = index_formtuple(itupdesc, attdata, nulls);
-		itup->t_tid = htup->t_self;
+	/* initialize the root page */
+	buffer = ReadBuffer(index, P_NEW);
+	RTInitBuffer(buffer, F_LEAF);
+	WriteBuffer(buffer);
 
-		/*
-		 * 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.
-		 */
+	/* build the index */
+	buildstate.indtuples = 0;
 
-		res = rtdoinsert(index, itup, &rtState);
-		pfree(itup);
-		pfree(res);
-	}
+	/* do the heap scan */
+	reltuples = IndexBuildHeapScan(heap, index, indexInfo,
+								   rtbuildCallback, (void *) &buildstate);
 
 	/* okay, all heap tuples are indexed */
-	heap_endscan(hscan);
-
-#ifndef OMIT_PARTIAL_INDEX
-	if (pred != NULL || oldPred != NULL)
-		ExecDropTupleTable(tupleTable, true);
-#endif	 /* OMIT_PARTIAL_INDEX */
-	FreeExprContext(econtext);
 
 	/*
 	 * Since we just counted the tuples in the heap, we update its stats
@@ -259,20 +151,57 @@ rtbuild(PG_FUNCTION_ARGS)
 
 		heap_close(heap, NoLock);
 		index_close(index);
-		UpdateStats(hrelid, nhtups);
-		UpdateStats(irelid, nitups);
-		if (oldPred != NULL)
-		{
-			if (nitups == nhtups)
-				pred = NULL;
-			UpdateIndexPredicate(irelid, oldPred, pred);
-		}
+		UpdateStats(hrelid, reltuples);
+		UpdateStats(irelid, buildstate.indtuples);
 	}
 
 	PG_RETURN_VOID();
 }
 
 /*
+ * Per-tuple callback from IndexBuildHeapScan
+ */
+static void
+rtbuildCallback(Relation index,
+				HeapTuple htup,
+				Datum *attdata,
+				char *nulls,
+				bool tupleIsAlive,
+				void *state)
+{
+	RTBuildState   *buildstate = (RTBuildState *) state;
+	IndexTuple	itup;
+	InsertIndexResult res;
+
+	/* form an index tuple and point it at the heap tuple */
+	itup = index_formtuple(RelationGetDescr(index), attdata, nulls);
+	itup->t_tid = htup->t_self;
+
+	/* rtree indexes don't index nulls, see notes in rtinsert */
+	if (IndexTupleHasNulls(itup))
+	{
+		pfree(itup);
+		return;
+	}
+
+	/*
+	 * 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, &buildstate->rtState);
+
+	if (res)
+		pfree(res);
+
+	buildstate->indtuples += 1;
+
+	pfree(itup);
+}
+
+/*
  *	rtinsert -- wrapper for rtree tuple insertion.
  *
  *	  This is the public interface routine for tuple insertion in rtrees.
@@ -285,10 +214,8 @@ rtinsert(PG_FUNCTION_ARGS)
 	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);
-
 #endif
 	InsertIndexResult res;
 	IndexTuple	itup;
@@ -297,12 +224,24 @@ rtinsert(PG_FUNCTION_ARGS)
 	/* generate an index tuple */
 	itup = index_formtuple(RelationGetDescr(r), datum, nulls);
 	itup->t_tid = *ht_ctid;
+
+	/*
+	 * Currently, rtrees do not support indexing NULLs; considerable
+	 * infrastructure work would have to be done to do anything reasonable
+	 * with a NULL.
+	 */
+	if (IndexTupleHasNulls(itup))
+	{
+		pfree(itup);
+		PG_RETURN_POINTER((InsertIndexResult) NULL);
+	}
+
 	initRtstate(&rtState, r);
 
 	/*
-	 * Notes in ExecUtils:ExecOpenIndices()
-	 *
-	 * RelationSetLockForWrite(r);
+	 * Since rtree is not marked "amconcurrent" in pg_am, caller should
+	 * have acquired exclusive lock on index relation.  We need no locking
+	 * here.
 	 */
 
 	res = rtdoinsert(r, itup, &rtState);
@@ -1104,40 +1043,92 @@ freestack(RTSTACK *s)
 	}
 }
 
+/*
+ * Bulk deletion of all index entries pointing to a set of heap tuples.
+ * The set of target tuples is specified via a callback routine that tells
+ * whether any given heap tuple (identified by ItemPointer) is being deleted.
+ *
+ * Result: a palloc'd struct containing statistical info for VACUUM displays.
+ */
 Datum
-rtdelete(PG_FUNCTION_ARGS)
+rtbulkdelete(PG_FUNCTION_ARGS)
 {
-	Relation	r = (Relation) PG_GETARG_POINTER(0);
-	ItemPointer tid = (ItemPointer) PG_GETARG_POINTER(1);
-	BlockNumber blkno;
-	OffsetNumber offnum;
-	Buffer		buf;
-	Page		page;
+	Relation	rel = (Relation) PG_GETARG_POINTER(0);
+	IndexBulkDeleteCallback callback = (IndexBulkDeleteCallback) PG_GETARG_POINTER(1);
+	void	   *callback_state = (void *) PG_GETARG_POINTER(2);
+	IndexBulkDeleteResult *result;
+	BlockNumber	num_pages;
+	double		tuples_removed;
+	double		num_index_tuples;
+	RetrieveIndexResult res;
+	IndexScanDesc iscan;
+
+	tuples_removed = 0;
+	num_index_tuples = 0;
 
 	/*
-	 * Notes in ExecUtils:ExecOpenIndices() Also note that only vacuum
-	 * deletes index tuples now...
-	 *
-	 * RelationSetLockForWrite(r);
+	 * Since rtree is not marked "amconcurrent" in pg_am, caller should
+	 * have acquired exclusive lock on index relation.  We need no locking
+	 * here.
 	 */
 
-	blkno = ItemPointerGetBlockNumber(tid);
-	offnum = ItemPointerGetOffsetNumber(tid);
+	/*
+	 * XXX generic implementation --- should be improved!
+	 */
 
-	/* adjust any scans that will be affected by this deletion */
-	rtadjscans(r, RTOP_DEL, blkno, offnum);
+	/* walk through the entire index */
+	iscan = index_beginscan(rel, false, 0, (ScanKey) NULL);
 
-	/* delete the index tuple */
-	buf = ReadBuffer(r, blkno);
-	page = BufferGetPage(buf);
+	while ((res = index_getnext(iscan, ForwardScanDirection))
+		   != (RetrieveIndexResult) NULL)
+	{
+		ItemPointer heapptr = &res->heap_iptr;
 
-	PageIndexTupleDelete(page, offnum);
+		if (callback(heapptr, callback_state))
+		{
+			ItemPointer indexptr = &res->index_iptr;
+			BlockNumber blkno;
+			OffsetNumber offnum;
+			Buffer		buf;
+			Page		page;
 
-	WriteBuffer(buf);
+			blkno = ItemPointerGetBlockNumber(indexptr);
+			offnum = ItemPointerGetOffsetNumber(indexptr);
 
-	PG_RETURN_VOID();
+			/* adjust any scans that will be affected by this deletion */
+			/* (namely, my own scan) */
+			rtadjscans(rel, RTOP_DEL, blkno, offnum);
+
+			/* delete the index tuple */
+			buf = ReadBuffer(rel, blkno);
+			page = BufferGetPage(buf);
+
+			PageIndexTupleDelete(page, offnum);
+
+			WriteBuffer(buf);
+
+			tuples_removed += 1;
+		}
+		else
+			num_index_tuples += 1;
+
+		pfree(res);
+	}
+
+	index_endscan(iscan);
+
+	/* return statistics */
+	num_pages = RelationGetNumberOfBlocks(rel);
+
+	result = (IndexBulkDeleteResult *) palloc(sizeof(IndexBulkDeleteResult));
+	result->num_pages = num_pages;
+	result->tuples_removed = tuples_removed;
+	result->num_index_tuples = num_index_tuples;
+
+	PG_RETURN_POINTER(result);
 }
 
+
 static void
 initRtstate(RTSTATE *rtstate, Relation index)
 {