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.

15 files changed, 323 insertions, 378 deletions

diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index ebdd045ce4f..f8196f86ab3 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -6,7 +6,7 @@
  * Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.53 1999/07/18 18:03:49 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.54 1999/09/18 19:06:39 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -51,16 +51,9 @@
  *	  transaction, since by assumption it is only called from outside any
  *	  transaction.
  *
- * Note that the system's use of pg_listener is confined to very short
- * intervals at the end of a transaction that contains NOTIFY statements,
- * or during the transaction caused by an inbound SIGUSR2.	So the fact that
- * pg_listener is a global resource shouldn't cause too much performance
- * problem.  But application authors ought to be discouraged from doing
- * LISTEN or UNLISTEN near the start of a long transaction --- that would
- * result in holding the pg_listener write lock for a long time, possibly
- * blocking unrelated activity.  It could even lead to deadlock against another
- * transaction that touches the same user tables and then tries to NOTIFY.
- * Probably best to do LISTEN or UNLISTEN outside of transaction blocks.
+ * Although we grab AccessExclusiveLock on pg_listener for any operation,
+ * the lock is never held very long, so it shouldn't cause too much of
+ * a performance problem.
  *
  * An application that listens on the same relname it notifies will get
  * NOTIFY messages for its own NOTIFYs.  These can be ignored, if not useful,
@@ -155,26 +148,21 @@ Async_Notify(char *relname)
 
 	TPRINTF(TRACE_NOTIFY, "Async_Notify: %s", relname);
 
-	/*
-	 * We allocate list memory from the global malloc pool to ensure that
-	 * it will live until we want to use it.  This is probably not
-	 * necessary any longer, since we will use it before the end of the
-	 * transaction. DLList only knows how to use malloc() anyway, but we
-	 * could probably palloc() the strings...
-	 */
 	if (!pendingNotifies)
 		pendingNotifies = DLNewList();
-	notifyName = strdup(relname);
-	DLAddHead(pendingNotifies, DLNewElem(notifyName));
-
-	/*
-	 * NOTE: we could check to see if pendingNotifies already has an entry
-	 * for relname, and thus avoid making duplicate entries.  However,
-	 * most apps probably don't notify the same name multiple times per
-	 * transaction, so we'd likely just be wasting cycles to make such a
-	 * check. AsyncExistsPendingNotify() doesn't really care whether the
-	 * list contains duplicates...
-	 */
+	/* no point in making duplicate entries in the list ... */
+	if (!AsyncExistsPendingNotify(relname))
+	{
+		/*
+		 * We allocate list memory from the global malloc pool to ensure
+		 * that it will live until we want to use it.  This is probably not
+		 * necessary any longer, since we will use it before the end of the
+		 * transaction. DLList only knows how to use malloc() anyway, but we
+		 * could probably palloc() the strings...
+		 */
+		notifyName = strdup(relname);
+		DLAddHead(pendingNotifies, DLNewElem(notifyName));
+	}
 }
 
 /*
@@ -212,8 +200,7 @@ Async_Listen(char *relname, int pid)
 
 	TPRINTF(TRACE_NOTIFY, "Async_Listen: %s", relname);
 
-	lRel = heap_openr(ListenerRelationName);
-	LockRelation(lRel, AccessExclusiveLock);
+	lRel = heap_openr(ListenerRelationName, AccessExclusiveLock);
 	tdesc = RelationGetDescr(lRel);
 
 	/* Detect whether we are already listening on this relname */
@@ -236,9 +223,8 @@ Async_Listen(char *relname, int pid)
 
 	if (alreadyListener)
 	{
+		heap_close(lRel, AccessExclusiveLock);
 		elog(NOTICE, "Async_Listen: We are already listening on %s", relname);
-		UnlockRelation(lRel, AccessExclusiveLock);
-		heap_close(lRel);
 		return;
 	}
 
@@ -262,8 +248,7 @@ Async_Listen(char *relname, int pid)
 	heap_insert(lRel, newtup);
 	pfree(newtup);
 
-	UnlockRelation(lRel, AccessExclusiveLock);
-	heap_close(lRel);
+	heap_close(lRel, AccessExclusiveLock);
 
 	/*
 	 * now that we are listening, make sure we will unlisten before dying.
@@ -308,18 +293,14 @@ Async_Unlisten(char *relname, int pid)
 
 	TPRINTF(TRACE_NOTIFY, "Async_Unlisten %s", relname);
 
+	lRel = heap_openr(ListenerRelationName, AccessExclusiveLock);
 	/* Note we assume there can be only one matching tuple. */
 	lTuple = SearchSysCacheTuple(LISTENREL, PointerGetDatum(relname),
 								 Int32GetDatum(pid),
 								 0, 0);
 	if (lTuple != NULL)
-	{
-		lRel = heap_openr(ListenerRelationName);
-		LockRelation(lRel, AccessExclusiveLock);
 		heap_delete(lRel, &lTuple->t_self, NULL);
-		UnlockRelation(lRel, AccessExclusiveLock);
-		heap_close(lRel);
-	}
+	heap_close(lRel, AccessExclusiveLock);
 
 	/*
 	 * We do not complain about unlistening something not being listened;
@@ -354,8 +335,7 @@ Async_UnlistenAll()
 
 	TPRINTF(TRACE_NOTIFY, "Async_UnlistenAll");
 
-	lRel = heap_openr(ListenerRelationName);
-	LockRelation(lRel, AccessExclusiveLock);
+	lRel = heap_openr(ListenerRelationName, AccessExclusiveLock);
 	tdesc = RelationGetDescr(lRel);
 
 	/* Find and delete all entries with my listenerPID */
@@ -369,8 +349,7 @@ Async_UnlistenAll()
 		heap_delete(lRel, &lTuple->t_self, NULL);
 
 	heap_endscan(sRel);
-	UnlockRelation(lRel, AccessExclusiveLock);
-	heap_close(lRel);
+	heap_close(lRel, AccessExclusiveLock);
 }
 
 /*
@@ -462,8 +441,7 @@ AtCommit_Notify()
 
 	TPRINTF(TRACE_NOTIFY, "AtCommit_Notify");
 
-	lRel = heap_openr(ListenerRelationName);
-	LockRelation(lRel, AccessExclusiveLock);
+	lRel = heap_openr(ListenerRelationName, AccessExclusiveLock);
 	tdesc = RelationGetDescr(lRel);
 	sRel = heap_beginscan(lRel, 0, SnapshotNow, 0, (ScanKey) NULL);
 
@@ -542,10 +520,13 @@ AtCommit_Notify()
 	heap_endscan(sRel);
 
 	/*
-	 * We do not do RelationUnsetLockForWrite(lRel) here, because the
-	 * transaction is about to be committed anyway.
+	 * We do NOT release the lock on pg_listener here; we need to hold it
+	 * until end of transaction (which is about to happen, anyway) to
+	 * ensure that notified backends see our tuple updates when they look.
+	 * Else they might disregard the signal, which would make the
+	 * application programmer very unhappy.
 	 */
-	heap_close(lRel);
+	heap_close(lRel, NoLock);
 
 	ClearPendingNotifies();
 
@@ -756,8 +737,7 @@ ProcessIncomingNotify(void)
 
 	StartTransactionCommand();
 
-	lRel = heap_openr(ListenerRelationName);
-	LockRelation(lRel, AccessExclusiveLock);
+	lRel = heap_openr(ListenerRelationName, AccessExclusiveLock);
 	tdesc = RelationGetDescr(lRel);
 
 	/* Scan only entries with my listenerPID */
@@ -794,10 +774,13 @@ ProcessIncomingNotify(void)
 	heap_endscan(sRel);
 
 	/*
-	 * We do not do RelationUnsetLockForWrite(lRel) here, because the
-	 * transaction is about to be committed anyway.
+	 * We do NOT release the lock on pg_listener here; we need to hold it
+	 * until end of transaction (which is about to happen, anyway) to
+	 * ensure that other backends see our tuple updates when they look.
+	 * Otherwise, a transaction started after this one might mistakenly
+	 * think it doesn't need to send this backend a new NOTIFY.
 	 */
-	heap_close(lRel);
+	heap_close(lRel, NoLock);
 
 	CommitTransactionCommand();
 
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 6f99a920f5a..d578fc263f0 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -14,7 +14,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/cluster.c,v 1.44 1999/07/17 20:16:51 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/cluster.c,v 1.45 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -101,29 +101,20 @@ cluster(char *oldrelname, char *oldindexname)
 
 	/*
 	 * Like vacuum, cluster spans transactions, so I'm going to handle it
-	 * in the same way.
+	 * in the same way: commit and restart transactions where needed.
+	 *
+	 * We grab exclusive access to the target rel and index for the
+	 * duration of the initial transaction.
 	 */
 
-	/* matches the StartTransaction in PostgresMain() */
-
-	OldHeap = heap_openr(oldrelname);
-	if (!RelationIsValid(OldHeap))
-	{
-		elog(ERROR, "cluster: unknown relation: \"%s\"",
-			 oldrelname);
-	}
-	OIDOldHeap = RelationGetRelid(OldHeap);		/* Get OID for the index
-												 * scan    */
+	OldHeap = heap_openr(oldrelname, AccessExclusiveLock);
+	OIDOldHeap = RelationGetRelid(OldHeap);
 
 	OldIndex = index_openr(oldindexname);		/* Open old index relation	*/
-	if (!RelationIsValid(OldIndex))
-	{
-		elog(ERROR, "cluster: unknown index: \"%s\"",
-			 oldindexname);
-	}
-	OIDOldIndex = RelationGetRelid(OldIndex);	/* OID for the index scan		  */
+	LockRelation(OldIndex, AccessExclusiveLock);
+	OIDOldIndex = RelationGetRelid(OldIndex);
 
