Add grantable MAINTAIN privilege and pg_maintain role.

Allows VACUUM, ANALYZE, REINDEX, REFRESH MATERIALIZED VIEW, CLUSTER,
and LOCK TABLE.

Effectively reverts 4441fc704d. Instead of creating separate
privileges for VACUUM, ANALYZE, and other maintenance commands, group
them together under a single MAINTAIN privilege.

Author: Nathan Bossart
Discussion: https://postgr.es/m/20221212210136.GA449764@nathanxps13
Discussion: https://postgr.es/m/45224.1670476523@sss.pgh.pa.us

src/backend/commands/analyze.c

@@ -167,7 +167,7 @@ analyze_rel(Oid relid, RangeVar *relation,
 	 */
 	if (!vacuum_is_permitted_for_relation(RelationGetRelid(onerel),
 										  onerel->rd_rel,
-										  VACOPT_ANALYZE))
+										  params->options & VACOPT_ANALYZE))
 	{
 		relation_close(onerel, ShareUpdateExclusiveLock);
 		return;

src/backend/commands/cluster.c

@@ -147,7 +147,8 @@ cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel)
 		tableOid = RangeVarGetRelidExtended(stmt->relation,
 											AccessExclusiveLock,
 											0,
-											RangeVarCallbackOwnsTable, NULL);
+											RangeVarCallbackMaintainsTable,
+											NULL);
 		rel = table_open(tableOid, NoLock);
 
 		/*
@@ -364,8 +365,9 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params)
 	 */
 	if (recheck)
 	{
-		/* Check that the user still owns the relation */
-		if (!object_ownercheck(RelationRelationId, tableOid, save_userid))
+		/* Check that the user still has privileges for the relation */
+		if (!object_ownercheck(RelationRelationId, tableOid, save_userid) &&
+			pg_class_aclcheck(tableOid, save_userid, ACL_MAINTAIN) != ACLCHECK_OK)
 		{
 			relation_close(OldHeap, AccessExclusiveLock);
 			goto out;
@@ -1612,7 +1614,7 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap,
 
 
 /*
- * Get a list of tables that the current user owns and
+ * Get a list of tables that the current user has privileges on and
  * have indisclustered set.  Return the list in a List * of RelToCluster
  * (stored in the specified memory context), each one giving the tableOid
  * and the indexOid on which the table is already clustered.
@@ -1629,8 +1631,8 @@ get_tables_to_cluster(MemoryContext cluster_context)
 	List	   *rtcs = NIL;
 
 	/*
-	 * Get all indexes that have indisclustered set and are owned by
-	 * appropriate user.
+	 * Get all indexes that have indisclustered set and that the current user
+	 * has the appropriate privileges for.
 	 */
 	indRelation = table_open(IndexRelationId, AccessShareLock);
 	ScanKeyInit(&entry,
@@ -1644,7 +1646,8 @@ get_tables_to_cluster(MemoryContext cluster_context)
 
 		index = (Form_pg_index) GETSTRUCT(indexTuple);
 
-		if (!object_ownercheck(RelationRelationId, index->indrelid, GetUserId()))
+		if (!object_ownercheck(RelationRelationId, index->indrelid, GetUserId()) &&
+			pg_class_aclcheck(index->indrelid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
 			continue;
 
 		/* Use a permanent memory context for the result list */
@@ -1694,6 +1697,7 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
 
 		/* Silently skip partitions which the user has no access to. */
 		if (!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
+			pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
 			(!object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) ||
 			 IsSharedRelation(relid)))
 			continue;

src/backend/commands/indexcmds.c

@@ -26,6 +26,7 @@
 #include "catalog/index.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_am.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_database.h"
 #include "catalog/pg_inherits.h"
@@ -2754,6 +2755,7 @@ RangeVarCallbackForReindexIndex(const RangeVar *relation,
 	char		relkind;
 	struct ReindexIndexCallbackState *state = arg;
 	LOCKMODE	table_lockmode;
+	Oid			table_oid;
 
 	/*
 	 * Lock level here should match table lock in reindex_index() for
@@ -2793,14 +2795,16 @@ RangeVarCallbackForReindexIndex(const RangeVar *relation,
 				 errmsg("\"%s\" is not an index", relation->relname)));
 
 	/* Check permissions */
-	if (!object_ownercheck(RelationRelationId, relId, GetUserId()))
-		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX, relation->relname);
+	table_oid = IndexGetRelation(relId, true);
+	if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
+		OidIsValid(table_oid) &&
+		pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
+					   relation->relname);
 
 	/* Lock heap before index to avoid deadlock. */
 	if (relId != oldRelId)
 	{
-		Oid			table_oid = IndexGetRelation(relId, true);
-
 		/*
 		 * If the OID isn't valid, it means the index was concurrently
 		 * dropped, which is not a problem for us; just return normally.
@@ -2835,7 +2839,7 @@ ReindexTable(RangeVar *relation, ReindexParams *params, bool isTopLevel)
 									   (params->options & REINDEXOPT_CONCURRENTLY) != 0 ?
 									   ShareUpdateExclusiveLock : ShareLock,
 									   0,
-									   RangeVarCallbackOwnsTable, NULL);
+									   RangeVarCallbackMaintainsTable, NULL);
 
 	if (get_rel_relkind(heapOid) == RELKIND_PARTITIONED_TABLE)
 		ReindexPartitions(heapOid, params, isTopLevel);
@@ -2917,7 +2921,8 @@ ReindexMultipleTables(const char *objectName, ReindexObjectType objectKind,
 	{
 		objectOid = get_namespace_oid(objectName, false);
 
-		if (!object_ownercheck(NamespaceRelationId, objectOid, GetUserId()))
+		if (!object_ownercheck(NamespaceRelationId, objectOid, GetUserId()) &&
+			!has_privs_of_role(GetUserId(), ROLE_PG_MAINTAIN))
 			aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
 						   objectName);
 	}
@@ -2929,7 +2934,8 @@ ReindexMultipleTables(const char *objectName, ReindexObjectType objectKind,
 			ereport(ERROR,
 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 					 errmsg("can only reindex the currently open database")));
-		if (!object_ownercheck(DatabaseRelationId, objectOid, GetUserId()))
+		if (!object_ownercheck(DatabaseRelationId, objectOid, GetUserId()) &&
+			!has_privs_of_role(GetUserId(), ROLE_PG_MAINTAIN))
 			aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
 						   get_database_name(objectOid));
 	}
@@ -3001,15 +3007,17 @@ ReindexMultipleTables(const char *objectName, ReindexObjectType objectKind,
 			continue;
 
 		/*
-		 * The table can be reindexed if the user is superuser, the table
-		 * owner, or the database/schema owner (but in the latter case, only
-		 * if it's not a shared relation).  object_ownercheck includes the
-		 * superuser case, and depending on objectKind we already know that
-		 * the user has permission to run REINDEX on this database or schema
-		 * per the permission checks at the beginning of this routine.
+		 * The table can be reindexed if the user has been granted MAINTAIN on
+		 * the table or the user is a superuser, the table owner, or the
+		 * database/schema owner (but in the latter case, only if it's not a
+		 * shared relation).  object_ownercheck includes the superuser case,
+		 * and depending on objectKind we already know that the user has
+		 * permission to run REINDEX on this database or schema per the
+		 * permission checks at the beginning of this routine.
 		 */
 		if (classtuple->relisshared &&
-			!object_ownercheck(RelationRelationId, relid, GetUserId()))
+			!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
+			pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
 			continue;
 
 		/*

src/backend/commands/lockcmds.c

@@ -300,6 +300,9 @@ LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid)
 	else
 		aclmask = ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE;
 
+	/* MAINTAIN privilege allows all lock modes */
+	aclmask |= ACL_MAINTAIN;
+
 	aclresult = pg_class_aclcheck(reloid, userid, aclmask);
 
 	return aclresult;

src/backend/commands/matview.c

@@ -165,7 +165,8 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
 	 */
 	matviewOid = RangeVarGetRelidExtended(stmt->relation,
 										  lockmode, 0,
-										  RangeVarCallbackOwnsTable, NULL);
+										  RangeVarCallbackMaintainsTable,
+										  NULL);
 	matviewRel = table_open(matviewOid, NoLock);
 	relowner = matviewRel->rd_rel->relowner;
 

src/backend/commands/tablecmds.c

@@ -16889,13 +16889,13 @@ AtEOSubXact_on_commit_actions(bool isCommit, SubTransactionId mySubid,
  * This is intended as a callback for RangeVarGetRelidExtended().  It allows
  * the relation to be locked only if (1) it's a plain or partitioned table,
  * materialized view, or TOAST table and (2) the current user is the owner (or
- * the superuser).  This meets the permission-checking needs of CLUSTER,
- * REINDEX TABLE, and REFRESH MATERIALIZED VIEW; we expose it here so that it
- * can be used by all.
+ * the superuser) or has been granted MAINTAIN.  This meets the
+ * permission-checking needs of CLUSTER, REINDEX TABLE, and REFRESH
+ * MATERIALIZED VIEW; we expose it here so that it can be used by all.
  */
 void
-RangeVarCallbackOwnsTable(const RangeVar *relation,
-						  Oid relId, Oid oldRelId, void *arg)
+RangeVarCallbackMaintainsTable(const RangeVar *relation,
+							   Oid relId, Oid oldRelId, void *arg)
 {
 	char		relkind;
 
@@ -16918,8 +16918,10 @@ RangeVarCallbackOwnsTable(const RangeVar *relation,
 				 errmsg("\"%s\" is not a table or materialized view", relation->relname)));
 
 	/* Check permissions */
-	if (!object_ownercheck(RelationRelationId, relId, GetUserId()))
-		aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relId)), relation->relname);
+	if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
+		pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
+					   relation->relname);
 }
 
 /*

src/backend/commands/vacuum.c

@@ -557,7 +557,6 @@ vacuum_is_permitted_for_relation(Oid relid, Form_pg_class reltuple,
 								 bits32 options)
 {
 	char	   *relname;
-	AclMode		mode = 0;
 
 	Assert((options & (VACOPT_VACUUM | VACOPT_ANALYZE)) != 0);
 
@@ -567,15 +566,11 @@ vacuum_is_permitted_for_relation(Oid relid, Form_pg_class reltuple,
 	 *   - the role is a superuser
 	 *   - the role owns the relation
 	 *   - the role owns the current database and the relation is not shared
-	 *   - the role has been granted privileges to vacuum/analyze the relation
+	 *   - the role has been granted the MAINTAIN privilege on the relation
 	 */
