fix_partition_and_toast_privs.patch

diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index 79ccddce55..022b5e9bd3 100644
--- a/src/backend/catalog/partition.c
+++ b/src/backend/catalog/partition.c
@@ -119,6 +119,20 @@ get_partition_parent_worker(Relation inhRel, Oid relid, bool *detach_pending)
 	return result;
 }
 
+Oid
+get_partition_root(Oid relid)
+{
+	List	   *ancestors = get_partition_ancestors(relid);
+	Oid			root = InvalidOid;
+
+	if (list_length(ancestors) > 0)
+		root = llast_oid(ancestors);
+
+	list_free(ancestors);
+
+	return root;
+}
+
 /*
  * get_partition_ancestors
  *		Obtain ancestors of given relation
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index 9bc10729b0..1ca2a65405 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -14,6 +14,7 @@
  */
 #include "postgres.h"
 
+#include "access/genam.h"
 #include "access/heapam.h"
 #include "access/toast_compression.h"
 #include "access/xact.h"
@@ -32,6 +33,7 @@
 #include "nodes/makefuncs.h"
 #include "storage/lock.h"
 #include "utils/builtins.h"
+#include "utils/fmgroids.h"
 #include "utils/rel.h"
 #include "utils/syscache.h"
 
@@ -413,3 +415,30 @@ needs_toast_table(Relation rel)
 	/* Otherwise, let the AM decide. */
 	return table_relation_needs_toast_table(rel);
 }
+
+Oid
+get_toast_parent(Oid relid)
+{
+	Relation	rel;
+	SysScanDesc scan;
+	ScanKeyData key[1];
+	Oid			result = InvalidOid;
+	HeapTuple	tuple;
+
+	rel = table_open(RelationRelationId, AccessShareLock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_class_reltoastrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(relid));
+
+	scan = systable_beginscan(rel, InvalidOid, false, NULL, 1, key);
+	tuple = systable_getnext(scan);
+	if (HeapTupleIsValid(tuple))
+		result = ((Form_pg_class) GETSTRUCT(tuple))->oid;
+	systable_endscan(scan);
+
+	table_close(rel, AccessShareLock);
+
+	return result;
+}
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 8966b75bd1..04681ce494 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -80,6 +80,7 @@ static void copy_table_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex,
 static List *get_tables_to_cluster(MemoryContext cluster_context);
 static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context,
 											   Oid indexOid);
+static bool cluster_is_permitted_for_relation(Oid relid, Oid userid);
 
 
 /*---------------------------------------------------------------------------
@@ -366,8 +367,7 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params)
 	if (recheck)
 	{
 		/* 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)
+		if (!cluster_is_permitted_for_relation(tableOid, save_userid))
 		{
 			relation_close(OldHeap, AccessExclusiveLock);
 			goto out;
@@ -1646,8 +1646,7 @@ get_tables_to_cluster(MemoryContext cluster_context)
 
 		index = (Form_pg_index) GETSTRUCT(indexTuple);
 
-		if (!object_ownercheck(RelationRelationId, index->indrelid, GetUserId()) &&
-			pg_class_aclcheck(index->indrelid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+		if (!cluster_is_permitted_for_relation(index->indrelid, GetUserId()))
 			continue;
 
 		/* Use a permanent memory context for the result list */
@@ -1695,12 +1694,11 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
 		if (get_rel_relkind(indexrelid) != RELKIND_INDEX)
 			continue;
 
-		/* 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;
+		/*
+		 * We already checked that the user has privileges to CLUSTER the
+		 * partitioned table when we locked it earlier, so there's no need to
+		 * check the privileges again here.
+		 */
 
 		/* Use a permanent memory context for the result list */
 		old_context = MemoryContextSwitchTo(cluster_context);
@@ -1715,3 +1713,10 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
 
 	return rtcs;
 }
+
+static bool
+cluster_is_permitted_for_relation(Oid relid, Oid userid)
+{
+	return pg_class_aclcheck(relid, userid, ACL_MAINTAIN) == ACLCHECK_OK ||
+		   has_parent_privs(relid, userid, ACL_MAINTAIN);
+}
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 7dc1aca8fe..d83e0158dd 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -2796,11 +2796,10 @@ RangeVarCallbackForReindexIndex(const RangeVar *relation,
 
 	/* Check permissions */
 	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);
+	if (OidIsValid(table_oid) &&
+		pg_class_aclcheck(table_oid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+		!has_parent_privs(table_oid, GetUserId(), ACL_MAINTAIN))
+		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX, relation->relname);
 
 	/* Lock heap before index to avoid deadlock. */
 	if (relId != oldRelId)
