ALTER TABLESPACE ... MOVE ... OWNED BY

Add the ability to specify the objects to move by who those objects are
owned by (as relowner) and change ALL to mean ALL objects.  This
makes the command always operate against a well-defined set of objects
and not have the objects-to-be-moved based on the role of the user
running the command.

Per discussion with Simon and Tom.

7 files changed, 90 insertions, 17 deletions

diff --git a/src/backend/commands/tablespace.c b/src/backend/commands/tablespace.c
index 05a89f0bde2..d73e5e826dc 100644
--- a/src/backend/commands/tablespace.c
+++ b/src/backend/commands/tablespace.c
@@ -67,6 +67,7 @@
 #include "commands/seclabel.h"
 #include "commands/tablecmds.h"
 #include "commands/tablespace.h"
+#include "commands/user.h"
 #include "common/relpath.h"
 #include "miscadmin.h"
 #include "postmaster/bgwriter.h"
@@ -994,6 +995,7 @@ AlterTableSpaceMove(AlterTableSpaceMoveStmt *stmt)
 	HeapTuple	tuple;
 	Oid			orig_tablespaceoid;
 	Oid			new_tablespaceoid;
+	List	   *role_oids = roleNamesToIds(stmt->roles);
 
 	/* Ensure we were not asked to move something we can't */
 	if (!stmt->move_all && stmt->objtype != OBJECT_TABLE &&
@@ -1075,14 +1077,10 @@ AlterTableSpaceMove(AlterTableSpaceMoveStmt *stmt)
 			relForm->relnamespace == PG_TOAST_NAMESPACE)
 			continue;
 
-		/*
-		 * Only move objects that we are considered an owner of and only
-		 * objects which can actually have a tablespace.
-		 */
-		if (!pg_class_ownercheck(relOid, GetUserId()) ||
-			(relForm->relkind != RELKIND_RELATION &&
-			 relForm->relkind != RELKIND_INDEX &&
-			 relForm->relkind != RELKIND_MATVIEW))
+		/* Only consider objects which live in tablespaces */
+		if (relForm->relkind != RELKIND_RELATION &&
+			relForm->relkind != RELKIND_INDEX &&
+			relForm->relkind != RELKIND_MATVIEW)
 			continue;
 
 		/* Check if we were asked to only move a certain type of object */
@@ -1095,6 +1093,21 @@ AlterTableSpaceMove(AlterTableSpaceMoveStmt *stmt)
 			  relForm->relkind != RELKIND_MATVIEW)))
 			continue;
 
+		/* Check if we are only moving objects owned by certain roles */
+		if (role_oids != NIL && !list_member_oid(role_oids, relForm->relowner))
+			continue;
+
+		/*
+		 * Handle permissions-checking here since we are locking the tables
+		 * and also to avoid doing a bunch of work only to fail part-way.
+		 * Note that permissions will also be checked by AlterTableInternal().
+		 *
+		 * Caller must be considered an owner on the table to move it.
+		 */
+		if (!pg_class_ownercheck(relOid, GetUserId()))
+			aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
+						   NameStr(relForm->relname));
+
 		if (stmt->nowait &&
 			!ConditionalLockRelationOid(relOid, AccessExclusiveLock))
 			ereport(ERROR,
diff --git a/src/backend/commands/user.c b/src/backend/commands/user.c
index f8cf2a1b462..bcdc392a817 100644
--- a/src/backend/commands/user.c
+++ b/src/backend/commands/user.c
@@ -48,7 +48,6 @@ extern bool Password_encryption;
 /* Hook to check passwords in CreateRole() and AlterRole() */
 check_password_hook_type check_password_hook = NULL;
 
-static List *roleNamesToIds(List *memberNames);
 static void AddRoleMems(const char *rolename, Oid roleid,
 			List *memberNames, List *memberIds,
 			Oid grantorId, bool admin_opt);
@@ -1302,7 +1301,7 @@ ReassignOwnedObjects(ReassignOwnedStmt *stmt)
  * Given a list of role names (as String nodes), generate a list of role OIDs
  * in the same order.
  */
-static List *
+List *
 roleNamesToIds(List *memberNames)
 {
 	List	   *result = NIL;
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index bb356d0b263..f90cb6797de 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -3404,6 +3404,9 @@ _copyAlterTableSpaceMoveStmt(const AlterTableSpaceMoveStmt *from)
 	AlterTableSpaceMoveStmt *newnode = makeNode(AlterTableSpaceMoveStmt);
 
 	COPY_STRING_FIELD(orig_tablespacename);
+	COPY_SCALAR_FIELD(objtype);
+	COPY_SCALAR_FIELD(move_all);
+	COPY_NODE_FIELD(roles);
 	COPY_STRING_FIELD(new_tablespacename);
 	COPY_SCALAR_FIELD(nowait);
 
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 5908d9abcf9..9438e7861d8 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1640,6 +1640,9 @@ _equalAlterTableSpaceMoveStmt(const AlterTableSpaceMoveStmt *a,
 							  const AlterTableSpaceMoveStmt *b)
 {
 	COMPARE_STRING_FIELD(orig_tablespacename);
+	COMPARE_SCALAR_FIELD(objtype);
+	COMPARE_SCALAR_FIELD(move_all);
+	COMPARE_NODE_FIELD(roles);
 	COMPARE_STRING_FIELD(new_tablespacename);
 	COMPARE_SCALAR_FIELD(nowait);
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 363a22c1690..0787eb7c5d4 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -7325,9 +7325,11 @@ RenameStmt: ALTER AGGREGATE func_name aggr_args RENAME TO name
 					AlterTableSpaceMoveStmt *n =
 						makeNode(AlterTableSpaceMoveStmt);
 					n->orig_tablespacename = $3;
+					n->objtype = -1;
+					n->move_all = true;
+					n->roles = NIL;
 					n->new_tablespacename = $7;
 					n->nowait = $8;
-					n->move_all = true;
 					$$ = (Node *)n;
 				}
 			| ALTER TABLESPACE name MOVE TABLES TO name opt_nowait
@@ -7335,10 +7337,11 @@ RenameStmt: ALTER AGGREGATE func_name aggr_args RENAME TO name
 					AlterTableSpaceMoveStmt *n =
 						makeNode(AlterTableSpaceMoveStmt);
 					n->orig_tablespacename = $3;
-					n->new_tablespacename = $7;
-					n->nowait = $8;
 					n->objtype = OBJECT_TABLE;
 					n->move_all = false;
+					n->roles = NIL;
+					n->new_tablespacename = $7;
+					n->nowait = $8;
 					$$ = (Node *)n;
 				}
 			| ALTER TABLESPACE name MOVE INDEXES TO name opt_nowait
@@ -7346,10 +7349,11 @@ RenameStmt: ALTER AGGREGATE func_name aggr_args RENAME TO name
 					AlterTableSpaceMoveStmt *n =
 						makeNode(AlterTableSpaceMoveStmt);
 					n->orig_tablespacename = $3;
-					n->new_tablespacename = $7;
-					n->nowait = $8;
 					n->objtype = OBJECT_INDEX;
 					n->move_all = false;
+					n->roles = NIL;
+					n->new_tablespacename = $7;
+					n->nowait = $8;
 					$$ = (Node *)n;
 				}
 			| ALTER TABLESPACE name MOVE MATERIALIZED VIEWS TO name opt_nowait