-	if (options & VACOPT_VACUUM)
-		mode |= ACL_VACUUM;
-	if (options & VACOPT_ANALYZE)
-		mode |= ACL_ANALYZE;
 	if (object_ownercheck(RelationRelationId, relid, GetUserId()) ||
 		(object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
-		pg_class_aclcheck(relid, GetUserId(), mode) == ACLCHECK_OK)
+		pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK)
 		return true;
 
 	relname = NameStr(reltuple->relname);
@@ -1800,9 +1795,7 @@ vac_truncate_clog(TransactionId frozenXID,
  *		be stale.
  *
  *		Returns true if it's okay to proceed with a requested ANALYZE
- *		operation on this table.  Note that if vacuuming fails because the user
- *		does not have the required privileges, this function returns true since
- *		the user might have been granted privileges to ANALYZE the relation.
+ *		operation on this table.
  *
  *		Doing one heap at a time incurs extra overhead, since we need to
  *		check that the heap exists again just before we vacuum it.  The
@@ -1902,12 +1895,12 @@ vacuum_rel(Oid relid, RangeVar *relation, VacuumParams *params)
 	 */
 	if (!vacuum_is_permitted_for_relation(RelationGetRelid(rel),
 										  rel->rd_rel,
-										  VACOPT_VACUUM))
+										  params->options & VACOPT_VACUUM))
 	{
 		relation_close(rel, lmode);
 		PopActiveSnapshot();
 		CommitTransactionCommand();
-		return true;	/* user might have the ANALYZE privilege */
+		return false;
 	}
 
 	/*