-	heap_close(OldHeap);
+	heap_close(OldHeap, NoLock); /* do NOT give up the locks */
 	index_close(OldIndex);
 
 	/*
@@ -132,7 +123,7 @@ cluster(char *oldrelname, char *oldindexname)
 	 * will get the lock after being blocked and add rows which won't be
 	 * present in the new table. Bleagh! I'd be best to try and ensure
 	 * that no-one's in the tables for the entire duration of this process
-	 * with a pg_vlock.
+	 * with a pg_vlock.  XXX Isn't the above comment now invalid?
 	 */
 	NewHeap = copy_heap(OIDOldHeap);
 	OIDNewHeap = RelationGetRelid(NewHeap);
@@ -171,7 +162,7 @@ cluster(char *oldrelname, char *oldindexname)
 	renamerel(NewIndexName, saveoldindexname);
 
 	/*
-	 * Again flush all the buffers.
+	 * Again flush all the buffers.  XXX perhaps not needed?
 	 */
 	CommitTransactionCommand();
 	StartTransactionCommand();
@@ -193,7 +184,7 @@ copy_heap(Oid OIDOldHeap)
 	 */
 	snprintf(NewName, NAMEDATALEN, "temp_%x", OIDOldHeap);
 
-	OldHeap = heap_open(OIDOldHeap);
+	OldHeap = heap_open(OIDOldHeap, AccessExclusiveLock);
 	OldHeapDesc = RelationGetDescr(OldHeap);
 
 	/*
@@ -209,10 +200,11 @@ copy_heap(Oid OIDOldHeap)
 	if (!OidIsValid(OIDNewHeap))
 		elog(ERROR, "clusterheap: cannot create temporary heap relation\n");
 
-	NewHeap = heap_open(OIDNewHeap);
+	/* XXX why are we bothering to do this: */
+	NewHeap = heap_open(OIDNewHeap, AccessExclusiveLock);
 
-	heap_close(NewHeap);
-	heap_close(OldHeap);
+	heap_close(NewHeap, AccessExclusiveLock);
+	heap_close(OldHeap, AccessExclusiveLock);
 
 	return NewHeap;
 }
@@ -233,7 +225,7 @@ copy_index(Oid OIDOldIndex, Oid OIDNewHeap)
 	int			natts;
 	FuncIndexInfo *finfo;
 
-	NewHeap = heap_open(OIDNewHeap);
+	NewHeap = heap_open(OIDNewHeap, AccessExclusiveLock);
 	OldIndex = index_open(OIDOldIndex);
 
 	/*
@@ -305,8 +297,8 @@ copy_index(Oid OIDOldIndex, Oid OIDNewHeap)
 				 Old_pg_index_Form->indisunique,
 				 Old_pg_index_Form->indisprimary);
 
-	heap_close(OldIndex);
-	heap_close(NewHeap);
+	index_close(OldIndex);
+	heap_close(NewHeap, AccessExclusiveLock);
 }
 
 
@@ -326,8 +318,8 @@ rebuildheap(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex)
 	 * Open the relations I need. Scan through the OldHeap on the OldIndex
 	 * and insert each tuple into the NewHeap.
 	 */
-	LocalNewHeap = (Relation) heap_open(OIDNewHeap);
-	LocalOldHeap = (Relation) heap_open(OIDOldHeap);
+	LocalNewHeap = heap_open(OIDNewHeap, AccessExclusiveLock);
+	LocalOldHeap = heap_open(OIDOldHeap, AccessExclusiveLock);
 	LocalOldIndex = (Relation) index_open(OIDOldIndex);
 
 	ScanDesc = index_beginscan(LocalOldIndex, false, 0, (ScanKey) NULL);
@@ -344,6 +336,6 @@ rebuildheap(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex)
 	index_endscan(ScanDesc);
 
 	index_close(LocalOldIndex);
-	heap_close(LocalOldHeap);
-	heap_close(LocalNewHeap);
+	heap_close(LocalOldHeap, AccessExclusiveLock);
+	heap_close(LocalNewHeap, AccessExclusiveLock);
 }
diff --git a/src/backend/commands/command.c b/src/backend/commands/command.c
index 7fe62cf1246..8872bcdfac7 100644
--- a/src/backend/commands/command.c
+++ b/src/backend/commands/command.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/command.c,v 1.53 1999/09/04 21:19:33 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/command.c,v 1.54 1999/09/18 19:06:40 tgl Exp $
  *
  * NOTES
  *	  The PortalExecutorHeapMemory crap needs to be eliminated
@@ -283,6 +283,7 @@ PerformAddAttribute(char *relationName,
 {
 	Relation	rel,
 				attrdesc;
+	Oid			myrelid;
 	HeapTuple	reltup;
 	HeapTuple	attributeTuple;
 	Form_pg_attribute attribute;
@@ -311,6 +312,14 @@ PerformAddAttribute(char *relationName,
 #endif
 
 	/*
+	 * Grab an exclusive lock on the target table, which we will NOT release
+	 * until end of transaction.
+	 */
+	rel = heap_openr(relationName, AccessExclusiveLock);
+	myrelid = RelationGetRelid(rel);
+	heap_close(rel, NoLock);	/* close rel but keep lock! */
+
+	/*
 	 * we can't add a not null attribute
 	 */
 	if (colDef->is_not_null)
@@ -331,20 +340,10 @@ PerformAddAttribute(char *relationName,
 	{
 		if (inherits)
 		{
-			Oid			myrelid,
-						childrelid;
+			Oid			childrelid;
 			List	   *child,
 					   *children;
 
-			rel = heap_openr(relationName);
-			if (!RelationIsValid(rel))
-			{
-				elog(ERROR, "PerformAddAttribute: unknown relation: \"%s\"",
-					 relationName);
-			}
-			myrelid = RelationGetRelid(rel);
-			heap_close(rel);
-
 			/* this routine is actually in the planner */
 			children = find_all_inheritors(lconsi(myrelid, NIL), NIL);
 
@@ -358,31 +357,23 @@ PerformAddAttribute(char *relationName,
 				childrelid = lfirsti(child);
 				if (childrelid == myrelid)
 					continue;
-				rel = heap_open(childrelid);
-				if (!RelationIsValid(rel))
-				{
-					elog(ERROR, "PerformAddAttribute: can't find catalog entry for inheriting class with oid %u",
-						 childrelid);
-				}
+				rel = heap_open(childrelid, AccessExclusiveLock);
 				PerformAddAttribute((rel->rd_rel->relname).data,
 									userName, false, colDef);
-				heap_close(rel);
+				heap_close(rel, AccessExclusiveLock);
 			}
 		}
 	}
 
-	rel = heap_openr(RelationRelationName);
+	rel = heap_openr(RelationRelationName, RowExclusiveLock);
 
 	reltup = SearchSysCacheTupleCopy(RELNAME,
 									 PointerGetDatum(relationName),
 									 0, 0, 0);
 
 	if (!HeapTupleIsValid(reltup))
-	{
-		heap_close(rel);
 		elog(ERROR, "PerformAddAttribute: relation \"%s\" not found",
 			 relationName);
-	}
 
 	/*
 	 * XXX is the following check sufficient?
@@ -391,23 +382,15 @@ PerformAddAttribute(char *relationName,
 	{
 		elog(ERROR, "PerformAddAttribute: index relation \"%s\" not changed",
 			 relationName);
-		return;
 	}
 
 	minattnum = ((Form_pg_class) GETSTRUCT(reltup))->relnatts;
 	maxatts = minattnum + 1;
 	if (maxatts > MaxHeapAttributeNumber)
-	{
-		pfree(reltup);
-		heap_close(rel);
 		elog(ERROR, "PerformAddAttribute: relations limited to %d attributes",
 			 MaxHeapAttributeNumber);
-	}
-
-	attrdesc = heap_openr(AttributeRelationName);
 
-	Assert(attrdesc);
-	Assert(RelationGetForm(attrdesc));
+	attrdesc = heap_openr(AttributeRelationName, RowExclusiveLock);
 
 	/*
 	 * Open all (if any) pg_attribute indices
@@ -438,12 +421,8 @@ PerformAddAttribute(char *relationName,
 								  0, 0);
 
 		if (HeapTupleIsValid(tup))
-		{
-			heap_close(attrdesc);
-			heap_close(rel);
 			elog(ERROR, "PerformAddAttribute: attribute \"%s\" already exists in class \"%s\"",
 				 colDef->colname, relationName);
-		}
 
 		/*
 		 * check to see if it is an array attribute.
@@ -490,7 +469,8 @@ PerformAddAttribute(char *relationName,
 
 	if (hasindex)
 		CatalogCloseIndices(Num_pg_attr_indices, idescs);
-	heap_close(attrdesc);
+
+	heap_close(attrdesc, RowExclusiveLock);
 
 	((Form_pg_class) GETSTRUCT(reltup))->relnatts = maxatts;
 	heap_replace(rel, &reltup->t_self, reltup, NULL);
@@ -501,7 +481,7 @@ PerformAddAttribute(char *relationName,
 	CatalogCloseIndices(Num_pg_class_indices, ridescs);
 
 	pfree(reltup);
-	heap_close(rel);
+	heap_close(rel, RowExclusiveLock);
 }
 
 void
@@ -510,9 +490,9 @@ LockTableCommand(LockStmt *lockstmt)
 	Relation	rel;
 	int			aclresult;
 
-	rel = heap_openr(lockstmt->relname);
-	if (rel == NULL)
-		elog(ERROR, "LOCK TABLE: relation %s can't be openned", lockstmt->relname);
+	rel = heap_openr(lockstmt->relname, NoLock);
+	if (! RelationIsValid(rel))
+		elog(ERROR, "Relation '%s' does not exist", lockstmt->relname);
 
 	if (lockstmt->mode == AccessShareLock)
 		aclresult = pg_aclcheck(lockstmt->relname, GetPgUserName(), ACL_RD);
@@ -524,4 +504,5 @@ LockTableCommand(LockStmt *lockstmt)
 
 	LockRelation(rel, lockstmt->mode);
 
+	heap_close(rel, NoLock);	/* close rel, keep lock */
 }
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 35f3d6da0b3..493d2a170e1 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -6,7 +6,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/copy.c,v 1.87 1999/09/11 22:28:11 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/copy.c,v 1.88 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -251,22 +251,22 @@ DoCopy(char *relname, bool binary, bool oids, bool from, bool pipe,
 	Relation	rel;
 	extern char *UserName;		/* defined in global.c */
 	const AclMode required_access = from ? ACL_WR : ACL_RD;
+	LOCKMODE	required_lock = from ? AccessExclusiveLock : AccessShareLock;
+	/* Note: AccessExclusive is probably overkill for copying to a relation,
+	 * but that's what the existing code grabs on the rel's indices.  If
+	 * this is relaxed then I think the index locks need relaxed also.
+	 */
 	int			result;
 
-	rel = heap_openr(relname);
-	if (rel == NULL)
-		elog(ERROR, "COPY command failed.  Class %s "
-			 "does not exist.", relname);
+	rel = heap_openr(relname, required_lock);
 
 	result = pg_aclcheck(relname, UserName, required_access);
 	if (result != ACLCHECK_OK)
 		elog(ERROR, "%s: %s", relname, aclcheck_error_strings[result]);
-	/* Above should not return */
-	else if (!superuser() && !pipe)
+	else if (!pipe && !superuser())
 		elog(ERROR, "You must have Postgres superuser privilege to do a COPY "
 			 "directly to or from a file.  Anyone can COPY to stdout or "
 			 "from stdin.  Psql's \\copy command also works for anyone.");
-	/* Above should not return. */
 	else
 	{
 		if (from)
@@ -342,6 +342,8 @@ DoCopy(char *relname, bool binary, bool oids, bool from, bool pipe,
 				pq_endcopyout(false);
 		}
 	}
+
+	heap_close(rel, required_lock);
 }
 
 
@@ -500,8 +502,6 @@ CopyTo(Relation rel, bool binary, bool oids, FILE *fp, char *delim)
 		pfree(elements);
 		pfree(typmod);
 	}
-
-	heap_close(rel);
 }
 
 static void
@@ -905,20 +905,19 @@ CopyFrom(Relation rel, bool binary, bool oids, FILE *fp, char *delim)
 		pfree(typmod);
 	}
 
