Fix misbehavior of DROP OWNED BY with duplicate polroles entries.

Ordinarily, a pg_policy.polroles array wouldn't list the same role
more than once; but CREATE POLICY does not prevent that.  If we
perform DROP OWNED BY on a role that is listed more than once,
RemoveRoleFromObjectPolicy either suffered an assertion failure
or encountered a tuple-updated-by-self error.  Rewrite it to cope
correctly with duplicate entries, and add a CommandCounterIncrement
call to prevent the other problem.

Per discussion, there's other cleanup that ought to happen here,
but this seems like the minimum essential fix.

Per bug #17062 from Alexander Lakhin.  It's been broken all along,
so back-patch to all supported branches.

Discussion: https://postgr.es/m/17062-11f471ae3199ca23@postgresql.org

Author: tglsfdc

src/backend/commands/policy.c

@@ -18,6 +18,7 @@
 #include "access/relation.h"
 #include "access/sysattr.h"
 #include "access/table.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
@@ -425,10 +426,9 @@ RemoveRoleFromObjectPolicy(Oid roleid, Oid classid, Oid policy_id)
 	Oid			relid;
 	Relation	rel;
 	ArrayType  *policy_roles;
-	int			num_roles;
 	Datum		roles_datum;
 	bool		attr_isnull;
-	bool		noperm = true;
+	bool		keep_policy = true;
 
 	Assert(classid == PolicyRelationId);
 
@@ -481,31 +481,20 @@ RemoveRoleFromObjectPolicy(Oid roleid, Oid classid, Oid policy_id)
 
 	policy_roles = DatumGetArrayTypePCopy(roles_datum);
 
-	/* We should be removing exactly one entry from the roles array */
-	num_roles = ARR_DIMS(policy_roles)[0] - 1;
-
-	Assert(num_roles >= 0);
-
 	/* Must own relation. */
-	if (pg_class_ownercheck(relid, GetUserId()))
-		noperm = false;			/* user is allowed to modify this policy */
-	else
+	if (!pg_class_ownercheck(relid, GetUserId()))
 		ereport(WARNING,
 				(errcode(ERRCODE_WARNING_PRIVILEGE_NOT_REVOKED),
 				 errmsg("role \"%s\" could not be removed from policy \"%s\" on \"%s\"",
 						GetUserNameFromId(roleid, false),
 						NameStr(((Form_pg_policy) GETSTRUCT(tuple))->polname),
 						RelationGetRelationName(rel))));
-
-	/*
-	 * If multiple roles exist on this policy, then remove the one we were
-	 * asked to and leave the rest.
-	 */
-	if (!noperm && num_roles > 0)
+	else
 	{
 		int			i,
 					j;
 		Oid		   *roles = (Oid *) ARR_DATA_PTR(policy_roles);
+		int			num_roles = ARR_DIMS(policy_roles)[0];
 		Datum	   *role_oids;
 		char	   *qual_value;
 		Node	   *qual_expr;
@@ -582,16 +571,22 @@ RemoveRoleFromObjectPolicy(Oid roleid, Oid classid, Oid policy_id)
 		else
 			with_check_qual = NULL;
 
-		/* Rebuild the roles array to then update the pg_policy tuple with */
+		/*
+		 * Rebuild the roles array, without any mentions of the target role.
+		 * Ordinarily there'd be exactly one, but we must cope with duplicate
+		 * mentions, since CREATE/ALTER POLICY historically have allowed that.
+		 */
 		role_oids = (Datum *) palloc(num_roles * sizeof(Datum));
-		for (i = 0, j = 0; i < ARR_DIMS(policy_roles)[0]; i++)
-			/* Copy over all of the roles which are not the one being removed */
+		for (i = 0, j = 0; i < num_roles; i++)
+		{
 			if (roles[i] != roleid)
 				role_oids[j++] = ObjectIdGetDatum(roles[i]);
+		}
+		num_roles = j;
 
-		/* We should have only removed the one role */
-		Assert(j == num_roles);
-
+		/* If any roles remain, update the policy entry. */
+		if (num_roles > 0)
+		{
 		/* This is the array for the new tuple */
 		role_ids = construct_array(role_oids, num_roles, OIDOID,
 								   sizeof(Oid), true, TYPALIGN_INT);
@@ -646,8 +641,17 @@ RemoveRoleFromObjectPolicy(Oid roleid, Oid classid, Oid policy_id)
 
 		heap_freetuple(new_tuple);
 
+		/* Make updates visible */
+		CommandCounterIncrement();
+
 		/* Invalidate Relation Cache */
 		CacheInvalidateRelcache(rel);
+		}
+		else
+		{
+			/* No roles would remain, so drop the policy instead */
+			keep_policy = false;
+		}
 	}
 
 	/* Clean up. */
@@ -657,7 +661,7 @@ RemoveRoleFromObjectPolicy(Oid roleid, Oid classid, Oid policy_id)
 
 	table_close(pg_policy_rel, RowExclusiveLock);
 
-	return (noperm || num_roles > 0);
+	return keep_policy;
 }
 
 /*

src/test/regress/expected/rowsecurity.out

@@ -3886,6 +3886,15 @@ ERROR:  policy "p1" for table "dob_t1" does not exist
 CREATE POLICY p1 ON dob_t1 TO regress_rls_dob_role1,regress_rls_dob_role2 USING (true);
 DROP OWNED BY regress_rls_dob_role1;
 DROP POLICY p1 ON dob_t1; -- should succeed
+-- same cases with duplicate polroles entries
+CREATE POLICY p1 ON dob_t1 TO regress_rls_dob_role1,regress_rls_dob_role1 USING (true);
+DROP OWNED BY regress_rls_dob_role1;
+DROP POLICY p1 ON dob_t1; -- should fail, already gone
+ERROR:  policy "p1" for table "dob_t1" does not exist
+CREATE POLICY p1 ON dob_t1 TO regress_rls_dob_role1,regress_rls_dob_role1,regress_rls_dob_role2 USING (true);
+DROP OWNED BY regress_rls_dob_role1;
+DROP POLICY p1 ON dob_t1; -- should succeed
+-- partitioned target
 CREATE POLICY p1 ON dob_t2 TO regress_rls_dob_role1,regress_rls_dob_role2 USING (true);
 DROP OWNED BY regress_rls_dob_role1;
 DROP POLICY p1 ON dob_t2; -- should succeed

src/test/regress/sql/rowsecurity.sql

@@ -1757,6 +1757,16 @@ CREATE POLICY p1 ON dob_t1 TO regress_rls_dob_role1,regress_rls_dob_role2 USING
 DROP OWNED BY regress_rls_dob_role1;
 DROP POLICY p1 ON dob_t1; -- should succeed
 
+-- same cases with duplicate polroles entries
+CREATE POLICY p1 ON dob_t1 TO regress_rls_dob_role1,regress_rls_dob_role1 USING (true);
+DROP OWNED BY regress_rls_dob_role1;
+DROP POLICY p1 ON dob_t1; -- should fail, already gone
+
+CREATE POLICY p1 ON dob_t1 TO regress_rls_dob_role1,regress_rls_dob_role1,regress_rls_dob_role2 USING (true);
+DROP OWNED BY regress_rls_dob_role1;
+DROP POLICY p1 ON dob_t1; -- should succeed
+
+-- partitioned target
 CREATE POLICY p1 ON dob_t2 TO regress_rls_dob_role1,regress_rls_dob_role2 USING (true);
 DROP OWNED BY regress_rls_dob_role1;
 DROP POLICY p1 ON dob_t2; -- should succeed