- 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

6 files changed, 207 insertions, 66 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);
 }
diff --git a/src/backend/parser/parse_type.c b/src/backend/parser/parse_type.c
index 8f7c34cbd3d..56379a28119 100644
--- a/src/backend/parser/parse_type.c
+++ b/src/backend/parser/parse_type.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_type.c,v 1.51 2003/02/09 06:56:28 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_type.c,v 1.52 2003/02/19 04:02:53 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -23,6 +23,7 @@
 #include "parser/parser.h"
 #include "parser/parse_expr.h"
 #include "parser/parse_type.h"
+#include "storage/lmgr.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
@@ -127,6 +128,15 @@ LookupTypeName(const TypeName *typename)
 		}
 	}
 
+	/*
+	 * Lock the type as having been read for remainder of the transaction
+	 *
+	 * XXX: There is a small time between the above and now when the type
+	 *		could dissapear.  We *should* recheck to confirm the type still
+	 *		exists, but won't for speed.
+	 */
+	LockObject(restype, RelOid_pg_type, AccessShareLock);
+
 	return restype;
 }
 
diff --git a/src/backend/storage/lmgr/deadlock.c b/src/backend/storage/lmgr/deadlock.c
index 61ab1962fb6..9191d40b1e7 100644
--- a/src/backend/storage/lmgr/deadlock.c
+++ b/src/backend/storage/lmgr/deadlock.c
@@ -12,7 +12,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/deadlock.c,v 1.17 2003/02/18 02:13:24 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/deadlock.c,v 1.18 2003/02/19 04:02:53 momjian Exp $
  *
  *	Interface:
  *
@@ -855,22 +855,25 @@ DeadLockReport(void)
 		else
 			nextpid = deadlockDetails[0].pid;
 