-	/* comments in execUtils.c */
 	if (has_index)
 	{
 		for (i = 0; i < n_indices; i++)
 		{
 			if (index_rels[i] == NULL)
 				continue;
+			/* see comments in ExecOpenIndices() in execUtils.c */
 			if ((index_rels[i])->rd_rel->relam != BTREE_AM_OID &&
 				(index_rels[i])->rd_rel->relam != HASH_AM_OID)
 				UnlockRelation(index_rels[i], AccessExclusiveLock);
 			index_close(index_rels[i]);
 		}
 	}
-	heap_close(rel);
 }
 
 
@@ -991,7 +990,7 @@ IsTypeByVal(Oid type)
 /*
  * Given the OID of a relation, return an array of index relation descriptors
  * and the number of index relations.  These relation descriptors are open
- * using heap_open().
+ * using index_open().
  *
  * Space for the array itself is palloc'ed.
  */
@@ -1017,7 +1016,7 @@ GetIndexRelations(Oid main_relation_oid,
 	int			i;
 	bool		isnull;
 
-	pg_index_rel = heap_openr(IndexRelationName);
+	pg_index_rel = heap_openr(IndexRelationName, AccessShareLock);
 	scandesc = heap_beginscan(pg_index_rel, 0, SnapshotNow, 0, NULL);
 	tupDesc = RelationGetDescr(pg_index_rel);
 
@@ -1044,7 +1043,7 @@ GetIndexRelations(Oid main_relation_oid,
 	}
 
 	heap_endscan(scandesc);
-	heap_close(pg_index_rel);
+	heap_close(pg_index_rel, AccessShareLock);
 
 	/* We cannot trust to relhasindex of the main_relation now, so... */
 	if (*n_indices == 0)
@@ -1055,7 +1054,7 @@ GetIndexRelations(Oid main_relation_oid,
 	for (i = 0, scan = head; i < *n_indices; i++, scan = scan->next)
 	{
 		(*index_rels)[i] = index_open(scan->index_rel_oid);
-		/* comments in execUtils.c */
+		/* see comments in ExecOpenIndices() in execUtils.c */
 		if ((*index_rels)[i] != NULL &&
 			((*index_rels)[i])->rd_rel->relam != BTREE_AM_OID &&
 			((*index_rels)[i])->rd_rel->relam != HASH_AM_OID)
diff --git a/src/backend/commands/creatinh.c b/src/backend/commands/creatinh.c
index b93a40e7ba3..c1069dd41e6 100644
--- a/src/backend/commands/creatinh.c
+++ b/src/backend/commands/creatinh.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/creatinh.c,v 1.45 1999/07/17 20:16:52 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/creatinh.c,v 1.46 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -202,14 +202,13 @@ MergeAttributes(List *schema, List *supers, List **supconstr)
 	 */
 	foreach(entry, schema)
 	{
-		List	   *rest;
 		ColumnDef  *coldef = lfirst(entry);
+		List	   *rest;
 
 		foreach(rest, lnext(entry))
 		{
-
 			/*
-			 * check for duplicated relation names
+			 * check for duplicated names within the new relation
 			 */
 			ColumnDef  *restdef = lfirst(rest);
 
@@ -246,17 +245,14 @@ MergeAttributes(List *schema, List *supers, List **supconstr)
 		TupleDesc	tupleDesc;
 		TupleConstr *constr;
 
-		relation = heap_openr(name);
-		if (relation == NULL)
-		{
-			elog(ERROR,
-				 "MergeAttr: Can't inherit from non-existent superclass '%s'", name);
-		}
-		if (relation->rd_rel->relkind == 'S')
-			elog(ERROR, "MergeAttr: Can't inherit from sequence superclass '%s'", name);
+		relation = heap_openr(name, AccessShareLock);
 		tupleDesc = RelationGetDescr(relation);
 		constr = tupleDesc->constr;
 
+		/* XXX shouldn't this test be stricter?  No indexes, for example? */
+		if (relation->rd_rel->relkind == 'S')
+			elog(ERROR, "MergeAttr: Can't inherit from sequence superclass '%s'", name);
+
 		for (attrno = relation->rd_rel->relnatts - 1; attrno >= 0; attrno--)
 		{
 			Form_pg_attribute attribute = tupleDesc->attrs[attrno];
@@ -340,9 +336,11 @@ MergeAttributes(List *schema, List *supers, List **supconstr)
 		}
 
 		/*
-		 * iteration cleanup and result collection
+		 * Close the parent rel, but keep our AccessShareLock on it until
+		 * xact commit.  That will prevent someone else from deleting or
+		 * ALTERing the parent before the child is committed.
 		 */
-		heap_close(relation);
+		heap_close(relation, NoLock);
 
 		/*
 		 * wants the inherited schema to appear in the order they are
@@ -386,7 +384,7 @@ StoreCatalogInheritance(Oid relationId, List *supers)
 	 * Catalog INHERITS information.
 	 * ----------------
 	 */
-	relation = heap_openr(InheritsRelationName);
+	relation = heap_openr(InheritsRelationName, RowExclusiveLock);
 	desc = RelationGetDescr(relation);
 
 	seqNumber = 1;
@@ -422,7 +420,7 @@ StoreCatalogInheritance(Oid relationId, List *supers)
 		seqNumber += 1;
 	}
 
-	heap_close(relation);
+	heap_close(relation, RowExclusiveLock);
 
 	/* ----------------
 	 * Catalog IPL information.
@@ -510,7 +508,7 @@ again:
 	 *	3.
 	 * ----------------
 	 */
-	relation = heap_openr(InheritancePrecidenceListRelationName);
+	relation = heap_openr(InheritancePrecidenceListRelationName, RowExclusiveLock);
 	desc = RelationGetDescr(relation);
 
 	seqNumber = 1;
@@ -537,7 +535,7 @@ again:
 		seqNumber += 1;
 	}
 
-	heap_close(relation);
+	heap_close(relation, RowExclusiveLock);
 }
 
 /*
diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c
index b779715fe8e..24eb5b531d3 100644
--- a/src/backend/commands/dbcommands.c
+++ b/src/backend/commands/dbcommands.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/dbcommands.c,v 1.39 1999/07/17 20:16:52 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/dbcommands.c,v 1.40 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -103,6 +103,8 @@ destroydb(char *dbname, CommandDest dest)
 	/* stop the vacuum daemon */
 	stop_vacuum(dbpath, dbname);
 
+	/* XXX what about stopping backends connected to the target database? */
+
 	path = ExpandDatabasePath(dbpath);
 	if (path == NULL)
 		elog(ERROR, "Unable to locate path '%s'"
@@ -189,6 +191,7 @@ check_permissions(char *command,
 	utup = SearchSysCacheTuple(USENAME,
 							   PointerGetDatum(userName),
 							   0, 0, 0);
+	Assert(utup);
 	*userIdP = ((Form_pg_shadow) GETSTRUCT(utup))->usesysid;
 	use_super = ((Form_pg_shadow) GETSTRUCT(utup))->usesuper;
 	use_createdb = ((Form_pg_shadow) GETSTRUCT(utup))->usecreatedb;
@@ -211,22 +214,13 @@ check_permissions(char *command,
 	/* Check to make sure database is owned by this user */
 
 	/*
-	 * need the reldesc to get the database owner out of dbtup and to set
-	 * a write lock on it.
+	 * Acquire exclusive lock on pg_database from the beginning, even though
+	 * we only need read access right here, to avoid potential deadlocks
+	 * from upgrading our lock later.  (Is this still necessary?  Could we
+	 * use something weaker than exclusive lock?)
 	 */
-	dbrel = heap_openr(DatabaseRelationName);
-
-	if (!RelationIsValid(dbrel))
-		elog(FATAL, "%s: cannot open relation \"%-.*s\"",
-			 command, DatabaseRelationName);
+	dbrel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 
-	/*
-	 * Acquire a write lock on pg_database from the beginning to avoid
-	 * upgrading a read lock to a write lock.  Upgrading causes long
-	 * delays when multiple 'createdb's or 'destroydb's are run simult.
-	 * -mer 7/3/91
-	 */
-	LockRelation(dbrel, AccessExclusiveLock);
 	dbtup = get_pg_dbtup(command, dbname, dbrel);
 	dbfound = HeapTupleIsValid(dbtup);
 
@@ -248,7 +242,8 @@ check_permissions(char *command,
 	else
 		*dbIdP = InvalidOid;
 
-	heap_close(dbrel);
+	/* We will keep the lock on dbrel until end of transaction. */
+	heap_close(dbrel, NoLock);
 
 	/*
 	 * Now be sure that the user is allowed to do this.
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 3b1da18783d..11e5c394239 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -4,7 +4,7 @@
  *
  * Copyright (c) 1994-5, Regents of the University of California
  *
- *	  $Id: explain.c,v 1.47 1999/09/11 19:06:36 tgl Exp $
+ *	  $Id: explain.c,v 1.48 1999/09/18 19:06:40 tgl Exp $
  *
  */
 
@@ -211,11 +211,14 @@ explain_outNode(StringInfo str, Plan *plan, int indent, ExplainState *es)
 			i = 0;
 			foreach(l, ((IndexScan *) plan)->indxid)
 			{
-				relation = RelationIdCacheGetRelation((int) lfirst(l));
+				relation = RelationIdGetRelation(lfirsti(l));
+				Assert(relation);
 				if (++i > 1)
 					appendStringInfo(str, ", ");
 				appendStringInfo(str,
 								 stringStringInfo((RelationGetRelationName(relation))->data));
+				/* drop relcache refcount from RelationIdGetRelation */
+				RelationDecrementReferenceCount(relation);
 			}
 		case T_SeqScan:
 			if (((Scan *) plan)->scanrelid > 0)
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 31d3419bee6..113854311d6 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/indexcmds.c,v 1.10 1999/08/22 20:14:37 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/indexcmds.c,v 1.11 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -324,15 +324,15 @@ ExtendIndex(char *indexRelationName, Expr *predicate, List *rangetable)
 		FIsetProcOid(funcInfo, tuple->t_data->t_oid);
 	}
 
-	heapRelation = heap_open(relationId);
+	heapRelation = heap_open(relationId, ShareLock);
 	indexRelation = index_open(indexId);
 
-	LockRelation(heapRelation, ShareLock);
-
 	InitIndexStrategy(numberOfAttributes, indexRelation, accessMethodId);
 
 	index_build(heapRelation, indexRelation, numberOfAttributes,
 				attributeNumberA, 0, NULL, funcInfo, predInfo);
+
+	/* heap and index rels are closed as a side-effect of index_build */
 }
 
 
diff --git a/src/backend/commands/proclang.c b/src/backend/commands/proclang.c
index 21fde2af579..b2fc76f090b 100644
--- a/src/backend/commands/proclang.c
+++ b/src/backend/commands/proclang.c
@@ -120,15 +120,14 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
 	values[i++] = ObjectIdGetDatum(procTup->t_data->t_oid);
 	values[i++] = (Datum) fmgr(F_TEXTIN, stmt->plcompiler);
 
-	rel = heap_openr(LanguageRelationName);
+	rel = heap_openr(LanguageRelationName, RowExclusiveLock);
 
 	tupDesc = rel->rd_att;
 	tup = heap_formtuple(tupDesc, values, nulls);
 
 	heap_insert(rel, tup);
 
-	heap_close(rel);
-	return;
+	heap_close(rel, RowExclusiveLock);
 }
 
 
@@ -160,6 +159,8 @@ DropProceduralLanguage(DropPLangStmt *stmt)
 	 */
 	case_translate_language_name(stmt->plname, languageName);
 
+	rel = heap_openr(LanguageRelationName, RowExclusiveLock);
+
 	langTup = SearchSysCacheTupleCopy(LANNAME,
 									  PointerGetDatum(languageName),
 									  0, 0, 0);
@@ -167,14 +168,11 @@ DropProceduralLanguage(DropPLangStmt *stmt)
 		elog(ERROR, "Language %s doesn't exist", languageName);
 
 	if (!((Form_pg_language) GETSTRUCT(langTup))->lanispl)
-	{
 		elog(ERROR, "Language %s isn't a created procedural language",
 			 languageName);
-	}
 
-	rel = heap_openr(LanguageRelationName);
 	heap_delete(rel, &langTup->t_self, NULL);
 
 	pfree(langTup);
-	heap_close(rel);
+	heap_close(rel, RowExclusiveLock);
 }
diff --git a/src/backend/commands/remove.c b/src/backend/commands/remove.c
index af20c79f17d..a73964cb02c 100644
--- a/src/backend/commands/remove.c
+++ b/src/backend/commands/remove.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/remove.c,v 1.36 1999/07/17 20:16:53 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/remove.c,v 1.37 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -75,13 +75,14 @@ RemoveOperator(char *operatorName,		/* operator name */
 	else
 		oprtype = 'r';
 
+	relation = heap_openr(OperatorRelationName, RowExclusiveLock);
+
 	tup = SearchSysCacheTupleCopy(OPRNAME,
 								  PointerGetDatum(operatorName),
 								  ObjectIdGetDatum(typeId1),
 								  ObjectIdGetDatum(typeId2),
 								  CharGetDatum(oprtype));
 
-	relation = heap_openr(OperatorRelationName);
 	if (HeapTupleIsValid(tup))
 	{
 #ifndef NO_SECURITY
@@ -117,7 +118,7 @@ RemoveOperator(char *operatorName,		/* operator name */
 		}
 	}
 	pfree(tup);
-	heap_close(relation);
+	heap_close(relation, RowExclusiveLock);
 }
 
 #ifdef NOTYET
@@ -141,7 +142,7 @@ SingleOpOperatorRemove(Oid typeOid)
 
 	ScanKeyEntryInitialize(&key[0],
 						   0, 0, F_OIDEQ, (Datum) typeOid);
-	rel = heap_openr(OperatorRelationName);
+	rel = heap_openr(OperatorRelationName, RowExclusiveLock);
 	for (i = 0; i < 3; ++i)
 	{
 		key[0].sk_attno = attnums[i];
@@ -150,7 +151,7 @@ SingleOpOperatorRemove(Oid typeOid)
 			heap_delete(rel, &tup->t_self, NULL);
 		heap_endscan(scan);
 	}
-	heap_close(rel);
+	heap_close(rel, RowExclusiveLock);
 }
 
 /*
@@ -187,7 +188,7 @@ AttributeAndRelationRemove(Oid typeOid)
 	oidptr = (struct oidlist *) palloc(sizeof(*oidptr));
 	oidptr->next = NULL;
 	optr = oidptr;
-	rel = heap_openr(AttributeRelationName);
+	rel = heap_openr(AttributeRelationName, AccessShareLock);
 	scan = heap_beginscan(rel, 0, SnapshotNow, 1, key);
 	while (HeapTupleIsValid(tup = heap_getnext(scan, 0)))
 	{
@@ -197,14 +198,15 @@ AttributeAndRelationRemove(Oid typeOid)
 	}
 	optr->next = NULL;
 	heap_endscan(scan);
-	heap_close(rel);
+	heap_close(rel, AccessShareLock);
 
+	optr = oidptr;
 
 	ScanKeyEntryInitialize(&key[0], 0,
 						   ObjectIdAttributeNumber,
 						   F_OIDEQ, (Datum) 0);
-	optr = oidptr;
-	rel = heap_openr(RelationRelationName);
+	/* get RowExclusiveLock because heap_destroy will need it */
+	rel = heap_openr(RelationRelationName, RowExclusiveLock);
 	while (PointerIsValid((char *) optr->next))
 	{
 		key[0].sk_argument = (Datum) (optr++)->reloid;
@@ -217,9 +219,9 @@ AttributeAndRelationRemove(Oid typeOid)
 			name = (((Form_pg_class) GETSTRUCT(tup))->relname).data;
 			heap_destroy_with_catalog(name);
 		}
+		heap_endscan(scan);
 	}
