Mega-commit to make heap_open/heap_openr/heap_close take an

additional argument specifying the kind of lock to acquire/release (or
'NoLock' to do no lock processing).  Ensure that all relations are locked
with some appropriate lock level before being examined --- this ensures
that relevant shared-inval messages have been processed and should prevent
problems caused by concurrent VACUUM.  Fix several bugs having to do with
mismatched increment/decrement of relation ref count and mismatched
heap_open/close (which amounts to the same thing).  A bogus ref count on
a relation doesn't matter much *unless* a SI Inval message happens to
arrive at the wrong time, which is probably why we got away with this
sloppiness for so long.  Repair missing grab of AccessExclusiveLock in
DROP TABLE, ALTER/RENAME TABLE, etc, as noted by Hiroshi.
Recommend 'make clean all' after pulling this update; I modified the
Relation struct layout slightly.
Will post further discussion to pghackers list shortly.

1 files changed, 70 insertions, 46 deletions

diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index c90c27fd5d..912996fb1f 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/catalog/index.c,v 1.89 1999/09/05 17:43:47 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/catalog/index.c,v 1.90 1999/09/18 19:06:33 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -378,7 +378,7 @@ AccessMethodObjectIdGetForm(Oid accessMethodObjectId)
 	 *	fetch the desired access method tuple
 	 * ----------------
 	 */
-	pg_am_desc = heap_openr(AccessMethodRelationName);
+	pg_am_desc = heap_openr(AccessMethodRelationName, AccessShareLock);
 	pg_am_scan = heap_beginscan(pg_am_desc, 0, SnapshotNow, 1, &key);
 
 	pg_am_tuple = heap_getnext(pg_am_scan, 0);
@@ -390,7 +390,7 @@ AccessMethodObjectIdGetForm(Oid accessMethodObjectId)
 	if (!HeapTupleIsValid(pg_am_tuple))
 	{
 		heap_endscan(pg_am_scan);
-		heap_close(pg_am_desc);
+		heap_close(pg_am_desc, AccessShareLock);
 		return NULL;
 	}
 
@@ -402,7 +402,7 @@ AccessMethodObjectIdGetForm(Oid accessMethodObjectId)
 	memcpy(aform, GETSTRUCT(pg_am_tuple), sizeof *aform);
 
 	heap_endscan(pg_am_scan);
-	heap_close(pg_am_desc);
+	heap_close(pg_am_desc, AccessShareLock);
 
 	return aform;
 }
@@ -456,7 +456,7 @@ UpdateRelationRelation(Relation indexRelation, char *temp_relname)
 	Oid			tupleOid;
 	Relation	idescs[Num_pg_class_indices];
 
-	pg_class = heap_openr(RelationRelationName);
+	pg_class = heap_openr(RelationRelationName, RowExclusiveLock);
 
 	/* XXX Natts_pg_class_fixed is a hack - see pg_class.h */
 	tuple = heap_addheader(Natts_pg_class_fixed,
@@ -491,7 +491,7 @@ UpdateRelationRelation(Relation indexRelation, char *temp_relname)
 
 	tupleOid = tuple->t_data->t_oid;
 	pfree(tuple);
-	heap_close(pg_class);
+	heap_close(pg_class, RowExclusiveLock);
 
 	return tupleOid;
 }
@@ -542,7 +542,7 @@ AppendAttributeTuples(Relation indexRelation, int numatts)
 	 *	XXX ADD INDEXING
 	 * ----------------
 	 */
-	pg_attribute = heap_openr(AttributeRelationName);
+	pg_attribute = heap_openr(AttributeRelationName, RowExclusiveLock);
 
 	/* ----------------
 	 *	initialize *null, *replace and *value
@@ -628,7 +628,7 @@ AppendAttributeTuples(Relation indexRelation, int numatts)
 
 	if (cur_tuple)
 		pfree(cur_tuple);
-	heap_close(pg_attribute);
+	heap_close(pg_attribute, RowExclusiveLock);
 	if (hasind)
 		CatalogCloseIndices(Num_pg_attr_indices, idescs);
 
@@ -734,7 +734,7 @@ UpdateIndexRelation(Oid indexoid,
 	 *	open the system catalog index relation
 	 * ----------------
 	 */
-	pg_index = heap_openr(IndexRelationName);
+	pg_index = heap_openr(IndexRelationName, RowExclusiveLock);
 
 	/* ----------------
 	 *	form a tuple to insert into pg_index
@@ -755,7 +755,7 @@ UpdateIndexRelation(Oid indexoid,
 	 *	close the relation and free the tuple
 	 * ----------------
 	 */