-		if (info->locktag.relId == XactLockTableId && info->locktag.dbId == 0)
+		if (info->locktag.objId == InvalidOid
+			&& info->locktag.classId == XactLockTableId
+			&& info->locktag.dbId == InvalidOid)
 		{
 			/* Lock is for transaction ID */
 			elog(NOTICE, "Proc %d waits for %s on transaction %u; blocked by %d",
 				 info->pid,
 				 GetLockmodeName(info->lockmode),
-				 info->locktag.objId.xid,
+				 info->locktag.objsubId.xid,
 				 nextpid);
 		}
 		else
 		{
 			/* Lock is for a relation */
-			elog(NOTICE, "Proc %d waits for %s on relation %u database %u; blocked by %d",
+			elog(NOTICE, "Proc %d waits for %s on object %u class %u database %u; blocked by %d",
 				 info->pid,
 				 GetLockmodeName(info->lockmode),
-				 info->locktag.relId,
+				 info->locktag.objId,
+				 info->locktag.classId,
 				 info->locktag.dbId,
 				 nextpid);
 		}
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index a4e9197edc7..a1011054fa3 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/lmgr.c,v 1.54 2002/08/01 05:18:33 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/lmgr.c,v 1.55 2003/02/19 04:02:53 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -126,9 +126,10 @@ LockRelation(Relation relation, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relation->rd_lockInfo.lockRelId.relId;
+	tag.objId = relation->rd_lockInfo.lockRelId.relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relation->rd_lockInfo.lockRelId.dbId;
-	tag.objId.blkno = InvalidBlockNumber;
+	tag.objsubId.blkno = InvalidBlockNumber;
 
 	if (!LockAcquire(LockTableId, &tag, GetCurrentTransactionId(),
 					 lockmode, false))
@@ -160,9 +161,10 @@ ConditionalLockRelation(Relation relation, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relation->rd_lockInfo.lockRelId.relId;
+	tag.objId = relation->rd_lockInfo.lockRelId.relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relation->rd_lockInfo.lockRelId.dbId;
-	tag.objId.blkno = InvalidBlockNumber;
+	tag.objsubId.blkno = InvalidBlockNumber;
 
 	if (!LockAcquire(LockTableId, &tag, GetCurrentTransactionId(),
 					 lockmode, true))
@@ -190,9 +192,10 @@ UnlockRelation(Relation relation, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relation->rd_lockInfo.lockRelId.relId;
+	tag.objId = relation->rd_lockInfo.lockRelId.relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relation->rd_lockInfo.lockRelId.dbId;
-	tag.objId.blkno = InvalidBlockNumber;
+	tag.objsubId.blkno = InvalidBlockNumber;
 
 	LockRelease(LockTableId, &tag, GetCurrentTransactionId(), lockmode);
 }
@@ -215,9 +218,10 @@ LockRelationForSession(LockRelId *relid, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relid->relId;
+	tag.objId = relid->relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relid->dbId;
-	tag.objId.blkno = InvalidBlockNumber;
+	tag.objsubId.blkno = InvalidBlockNumber;
 
 	if (!LockAcquire(LockTableId, &tag, InvalidTransactionId,
 					 lockmode, false))
@@ -233,9 +237,10 @@ UnlockRelationForSession(LockRelId *relid, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relid->relId;
+	tag.objId = relid->relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relid->dbId;
-	tag.objId.blkno = InvalidBlockNumber;
+	tag.objsubId.blkno = InvalidBlockNumber;
 
 	LockRelease(LockTableId, &tag, InvalidTransactionId, lockmode);
 }
@@ -253,9 +258,10 @@ LockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relation->rd_lockInfo.lockRelId.relId;
+	tag.objId = relation->rd_lockInfo.lockRelId.relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relation->rd_lockInfo.lockRelId.dbId;
-	tag.objId.blkno = blkno;
+	tag.objsubId.blkno = blkno;
 
 	if (!LockAcquire(LockTableId, &tag, GetCurrentTransactionId(),
 					 lockmode, false))
@@ -271,9 +277,10 @@ UnlockPage(Relation relation, BlockNumber blkno, LOCKMODE lockmode)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = relation->rd_lockInfo.lockRelId.relId;
+	tag.objId = relation->rd_lockInfo.lockRelId.relId;
+	tag.classId = RelOid_pg_class;
 	tag.dbId = relation->rd_lockInfo.lockRelId.dbId;
-	tag.objId.blkno = blkno;
+	tag.objsubId.blkno = blkno;
 
 	LockRelease(LockTableId, &tag, GetCurrentTransactionId(), lockmode);
 }
@@ -294,9 +301,10 @@ XactLockTableInsert(TransactionId xid)
 	LOCKTAG		tag;
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = XactLockTableId;
+	tag.objId = InvalidOid;
+	tag.classId = XactLockTableId;
 	tag.dbId = InvalidOid;		/* xids are globally unique */
-	tag.objId.xid = xid;
+	tag.objsubId.xid = xid;
 
 	if (!LockAcquire(LockTableId, &tag, xid,
 					 ExclusiveLock, false))
@@ -317,9 +325,10 @@ XactLockTableWait(TransactionId xid)
 	Assert(!TransactionIdEquals(xid, myxid));
 
 	MemSet(&tag, 0, sizeof(tag));
-	tag.relId = XactLockTableId;
+	tag.objId = InvalidOid;
+	tag.classId = XactLockTableId;
 	tag.dbId = InvalidOid;
-	tag.objId.xid = xid;
+	tag.objsubId.xid = xid;
 
 	if (!LockAcquire(LockTableId, &tag, myxid,
 					 ShareLock, false))
@@ -334,3 +343,59 @@ XactLockTableWait(TransactionId xid)
 	if (!TransactionIdDidCommit(xid) && !TransactionIdDidAbort(xid))
 		TransactionIdAbort(xid);
 }
+
+/*
+ * LockObject
+ *
+ * Lock an arbitrary database object.  A standard relation lock would lock the
+ * classId of RelOid_pg_class and objId of the relations OID within the pg_class
+ * table.  LockObject allows classId to be specified by the caller, thus allowing
+ * locks on any row in any system table.
+ *
+ * If classId is NOT a system table (protected from removal), an additional lock
+ * should be held on the relation to prevent it from being dropped.
+ */
+void
+LockObject(Oid objId, Oid classId, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	MemSet(&tag, 0, sizeof(tag));
+	tag.objId = objId;
+	tag.classId = classId;
+	tag.dbId = MyDatabaseId;
+	tag.objsubId.blkno = InvalidBlockNumber;
+
+	/* Only two reasonable lock types */
+	Assert(lockmode == AccessShareLock || lockmode == AccessExclusiveLock);
+
+	if (!LockAcquire(LockTableId, &tag, GetCurrentTransactionId(),
+					 lockmode, false))
+		elog(ERROR, "LockObject: LockAcquire failed");
+}
+
+/*
+ * UnlockObject
+ */
+void
+UnlockObject(Oid objId, Oid classId, LOCKMODE lockmode)
+{
+	LOCKTAG		tag;
+
+	/* NoLock is a no-op */
+	if (lockmode == NoLock)
+		return;
+
+	MemSet(&tag, 0, sizeof(tag));
+	tag.objId = objId;
+	tag.classId = classId;
+	tag.dbId = MyDatabaseId;
+	tag.objsubId.blkno = InvalidBlockNumber;
+
+	/* Only two reasonable lock types */
+	Assert(lockmode == AccessShareLock
+		  || lockmode == AccessExclusiveLock);
+
+	LockRelease(LockTableId, &tag, GetCurrentTransactionId(), lockmode);
+}
+
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index d0c5a055f9e..959fe7877a0 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/lock.c,v 1.120 2003/02/18 02:13:24 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/lmgr/lock.c,v 1.121 2003/02/19 04:02:53 momjian Exp $
  *
  * NOTES
  *	  Outside modules can create a lock table and acquire/release
@@ -97,8 +97,8 @@ LOCK_DEBUG_ENABLED(const LOCK *lock)
 	return
 		(((LOCK_LOCKMETHOD(*lock) == DEFAULT_LOCKMETHOD && Trace_locks)
 	   || (LOCK_LOCKMETHOD(*lock) == USER_LOCKMETHOD && Trace_userlocks))
-		 && (lock->tag.relId >= (Oid) Trace_lock_oidmin))
-		|| (Trace_lock_table && (lock->tag.relId == Trace_lock_table));
+		 && (lock->tag.objId >= (Oid) Trace_lock_oidmin))
+		|| (Trace_lock_table && (lock->tag.objId == Trace_lock_table));
 }
 
 