-	heap_endscan(scan);
-	heap_close(rel);
+	heap_close(rel, RowExclusiveLock);
 }
 
 #endif	 /* NOTYET */
@@ -245,18 +247,17 @@ RemoveType(char *typeName)		/* type name to be removed */
 			 typeName);
 #endif
 
-	relation = heap_openr(TypeRelationName);
+	relation = heap_openr(TypeRelationName, RowExclusiveLock);
+
 	tup = SearchSysCacheTuple(TYPNAME,
 							  PointerGetDatum(typeName),
 							  0, 0, 0);
-
 	if (!HeapTupleIsValid(tup))
 	{
-		heap_close(relation);
+		heap_close(relation, RowExclusiveLock);
 		elog(ERROR, "RemoveType: type '%s' does not exist", typeName);
 	}
 
-	relation = heap_openr(TypeRelationName);
 	typeOid = tup->t_data->t_oid;
 	heap_delete(relation, &tup->t_self, NULL);
 
@@ -267,14 +268,13 @@ RemoveType(char *typeName)		/* type name to be removed */
 							  0, 0, 0);
 	if (!HeapTupleIsValid(tup))
 	{
-		heap_close(relation);
-		elog(ERROR, "RemoveType: type '%s' does not exist", typeName);
+		heap_close(relation, RowExclusiveLock);
+		elog(ERROR, "RemoveType: type '%s' does not exist", shadow_type);
 	}
 
