Add locking clause for SB views for update/delete

In expand_security_qual(), we were handling locking correctly when a
PlanRowMark existed, but not when we were working with the target
relation (which doesn't have any PlanRowMarks, but the subquery created
for the security barrier quals still needs to lock the rows under it).

Noted by Etsuro Fujita when working with the Postgres FDW, which wasn't
properly issuing a SELECT ... FOR UPDATE to the remote side under a
DELETE.

Back-patch to 9.4 where updatable security barrier views were
introduced.

Per discussion with Etsuro and Dean Rasheed.

src/backend/optimizer/prep/prepsecurity.c

@@ -37,7 +37,7 @@ typedef struct
 } security_barrier_replace_vars_context;
 
 static void expand_security_qual(PlannerInfo *root, List *tlist, int rt_index,
-					 RangeTblEntry *rte, Node *qual);
+					 RangeTblEntry *rte, Node *qual, bool targetRelation);
 
 static void security_barrier_replace_vars(Node *node,
 							  security_barrier_replace_vars_context *context);
@@ -63,6 +63,7 @@ expand_security_quals(PlannerInfo *root, List *tlist)
 	Query	   *parse = root->parse;
 	int			rt_index;
 	ListCell   *cell;
+	bool		targetRelation = false;
 
 	/*
 	 * Process each RTE in the rtable list.
@@ -98,6 +99,15 @@ expand_security_quals(PlannerInfo *root, List *tlist)
 		{
 			RangeTblEntry *newrte = copyObject(rte);
 
+			/*
+			 * We need to let expand_security_qual know if this is the target
+			 * relation, as it has additional work to do in that case.
+			 *
+			 * Capture that information here as we're about to replace
+			 * parse->resultRelation.
+			 */
+			targetRelation = true;
+
 			parse->rtable = lappend(parse->rtable, newrte);
 			parse->resultRelation = list_length(parse->rtable);
 
@@ -147,7 +157,8 @@ expand_security_quals(PlannerInfo *root, List *tlist)
 			rte->securityQuals = list_delete_first(rte->securityQuals);
 
 			ChangeVarNodes(qual, rt_index, 1, 0);
-			expand_security_qual(root, tlist, rt_index, rte, qual);
+			expand_security_qual(root, tlist, rt_index, rte, qual,
+								 targetRelation);
 		}
 	}
 }