@@ -107,12 +107,12 @@ LOCK_PRINT(const char *where, const LOCK *lock, LOCKMODE type)
 {
 	if (LOCK_DEBUG_ENABLED(lock))
 		elog(LOG,
-			 "%s: lock(%lx) tbl(%d) rel(%u) db(%u) obj(%u) grantMask(%x) "
+			 "%s: lock(%lx) tbl(%d) obj(%u) class(%u) db(%u) objsub(%u) grantMask(%x) "
 			 "req(%d,%d,%d,%d,%d,%d,%d)=%d "
 			 "grant(%d,%d,%d,%d,%d,%d,%d)=%d wait(%d) type(%s)",
 			 where, MAKE_OFFSET(lock),
-			 lock->tag.lockmethod, lock->tag.relId, lock->tag.dbId,
-			 lock->tag.objId.blkno, lock->grantMask,
+			 lock->tag.lockmethod, lock->tag.objId, lock->tag.classId, lock->tag.dbId,
+			 lock->tag.objsubId.blkno, lock->grantMask,
 			 lock->requested[1], lock->requested[2], lock->requested[3],
 			 lock->requested[4], lock->requested[5], lock->requested[6],
 			 lock->requested[7], lock->nRequested,
@@ -129,16 +129,16 @@ PROCLOCK_PRINT(const char *where, const PROCLOCK *proclockP)
 	if (
 	(((PROCLOCK_LOCKMETHOD(*proclockP) == DEFAULT_LOCKMETHOD && Trace_locks)
 	  || (PROCLOCK_LOCKMETHOD(*proclockP) == USER_LOCKMETHOD && Trace_userlocks))
-	 && (((LOCK *) MAKE_PTR(proclockP->tag.lock))->tag.relId >= (Oid) Trace_lock_oidmin))
-		|| (Trace_lock_table && (((LOCK *) MAKE_PTR(proclockP->tag.lock))->tag.relId == Trace_lock_table))
+	 && (((LOCK *) MAKE_PTR(proclockP->tag.lock))->tag.objId >= (Oid) Trace_lock_oidmin))
+		|| (Trace_lock_table && (((LOCK *) MAKE_PTR(proclockP->tag.lock))->tag.objId == Trace_lock_table))
 		)
 		elog(LOG,
 			 "%s: proclock(%lx) lock(%lx) tbl(%d) proc(%lx) xid(%u) hold(%d,%d,%d,%d,%d,%d,%d)=%d",
 			 where, MAKE_OFFSET(proclockP), proclockP->tag.lock,
 			 PROCLOCK_LOCKMETHOD(*(proclockP)),
 			 proclockP->tag.proc, proclockP->tag.xid,
-		   proclockP->holding[1], proclockP->holding[2], proclockP->holding[3],
-		   proclockP->holding[4], proclockP->holding[5], proclockP->holding[6],
+			 proclockP->holding[1], proclockP->holding[2], proclockP->holding[3],
+			 proclockP->holding[4], proclockP->holding[5], proclockP->holding[6],
 			 proclockP->holding[7], proclockP->nHolding);
 }
 
