Refactor ALTER some-obj RENAME implementation

Remove duplicate implementations of catalog munging and miscellaneous
privilege checks.  Instead rely on already existing data in
objectaddress.c to do the work.

Author: KaiGai Kohei, changes by me
Reviewed by: Robert Haas, Álvaro Herrera, Dimitri Fontaine

1 files changed, 283 insertions, 91 deletions

diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index b565e9b2c33..c2d4bb3ed4d 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -21,9 +21,15 @@
 #include "catalog/namespace.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_conversion.h"
+#include "catalog/pg_event_trigger.h"
+#include "catalog/pg_foreign_data_wrapper.h"
+#include "catalog/pg_foreign_server.h"
+#include "catalog/pg_language.h"
 #include "catalog/pg_largeobject.h"
 #include "catalog/pg_largeobject_metadata.h"
 #include "catalog/pg_namespace.h"
+#include "catalog/pg_opclass.h"
+#include "catalog/pg_opfamily.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_ts_config.h"
 #include "catalog/pg_ts_dict.h"
@@ -43,6 +49,7 @@
 #include "commands/trigger.h"
 #include "commands/typecmds.h"
 #include "commands/user.h"
+#include "parser/parse_func.h"
 #include "miscadmin.h"
 #include "tcop/utility.h"
 #include "utils/builtins.h"
@@ -56,49 +63,246 @@
 static Oid AlterObjectNamespace_internal(Relation rel, Oid objid, Oid nspOid);
 
 /*
- * Executes an ALTER OBJECT / RENAME TO statement.	Based on the object
- * type, the function appropriate to that type is executed.
+ * Raise an error to the effect that an object of the given name is already
+ * present in the given namespace.
  */
-Oid
-ExecRenameStmt(RenameStmt *stmt)
+static void
+report_name_conflict(Oid classId, const char *name)
 {
-	switch (stmt->renameType)
+	char   *msgfmt;
+
+	switch (classId)
 	{
-		case OBJECT_AGGREGATE:
-			return RenameAggregate(stmt->object, stmt->objarg, stmt->newname);
+		case EventTriggerRelationId:
+			msgfmt = gettext_noop("event trigger \"%s\" already exists");
+			break;
+		case ForeignDataWrapperRelationId:
+			msgfmt = gettext_noop("foreign-data wrapper \"%s\" already exists");
+			break;
+		case ForeignServerRelationId:
+			msgfmt = gettext_noop("server \"%s\" already exists");
+			break;
+		case LanguageRelationId:
+			msgfmt = gettext_noop("language \"%s\" already exists");
+			break;
+		default:
+			elog(ERROR, "unsupported object class %u", classId);
+			break;
+	}
 
-		case OBJECT_COLLATION:
-			return RenameCollation(stmt->object, stmt->newname);
+	ereport(ERROR,
+			(errcode(ERRCODE_DUPLICATE_OBJECT),
+			 errmsg(msgfmt, name)));
+}
 
-		case OBJECT_CONSTRAINT:
-			return RenameConstraint(stmt);
+static void
+report_namespace_conflict(Oid classId, const char *name, Oid nspOid)
+{
+	char   *msgfmt;
 
-		case OBJECT_CONVERSION:
-			return RenameConversion(stmt->object, stmt->newname);
+	Assert(OidIsValid(nspOid));
 
-		case OBJECT_DATABASE:
-			return RenameDatabase(stmt->subname, stmt->newname);
+	switch (classId)
+	{
+		case ConversionRelationId:
+			Assert(OidIsValid(nspOid));
+			msgfmt = gettext_noop("conversion \"%s\" already exists in schema \"%s\"");
+			break;
+		case TSParserRelationId:
+			Assert(OidIsValid(nspOid));
+			msgfmt = gettext_noop("text search parser \"%s\" already exists in schema \"%s\"");
+			break;
+		case TSDictionaryRelationId:
+			Assert(OidIsValid(nspOid));
+			msgfmt = gettext_noop("text search dictionary \"%s\" already exists in schema \"%s\"");
+			break;
+		case TSTemplateRelationId:
+			Assert(OidIsValid(nspOid));
+			msgfmt = gettext_noop("text search template \"%s\" already exists in schema \"%s\"");
+			break;
+		case TSConfigRelationId:
+			Assert(OidIsValid(nspOid));
+			msgfmt = gettext_noop("text search configuration \"%s\" already exists in schema \"%s\"");
+			break;
+		default:
+			elog(ERROR, "unsupported object class %u", classId);
+			break;
+	}
 
-		case OBJECT_FDW:
-			return RenameForeignDataWrapper(stmt->subname, stmt->newname);
+	ereport(ERROR,
+			(errcode(ERRCODE_DUPLICATE_OBJECT),
+			 errmsg(msgfmt, name, get_namespace_name(nspOid))));
+}
 