@@ -160,7 +171,7 @@ expand_security_quals(PlannerInfo *root, List *tlist)
  */
 static void
 expand_security_qual(PlannerInfo *root, List *tlist, int rt_index,
-					 RangeTblEntry *rte, Node *qual)
+					 RangeTblEntry *rte, Node *qual, bool targetRelation)
 {
 	Query	   *parse = root->parse;
 	Oid			relid = rte->relid;
@@ -219,10 +230,11 @@ expand_security_qual(PlannerInfo *root, List *tlist, int rt_index,
 			 * Now deal with any PlanRowMark on this RTE by requesting a lock
 			 * of the same strength on the RTE copied down to the subquery.
 			 *
-			 * Note that we can't push the user-defined quals down since they
-			 * may included untrusted functions and that means that we will
-			 * end up locking all rows which pass the securityQuals, even if
-			 * those rows don't pass the user-defined quals.  This is
+			 * Note that we can only push down user-defined quals if they are
+			 * only using leakproof (and therefore trusted) functions and
+			 * operators.  As a result, we may end up locking more rows than
+			 * strictly necessary (and, in the worst case, we could end up
+			 * locking all rows which pass the securityQuals).  This is
 			 * currently documented behavior, but it'd be nice to come up with
 			 * a better solution some day.
 			 */
@@ -255,6 +267,15 @@ expand_security_qual(PlannerInfo *root, List *tlist, int rt_index,
 				root->rowMarks = list_delete(root->rowMarks, rc);
 			}
 
+			/*
+			 * When we are replacing the target relation with a subquery, we
+			 * need to make sure to add a locking clause explicitly to the
+			 * generated subquery since there won't be any row marks against
+			 * the target relation itself.
+			 */
+			if (targetRelation)
+				applyLockingClause(subquery, 1, LCS_FORUPDATE,
+								   LockWaitBlock, false);
 			/*
 			 * Replace any variables in the outer query that refer to the
 			 * original relation RTE with references to columns that we will

src/test/regress/expected/rowsecurity.out

@@ -1034,22 +1034,25 @@ EXPLAIN (COSTS OFF) EXECUTE p2(2);
 --
 SET SESSION AUTHORIZATION rls_regress_user1;
 EXPLAIN (COSTS OFF) UPDATE t1 SET b = b || b WHERE f_leak(b);
-             QUERY PLAN              
--------------------------------------
+                QUERY PLAN                 
+-------------------------------------------
  Update on t1 t1_3
    ->  Subquery Scan on t1
          Filter: f_leak(t1.b)
-         ->  Seq Scan on t1 t1_4
-               Filter: ((a % 2) = 0)
+         ->  LockRows
+               ->  Seq Scan on t1 t1_4
+                     Filter: ((a % 2) = 0)
    ->  Subquery Scan on t1_1
          Filter: f_leak(t1_1.b)
-         ->  Seq Scan on t2
-               Filter: ((a % 2) = 0)
+         ->  LockRows
+               ->  Seq Scan on t2
+                     Filter: ((a % 2) = 0)
    ->  Subquery Scan on t1_2
          Filter: f_leak(t1_2.b)
-         ->  Seq Scan on t3
-               Filter: ((a % 2) = 0)
-(13 rows)
+         ->  LockRows
+               ->  Seq Scan on t3
+                     Filter: ((a % 2) = 0)
+(16 rows)
 
 UPDATE t1 SET b = b || b WHERE f_leak(b);
 NOTICE:  f_leak => bbb
@@ -1058,14 +1061,15 @@ NOTICE:  f_leak => bcd
 NOTICE:  f_leak => def
 NOTICE:  f_leak => yyy
 EXPLAIN (COSTS OFF) UPDATE only t1 SET b = b || '_updt' WHERE f_leak(b);
-             QUERY PLAN              
--------------------------------------
+                QUERY PLAN                 
+-------------------------------------------
  Update on t1 t1_1
    ->  Subquery Scan on t1
          Filter: f_leak(t1.b)
-         ->  Seq Scan on t1 t1_2
-               Filter: ((a % 2) = 0)
-(5 rows)
+         ->  LockRows
+               ->  Seq Scan on t1 t1_2
+                     Filter: ((a % 2) = 0)
+(6 rows)
 
 UPDATE only t1 SET b = b || '_updt' WHERE f_leak(b);
 NOTICE:  f_leak => bbbbbb
@@ -1131,32 +1135,36 @@ SELECT * FROM t1;
 SET SESSION AUTHORIZATION rls_regress_user1;
 SET row_security TO ON;
 EXPLAIN (COSTS OFF) DELETE FROM only t1 WHERE f_leak(b);
-             QUERY PLAN              
--------------------------------------
+                QUERY PLAN                 
+-------------------------------------------
  Delete on t1 t1_1
    ->  Subquery Scan on t1
          Filter: f_leak(t1.b)
-         ->  Seq Scan on t1 t1_2
-               Filter: ((a % 2) = 0)
-(5 rows)
+         ->  LockRows
+               ->  Seq Scan on t1 t1_2
+                     Filter: ((a % 2) = 0)
+(6 rows)
 
 EXPLAIN (COSTS OFF) DELETE FROM t1 WHERE f_leak(b);
-             QUERY PLAN              
--------------------------------------
+                QUERY PLAN                 
+-------------------------------------------
  Delete on t1 t1_3
    ->  Subquery Scan on t1
          Filter: f_leak(t1.b)
-         ->  Seq Scan on t1 t1_4
-               Filter: ((a % 2) = 0)
+         ->  LockRows
+               ->  Seq Scan on t1 t1_4
+                     Filter: ((a % 2) = 0)
    ->  Subquery Scan on t1_1
          Filter: f_leak(t1_1.b)
-         ->  Seq Scan on t2
-               Filter: ((a % 2) = 0)
+         ->  LockRows
+               ->  Seq Scan on t2
+                     Filter: ((a % 2) = 0)
    ->  Subquery Scan on t1_2
          Filter: f_leak(t1_2.b)
-         ->  Seq Scan on t3
-               Filter: ((a % 2) = 0)
-(13 rows)
+         ->  LockRows
+               ->  Seq Scan on t3
+                     Filter: ((a % 2) = 0)
+(16 rows)
 
 DELETE FROM only t1 WHERE f_leak(b) RETURNING oid, *, t1;
 NOTICE:  f_leak => bbbbbb_updt