Create a syscache for pg_database-indexed-by-oid, and make use of it

in various places that were previously doing ad hoc pg_database searches.
This may speed up database-related privilege checks a little bit, but
the main motivation is to eliminate the performance reason for having
ReverifyMyDatabase do such a lot of stuff (viz, avoiding repeat scans
of pg_database during backend startup).  The locking reason for having
that routine is about to go away, and it'd be good to have the option
to break it up.

Author: tglsfdc

src/backend/catalog/aclchk.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/aclchk.c,v 1.127 2006/04/30 21:15:33 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/aclchk.c,v 1.128 2006/05/03 22:45:26 tgl Exp $
  *
  * NOTES
  *	  See acl.h.
@@ -34,6 +34,7 @@
 #include "catalog/pg_proc.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_type.h"
+#include "commands/dbcommands.h"
 #include "miscadmin.h"
 #include "parser/parse_func.h"
 #include "utils/acl.h"
@@ -412,8 +413,8 @@ objectNamesToOids(GrantObjectType objtype, List *objnames)
 		case ACL_OBJECT_SEQUENCE:
 			foreach(cell, objnames)
 			{
-				Oid			relOid;
 				RangeVar   *relvar = (RangeVar *) lfirst(cell);
+				Oid			relOid;
 
 				relOid = RangeVarGetRelid(relvar, false);
 				objects = lappend_oid(objects, relOid);
@@ -423,32 +424,15 @@ objectNamesToOids(GrantObjectType objtype, List *objnames)
 			foreach(cell, objnames)
 			{
 				char	   *dbname = strVal(lfirst(cell));
-				ScanKeyData entry[1];
-				HeapScanDesc scan;
-				HeapTuple	tuple;
-				Relation	relation;
-
-				relation = heap_open(DatabaseRelationId, AccessShareLock);
+				Oid			dbid;
 
-				/*
-				 * There's no syscache for pg_database, so we must look the
-				 * hard way.
-				 */
-				ScanKeyInit(&entry[0],
-							Anum_pg_database_datname,
-							BTEqualStrategyNumber, F_NAMEEQ,
-							CStringGetDatum(dbname));
-				scan = heap_beginscan(relation, SnapshotNow, 1, entry);
-				tuple = heap_getnext(scan, ForwardScanDirection);
-				if (!HeapTupleIsValid(tuple))
+				dbid = get_database_oid(dbname);
+				if (!OidIsValid(dbid))
 					ereport(ERROR,
 							(errcode(ERRCODE_UNDEFINED_DATABASE),
-						  errmsg("database \"%s\" does not exist", dbname)));
-				objects = lappend_oid(objects, HeapTupleGetOid(tuple));
-
-				heap_close(relation, AccessShareLock);
-
-				heap_endscan(scan);
+							 errmsg("database \"%s\" does not exist",
+									dbname)));
+				objects = lappend_oid(objects, dbid);
 			}
 			break;
 		case ACL_OBJECT_FUNCTION:
@@ -474,7 +458,8 @@ objectNamesToOids(GrantObjectType objtype, List *objnames)
 				if (!HeapTupleIsValid(tuple))
 					ereport(ERROR,
 							(errcode(ERRCODE_UNDEFINED_OBJECT),
-						errmsg("language \"%s\" does not exist", langname)));
+							 errmsg("language \"%s\" does not exist",
+									langname)));
 
 				objects = lappend_oid(objects, HeapTupleGetOid(tuple));
 
@@ -493,7 +478,8 @@ objectNamesToOids(GrantObjectType objtype, List *objnames)
 				if (!HeapTupleIsValid(tuple))
 					ereport(ERROR,
 							(errcode(ERRCODE_UNDEFINED_SCHEMA),
-						   errmsg("schema \"%s\" does not exist", nspname)));
+							 errmsg("schema \"%s\" does not exist",
+									nspname)));
 
 				objects = lappend_oid(objects, HeapTupleGetOid(tuple));
 
@@ -764,22 +750,13 @@ ExecGrant_Database(InternalGrant *istmt)
 		int			nnewmembers;
 		Oid		   *oldmembers;
 		Oid		   *newmembers;
-		ScanKeyData entry[1];
-		SysScanDesc scan;
 		HeapTuple	tuple;
 