-	typeOid = tup->t_data->t_oid;
 	heap_delete(relation, &tup->t_self, NULL);
 
-	heap_close(relation);
+	heap_close(relation, RowExclusiveLock);
 }
 
 /*
@@ -328,7 +328,7 @@ RemoveFunction(char *functionName,		/* function name to be removed */
 	}
 #endif
 
-	relation = heap_openr(ProcedureRelationName);
+	relation = heap_openr(ProcedureRelationName, RowExclusiveLock);
 	tup = SearchSysCacheTuple(PRONAME,
 							  PointerGetDatum(functionName),
 							  Int32GetDatum(nargs),
@@ -337,19 +337,19 @@ RemoveFunction(char *functionName,		/* function name to be removed */
 
 	if (!HeapTupleIsValid(tup))
 	{
-		heap_close(relation);
+		heap_close(relation, RowExclusiveLock);
 		func_error("RemoveFunction", functionName, nargs, argList, NULL);
 	}
 
 	if ((((Form_pg_proc) GETSTRUCT(tup))->prolang) == INTERNALlanguageId)
 	{
-		heap_close(relation);
+		heap_close(relation, RowExclusiveLock);
 		elog(ERROR, "RemoveFunction: function \"%s\" is built-in", functionName);
 	}
 
 	heap_delete(relation, &tup->t_self, NULL);
 
-	heap_close(relation);
+	heap_close(relation, RowExclusiveLock);
 }
 
 void
@@ -398,7 +398,7 @@ RemoveAggregate(char *aggName, char *aggType)
 	}
 #endif
 
-	relation = heap_openr(AggregateRelationName);
+	relation = heap_openr(AggregateRelationName, RowExclusiveLock);
 	tup = SearchSysCacheTuple(AGGNAME,
 							  PointerGetDatum(aggName),
 							  ObjectIdGetDatum(basetypeID),
@@ -406,7 +406,7 @@ RemoveAggregate(char *aggName, char *aggType)
 
 	if (!HeapTupleIsValid(tup))
 	{
-		heap_close(relation);
+		heap_close(relation, RowExclusiveLock);
 		if (aggType)
 		{
 			elog(ERROR, "RemoveAggregate: aggregate '%s' for '%s' does not exist",
@@ -420,5 +420,5 @@ RemoveAggregate(char *aggName, char *aggType)
 	}
 	heap_delete(relation, &tup->t_self, NULL);
 
-	heap_close(relation);
+	heap_close(relation, RowExclusiveLock);
 }
diff --git a/src/backend/commands/rename.c b/src/backend/commands/rename.c
index 1a93027b4a3..3a822bd4e49 100644
--- a/src/backend/commands/rename.c
+++ b/src/backend/commands/rename.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/rename.c,v 1.32 1999/07/17 20:16:53 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/Attic/rename.c,v 1.33 1999/09/18 19:06:40 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -48,6 +48,7 @@ renameatt(char *relname,
 		  char *userName,
 		  int recurse)
 {
+	Relation	targetrelation;
 	Relation	attrelation;
 	HeapTuple	reltup,
 				oldatttup,
@@ -72,6 +73,14 @@ renameatt(char *relname,
 #endif
 
 	/*
+	 * Grab an exclusive lock on the target table, which we will NOT release
+	 * until end of transaction.
+	 */
+	targetrelation = heap_openr(relname, AccessExclusiveLock);
+	relid = RelationGetRelid(targetrelation);
+	heap_close(targetrelation, NoLock);	/* close rel but keep lock! */
+
+	/*
 	 * if the 'recurse' flag is set then we are supposed to rename this
 	 * attribute in all classes that inherit from 'relname' (as well as in
 	 * 'relname').
@@ -82,16 +91,11 @@ renameatt(char *relname,
 	 */
 	if (recurse)
 	{
-		Oid			myrelid,
-					childrelid;
 		List	   *child,
 				   *children;
 
-		if ((myrelid = RelnameFindRelid(relname)) == InvalidOid)
-			elog(ERROR, "renameatt: unknown relation: \"%s\"", relname);
-
 		/* this routine is actually in the planner */
-		children = find_all_inheritors(lconsi(myrelid, NIL), NIL);
+		children = find_all_inheritors(lconsi(relid, NIL), NIL);
 
 		/*
 		 * find_all_inheritors does the recursive search of the
@@ -100,10 +104,11 @@ renameatt(char *relname,
 		 */
 		foreach(child, children)
 		{
+			Oid			childrelid;
 			char		childname[NAMEDATALEN];
 
 			childrelid = lfirsti(child);
-			if (childrelid == myrelid)
+			if (childrelid == relid)
 				continue;
 			reltup = SearchSysCacheTuple(RELOID,
 										 ObjectIdGetDatum(childrelid),
@@ -117,14 +122,12 @@ renameatt(char *relname,
 			StrNCpy(childname,
 					((Form_pg_class) GETSTRUCT(reltup))->relname.data,
 					NAMEDATALEN);
-			/* no more recursion! */
+			/* note we need not recurse again! */
 			renameatt(childname, oldattname, newattname, userName, 0);
 		}
 	}
 
-
-	if ((relid = RelnameFindRelid(relname)) == InvalidOid)
-		elog(ERROR, "renameatt: relation \"%s\" nonexistent", relname);
+	attrelation = heap_openr(AttributeRelationName, RowExclusiveLock);
 
 	oldatttup = SearchSysCacheTupleCopy(ATTNAME,
 										ObjectIdGetDatum(relid),
@@ -150,7 +153,6 @@ renameatt(char *relname,
 	StrNCpy((((Form_pg_attribute) (GETSTRUCT(oldatttup)))->attname.data),
 			newattname, NAMEDATALEN);
 
-	attrelation = heap_openr(AttributeRelationName);
 	heap_replace(attrelation, &oldatttup->t_self, oldatttup, NULL);
 
 	/* keep system catalog indices current */
@@ -159,7 +161,7 @@ renameatt(char *relname,
 	CatalogCloseIndices(Num_pg_attr_indices, irelations);
 
 	pfree(oldatttup);
-	heap_close(attrelation);
+	heap_close(attrelation, RowExclusiveLock);
 }
 
 /*
@@ -182,6 +184,7 @@ void
 renamerel(char *oldrelname, char *newrelname)
 {
 	int			i;
+	Relation	targetrelation;
 	Relation	relrelation;	/* for RELATION relation */
 	HeapTuple	oldreltup;
 	char		oldpath[MAXPGPATH],
@@ -198,6 +201,15 @@ renamerel(char *oldrelname, char *newrelname)
 		elog(ERROR, "renamerel: Illegal class name: \"%s\" -- pg_ is reserved for system catalogs",
 			 newrelname);
 
+	/*
+	 * Grab an exclusive lock on the target table, which we will NOT release
+	 * until end of transaction.
+	 */
+	targetrelation = heap_openr(oldrelname, AccessExclusiveLock);
+	heap_close(targetrelation, NoLock);	/* close rel but keep lock! */
+
+	relrelation = heap_openr(RelationRelationName, RowExclusiveLock);
+
 	oldreltup = SearchSysCacheTupleCopy(RELNAME,
 										PointerGetDatum(oldrelname),
 										0, 0, 0);
@@ -207,12 +219,17 @@ renamerel(char *oldrelname, char *newrelname)
 	if (RelnameFindRelid(newrelname) != InvalidOid)
 		elog(ERROR, "renamerel: relation \"%s\" exists", newrelname);
 
+	/*
+	 * XXX need to close relation and flush dirty buffers here!
+	 */
+
 	/* rename the path first, so if this fails the rename's not done */
 	strcpy(oldpath, relpath(oldrelname));
 	strcpy(newpath, relpath(newrelname));
 	if (rename(oldpath, newpath) < 0)
 		elog(ERROR, "renamerel: unable to rename file: %s", oldpath);
 
+	/* rename additional segments of relation, too */
 	for (i = 1;; i++)
 	{
 		sprintf(toldpath, "%s.%d", oldpath, i);
@@ -225,7 +242,6 @@ renamerel(char *oldrelname, char *newrelname)
 			newrelname, NAMEDATALEN);
 
 	/* insert fixed rel tuple */
-	relrelation = heap_openr(RelationRelationName);
 	heap_replace(relrelation, &oldreltup->t_self, oldreltup, NULL);
 
 	/* keep the system catalog indices current */
@@ -233,5 +249,5 @@ renamerel(char *oldrelname, char *newrelname)
 	CatalogIndexInsert(irelations, Num_pg_class_indices, relrelation, oldreltup);
 	CatalogCloseIndices(Num_pg_class_indices, irelations);
 
-	heap_close(relrelation);
+	heap_close(relrelation, RowExclusiveLock);
 }
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index b7f959cd65e..be47d32f9f9 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -153,10 +153,7 @@ DefineSequence(CreateSeqStmt *seq)
 
 	DefineRelation(stmt, RELKIND_SEQUENCE);
 
-	rel = heap_openr(seq->seqname);
-	Assert(RelationIsValid(rel));
-
-	LockRelation(rel, AccessExclusiveLock);
+	rel = heap_openr(seq->seqname, AccessExclusiveLock);
 
 	tupDesc = RelationGetDescr(rel);
 
@@ -179,11 +176,7 @@ DefineSequence(CreateSeqStmt *seq)
 	if (WriteBuffer(buf) == STATUS_ERROR)
 		elog(ERROR, "DefineSequence: WriteBuffer failed");
 
-	UnlockRelation(rel, AccessExclusiveLock);
-	heap_close(rel);
-
-	return;
-
+	heap_close(rel, AccessExclusiveLock);
 }
 
 
@@ -422,12 +415,7 @@ init_sequence(char *caller, char *name)
 		temp = elm;
 	}
 
-	temp->rel = heap_openr(name);
-
-	if (!RelationIsValid(temp->rel))
-		elog(ERROR, "%s.%s: sequence does not exist", name, caller);
-
-	LockRelation(temp->rel, AccessShareLock);
+	temp->rel = heap_openr(name, AccessShareLock);
 
 	if (temp->rel->rd_rel->relkind != RELKIND_SEQUENCE)
 		elog(ERROR, "%s.%s: %s is not sequence !", name, caller, name);
@@ -453,7 +441,6 @@ init_sequence(char *caller, char *name)
 	}
 
 	return elm;
-
 }
 
 
@@ -467,20 +454,15 @@ CloseSequences(void)
 	SeqTable	elm;
 	Relation	rel;
 
-	for (elm = seqtab; elm != (SeqTable) NULL;)
+	for (elm = seqtab; elm != (SeqTable) NULL; elm = elm->next)
 	{
 		if (elm->rel != (Relation) NULL)		/* opened in current xact */
 		{
 			rel = elm->rel;
 			elm->rel = (Relation) NULL;
-			UnlockRelation(rel, AccessShareLock);
-			heap_close(rel);
+			heap_close(rel, AccessShareLock);
 		}
-		elm = elm->next;
 	}
-
-	return;
-
 }
 
 
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index 55f67711c77..aa7a0b56c1d 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -63,11 +63,7 @@ CreateTrigger(CreateTrigStmt *stmt)
 		elog(ERROR, "%s: %s", stmt->relname, aclcheck_error_strings[ACLCHECK_NOT_OWNER]);
 #endif
 
-	rel = heap_openr(stmt->relname);
-	if (!RelationIsValid(rel))
-		elog(ERROR, "CreateTrigger: there is no relation %s", stmt->relname);
-
-	LockRelation(rel, AccessExclusiveLock);
+	rel = heap_openr(stmt->relname, AccessExclusiveLock);
 
 	TRIGGER_CLEAR_TYPE(tgtype);
 	if (stmt->before)
@@ -103,8 +99,7 @@ CreateTrigger(CreateTrigStmt *stmt)
 	}
 
 	/* Scan pg_trigger */
-	tgrel = heap_openr(TriggerRelationName);
-	LockRelation(tgrel, AccessExclusiveLock);
+	tgrel = heap_openr(TriggerRelationName, RowExclusiveLock);
 	ScanKeyEntryInitialize(&key, 0, Anum_pg_trigger_tgrelid,
 						   F_OIDEQ, RelationGetRelid(rel));
 	tgscan = heap_beginscan(tgrel, 0, SnapshotNow, 1, &key);
@@ -203,20 +198,19 @@ CreateTrigger(CreateTrigStmt *stmt)
 	CatalogIndexInsert(idescs, Num_pg_trigger_indices, tgrel, tuple);
 	CatalogCloseIndices(Num_pg_trigger_indices, idescs);
 	pfree(tuple);
-	UnlockRelation(tgrel, AccessExclusiveLock);
-	heap_close(tgrel);
+	heap_close(tgrel, RowExclusiveLock);
 
 	pfree(DatumGetPointer(values[Anum_pg_trigger_tgname - 1]));
 	pfree(DatumGetPointer(values[Anum_pg_trigger_tgargs - 1]));
 
 	/* update pg_class */
+	pgrel = heap_openr(RelationRelationName, RowExclusiveLock);
 	tuple = SearchSysCacheTupleCopy(RELNAME,
 									PointerGetDatum(stmt->relname),
 									0, 0, 0);
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "CreateTrigger: relation %s not found in pg_class", stmt->relname);
 