@@ -417,8 +417,9 @@ LockMethodTableRename(LOCKMETHOD lockmethod)
  *
  *		lockmethod						1				2
  *		tag.dbId						database oid	database oid
- *		tag.relId						rel oid or 0	0
- *		tag.objId						block id		lock id2
+ *		tag.classId						class oid		0
+ *		tag.objId						rel oid or 0	0
+ *		tag.objsubId					block id		lock id2
  *										or xact id
  *		tag.offnum						0				lock id1
  *		proclock.xid					xid or 0		0
@@ -449,7 +450,7 @@ LockAcquire(LOCKMETHOD lockmethod, LOCKTAG *locktag,
 #ifdef LOCK_DEBUG
 	if (lockmethod == USER_LOCKMETHOD && Trace_userlocks)
 		elog(LOG, "LockAcquire: user lock [%u] %s",
-			 locktag->objId.blkno, lock_mode_names[lockmode]);
+			 locktag->objsubId.blkno, lock_mode_names[lockmode]);
 #endif
 
 	/* ???????? This must be changed when short term locks will be used */
@@ -572,7 +573,7 @@ LockAcquire(LOCKMETHOD lockmethod, LOCKTAG *locktag,
 				elog(LOG, "Deadlock risk: raising lock level"
 					 " from %s to %s on object %u/%u/%u",
 					 lock_mode_names[i], lock_mode_names[lockmode],
-				 lock->tag.relId, lock->tag.dbId, lock->tag.objId.blkno);
+				 lock->tag.objId, lock->tag.dbId, lock->tag.objsubId.blkno);
 				break;
 			}
 		}
@@ -993,7 +994,7 @@ LockRelease(LOCKMETHOD lockmethod, LOCKTAG *locktag,
 
 #ifdef LOCK_DEBUG
 	if (lockmethod == USER_LOCKMETHOD && Trace_userlocks)
-		elog(LOG, "LockRelease: user lock tag [%u] %d", locktag->objId.blkno, lockmode);
+		elog(LOG, "LockRelease: user lock tag [%u] %d", locktag->objsubId.blkno, lockmode);
 #endif
 
 	/* ???????? This must be changed when short term locks will be used */
diff --git a/src/backend/utils/adt/lockfuncs.c b/src/backend/utils/adt/lockfuncs.c
index 37788d8ff0d..25bc726cba2 100644
--- a/src/backend/utils/adt/lockfuncs.c
+++ b/src/backend/utils/adt/lockfuncs.c
@@ -6,7 +6,7 @@
  * Copyright (c) 2002, PostgreSQL Global Development Group
  *
  * IDENTIFICATION
- *		$Header: /cvsroot/pgsql/src/backend/utils/adt/lockfuncs.c,v 1.8 2003/02/18 02:13:24 momjian Exp $
+ *		$Header: /cvsroot/pgsql/src/backend/utils/adt/lockfuncs.c,v 1.9 2003/02/19 04:02:54 momjian Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -53,18 +53,20 @@ pg_lock_status(PG_FUNCTION_ARGS)
 
 		/* build tupdesc for result tuples */
 		/* this had better match pg_locks view in initdb.sh */
-		tupdesc = CreateTemplateTupleDesc(6, false);
-		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "relation",
+		tupdesc = CreateTemplateTupleDesc(7, false);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 1, "object",
 						   OIDOID, -1, 0, false);
