Have CLUSTER ignore partitions not owned by caller

If a partitioned table has partitions owned by roles other than the
owner of the partitioned table, don't include them in the to-be-
clustered list.  This is similar to what VACUUM FULL does (except we do
it sooner, because there is no reason to postpone it).  Add a simple
test to verify that only owned partitions are clustered.

While at it, change memory context switch-and-back to occur once per
partition instead of outside of the loop.

Author: Justin Pryzby <pryzby@telsasoft.com>
Reviewed-by: Zhihong Yu <zyu@yugabyte.com>
Reviewed-by: Michael Paquier <michael@paquier.xyz>
Discussion: https://postgr.es/m/20220411140609.GF26620@telsasoft.com

src/backend/commands/cluster.c

@@ -1647,10 +1647,8 @@ get_tables_to_cluster(MemoryContext cluster_context)
  * Given an index on a partitioned table, return a list of RelToCluster for
  * all the children leaves tables/indexes.
  *
- * Caller must hold lock on the table containing the index.
- *
  * Like expand_vacuum_rel, but here caller must hold AccessExclusiveLock
- * on the table already.
+ * on the table containing the index.
  */
 static List *
 get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
@@ -1663,9 +1661,6 @@ get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid)
 	/* Do not lock the children until they're processed */
 	inhoids = find_all_inheritors(indexOid, NoLock, NULL);
 
-	/* Use a permanent memory context for the result list */
-	old_context = MemoryContextSwitchTo(cluster_context);
-
 	foreach(lc, inhoids)
 	{
 		Oid			indexrelid = lfirst_oid(lc);
@@ -1676,12 +1671,22 @@ 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 (!pg_class_ownercheck(relid, GetUserId()) &&
+			(!pg_database_ownercheck(MyDatabaseId, GetUserId()) ||
+			 IsSharedRelation(relid)))
+			continue;
+
+		/* Use a permanent memory context for the result list */
+		old_context = MemoryContextSwitchTo(cluster_context);
+
 		rtc = (RelToCluster *) palloc(sizeof(RelToCluster));
 		rtc->tableOid = relid;
 		rtc->indexOid = indexrelid;
 		rtcs = lappend(rtcs, rtc);
+
+		MemoryContextSwitchTo(old_context);
 	}
 
-	MemoryContextSwitchTo(old_context);
 	return rtcs;
 }

src/test/regress/expected/cluster.out

@@ -493,6 +493,31 @@ ERROR:  cannot mark index clustered in partitioned table
 ALTER TABLE clstrpart CLUSTER ON clstrpart_idx;
 ERROR:  cannot mark index clustered in partitioned table
 DROP TABLE clstrpart;
+-- Ownership of partitions is checked
+CREATE TABLE ptnowner(i int unique) PARTITION BY LIST (i);
+CREATE INDEX ptnowner_i_idx ON ptnowner(i);
+CREATE TABLE ptnowner1 PARTITION OF ptnowner FOR VALUES IN (1);
+CREATE ROLE regress_ptnowner;
+CREATE TABLE ptnowner2 PARTITION OF ptnowner FOR VALUES IN (2);
+ALTER TABLE ptnowner1 OWNER TO regress_ptnowner;
+ALTER TABLE ptnowner OWNER TO regress_ptnowner;
+CREATE TEMP TABLE ptnowner_oldnodes AS
+  SELECT oid, relname, relfilenode FROM pg_partition_tree('ptnowner') AS tree
+  JOIN pg_class AS c ON c.oid=tree.relid;
+SET SESSION AUTHORIZATION regress_ptnowner;
+CLUSTER ptnowner USING ptnowner_i_idx;
+RESET SESSION AUTHORIZATION;
+SELECT a.relname, a.relfilenode=b.relfilenode FROM pg_class a
+  JOIN ptnowner_oldnodes b USING (oid) ORDER BY a.relname COLLATE "C";
+  relname  | ?column? 
+-----------+----------
+ ptnowner  | t
+ ptnowner1 | f
+ ptnowner2 | t
+(3 rows)
+
+DROP TABLE ptnowner;
+DROP ROLE regress_ptnowner;
 -- Test CLUSTER with external tuplesorting
 create table clstr_4 as select * from tenk1;
 create index cluster_sort on clstr_4 (hundred, thousand, tenthous);

src/test/regress/sql/cluster.sql

@@ -229,6 +229,25 @@ ALTER TABLE clstrpart SET WITHOUT CLUSTER;
 ALTER TABLE clstrpart CLUSTER ON clstrpart_idx;
 DROP TABLE clstrpart;
 
+-- Ownership of partitions is checked
+CREATE TABLE ptnowner(i int unique) PARTITION BY LIST (i);
+CREATE INDEX ptnowner_i_idx ON ptnowner(i);
+CREATE TABLE ptnowner1 PARTITION OF ptnowner FOR VALUES IN (1);
+CREATE ROLE regress_ptnowner;
+CREATE TABLE ptnowner2 PARTITION OF ptnowner FOR VALUES IN (2);
+ALTER TABLE ptnowner1 OWNER TO regress_ptnowner;
+ALTER TABLE ptnowner OWNER TO regress_ptnowner;
+CREATE TEMP TABLE ptnowner_oldnodes AS
+  SELECT oid, relname, relfilenode FROM pg_partition_tree('ptnowner') AS tree
+  JOIN pg_class AS c ON c.oid=tree.relid;
+SET SESSION AUTHORIZATION regress_ptnowner;
+CLUSTER ptnowner USING ptnowner_i_idx;
+RESET SESSION AUTHORIZATION;
+SELECT a.relname, a.relfilenode=b.relfilenode FROM pg_class a
+  JOIN ptnowner_oldnodes b USING (oid) ORDER BY a.relname COLLATE "C";
+DROP TABLE ptnowner;
+DROP ROLE regress_ptnowner;
+
 -- Test CLUSTER with external tuplesorting
 
 create table clstr_4 as select * from tenk1;