-	pgrel = heap_openr(RelationRelationName);
 	((Form_pg_class) GETSTRUCT(tuple))->reltriggers = found + 1;
 	RelationInvalidateHeapTuple(pgrel, tuple);
 	heap_replace(pgrel, &tuple->t_self, tuple, NULL);
@@ -224,7 +218,7 @@ CreateTrigger(CreateTrigStmt *stmt)
 	CatalogIndexInsert(ridescs, Num_pg_class_indices, pgrel, tuple);
 	CatalogCloseIndices(Num_pg_class_indices, ridescs);
 	pfree(tuple);
-	heap_close(pgrel);
+	heap_close(pgrel, RowExclusiveLock);
 
 	CommandCounterIncrement();
 	oldcxt = MemoryContextSwitchTo((MemoryContext) CacheCxt);
@@ -232,8 +226,8 @@ CreateTrigger(CreateTrigStmt *stmt)
 	rel->rd_rel->reltriggers = found + 1;
 	RelationBuildTriggers(rel);
 	MemoryContextSwitchTo(oldcxt);
-	heap_close(rel);
-	return;
+	/* Keep lock on target rel until end of xact */
+	heap_close(rel, NoLock);
 }
 
 void
@@ -255,14 +249,9 @@ DropTrigger(DropTrigStmt *stmt)
 		elog(ERROR, "%s: %s", stmt->relname, aclcheck_error_strings[ACLCHECK_NOT_OWNER]);
 #endif
 
-	rel = heap_openr(stmt->relname);
-	if (!RelationIsValid(rel))
-		elog(ERROR, "DropTrigger: there is no relation %s", stmt->relname);
+	rel = heap_openr(stmt->relname, AccessExclusiveLock);
 
-	LockRelation(rel, AccessExclusiveLock);
-
-	tgrel = heap_openr(TriggerRelationName);
-	LockRelation(tgrel, AccessExclusiveLock);
+	tgrel = heap_openr(TriggerRelationName, RowExclusiveLock);
 	ScanKeyEntryInitialize(&key, 0, Anum_pg_trigger_tgrelid,
 						   F_OIDEQ, RelationGetRelid(rel));
 	tgscan = heap_beginscan(tgrel, 0, SnapshotNow, 1, &key);
@@ -282,20 +271,19 @@ DropTrigger(DropTrigStmt *stmt)
 		elog(ERROR, "DropTrigger: there is no trigger %s on relation %s",
 			 stmt->trigname, stmt->relname);
 	if (tgfound > 1)
-		elog(NOTICE, "DropTrigger: found (and deleted) %d trigger %s on relation %s",
+		elog(NOTICE, "DropTrigger: found (and deleted) %d triggers %s on relation %s",
 			 tgfound, stmt->trigname, stmt->relname);
 	heap_endscan(tgscan);
-	UnlockRelation(tgrel, AccessExclusiveLock);
-	heap_close(tgrel);
+	heap_close(tgrel, RowExclusiveLock);
 
+	/* update pg_class */
+	pgrel = heap_openr(RelationRelationName, RowExclusiveLock);
 	tuple = SearchSysCacheTupleCopy(RELNAME,
 									PointerGetDatum(stmt->relname),
 									0, 0, 0);
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "DropTrigger: relation %s not found in pg_class", stmt->relname);
 
-	/* update pg_class */
-	pgrel = heap_openr(RelationRelationName);
 	((Form_pg_class) GETSTRUCT(tuple))->reltriggers = found;
 	RelationInvalidateHeapTuple(pgrel, tuple);
 	heap_replace(pgrel, &tuple->t_self, tuple, NULL);
@@ -303,7 +291,7 @@ DropTrigger(DropTrigStmt *stmt)
 	CatalogIndexInsert(ridescs, Num_pg_class_indices, pgrel, tuple);
 	CatalogCloseIndices(Num_pg_class_indices, ridescs);
 	pfree(tuple);
-	heap_close(pgrel);
+	heap_close(pgrel, RowExclusiveLock);
 
 	CommandCounterIncrement();
 	oldcxt = MemoryContextSwitchTo((MemoryContext) CacheCxt);
@@ -312,8 +300,8 @@ DropTrigger(DropTrigStmt *stmt)
 	if (found > 0)
 		RelationBuildTriggers(rel);
 	MemoryContextSwitchTo(oldcxt);
-	heap_close(rel);
-	return;
+	/* Keep lock on target rel until end of xact */
+	heap_close(rel, NoLock);
 }
 
 void
@@ -324,8 +312,7 @@ RelationRemoveTriggers(Relation rel)
 	ScanKeyData key;
 	HeapTuple	tup;
 
-	tgrel = heap_openr(TriggerRelationName);
-	LockRelation(tgrel, AccessExclusiveLock);
+	tgrel = heap_openr(TriggerRelationName, RowExclusiveLock);
 	ScanKeyEntryInitialize(&key, 0, Anum_pg_trigger_tgrelid,
 						   F_OIDEQ, RelationGetRelid(rel));
 
@@ -335,9 +322,7 @@ RelationRemoveTriggers(Relation rel)
 		heap_delete(tgrel, &tup->t_self, NULL);
 
 	heap_endscan(tgscan);
-	UnlockRelation(tgrel, AccessExclusiveLock);
-	heap_close(tgrel);
-
+	heap_close(tgrel, RowExclusiveLock);
 }
 
 void
@@ -367,7 +352,7 @@ RelationBuildTriggers(Relation relation)
 						   (RegProcedure) F_OIDEQ,
 						   ObjectIdGetDatum(RelationGetRelid(relation)));
 
-	tgrel = heap_openr(TriggerRelationName);
+	tgrel = heap_openr(TriggerRelationName, AccessShareLock);
 	irel = index_openr(TriggerRelidIndex);
 	sd = index_beginscan(irel, false, 1, &skey);
 
@@ -441,7 +426,7 @@ RelationBuildTriggers(Relation relation)
 	index_endscan(sd);
 	pfree(sd);
 	index_close(irel);
