- Modifies LOCKTAG to include a 'classId'. Relation receive a classId of

RelOid_pg_class, and transaction locks XactLockTableId. RelId is renamed
to objId.

- LockObject() and UnlockObject() functions created, and their use
sprinkled throughout the code to do descent locking for domains and
types. They accept lock modes AccessShare and AccessExclusive, as we
only really need a 'read' and 'write' lock at the moment.  Most locking
cases are held until the end of the transaction.

This fixes the cases Tom mentioned earlier in regards to locking with
Domains.  If the patch is good, I'll work on cleaning up issues with
other database objects that have this problem (most of them).

Rod Taylor

1 files changed, 64 insertions, 8 deletions

diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index ad954605373..7ae852672c5 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/typecmds.c,v 1.30 2003/02/03 21:15:43 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/typecmds.c,v 1.31 2003/02/19 04:02:53 momjian Exp $
  *
  * DESCRIPTION
  *	  The "DefineFoo" routines take the parse tree and pick out the
@@ -353,6 +353,12 @@ RemoveType(List *names, DropBehavior behavior)
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Grab an exclusive lock on the type id, the SearchSysCache confirms
+	 * the type still exists after locking
+	 */
+	LockObject(typeoid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCache(TYPEOID,
 						 ObjectIdGetDatum(typeoid),
 						 0, 0, 0);
@@ -376,6 +382,9 @@ RemoveType(List *names, DropBehavior behavior)
 	object.objectSubId = 0;
 
 	performDeletion(&object, behavior);
+
+	/* Hold the lock until the end of the transaction */
+	UnlockObject(typeoid, RelOid_pg_type, NoLock);
 }
 
 
