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.

2 files changed, 53 insertions, 15 deletions

diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index 869e5e1e69..6ad242c0c3 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/index/indexam.c,v 1.35 1999/07/16 04:58:28 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/index/indexam.c,v 1.36 1999/09/18 19:06:04 tgl Exp $
  *
  * INTERFACE ROUTINES
  *		index_open		- open an index relation by relationId
@@ -129,26 +129,49 @@
  *		index_open - open an index relation by relationId
  *
  *		presently the relcache routines do all the work we need
- *		to open/close index relations.
+ *		to open/close index relations.  However, callers of index_open
+ *		expect it to succeed, so we need to check for a failure return.
+ *
+ *		Note: we acquire no lock on the index.  An AccessShareLock is
+ *		acquired by index_beginscan (and released by index_endscan).
  * ----------------
  */
 Relation
 index_open(Oid relationId)
 {
-	return RelationIdGetRelation(relationId);
+	Relation	r;
+
+	r = RelationIdGetRelation(relationId);
+
+	if (! RelationIsValid(r))
+		elog(ERROR, "Index %u does not exist", relationId);
+
+	if (r->rd_rel->relkind != RELKIND_INDEX)
+		elog(ERROR, "%s is not an index relation", r->rd_rel->relname.data);
+
+	return r;
 }
 
 /* ----------------
  *		index_openr - open a index relation by name
  *
- *		presently the relcache routines do all the work we need
- *		to open/close index relations.
+ *		As above, but lookup by name instead of OID.
  * ----------------
  */
 Relation
 index_openr(char *relationName)
 {
-	return RelationNameGetRelation(relationName);
+	Relation	r;
+
+	r = RelationNameGetRelation(relationName);
+
+	if (! RelationIsValid(r))
+		elog(ERROR, "Index '%s' does not exist", relationName);
+
+	if (r->rd_rel->relkind != RELKIND_INDEX)
+		elog(ERROR, "%s is not an index relation", r->rd_rel->relname.data);
+
+	return r;
 }
 
 /* ----------------
@@ -223,6 +246,16 @@ index_beginscan(Relation relation,
 	RELATION_CHECKS;
 	GET_REL_PROCEDURE(beginscan, ambeginscan);
 
+	RelationIncrementReferenceCount(relation);
+
+	/* ----------------
+	 *	Acquire AccessShareLock for the duration of the scan
+	 *
+	 *	Note: we could get an SI inval message here and consequently have
+	 *	to rebuild the relcache entry.  The refcount increment above
+	 *	ensures that we will rebuild it and not just flush it...
+	 * ----------------
+	 */
 	LockRelation(relation, AccessShareLock);
 
 	scandesc = (IndexScanDesc)
@@ -260,7 +293,11 @@ index_endscan(IndexScanDesc scan)
 
 	fmgr(procedure, scan);
 
+	/* Release lock and refcount acquired by index_beginscan */
+
 	UnlockRelation(scan->relation, AccessShareLock);
+
+	RelationDecrementReferenceCount(scan->relation);
 }
 
 /* ----------------
diff --git a/src/backend/access/index/istrat.c b/src/backend/access/index/istrat.c
index 1cdaadbc0c..cab6e58460 100644
--- a/src/backend/access/index/istrat.c
+++ b/src/backend/access/index/istrat.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/index/Attic/istrat.c,v 1.35 1999/07/16 04:58:28 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/index/Attic/istrat.c,v 1.36 1999/09/18 19:06:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -558,7 +558,7 @@ IndexSupportInitialize(IndexStrategy indexStrategy,
 							   F_OIDEQ,
 							   ObjectIdGetDatum(indexObjectId));
 
-		relation = heap_openr(IndexRelationName);
+		relation = heap_openr(IndexRelationName, AccessShareLock);
 		scan = heap_beginscan(relation, false, SnapshotNow, 1, entry);
 		tuple = heap_getnext(scan, 0);
 	}
@@ -591,7 +591,7 @@ IndexSupportInitialize(IndexStrategy indexStrategy,
 	if (IsBootstrapProcessingMode())
 	{
 		heap_endscan(scan);
-		heap_close(relation);
+		heap_close(relation, AccessShareLock);
 	}
 
 	/* if support routines exist for this access method, load them */
@@ -604,7 +604,8 @@ IndexSupportInitialize(IndexStrategy indexStrategy,
 		ScanKeyEntryInitialize(&entry[1], 0, Anum_pg_amproc_amopclaid,
 							   F_OIDEQ, 0);
 
-		relation = heap_openr(AccessMethodProcedureRelationName);
+		relation = heap_openr(AccessMethodProcedureRelationName,
+							  AccessShareLock);
 
 		for (attributeNumber = 1; attributeNumber <= maxAttributeNumber;
 			 attributeNumber++)
@@ -631,7 +632,7 @@ IndexSupportInitialize(IndexStrategy indexStrategy,
 
 			heap_endscan(scan);
 		}
-		heap_close(relation);
+		heap_close(relation, AccessShareLock);
 	}
 
 	ScanKeyEntryInitialize(&entry[0], 0,
@@ -643,8 +644,8 @@ IndexSupportInitialize(IndexStrategy indexStrategy,
 						   Anum_pg_amop_amopclaid,
 						   F_OIDEQ, 0);
 
-	relation = heap_openr(AccessMethodOperatorRelationName);
-	operatorRelation = heap_openr(OperatorRelationName);
+	relation = heap_openr(AccessMethodOperatorRelationName, AccessShareLock);
+	operatorRelation = heap_openr(OperatorRelationName, AccessShareLock);
 
 	for (attributeNumber = maxAttributeNumber; attributeNumber > 0;
 		 attributeNumber--)
@@ -676,8 +677,8 @@ IndexSupportInitialize(IndexStrategy indexStrategy,
 		heap_endscan(scan);
 	}
 
-	heap_close(operatorRelation);
-	heap_close(relation);
+	heap_close(operatorRelation, AccessShareLock);
+	heap_close(relation, AccessShareLock);
 }
 
 /* ----------------