-	heap_close(tgrel);
+	heap_close(tgrel, AccessShareLock);
 
 	/* Build trigdesc */
 	trigdesc->triggers = triggers;
diff --git a/src/backend/commands/user.c b/src/backend/commands/user.c
index 1b0d972839e..c05b6da5856 100644
--- a/src/backend/commands/user.c
+++ b/src/backend/commands/user.c
@@ -5,7 +5,7 @@
  *
  * Copyright (c) 1994, Regents of the University of California
  *
- * $Id: user.c,v 1.33 1999/07/30 18:09:47 momjian Exp $
+ * $Id: user.c,v 1.34 1999/09/18 19:06:41 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -36,13 +36,15 @@ static void CheckPgUserAclNotNull(void);
  *
  * copy the modified contents of pg_shadow to a file used by the postmaster
  * for user authentication.  The file is stored as $PGDATA/pg_pwd.
+ *
+ * NB: caller is responsible for ensuring that only one backend can
+ * execute this routine at a time.  Acquiring AccessExclusiveLock on
+ * pg_shadow is the standard way to do that.
  *---------------------------------------------------------------------
  */
-static
-void
+static void
 UpdatePgPwdFile(char *sql, CommandDest dest)
 {
-
 	char	   *filename,
 			   *tempname;
 	int			bufsize;
@@ -125,17 +127,13 @@ DefineUser(CreateUserStmt *stmt, CommandDest dest)
 
 	/*
 	 * Scan the pg_shadow relation to be certain the user doesn't already
-	 * exist.
+	 * exist.  Note we secure exclusive lock, because we also need to be
+	 * sure of what the next usesysid should be, and we need to protect
+	 * our update of the flat password file.
 	 */
-	pg_shadow_rel = heap_openr(ShadowRelationName);
+	pg_shadow_rel = heap_openr(ShadowRelationName, AccessExclusiveLock);
 	pg_shadow_dsc = RelationGetDescr(pg_shadow_rel);
 
-	/*
-	 * Secure a write lock on pg_shadow so we can be sure of what the next
-	 * usesysid should be.
-	 */
-	LockRelation(pg_shadow_rel, AccessExclusiveLock);
-
 	scan = heap_beginscan(pg_shadow_rel, false, SnapshotNow, 0, NULL);
 	while (HeapTupleIsValid(tuple = heap_getnext(scan, 0)))
 	{
@@ -152,8 +150,7 @@ DefineUser(CreateUserStmt *stmt, CommandDest dest)
 
 	if (exists)
 	{
-		UnlockRelation(pg_shadow_rel, AccessExclusiveLock);
-		heap_close(pg_shadow_rel);
+		heap_close(pg_shadow_rel, AccessExclusiveLock);
 		UserAbortTransactionBlock();
 		elog(ERROR,
 		 "defineUser: user \"%s\" has already been created", stmt->user);
@@ -165,6 +162,12 @@ DefineUser(CreateUserStmt *stmt, CommandDest dest)
 	 *
 	 * XXX Ugly as this code is, it still fails to cope with ' or \ in any of
 	 * the provided strings.
+	 *
+	 * XXX This routine would be *lots* better if it inserted the new
+	 * tuple with formtuple/heap_insert.  For one thing, all of the
+	 * transaction-block gamesmanship could be eliminated, because
+	 * it's only there to make the world safe for a recursive call
+	 * to pg_exec_query_dest().
 	 */
 	snprintf(sql, SQL_LENGTH,
 			 "insert into %s (usename,usesysid,usecreatedb,usetrace,"
@@ -189,17 +192,21 @@ DefineUser(CreateUserStmt *stmt, CommandDest dest)
 	pg_exec_query_dest(sql, dest, false);
 
 	/*
-	 * Add the stuff here for groups.
+	 * Add stuff here for groups?
 	 */
 
+	/*
+	 * Write the updated pg_shadow data to the flat password file.
+	 * Because we are still holding AccessExclusiveLock on pg_shadow,
+	 * we can be sure no other backend will try to write the flat
+	 * file at the same time.
+	 */
 	UpdatePgPwdFile(sql, dest);
 
 	/*
-	 * This goes after the UpdatePgPwdFile to be certain that two backends
-	 * to not attempt to write to the pg_pwd file at the same time.
+	 * Now we can clean up.
 	 */
-	UnlockRelation(pg_shadow_rel, AccessExclusiveLock);
-	heap_close(pg_shadow_rel);
+	heap_close(pg_shadow_rel, AccessExclusiveLock);
 
 	if (IsTransactionBlock() && !inblock)
 		EndTransactionBlock();
@@ -237,70 +244,79 @@ AlterUser(AlterUserStmt *stmt, CommandDest dest)
 
 	/*
 	 * Scan the pg_shadow relation to be certain the user exists.
+	 * Note we secure exclusive lock to protect our update of the
+	 * flat password file.
 	 */
-	pg_shadow_rel = heap_openr(ShadowRelationName);
+	pg_shadow_rel = heap_openr(ShadowRelationName, AccessExclusiveLock);
 	pg_shadow_dsc = RelationGetDescr(pg_shadow_rel);
 
-	/*
-	 * Secure a write lock on pg_shadow so we can be sure that when the
-	 * dump of the pg_pwd file is done, there is not another backend doing
-	 * the same.
-	 */
-	LockRelation(pg_shadow_rel, AccessExclusiveLock);
-
 	tuple = SearchSysCacheTuple(USENAME,
 								PointerGetDatum(stmt->user),
 								0, 0, 0);
 	if (!HeapTupleIsValid(tuple))
 	{
-		UnlockRelation(pg_shadow_rel, AccessExclusiveLock);
-		heap_close(pg_shadow_rel);
-		UserAbortTransactionBlock();	/* needed? */
+		heap_close(pg_shadow_rel, AccessExclusiveLock);
+		UserAbortTransactionBlock();
 		elog(ERROR, "alterUser: user \"%s\" does not exist", stmt->user);
-		return;
 	}
 
 	/*
 	 * Create the update statement to modify the user.
+	 *
+	 * XXX see diatribe in preceding routine.  This code is just as bogus.
 	 */
 	snprintf(sql, SQL_LENGTH, "update %s set", ShadowRelationName);
 
 	if (stmt->password)
-		snprintf(sql, SQL_LENGTH, "%s passwd = '%s'", pstrdup(sql), stmt->password);
+		snprintf(sql + strlen(sql), SQL_LENGTH - strlen(sql),
+				 " passwd = '%s'", stmt->password);
 
 	if (stmt->createdb)
 	{
-		snprintf(sql, SQL_LENGTH, "%s %susecreatedb='%s'",
-				 pstrdup(sql), stmt->password ? "," : "",
+		snprintf(sql + strlen(sql), SQL_LENGTH - strlen(sql),
+				 "%s usecreatedb='%s'",
+				 stmt->password ? "," : "",
 				 *stmt->createdb ? "t" : "f");
 	}
 
 	if (stmt->createuser)
 	{
-		snprintf(sql, SQL_LENGTH, "%s %susesuper='%s'",
-			 pstrdup(sql), (stmt->password || stmt->createdb) ? "," : "",
+		snprintf(sql + strlen(sql), SQL_LENGTH - strlen(sql),
+				 "%s usesuper='%s'",
+				 (stmt->password || stmt->createdb) ? "," : "",
 				 *stmt->createuser ? "t" : "f");
 	}
 
 	if (stmt->validUntil)
 	{
-		snprintf(sql, SQL_LENGTH, "%s %svaluntil='%s'",
-				 pstrdup(sql),
+		snprintf(sql + strlen(sql), SQL_LENGTH - strlen(sql),
+				 "%s valuntil='%s'",
 		(stmt->password || stmt->createdb || stmt->createuser) ? "," : "",
 				 stmt->validUntil);
 	}
 
-	snprintf(sql, SQL_LENGTH, "%s where usename = '%s'",
-			 pstrdup(sql), stmt->user);
+	snprintf(sql + strlen(sql), SQL_LENGTH - strlen(sql),
+			 " where usename = '%s'",
+			 stmt->user);
 
 	pg_exec_query_dest(sql, dest, false);
 
-	/* do the pg_group stuff here */
+	/*
+	 * Add stuff here for groups?
+	 */
 
+	/*
+	 * Write the updated pg_shadow data to the flat password file.
+	 * Because we are still holding AccessExclusiveLock on pg_shadow,
+	 * we can be sure no other backend will try to write the flat
+	 * file at the same time.
+	 */
 	UpdatePgPwdFile(sql, dest);
 
-	UnlockRelation(pg_shadow_rel, AccessExclusiveLock);
-	heap_close(pg_shadow_rel);
+	/*
+	 * Now we can clean up.
+	 */
+	heap_close(pg_shadow_rel, AccessExclusiveLock);
 
 	if (IsTransactionBlock() && !inblock)
 		EndTransactionBlock();
@@ -310,7 +326,6 @@ AlterUser(AlterUserStmt *stmt, CommandDest dest)
 extern void
 RemoveUser(char *user, CommandDest dest)
 {
-
 	char	   *pg_shadow;
 	Relation	pg_shadow_rel,
 				pg_rel;
@@ -318,7 +333,7 @@ RemoveUser(char *user, CommandDest dest)
 	HeapScanDesc scan;
 	HeapTuple	tuple;
 	Datum		datum;
-	char		sql[512];
+	char		sql[SQL_LENGTH];
 	bool		n,
 				inblock;
 	int32		usesysid;
@@ -341,27 +356,19 @@ RemoveUser(char *user, CommandDest dest)
 	}
 
 	/*
-	 * Perform a scan of the pg_shadow relation to find the usesysid of
-	 * the user to be deleted.	If it is not found, then return a warning
-	 * message.
+	 * Scan the pg_shadow relation to find the usesysid of the user to be
+	 * deleted.  Note we secure exclusive lock, because we need to protect
+	 * our update of the flat password file.
 	 */
-	pg_shadow_rel = heap_openr(ShadowRelationName);
+	pg_shadow_rel = heap_openr(ShadowRelationName, AccessExclusiveLock);
 	pg_dsc = RelationGetDescr(pg_shadow_rel);
 
-	/*
-	 * Secure a write lock on pg_shadow so we can be sure that when the
-	 * dump of the pg_pwd file is done, there is not another backend doing
-	 * the same.
-	 */
-	LockRelation(pg_shadow_rel, AccessExclusiveLock);
-
 	tuple = SearchSysCacheTuple(USENAME,
 								PointerGetDatum(user),
 								0, 0, 0);
 	if (!HeapTupleIsValid(tuple))
 	{
-		UnlockRelation(pg_shadow_rel, AccessExclusiveLock);
-		heap_close(pg_shadow_rel);
+		heap_close(pg_shadow_rel, AccessExclusiveLock);
 		UserAbortTransactionBlock();
 		elog(ERROR, "removeUser: user \"%s\" does not exist", user);
 	}
@@ -372,7 +379,7 @@ RemoveUser(char *user, CommandDest dest)
 	 * Perform a scan of the pg_database relation to find the databases
 	 * owned by usesysid.  Then drop them.
 	 */
-	pg_rel = heap_openr(DatabaseRelationName);
+	pg_rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 	pg_dsc = RelationGetDescr(pg_rel);
 
 	scan = heap_beginscan(pg_rel, false, SnapshotNow, 0, NULL);
@@ -383,7 +390,7 @@ RemoveUser(char *user, CommandDest dest)
 		if ((int) datum == usesysid)
 		{
 			datum = heap_getattr(tuple, Anum_pg_database_datname, pg_dsc, &n);
-			if (memcmp((void *) datum, "template1", 9))
+			if (memcmp((void *) datum, "template1", 9) != 0)
 			{
 				dbase =
 					(char **) repalloc((void *) dbase, sizeof(char *) * (ndbase + 1));
@@ -394,12 +401,12 @@ RemoveUser(char *user, CommandDest dest)
 		}
 	}
 	heap_endscan(scan);
-	heap_close(pg_rel);
+	heap_close(pg_rel, AccessExclusiveLock);
 
 	while (ndbase--)
 	{
 		elog(NOTICE, "Dropping database %s", dbase[ndbase]);
-		snprintf(sql, SQL_LENGTH, "drop database %s", dbase[ndbase]);
+		snprintf(sql, SQL_LENGTH, "DROP DATABASE %s", dbase[ndbase]);
 		pfree((void *) dbase[ndbase]);
 		pg_exec_query_dest(sql, dest, false);
 	}
@@ -431,10 +438,18 @@ RemoveUser(char *user, CommandDest dest)
 		"delete from %s where usename = '%s'", ShadowRelationName, user);
 	pg_exec_query_dest(sql, dest, false);
 
+	/*
+	 * Write the updated pg_shadow data to the flat password file.
+	 * Because we are still holding AccessExclusiveLock on pg_shadow,
+	 * we can be sure no other backend will try to write the flat
+	 * file at the same time.
+	 */
 	UpdatePgPwdFile(sql, dest);
 
-	UnlockRelation(pg_shadow_rel, AccessExclusiveLock);
-	heap_close(pg_shadow_rel);
+	/*
+	 * Now we can clean up.
+	 */
+	heap_close(pg_shadow_rel, AccessExclusiveLock);
 
 	if (IsTransactionBlock() && !inblock)
 		EndTransactionBlock();
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 55ec864fca5..3027763b468 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -7,7 +7,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.119 1999/08/25 12:20:57 ishii Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.120 1999/09/18 19:06:41 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -304,7 +304,7 @@ vc_getrels(NameData *VacRelP)
 	portalmem = PortalGetVariableMemory(vc_portal);
 	vrl = cur = (VRelList) NULL;
 
-	rel = heap_openr(RelationRelationName);
+	rel = heap_openr(RelationRelationName, AccessShareLock);
 	tupdesc = RelationGetDescr(rel);
 
 	scan = heap_beginscan(rel, false, SnapshotNow, 1, &key);
@@ -343,9 +343,8 @@ vc_getrels(NameData *VacRelP)
 	if (found == false)
 		elog(NOTICE, "Vacuum: table not found");
 
-
 	heap_endscan(scan);
-	heap_close(rel);
+	heap_close(rel, AccessShareLock);
 
 	CommitTransactionCommand();
 
@@ -395,8 +394,10 @@ vc_vacone(Oid relid, bool analyze, List *va_cols)
 		return;
 	}
 
-	/* now open the class and vacuum it */
-	onerel = heap_open(relid);
+	/*
+	 * Open the class, get an exclusive lock on it, and vacuum it
+	 */
+	onerel = heap_open(relid, AccessExclusiveLock);
 
 	vacrelstats = (VRelStats *) palloc(sizeof(VRelStats));
 	vacrelstats->relid = relid;
@@ -509,9 +510,6 @@ vc_vacone(Oid relid, bool analyze, List *va_cols)
 		vacrelstats->vacattrstats = (VacAttrStats *) NULL;
 	}
 
-	/* we require the relation to be locked until the indices are cleaned */
-	LockRelation(onerel, AccessExclusiveLock);
-
 	GetXmaxRecent(&XmaxRecent);
 
 	/* scan it */
@@ -565,13 +563,13 @@ vc_vacone(Oid relid, bool analyze, List *va_cols)
 			pfree(fraged_pages.vpl_pagedesc);
 	}
 
-	/* all done with this class */
-	heap_close(onerel);
-
 	/* update statistics in pg_class */
 	vc_updstats(vacrelstats->relid, vacrelstats->num_pages,
 			vacrelstats->num_tuples, vacrelstats->hasindex, vacrelstats);
 
+	/* all done with this class, but hold lock until commit */
+	heap_close(onerel, NoLock);
+
 	/* next command frees attribute stats */
 	CommitTransactionCommand();
 }
@@ -2281,6 +2279,8 @@ vc_updstats(Oid relid, int num_pages, int num_tuples, bool hasindex, VRelStats *
 	/*
 	 * update number of tuples and number of pages in pg_class
 	 */
+	rd = heap_openr(RelationRelationName, RowExclusiveLock);
+
 	ctup = SearchSysCacheTupleCopy(RELOID,
 								   ObjectIdGetDatum(relid),
 								   0, 0, 0);
@@ -2288,8 +2288,6 @@ vc_updstats(Oid relid, int num_pages, int num_tuples, bool hasindex, VRelStats *
 		elog(ERROR, "pg_class entry for relid %u vanished during vacuuming",
 			 relid);
 
-	rd = heap_openr(RelationRelationName);
-
 	/* get the buffer cache tuple */
 	rtup.t_self = ctup->t_self;
 	heap_fetch(rd, SnapshotNow, &rtup, &buffer);
@@ -2306,8 +2304,8 @@ vc_updstats(Oid relid, int num_pages, int num_tuples, bool hasindex, VRelStats *
 		VacAttrStats *vacattrstats = vacrelstats->vacattrstats;
 		int			natts = vacrelstats->va_natts;
 
-		ad = heap_openr(AttributeRelationName);
-		sd = heap_openr(StatisticRelationName);
+		ad = heap_openr(AttributeRelationName, RowExclusiveLock);
+		sd = heap_openr(StatisticRelationName, RowExclusiveLock);
 		ScanKeyEntryInitialize(&askey, 0, Anum_pg_attribute_attrelid,
 							   F_INT4EQ, relid);
 
@@ -2458,8 +2456,8 @@ vc_updstats(Oid relid, int num_pages, int num_tuples, bool hasindex, VRelStats *
 			}
 		}
 		heap_endscan(scan);
-		heap_close(ad);
-		heap_close(sd);
+		heap_close(ad, RowExclusiveLock);
+		heap_close(sd, RowExclusiveLock);
 	}
 
 	/*
@@ -2469,7 +2467,7 @@ vc_updstats(Oid relid, int num_pages, int num_tuples, bool hasindex, VRelStats *
 
 	WriteBuffer(buffer);
 
-	heap_close(rd);
+	heap_close(rd, RowExclusiveLock);
 }
 
 /*
@@ -2484,7 +2482,7 @@ vc_delhilowstats(Oid relid, int attcnt, int *attnums)
 	HeapTuple	tuple;
 	ScanKeyData key;
 
-	pgstatistic = heap_openr(StatisticRelationName);
+	pgstatistic = heap_openr(StatisticRelationName, RowExclusiveLock);
 
 	if (relid != InvalidOid)
 	{
@@ -2515,7 +2513,7 @@ vc_delhilowstats(Oid relid, int attcnt, int *attnums)
 	}
 
 	heap_endscan(scan);
-	heap_close(pgstatistic);
+	heap_close(pgstatistic, RowExclusiveLock);
 }
 
 /*
@@ -2721,7 +2719,7 @@ vc_getindices(Oid relid, int *nindices, Relation **Irel)
 	ioid = (Oid *) palloc(10 * sizeof(Oid));
 
 	/* prepare a heap scan on the pg_index relation */
-	pgindex = heap_openr(IndexRelationName);
+	pgindex = heap_openr(IndexRelationName, AccessShareLock);
 	tupdesc = RelationGetDescr(pgindex);
 
 	ScanKeyEntryInitialize(&key, 0x0, Anum_pg_index_indrelid,
@@ -2741,7 +2739,7 @@ vc_getindices(Oid relid, int *nindices, Relation **Irel)
 	}
 
 	heap_endscan(scan);
-	heap_close(pgindex);
+	heap_close(pgindex, AccessShareLock);
 
 	if (i == 0)
 	{							/* No one index found */