-	heap_close(pg_index);
+	heap_close(pg_index, RowExclusiveLock);
 	pfree(predText);
 	pfree(indexForm);
 	pfree(tuple);
@@ -810,7 +810,7 @@ UpdateIndexPredicate(Oid indexoid, Node *oldPred, Node *predicate)
 		predText = (text *) fmgr(F_TEXTIN, "");
 
 	/* open the index system catalog relation */
-	pg_index = heap_openr(IndexRelationName);
+	pg_index = heap_openr(IndexRelationName, RowExclusiveLock);
 
 	tuple = SearchSysCacheTuple(INDEXRELID,
 								ObjectIdGetDatum(indexoid),
@@ -832,7 +832,7 @@ UpdateIndexPredicate(Oid indexoid, Node *oldPred, Node *predicate)
 	heap_replace(pg_index, &newtup->t_self, newtup, NULL);
 
 	pfree(newtup);
-	heap_close(pg_index);
+	heap_close(pg_index, RowExclusiveLock);
 	pfree(predText);
 }
 
@@ -956,12 +956,10 @@ index_create(char *heapRelationName,
 	 */
 	heapoid = GetHeapRelationOid(heapRelationName, indexRelationName, istemp);
 
-	heapRelation = heap_open(heapoid);
-
 	/*
-	 * Only SELECT ... FOR UPDATE are allowed
+	 * Only SELECT ... FOR UPDATE are allowed while doing this
 	 */
-	LockRelation(heapRelation, ShareLock);
+	heapRelation = heap_open(heapoid, ShareLock);
 
 	/* ----------------
 	 *	  construct new tuple descriptor
@@ -1070,16 +1068,20 @@ index_create(char *heapRelationName,
 	 * the index yet.  We'll be creating more indices and classes later,
 	 * so we delay filling them in until just before we're done with
 	 * bootstrapping.  Otherwise, we call the routine that constructs the
-	 * index.  The heap and index relations are closed by index_build().
+	 * index.
+	 *
+	 * In normal processing mode, the heap and index relations are closed
+	 * by index_build() --- but we continue to hold the ShareLock on the
+	 * heap that we acquired above, until end of transaction.
 	 */
 	if (IsBootstrapProcessingMode())
 	{
 		index_register(heapRelationName, indexRelationName, numatts, attNums,
 					   parameterCount, parameter, funcInfo, predInfo);
+		/* XXX shouldn't we close the heap and index rels here? */
 	}
 	else
 	{
-		heapRelation = heap_openr(heapRelationName);
 		index_build(heapRelation, indexRelation, numatts, attNums,
 					parameterCount, parameter, funcInfo, predInfo);
 	}
@@ -1103,9 +1105,13 @@ index_destroy(Oid indexId)
 
 	Assert(OidIsValid(indexId));
 
-	/* Open now to obtain lock by referencing table?  bjm */
 	userindexRelation = index_open(indexId);
 
+	/*
+	 * Get exclusive lock to ensure no one else is scanning this index.
+	 */
+	LockRelation(userindexRelation, AccessExclusiveLock);
+
 	/* ----------------
 	 *	We do not allow DROP INDEX within a transaction block, because
 	 *	if the transaction is later rolled back there would be no way to
@@ -1121,7 +1127,7 @@ index_destroy(Oid indexId)
 	 * fix RELATION relation
 	 * ----------------
 	 */
-	relationRelation = heap_openr(RelationRelationName);
+	relationRelation = heap_openr(RelationRelationName, RowExclusiveLock);
 
 	tuple = SearchSysCacheTupleCopy(RELOID,
 									ObjectIdGetDatum(indexId),
@@ -1131,13 +1137,13 @@ index_destroy(Oid indexId)
 
 	heap_delete(relationRelation, &tuple->t_self, NULL);
 	pfree(tuple);
-	heap_close(relationRelation);
+	heap_close(relationRelation, RowExclusiveLock);
 
 	/* ----------------
 	 * fix ATTRIBUTE relation
 	 * ----------------
 	 */
-	attributeRelation = heap_openr(AttributeRelationName);
+	attributeRelation = heap_openr(AttributeRelationName, RowExclusiveLock);
 
 	attnum = 1;					/* indexes start at 1 */
 
@@ -1150,7 +1156,7 @@ index_destroy(Oid indexId)
 		pfree(tuple);
 		attnum++;
 	}
-	heap_close(attributeRelation);
+	heap_close(attributeRelation, RowExclusiveLock);
 
 	/* does something only if it is a temp index */
 	remove_temp_relation(indexId);
@@ -1159,16 +1165,16 @@ index_destroy(Oid indexId)
 	 * fix INDEX relation
 	 * ----------------
 	 */
+	indexRelation = heap_openr(IndexRelationName, RowExclusiveLock);
+
 	tuple = SearchSysCacheTupleCopy(INDEXRELID,
 									ObjectIdGetDatum(indexId),
 									0, 0, 0);
 	Assert(HeapTupleIsValid(tuple));
 
-	indexRelation = heap_openr(IndexRelationName);
-
 	heap_delete(indexRelation, &tuple->t_self, NULL);
 	pfree(tuple);
-	heap_close(indexRelation);
+	heap_close(indexRelation, RowExclusiveLock);
 
 	/*
 	 * flush cache and physically remove the file
@@ -1179,7 +1185,8 @@ index_destroy(Oid indexId)
 		elog(ERROR, "index_destroy: unlink: %m");
 
 	index_close(userindexRelation);
-	RelationForgetRelation(RelationGetRelid(userindexRelation));
+
+	RelationForgetRelation(indexId);
 }
 
 /* ----------------------------------------------------------------
@@ -1264,16 +1271,21 @@ UpdateStats(Oid relid, long reltuples, bool hasindex)
 	 * ----------------
 	 */
 
+	/*
+	 * Can't use heap_open here since we don't know if it's an index...
+	 */
 	whichRel = RelationIdGetRelation(relid);
 
 	if (!RelationIsValid(whichRel))
 		elog(ERROR, "UpdateStats: cannot open relation id %u", relid);
 
+	LockRelation(whichRel, ShareLock);
+
 	/* ----------------
 	 * Find the RELATION relation tuple for the given relation.
 	 * ----------------
 	 */
-	pg_class = heap_openr(RelationRelationName);
+	pg_class = heap_openr(RelationRelationName, RowExclusiveLock);
 	if (!RelationIsValid(pg_class))
 		elog(ERROR, "UpdateStats: could not open RELATION relation");
 
@@ -1300,7 +1312,7 @@ UpdateStats(Oid relid, long reltuples, bool hasindex)
 	{
 		if (IsBootstrapProcessingMode())
 			heap_endscan(pg_class_scan);
-		heap_close(pg_class);
+		heap_close(pg_class, RowExclusiveLock);
 		elog(ERROR, "UpdateStats: cannot scan RELATION relation");
 	}
 
@@ -1391,8 +1403,9 @@ UpdateStats(Oid relid, long reltuples, bool hasindex)
 	else
 		heap_endscan(pg_class_scan);
 
-	heap_close(pg_class);
-	heap_close(whichRel);
+	heap_close(pg_class, RowExclusiveLock);
+	/* Cheating a little bit since we didn't open it with heap_open... */
+	heap_close(whichRel, ShareLock);
 }
 
 