-		/* There's no syscache for pg_database, so must look the hard way */
-		ScanKeyInit(&entry[0],
-					ObjectIdAttributeNumber,
-					BTEqualStrategyNumber, F_OIDEQ,
-					ObjectIdGetDatum(datId));
-		scan = systable_beginscan(relation, DatabaseOidIndexId, true,
-								  SnapshotNow, 1, entry);
-
-		tuple = systable_getnext(scan);
-
+		tuple = SearchSysCache(DATABASEOID,
+							   ObjectIdGetDatum(datId),
+							   0, 0, 0);
 		if (!HeapTupleIsValid(tuple))
-			elog(ERROR, "could not find tuple for database %u", datId);
+			elog(ERROR, "cache lookup failed for database %u", datId);
 
 		pg_database_tuple = (Form_pg_database) GETSTRUCT(tuple);
 
@@ -847,7 +824,7 @@ ExecGrant_Database(InternalGrant *istmt)
 							  noldmembers, oldmembers,
 							  nnewmembers, newmembers);
 
-		systable_endscan(scan);
+		ReleaseSysCache(tuple);
 
 		pfree(new_acl);
 
@@ -1657,61 +1634,31 @@ pg_database_aclmask(Oid db_oid, Oid roleid,
 					AclMode mask, AclMaskHow how)
 {
 	AclMode		result;
-	Relation	pg_database;
-	ScanKeyData entry[1];
-	SysScanDesc scan;
 	HeapTuple	tuple;
+	Datum		aclDatum;
+	bool		isNull;
+	Acl		   *acl;
+	Oid			ownerId;
 
 	/* Superusers bypass all permission checking. */
 	if (superuser_arg(roleid))
 		return mask;
 
 	/*
 	 * Get the database's ACL from pg_database
-	 *
-	 * There's no syscache for pg_database, so must look the hard way
 	 */
-	pg_database = heap_open(DatabaseRelationId, AccessShareLock);
-	ScanKeyInit(&entry[0],
-				ObjectIdAttributeNumber,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(db_oid));
-	scan = systable_beginscan(pg_database, DatabaseOidIndexId, true,
-							  SnapshotNow, 1, entry);
-	tuple = systable_getnext(scan);
+	tuple = SearchSysCache(DATABASEOID,
+						   ObjectIdGetDatum(db_oid),
+						   0, 0, 0);
 	if (!HeapTupleIsValid(tuple))
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_DATABASE),
 				 errmsg("database with OID %u does not exist", db_oid)));
 
-	result = pg_database_tuple_aclmask(tuple, RelationGetDescr(pg_database),
-									   roleid, mask, how);
-
-	systable_endscan(scan);
-	heap_close(pg_database, AccessShareLock);
-
-	return result;
-}
-
-/*
- * This is split out so that ReverifyMyDatabase can perform an ACL check
- * without a whole extra search of pg_database
- */
-AclMode
-pg_database_tuple_aclmask(HeapTuple db_tuple, TupleDesc tupdesc,
-						  Oid roleid, AclMode mask, AclMaskHow how)
-{
-	AclMode		result;
-	Datum		aclDatum;
-	bool		isNull;
-	Acl		   *acl;
-	Oid			ownerId;
-
-	ownerId = ((Form_pg_database) GETSTRUCT(db_tuple))->datdba;
-
-	aclDatum = heap_getattr(db_tuple, Anum_pg_database_datacl,
-							tupdesc, &isNull);
+	ownerId = ((Form_pg_database) GETSTRUCT(tuple))->datdba;
 
+	aclDatum = SysCacheGetAttr(DATABASEOID, tuple, Anum_pg_database_datacl,
+							   &isNull);
 	if (isNull)
 	{
 		/* No ACL, so build default ACL */
@@ -1730,6 +1677,8 @@ pg_database_tuple_aclmask(HeapTuple db_tuple, TupleDesc tupdesc,
 	if (acl && (Pointer) acl != DatumGetPointer(aclDatum))
 		pfree(acl);
 
+	ReleaseSysCache(tuple);
+
 	return result;
 }
 
@@ -2298,36 +2247,24 @@ pg_opclass_ownercheck(Oid opc_oid, Oid roleid)
 bool
 pg_database_ownercheck(Oid db_oid, Oid roleid)
 {
-	Relation	pg_database;
-	ScanKeyData entry[1];
-	SysScanDesc scan;
-	HeapTuple	dbtuple;
+	HeapTuple	tuple;
 	Oid			dba;
 
 	/* Superusers bypass all permission checking. */
 	if (superuser_arg(roleid))
 		return true;
 
-	/* There's no syscache for pg_database, so must look the hard way */
-	pg_database = heap_open(DatabaseRelationId, AccessShareLock);
-	ScanKeyInit(&entry[0],
-				ObjectIdAttributeNumber,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(db_oid));
-	scan = systable_beginscan(pg_database, DatabaseOidIndexId, true,
-							  SnapshotNow, 1, entry);
-
-	dbtuple = systable_getnext(scan);
-
-	if (!HeapTupleIsValid(dbtuple))
+	tuple = SearchSysCache(DATABASEOID,
+						   ObjectIdGetDatum(db_oid),
+						   0, 0, 0);
+	if (!HeapTupleIsValid(tuple))
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_DATABASE),
 				 errmsg("database with OID %u does not exist", db_oid)));
 