@@ -3017,7 +3016,8 @@ ReindexMultipleTables(const char *objectName, ReindexObjectType objectKind,
 		 */
 		if (classtuple->relisshared &&
 			!object_ownercheck(RelationRelationId, relid, GetUserId()) &&
-			pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+			pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+			!has_parent_privs(relid, GetUserId(), ACL_MAINTAIN))
 			continue;
 
 		/*
diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c
index e294efc67c..425e94f27c 100644
--- a/src/backend/commands/lockcmds.c
+++ b/src/backend/commands/lockcmds.c
@@ -17,6 +17,7 @@
 #include "access/table.h"
 #include "access/xact.h"
 #include "catalog/namespace.h"
+#include "catalog/partition.h"
 #include "catalog/pg_inherits.h"
 #include "commands/lockcmds.h"
 #include "miscadmin.h"
@@ -305,5 +306,17 @@ LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid)
 
 	aclresult = pg_class_aclcheck(reloid, userid, aclmask);
 
+	/*
+	 * If this is a partition, check the permissions on its partitioned table,
+	 * too.
+	 */
+	if (aclresult != ACLCHECK_OK && get_rel_relispartition(reloid))
+	{
+		Oid			root = get_partition_root(reloid);
+
+		if (OidIsValid(root))
+			aclresult = pg_class_aclcheck(root, userid, ACL_MAINTAIN);
+	}
+
 	return aclresult;
 }
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 56dc995713..a2763b43df 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16918,12 +16918,42 @@ RangeVarCallbackMaintainsTable(const RangeVar *relation,
 				 errmsg("\"%s\" is not a table or materialized view", relation->relname)));
 
 	/* Check permissions */
-	if (!object_ownercheck(RelationRelationId, relId, GetUserId()) &&
-		pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK)
+	if (pg_class_aclcheck(relId, GetUserId(), ACL_MAINTAIN) != ACLCHECK_OK &&
+		!has_parent_privs(relId, GetUserId(), ACL_MAINTAIN))
 		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
 					   relation->relname);
 }
 
+bool
+has_parent_privs(Oid relid, Oid userid, AclMode acl)
+{
+	/*
+	 * If this is a partition, check the permissions on its partitioned table.
+	 */
+	if (get_rel_relispartition(relid))
+	{
+		Oid			root = get_partition_root(relid);
+
+		if (OidIsValid(root) &&
+			pg_class_aclcheck(root, userid, acl) == ACLCHECK_OK)
+			return true;
+	}
+
+	/*
+	 * If this is a TOAST table, check the permissions on the main relation.
+	 */
+	if (get_rel_relkind(relid) == RELKIND_TOASTVALUE)
+	{
+		Oid			parent = get_toast_parent(relid);
+
+		if (OidIsValid(parent) &&
+			pg_class_aclcheck(parent, userid, acl) == ACLCHECK_OK)
+			return true;
+	}
+
+	return false;
+}
+
 /*
  * Callback to RangeVarGetRelidExtended() for TRUNCATE processing.
  */
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index 293b84bbca..9360564945 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -41,6 +41,7 @@
 #include "catalog/pg_namespace.h"
 #include "commands/cluster.h"
 #include "commands/defrem.h"
+#include "commands/tablecmds.h"
 #include "commands/vacuum.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