@@ -1600,20 +1613,31 @@ DefaultBuild(Relation heapRelation,
 	pfree(datum);
 
 	/*
-	 * Okay, now update the reltuples and relpages statistics for both the
-	 * heap relation and the index.  These statistics are used by the
-	 * planner to choose a scan type.  They are maintained generally by
-	 * the vacuum daemon, but we update them here to make the index useful
-	 * as soon as possible.
-	 */
-	UpdateStats(RelationGetRelid(heapRelation), reltuples, true);
-	UpdateStats(RelationGetRelid(indexRelation), indtuples, false);
-	if (oldPred != NULL)
+	 * 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())
 	{
-		if (indtuples == reltuples)
-			predicate = NULL;
-		UpdateIndexPredicate(RelationGetRelid(indexRelation),
-							 oldPred, predicate);
+		Oid		hrelid = RelationGetRelid(heapRelation);
+		Oid		irelid = RelationGetRelid(indexRelation);
+
+		heap_close(heapRelation, NoLock);
+		index_close(indexRelation);
+		UpdateStats(hrelid, reltuples, true);
+		UpdateStats(irelid, indtuples, false);
+		if (oldPred != NULL)
+		{
+			if (indtuples == reltuples)
+				predicate = NULL;
+			UpdateIndexPredicate(irelid, oldPred, predicate);
+		}
 	}
 }
 
@@ -1715,7 +1739,7 @@ IndexIsUniqueNoCache(Oid indexId)
 	Form_pg_index index;
 	bool		isunique;
 
-	pg_index = heap_openr(IndexRelationName);
+	pg_index = heap_openr(IndexRelationName, AccessShareLock);
 
 	ScanKeyEntryInitialize(&skey[0], (bits16) 0x0,
 						   Anum_pg_index_indexrelid,
@@ -1734,6 +1758,6 @@ IndexIsUniqueNoCache(Oid indexId)
 	isunique = index->indisunique;
 
 	heap_endscan(scandesc);
-	heap_close(pg_index);
+	heap_close(pg_index, AccessShareLock);
 	return isunique;
 }