-		case OBJECT_FOREIGN_SERVER:
-			return RenameForeignServer(stmt->subname, stmt->newname);
+/*
+ * AlterObjectRename_internal
+ *
+ * Generic function to rename the given object, for simple cases (won't
+ * work for tables, nor other cases where we need to do more than change
+ * the name column of a single catalog entry).
+ *
+ * rel: catalog relation containing object (RowExclusiveLock'd by caller)
+ * objectId: OID of object to be renamed
+ * new_name: CString representation of new name
+ */
+static void
+AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name)
+{
+	Oid			classId = RelationGetRelid(rel);
+	int			oidCacheId = get_object_catcache_oid(classId);
+	int			nameCacheId = get_object_catcache_name(classId);
+	AttrNumber	Anum_name = get_object_attnum_name(classId);
+	AttrNumber	Anum_namespace = get_object_attnum_namespace(classId);
+	AttrNumber	Anum_owner = get_object_attnum_owner(classId);
+	AclObjectKind acl_kind = get_object_aclkind(classId);
+	HeapTuple	oldtup;
+	HeapTuple	newtup;
+	Datum		datum;
+	bool		isnull;
+	Oid			namespaceId;
+	Oid			ownerId;
+	char	   *old_name;
+	AclResult	aclresult;
+	Datum	   *values;
+	bool	   *nulls;
+	bool	   *replaces;
 
-		case OBJECT_EVENT_TRIGGER:
-			return RenameEventTrigger(stmt->subname, stmt->newname);
+	oldtup = SearchSysCache1(oidCacheId, ObjectIdGetDatum(objectId));
+	if (!HeapTupleIsValid(oldtup))
+		elog(ERROR, "cache lookup failed for object %u of catalog \"%s\"",
+			 objectId, RelationGetRelationName(rel));
 
-		case OBJECT_FUNCTION:
-			return RenameFunction(stmt->object, stmt->objarg, stmt->newname);
+	datum = heap_getattr(oldtup, Anum_name,
+						 RelationGetDescr(rel), &isnull);
+	Assert(!isnull);
+	old_name = NameStr(*(DatumGetName(datum)));
 
-		case OBJECT_LANGUAGE:
-			return RenameLanguage(stmt->subname, stmt->newname);
+	/* Get OID of namespace */
+	if (Anum_namespace > 0)
+	{
+		datum = heap_getattr(oldtup, Anum_namespace,
+							 RelationGetDescr(rel), &isnull);
+		Assert(!isnull);
+		namespaceId = DatumGetObjectId(datum);
+	}
+	else
+		namespaceId = InvalidOid;
 
-		case OBJECT_OPCLASS:
-			return RenameOpClass(stmt->object, stmt->subname, stmt->newname);
+	/* Permission checks ... superusers can always do it */
+	if (!superuser())
+	{
+		/* Fail if object does not have an explicit owner */
+		if (Anum_owner <= 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 (errmsg("must be superuser to rename %s",
+							 getObjectDescriptionOids(classId, objectId)))));
 
-		case OBJECT_OPFAMILY:
-			return RenameOpFamily(stmt->object, stmt->subname, stmt->newname);
+		/* Otherwise, must be owner of the existing object */
+		datum = heap_getattr(oldtup, Anum_owner,
+							 RelationGetDescr(rel), &isnull);
+		Assert(!isnull);
+		ownerId = DatumGetObjectId(datum);
+
+		if (!has_privs_of_role(GetUserId(), DatumGetObjectId(ownerId)))
+			aclcheck_error(ACLCHECK_NOT_OWNER, acl_kind, old_name);
+
+		/* User must have CREATE privilege on the namespace */
+		if (OidIsValid(namespaceId))
+		{
+			aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(),
+											  ACL_CREATE);
+			if (aclresult != ACLCHECK_OK)
+				aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
+							   get_namespace_name(namespaceId));
+		}
+	}
+
+	/*
+     * Check for duplicate name (more friendly than unique-index failure).
+     * Since this is just a friendliness check, we can just skip it in cases
+     * where there isn't suitable support.
+     */
+	if (classId == ProcedureRelationId)
+	{
+		Form_pg_proc proc = (Form_pg_proc) GETSTRUCT(oldtup);
+
+		IsThereFunctionInNamespace(new_name, proc->pronargs,
+								   proc->proargtypes, proc->pronamespace);
+	}
+	else if (classId == CollationRelationId)
+	{
+		Form_pg_collation coll = (Form_pg_collation) GETSTRUCT(oldtup);
+
+		IsThereCollationInNamespace(new_name, coll->collnamespace);
+	}
+	else if (classId == OperatorClassRelationId)
+	{
+		Form_pg_opclass opc = (Form_pg_opclass) GETSTRUCT(oldtup);
+
+		IsThereOpClassInNamespace(new_name, opc->opcmethod,
+								  opc->opcnamespace);
+	}
+	else if (classId == OperatorFamilyRelationId)
+	{
+		Form_pg_opfamily opf = (Form_pg_opfamily) GETSTRUCT(oldtup);
+
+		IsThereOpFamilyInNamespace(new_name, opf->opfmethod,
+								   opf->opfnamespace);
+	}
+	else if (nameCacheId >= 0)
+	{
+		if (OidIsValid(namespaceId))
+		{
+			if (SearchSysCacheExists2(nameCacheId,
+									  CStringGetDatum(new_name),
+									  ObjectIdGetDatum(namespaceId)))
+				report_namespace_conflict(classId, new_name, namespaceId);
+		}
+		else
+		{
+			if (SearchSysCacheExists1(nameCacheId,
+									  CStringGetDatum(new_name)))
+				report_name_conflict(classId, new_name);
+		}
+	}
+
+	/* Build modified tuple */
+	values = palloc0(RelationGetNumberOfAttributes(rel) * sizeof(Datum));
+	nulls = palloc0(RelationGetNumberOfAttributes(rel) * sizeof(bool));
+	replaces = palloc0(RelationGetNumberOfAttributes(rel) * sizeof(bool));
+	values[Anum_name - 1] = PointerGetDatum(new_name);
+	replaces[Anum_name - 1] = true;
+	newtup = heap_modify_tuple(oldtup, RelationGetDescr(rel),
+							   values, nulls, replaces);
+
+	/* Perform actual update */
+	simple_heap_update(rel, &oldtup->t_self, newtup);
+	CatalogUpdateIndexes(rel, newtup);
+
+	/* Release memory */
+	pfree(values);
+	pfree(nulls);
+	pfree(replaces);
+	heap_freetuple(newtup);
+
+	ReleaseSysCache(oldtup);
+}
+
+/*
+ * Executes an ALTER OBJECT / RENAME TO statement.	Based on the object
+ * type, the function appropriate to that type is executed.
+ */
+Oid
+ExecRenameStmt(RenameStmt *stmt)
+{
+	switch (stmt->renameType)
+	{
+		case OBJECT_CONSTRAINT:
+			return RenameConstraint(stmt);
+
+		case OBJECT_DATABASE:
+			return RenameDatabase(stmt->subname, stmt->newname);
 
 		case OBJECT_ROLE:
 			return RenameRole(stmt->subname, stmt->newname);
@@ -123,21 +327,43 @@ ExecRenameStmt(RenameStmt *stmt)
 		case OBJECT_TRIGGER:
 			return renametrig(stmt);
 
-		case OBJECT_TSPARSER:
-			return RenameTSParser(stmt->object, stmt->newname);
+		case OBJECT_DOMAIN:
+		case OBJECT_TYPE:
+			return RenameType(stmt);
 
+		case OBJECT_AGGREGATE:
+		case OBJECT_COLLATION:
+		case OBJECT_CONVERSION:
+		case OBJECT_EVENT_TRIGGER:
+		case OBJECT_FDW:
+		case OBJECT_FOREIGN_SERVER:
+		case OBJECT_FUNCTION:
+		case OBJECT_OPCLASS:
+		case OBJECT_OPFAMILY:
+		case OBJECT_LANGUAGE:
+		case OBJECT_TSCONFIGURATION:
 		case OBJECT_TSDICTIONARY:
-			return RenameTSDictionary(stmt->object, stmt->newname);
-
+		case OBJECT_TSPARSER:
 		case OBJECT_TSTEMPLATE:
-			return RenameTSTemplate(stmt->object, stmt->newname);
+			{
+				ObjectAddress	address;
+				Relation		catalog;
+				Relation		relation;
 
-		case OBJECT_TSCONFIGURATION:
-			return RenameTSConfiguration(stmt->object, stmt->newname);
+				address = get_object_address(stmt->renameType,
+											 stmt->object, stmt->objarg,
+											 &relation,
+											 AccessExclusiveLock, false);
+				Assert(relation == NULL);
 
-		case OBJECT_DOMAIN:
-		case OBJECT_TYPE:
-			return RenameType(stmt);
+				catalog = heap_open(address.classId, RowExclusiveLock);
+				AlterObjectRename_internal(catalog,
+										   address.objectId,
+										   stmt->newname);
+				heap_close(catalog, RowExclusiveLock);
+
+				return address.objectId;
+			}
 
 		default:
 			elog(ERROR, "unrecognized rename stmt type: %d",
@@ -288,43 +514,6 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 }
 
 /*
- * Raise an error to the effect that an object of the given name is already
- * present in the given namespace.
- */
-static void
-report_namespace_conflict(Oid classId, const char *name, Oid nspOid)
-{
-	char   *msgfmt;
-
-	Assert(OidIsValid(nspOid));
-	switch (classId)
-	{
-		case ConversionRelationId:
-			msgfmt = gettext_noop("conversion \"%s\" already exists in schema \"%s\"");
-			break;
-		case TSParserRelationId:
-			msgfmt = gettext_noop("text search parser \"%s\" already exists in schema \"%s\"");
-			break;
-		case TSDictionaryRelationId:
-			msgfmt = gettext_noop("text search dictionary \"%s\" already exists in schema \"%s\"");
-			break;
-		case TSTemplateRelationId:
-			msgfmt = gettext_noop("text search template \"%s\" already exists in schema \"%s\"");
-			break;
-		case TSConfigRelationId:
-			msgfmt = gettext_noop("text search configuration \"%s\" already exists in schema \"%s\"");
-			break;
-		default:
-			elog(ERROR, "unsupported object class %u", classId);
-			break;
-	}
-
-	ereport(ERROR,
-			(errcode(ERRCODE_DUPLICATE_OBJECT),
-			 errmsg(msgfmt, name, get_namespace_name(nspOid))));
-}
-
-/*
  * Generic function to change the namespace of a given object, for simple
  * cases (won't work for tables, nor other cases where we need to do more
  * than change the namespace column of a single catalog entry).
@@ -403,31 +592,34 @@ AlterObjectNamespace_internal(Relation rel, Oid objid, Oid nspOid)
 	/*
 	 * Check for duplicate name (more friendly than unique-index failure).
 	 * Since this is just a friendliness check, we can just skip it in cases
-	 * where there isn't a suitable syscache available.
+	 * where there isn't suitable support.
 	 */
 	if (classId == ProcedureRelationId)
 	{
-		HeapTuple	tup;
-		Form_pg_proc proc;
-
-		tup = SearchSysCacheCopy1(PROCOID, ObjectIdGetDatum(objid));
-		if (!HeapTupleIsValid(tup))
-			elog(ERROR, "cache lookup failed for function %u", objid);
-		proc = (Form_pg_proc) GETSTRUCT(tup);
+		Form_pg_proc proc = (Form_pg_proc) GETSTRUCT(tup);
 
 		IsThereFunctionInNamespace(NameStr(proc->proname), proc->pronargs,
 								   proc->proargtypes, nspOid);
-		heap_freetuple(tup);
 	}
 	else if (classId == CollationRelationId)
 	{
-		char *collname;
+		Form_pg_collation coll = (Form_pg_collation) GETSTRUCT(tup);
+
+		IsThereCollationInNamespace(NameStr(coll->collname), nspOid);
+	}
+	else if (classId == OperatorClassRelationId)
+	{
+		Form_pg_opclass opc = (Form_pg_opclass) GETSTRUCT(tup);
+
+		IsThereOpClassInNamespace(NameStr(opc->opcname),
+								  opc->opcmethod, nspOid);
+	}
+	else if (classId == OperatorFamilyRelationId)
+	{
+		Form_pg_opfamily opf = (Form_pg_opfamily) GETSTRUCT(tup);
 
-		collname = get_collation_name(objid);
-		if (!collname)
-			elog(ERROR, "cache lookup failed for collation %u", objid);
-		IsThereCollationInNamespace(collname, nspOid);
-		pfree(collname);
+		IsThereOpFamilyInNamespace(NameStr(opf->opfname),
+								   opf->opfmethod, nspOid);
 	}
 	else if (nameCacheId >= 0 &&
 			 SearchSysCacheExists2(nameCacheId, name,