@@ -7357,10 +7361,59 @@ RenameStmt: ALTER AGGREGATE func_name aggr_args RENAME TO name
 					AlterTableSpaceMoveStmt *n =
 						makeNode(AlterTableSpaceMoveStmt);
 					n->orig_tablespacename = $3;
+					n->objtype = OBJECT_MATVIEW;
+					n->move_all = false;
+					n->roles = NIL;
 					n->new_tablespacename = $8;
 					n->nowait = $9;
+					$$ = (Node *)n;
+				}
+			| ALTER TABLESPACE name MOVE ALL OWNED BY role_list TO name opt_nowait
+				{
+					AlterTableSpaceMoveStmt *n =
+						makeNode(AlterTableSpaceMoveStmt);
+					n->orig_tablespacename = $3;
+					n->objtype = -1;
+					n->move_all = true;
+					n->roles = $8;
+					n->new_tablespacename = $10;
+					n->nowait = $11;
+					$$ = (Node *)n;
+				}
+			| ALTER TABLESPACE name MOVE TABLES OWNED BY role_list TO name opt_nowait
+				{
+					AlterTableSpaceMoveStmt *n =
+						makeNode(AlterTableSpaceMoveStmt);
+					n->orig_tablespacename = $3;
+					n->objtype = OBJECT_TABLE;
+					n->move_all = false;
+					n->roles = $8;
+					n->new_tablespacename = $10;
+					n->nowait = $11;
+					$$ = (Node *)n;
+				}
+			| ALTER TABLESPACE name MOVE INDEXES OWNED BY role_list TO name opt_nowait
+				{
+					AlterTableSpaceMoveStmt *n =
+						makeNode(AlterTableSpaceMoveStmt);
+					n->orig_tablespacename = $3;
+					n->objtype = OBJECT_INDEX;
+					n->move_all = false;
+					n->roles = $8;
+					n->new_tablespacename = $10;
+					n->nowait = $11;
+					$$ = (Node *)n;
+				}
+			| ALTER TABLESPACE name MOVE MATERIALIZED VIEWS OWNED BY role_list TO name opt_nowait
+				{
+					AlterTableSpaceMoveStmt *n =
+						makeNode(AlterTableSpaceMoveStmt);
+					n->orig_tablespacename = $3;
 					n->objtype = OBJECT_MATVIEW;
 					n->move_all = false;
+					n->roles = $9;
+					n->new_tablespacename = $11;
+					n->nowait = $12;
 					$$ = (Node *)n;
 				}
 			| ALTER TABLESPACE name SET reloptions
diff --git a/src/include/commands/user.h b/src/include/commands/user.h
index 9e73a195e3f..d76685182f8 100644
--- a/src/include/commands/user.h
+++ b/src/include/commands/user.h
@@ -30,5 +30,6 @@ extern void GrantRole(GrantRoleStmt *stmt);
 extern Oid	RenameRole(const char *oldname, const char *newname);
 extern void DropOwnedObjects(DropOwnedStmt *stmt);
 extern void ReassignOwnedObjects(ReassignOwnedStmt *stmt);
+extern List *roleNamesToIds(List *memberNames);
 
 #endif   /* USER_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 846c31aebdf..ad58b3949b6 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -1691,10 +1691,11 @@ typedef struct AlterTableSpaceMoveStmt
 {
 	NodeTag		type;
 	char	   *orig_tablespacename;
+	ObjectType	objtype;		/* set to -1 if move_all is true */
+	bool		move_all;		/* move all, or just objtype objects? */
+	List	   *roles;			/* List of roles to move objects of */
 	char	   *new_tablespacename;
-	ObjectType	objtype;
 	bool		nowait;
-	bool		move_all;
 } AlterTableSpaceMoveStmt;
 
 /* ----------------------