-	dba = ((Form_pg_database) GETSTRUCT(dbtuple))->datdba;
+	dba = ((Form_pg_database) GETSTRUCT(tuple))->datdba;
 
-	systable_endscan(scan);
-	heap_close(pg_database, AccessShareLock);
+	ReleaseSysCache(tuple);
 
 	return has_privs_of_role(roleid, dba);
 }

src/backend/commands/dbcommands.c

@@ -15,7 +15,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.179 2006/03/29 21:17:38 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.180 2006/05/03 22:45:26 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -556,8 +556,6 @@ dropdb(const char *dbname, bool missing_ok)
 	Oid			db_id;
 	bool		db_istemplate;
 	Relation	pgdbrel;
-	SysScanDesc pgdbscan;
-	ScanKeyData key;
 	HeapTuple	tup;
 
 	PreventTransactionChain((void *) dbname, "DROP DATABASE");
@@ -629,31 +627,17 @@ dropdb(const char *dbname, bool missing_ok)
 						dbname)));
 
 	/*
-	 * Find the database's tuple by OID (should be unique).
+	 * Remove the database's tuple from pg_database.
 	 */
-	ScanKeyInit(&key,
-				ObjectIdAttributeNumber,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(db_id));
-
-	pgdbscan = systable_beginscan(pgdbrel, DatabaseOidIndexId, true,
-								  SnapshotNow, 1, &key);
-
-	tup = systable_getnext(pgdbscan);
+	tup = SearchSysCache(DATABASEOID,
+						 ObjectIdGetDatum(db_id),
+						 0, 0, 0);
 	if (!HeapTupleIsValid(tup))
-	{
-		/*
-		 * This error should never come up since the existence of the database
-		 * is checked earlier
-		 */
-		elog(ERROR, "database \"%s\" doesn't exist despite earlier reports to the contrary",
-			 dbname);
-	}
+		elog(ERROR, "cache lookup failed for database %u", db_id);
 
-	/* Remove the database's tuple from pg_database */
 	simple_heap_delete(pgdbrel, &tup->t_self);
 
-	systable_endscan(pgdbscan);
+	ReleaseSysCache(tup);
 
 	/*
 	 * Delete any comments associated with the database
@@ -1262,7 +1246,10 @@ get_database_oid(const char *dbname)
 	HeapTuple	dbtuple;
 	Oid			oid;
 
-	/* There's no syscache for pg_database, so must look the hard way */
+	/*
+	 * There's no syscache for pg_database indexed by name,
+	 * so we must look the hard way.
+	 */
 	pg_database = heap_open(DatabaseRelationId, AccessShareLock);
 	ScanKeyInit(&entry[0],
 				Anum_pg_database_datname,
@@ -1296,32 +1283,20 @@ get_database_oid(const char *dbname)
 char *
 get_database_name(Oid dbid)
 {
-	Relation	pg_database;
-	ScanKeyData entry[1];
-	SysScanDesc scan;
 	HeapTuple	dbtuple;
 	char	   *result;
 
-	/* There's no syscache for pg_database, so must look the hard way */
-	pg_database = heap_open(DatabaseRelationId, AccessShareLock);
-	ScanKeyInit(&entry[0],
-				ObjectIdAttributeNumber,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(dbid));
-	scan = systable_beginscan(pg_database, DatabaseOidIndexId, true,
-							  SnapshotNow, 1, entry);
-
-	dbtuple = systable_getnext(scan);
-
-	/* We assume that there can be at most one matching tuple */
+	dbtuple = SearchSysCache(DATABASEOID,
+							 ObjectIdGetDatum(dbid),
+							 0, 0, 0);
 	if (HeapTupleIsValid(dbtuple))
+	{
 		result = pstrdup(NameStr(((Form_pg_database) GETSTRUCT(dbtuple))->datname));
+		ReleaseSysCache(dbtuple);
+	}
 	else
 		result = NULL;
 
-	systable_endscan(scan);
-	heap_close(pg_database, AccessShareLock);
-
 	return result;
 }