-		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "database",
+		TupleDescInitEntry(tupdesc, (AttrNumber) 2, "class",
 						   OIDOID, -1, 0, false);
-		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "transaction",
+		TupleDescInitEntry(tupdesc, (AttrNumber) 3, "database",
+						   OIDOID, -1, 0, false);
+		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "transaction",
 						   XIDOID, -1, 0, false);
-		TupleDescInitEntry(tupdesc, (AttrNumber) 4, "pid",
+		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "pid",
 						   INT4OID, -1, 0, false);
-		TupleDescInitEntry(tupdesc, (AttrNumber) 5, "mode",
+		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "mode",
 						   TEXTOID, -1, 0, false);
-		TupleDescInitEntry(tupdesc, (AttrNumber) 6, "granted",
+		TupleDescInitEntry(tupdesc, (AttrNumber) 7, "granted",
 						   BOOLOID, -1, 0, false);
 
 		funcctx->slot = TupleDescGetSlot(tupdesc);
@@ -93,8 +95,8 @@ pg_lock_status(PG_FUNCTION_ARGS)
 		PGPROC	   *proc;
 		bool		granted;
 		LOCKMODE	mode;
-		Datum		values[6];
-		char		nulls[6];
+		Datum		values[7];
+		char		nulls[7];
 		HeapTuple	tuple;
 		Datum		result;
 
@@ -152,26 +154,30 @@ pg_lock_status(PG_FUNCTION_ARGS)
 		MemSet(values, 0, sizeof(values));
 		MemSet(nulls, ' ', sizeof(nulls));
 
-		if (lock->tag.relId == XactLockTableId && lock->tag.dbId == 0)
+		if (lock->tag.objId == InvalidOid
+			&& lock->tag.classId == XactLockTableId
+			&& lock->tag.dbId == InvalidOid)
 		{
 			/* Lock is for transaction ID */
 			nulls[0] = 'n';
 			nulls[1] = 'n';
-			values[2] = TransactionIdGetDatum(lock->tag.objId.xid);
+			nulls[2] = 'n';
+			values[3] = TransactionIdGetDatum(lock->tag.objsubId.xid);
 		}
 		else
 		{
 			/* Lock is for a relation */
-			values[0] = ObjectIdGetDatum(lock->tag.relId);
-			values[1] = ObjectIdGetDatum(lock->tag.dbId);
-			nulls[2] = 'n';
+			values[0] = ObjectIdGetDatum(lock->tag.objId);
+			values[1] = ObjectIdGetDatum(lock->tag.classId);
+			values[2] = ObjectIdGetDatum(lock->tag.dbId);
+			nulls[3] = 'n';
 
 		}
 
-		values[3] = Int32GetDatum(proc->pid);
-		values[4] = DirectFunctionCall1(textin,
+		values[4] = Int32GetDatum(proc->pid);
+		values[5] = DirectFunctionCall1(textin,
 								 CStringGetDatum(GetLockmodeName(mode)));
-		values[5] = BoolGetDatum(granted);
+		values[6] = BoolGetDatum(granted);
 
 		tuple = heap_formtuple(funcctx->slot->ttc_tupleDescriptor,
 							   values, nulls);