@@ -680,7 +689,7 @@ void
 RemoveDomain(List *names, DropBehavior behavior)
 {
 	TypeName   *typename;
-	Oid			typeoid;
+	Oid			domainoid;
 	HeapTuple	tup;
 	char		typtype;
 	ObjectAddress object;
@@ -692,20 +701,26 @@ RemoveDomain(List *names, DropBehavior behavior)
 	typename->arrayBounds = NIL;
 
 	/* Use LookupTypeName here so that shell types can be removed. */
-	typeoid = LookupTypeName(typename);
-	if (!OidIsValid(typeoid))
+	domainoid = LookupTypeName(typename);
+	if (!OidIsValid(domainoid))
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Lock the domain.  The SearchSysCache confirms the domain still exists
+	 * after locking
+	 */
+	LockObject(domainoid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCache(TYPEOID,
-						 ObjectIdGetDatum(typeoid),
+						 ObjectIdGetDatum(domainoid),
 						 0, 0, 0);
 	if (!HeapTupleIsValid(tup))
 		elog(ERROR, "RemoveDomain: type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
 	/* Permission check: must own type or its namespace */
-	if (!pg_type_ownercheck(typeoid, GetUserId()) &&
+	if (!pg_type_ownercheck(domainoid, GetUserId()) &&
 		!pg_namespace_ownercheck(((Form_pg_type) GETSTRUCT(tup))->typnamespace,
 								 GetUserId()))
 		aclcheck_error(ACLCHECK_NOT_OWNER, TypeNameToString(typename));
@@ -723,10 +738,13 @@ RemoveDomain(List *names, DropBehavior behavior)
 	 * Do the deletion
 	 */
 	object.classId = RelOid_pg_type;
-	object.objectId = typeoid;
+	object.objectId = domainoid;
 	object.objectSubId = 0;
 
 	performDeletion(&object, behavior);
+
+	/* Hold the lock until the end of the transaction */
+	UnlockObject(domainoid, RelOid_pg_type, NoLock);
 }
 
 
@@ -941,6 +959,12 @@ AlterDomainDefault(List *names, Node *defaultRaw)
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Lock the domain.  The SearchSysCacheCopy confirms the type
+	 * still exists after locking
+	 */
+	LockObject(domainoid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCacheCopy(TYPEOID,
 							 ObjectIdGetDatum(domainoid),
 							 0, 0, 0);
@@ -1025,6 +1049,7 @@ AlterDomainDefault(List *names, Node *defaultRaw)
 	/* Clean up */
 	heap_close(rel, NoLock);
 	heap_freetuple(newtuple);
+	UnlockObject(domainoid, RelOid_pg_type, NoLock);
 };
 
 /*
@@ -1056,6 +1081,12 @@ AlterDomainNotNull(List *names, bool notNull)
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Lock the domain.  The SearchSysCacheCopy confirms the domain
+	 * still exists after locking
+	 */
+	LockObject(domainoid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCacheCopy(TYPEOID,
 							 ObjectIdGetDatum(domainoid),
 							 0, 0, 0);
@@ -1137,6 +1168,7 @@ AlterDomainNotNull(List *names, bool notNull)
 	/* Clean up */
 	heap_freetuple(tup);
 	heap_close(typrel, RowExclusiveLock);
+	UnlockObject(domainoid, RelOid_pg_type, NoLock);
 }
 
 /*
@@ -1172,6 +1204,12 @@ AlterDomainDropConstraint(List *names, const char *constrName, DropBehavior beha
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Lock the domain.  The SearchSysCacheCopy confirms the type still
+	 * exists after locking.
+	 */
+	LockObject(domainoid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCacheCopy(TYPEOID,
 							 ObjectIdGetDatum(domainoid),
 							 0, 0, 0);
@@ -1219,6 +1257,7 @@ AlterDomainDropConstraint(List *names, const char *constrName, DropBehavior beha
 	heap_close(conrel, RowExclusiveLock);
 
 	heap_close(rel, NoLock);
+	UnlockObject(domainoid, RelOid_pg_type, NoLock);
 };
 
 /*
@@ -1259,6 +1298,12 @@ AlterDomainAddConstraint(List *names, Node *newConstraint)
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Lock the domain.  The SearchSysCacheCopy confirms the domain
+	 * still exists after locking.
+	 */
+	LockObject(domainoid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCacheCopy(TYPEOID,
 							 ObjectIdGetDatum(domainoid),
 							 0, 0, 0);
@@ -1393,6 +1438,7 @@ AlterDomainAddConstraint(List *names, Node *newConstraint)
 
 	/* Clean up */
 	heap_close(typrel, RowExclusiveLock);
+	UnlockObject(domainoid, RelOid_pg_type, NoLock);
 }
 
 /*
@@ -1696,7 +1742,10 @@ GetDomainConstraints(Oid typeOid)
 		Form_pg_type typTup;
 		ScanKeyData key[1];
 		SysScanDesc scan;
-		
+
+		/* Lock the domain */
+		LockObject(typeOid, RelOid_pg_type, AccessShareLock);
+
 		tup = SearchSysCache(TYPEOID,
 							 ObjectIdGetDatum(typeOid),
 							 0, 0, 0);
@@ -1824,6 +1873,12 @@ AlterTypeOwner(List *names, AclId newOwnerSysId)
 		elog(ERROR, "Type \"%s\" does not exist",
 			 TypeNameToString(typename));
 
+	/*
+	 * Lock the type.  The SearchSysCacheCopy serves to confirm the
+	 * domain still exists after locking
+	 */
+	LockObject(typeOid, RelOid_pg_type, AccessExclusiveLock);
+
 	tup = SearchSysCacheCopy(TYPEOID,
 							 ObjectIdGetDatum(typeOid),
 							 0, 0, 0);
@@ -1846,4 +1901,5 @@ AlterTypeOwner(List *names, AclId newOwnerSysId)
 
 	/* Clean up */
 	heap_close(rel, RowExclusiveLock);
+	UnlockObject(typeOid, RelOid_pg_type, NoLock);
 }