@@ -568,9 +569,9 @@ vacuum_is_permitted_for_relation(Oid relid, Form_pg_class reltuple,
 	 *   - the role owns the current database and the relation is not shared
 	 *   - the role has been granted the MAINTAIN privilege on the relation
 	 */
-	if (object_ownercheck(RelationRelationId, relid, GetUserId()) ||
-		(object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
-		pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK)
+	if ((object_ownercheck(DatabaseRelationId, MyDatabaseId, GetUserId()) && !reltuple->relisshared) ||
+		pg_class_aclcheck(relid, GetUserId(), ACL_MAINTAIN) == ACLCHECK_OK ||
+		has_parent_privs(relid, GetUserId(), ACL_MAINTAIN))
 		return true;
 
 	relname = NameStr(reltuple->relname);
diff --git a/src/include/catalog/partition.h b/src/include/catalog/partition.h
index b29df6690c..41144abf3f 100644
--- a/src/include/catalog/partition.h
+++ b/src/include/catalog/partition.h
@@ -20,6 +20,7 @@
 #define HASH_PARTITION_SEED UINT64CONST(0x7A5B22367996DCFD)
 
 extern Oid	get_partition_parent(Oid relid, bool even_if_detached);
+extern Oid	get_partition_root(Oid relid);
 extern List *get_partition_ancestors(Oid relid);
 extern Oid	index_get_partition(Relation partition, Oid indexId);
 extern List *map_partition_varattnos(List *expr, int fromrel_varno,
diff --git a/src/include/catalog/toasting.h b/src/include/catalog/toasting.h
index 51e3ba7b69..2afdec5c4a 100644
--- a/src/include/catalog/toasting.h
+++ b/src/include/catalog/toasting.h
@@ -26,5 +26,6 @@ extern void AlterTableCreateToastTable(Oid relOid, Datum reloptions,
 									   LOCKMODE lockmode);
 extern void BootstrapToastTable(char *relName,
 								Oid toastOid, Oid toastIndexOid);
+extern Oid get_toast_parent(Oid relid);
 
 #endif							/* TOASTING_H */
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index 07eac9a26c..959ae53aa8 100644
--- a/src/include/commands/tablecmds.h
+++ b/src/include/commands/tablecmds.h
@@ -98,6 +98,7 @@ extern void AtEOSubXact_on_commit_actions(bool isCommit,
 extern void RangeVarCallbackMaintainsTable(const RangeVar *relation,
 										   Oid relId, Oid oldRelId,
 										   void *arg);
+extern bool has_parent_privs(Oid relid, Oid userid, AclMode acl);
 
 extern void RangeVarCallbackOwnsRelation(const RangeVar *relation,
 										 Oid relId, Oid oldRelId, void *arg);
diff --git a/src/test/isolation/expected/cluster-conflict-partition.out b/src/test/isolation/expected/cluster-conflict-partition.out
index 7acb675c97..8d21276996 100644
--- a/src/test/isolation/expected/cluster-conflict-partition.out
+++ b/src/test/isolation/expected/cluster-conflict-partition.out
@@ -22,14 +22,16 @@ starting permutation: s1_begin s1_lock_child s2_auth s2_cluster s1_commit s2_res
 step s1_begin: BEGIN;
 step s1_lock_child: LOCK cluster_part_tab1 IN SHARE UPDATE EXCLUSIVE MODE;
 step s2_auth: SET ROLE regress_cluster_part;
-step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind;
+step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind; <waiting ...>
 step s1_commit: COMMIT;
+step s2_cluster: <... completed>
 step s2_reset: RESET ROLE;
 
 starting permutation: s1_begin s2_auth s1_lock_child s2_cluster s1_commit s2_reset
 step s1_begin: BEGIN;
 step s2_auth: SET ROLE regress_cluster_part;
 step s1_lock_child: LOCK cluster_part_tab1 IN SHARE UPDATE EXCLUSIVE MODE;
-step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind;
+step s2_cluster: CLUSTER cluster_part_tab USING cluster_part_ind; <waiting ...>
 step s1_commit: COMMIT;
+step s2_cluster: <... completed>
 step s2_reset: RESET ROLE;
diff --git a/src/test/regress/expected/cluster.out b/src/test/regress/expected/cluster.out
index 542c2e098c..332210e81f 100644
--- a/src/test/regress/expected/cluster.out
+++ b/src/test/regress/expected/cluster.out
@@ -513,7 +513,7 @@ SELECT a.relname, a.relfilenode=b.relfilenode FROM pg_class a
 -----------+----------
  ptnowner  | t
  ptnowner1 | f
- ptnowner2 | t
+ ptnowner2 | f
 (3 rows)
 
 DROP TABLE ptnowner;
diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index 0035d158b7..41ff4ba9cf 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -347,20 +347,14 @@ ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
 ALTER TABLE vacowned_part1 OWNER TO regress_vacuum;
 SET ROLE regress_vacuum;
 VACUUM vacowned_parted;
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 VACUUM vacowned_part1;
 VACUUM vacowned_part2;
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 ANALYZE vacowned_parted;
-WARNING:  permission denied to analyze "vacowned_part2", skipping it
 ANALYZE vacowned_part1;
 ANALYZE vacowned_part2;
-WARNING:  permission denied to analyze "vacowned_part2", skipping it
 VACUUM (ANALYZE) vacowned_parted;
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 VACUUM (ANALYZE) vacowned_part1;
 VACUUM (ANALYZE) vacowned_part2;
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 RESET ROLE;
 -- Only one partition owned by other user.
 ALTER TABLE vacowned_parted OWNER TO CURRENT_USER;
@@ -389,26 +383,14 @@ ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
 ALTER TABLE vacowned_part1 OWNER TO CURRENT_USER;
 SET ROLE regress_vacuum;
 VACUUM vacowned_parted;
-WARNING:  permission denied to vacuum "vacowned_part1", skipping it
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 VACUUM vacowned_part1;
-WARNING:  permission denied to vacuum "vacowned_part1", skipping it
 VACUUM vacowned_part2;
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 ANALYZE vacowned_parted;
-WARNING:  permission denied to analyze "vacowned_part1", skipping it
-WARNING:  permission denied to analyze "vacowned_part2", skipping it
 ANALYZE vacowned_part1;
-WARNING:  permission denied to analyze "vacowned_part1", skipping it
 ANALYZE vacowned_part2;
-WARNING:  permission denied to analyze "vacowned_part2", skipping it
 VACUUM (ANALYZE) vacowned_parted;
-WARNING:  permission denied to vacuum "vacowned_part1", skipping it
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 VACUUM (ANALYZE) vacowned_part1;
-WARNING:  permission denied to vacuum "vacowned_part1", skipping it
 VACUUM (ANALYZE) vacowned_part2;
-WARNING:  permission denied to vacuum "vacowned_part2", skipping it
 RESET ROLE;
 DROP TABLE vacowned;
 DROP TABLE vacowned_parted;