Improve behavior of concurrent ANALYZE/VACUUM

From 3faa871f439bf0b9477c978557f1b90604ffa5d3 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Fri, 17 Aug 2018 15:24:30 +0900
Subject: [PATCH] Improve VACUUM and ANALYZE by avoiding early lock queue

A caller of VACUUM can perform early lookup obtention which can cause
other sessions to block on the request done, causing potentially DOS
attacks as even a non-privileged user can attempt a truncation of a
critical catalog table to block even all incoming connection attempts.

Contrary to TRUNCATE, a client could attempt a system-wide VACUUM after
building the list of relations to VACUUM, which can cause vacuum_rel()
to try to lock the relation but the thing would just lock.  When the
client specifies a list of relations and the relation needs to be
skipped, fail hard so as there is no conflict with any relation a user
has no rights to work on.

vacuum_rel() already had the sanity checks needed, except that those
were applied too late.  This commit refactors the code so as relation
skips are checked beforehand, making it safer to avoid too early lock,
for both manual VACUUM with and without a list of relations specified.

Reported-by: Lloyd Albin, Jeremy Schneider
Author: Michael Paquier
Reviewed by: Nathan Bossart, Kyotaro Horiguchi
Discussion: https://postgr.es/m/152512087100.19803.12733865831237526317@wrigleys.postgresql.org
---
 src/backend/commands/analyze.c                |  28 ++--
 src/backend/commands/vacuum.c                 | 152 +++++++++++++-----
 src/include/commands/vacuum.h                 |   3 +
 .../isolation/expected/vacuum-conflict.out    | 149 +++++++++++++++++
 src/test/isolation/isolation_schedule         |   1 +
 src/test/isolation/specs/vacuum-conflict.spec |  51 ++++++
 src/test/regress/expected/vacuum.out          |  31 ++++
 src/test/regress/sql/vacuum.sql               |  20 +++
 8 files changed, 379 insertions(+), 56 deletions(-)
 create mode 100644 src/test/isolation/expected/vacuum-conflict.out
 create mode 100644 src/test/isolation/specs/vacuum-conflict.spec

diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 3e148f03d0..4f0753e02a 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -196,27 +196,17 @@ analyze_rel(Oid relid, RangeVar *relation, int options,
 	}
 
 	/*
-	 * Check permissions --- this should match vacuum's check!
+	 * Check if relation needs to be skipped based on ownership.  This
+	 * check happens also when building the relation list to analyze
+	 * for a manual operation, and needs to be done additionally here
+	 * as ANALYZE could happen across multiple transactions where relation
+	 * ownership could have changed in-between.  Make sure to generate
+	 * only logs for ANALYZE in this case.
 	 */
-	if (!(pg_class_ownercheck(RelationGetRelid(onerel), GetUserId()) ||
-		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !onerel->rd_rel->relisshared)))
+	if (!vacuum_is_relation_owner(RelationGetRelid(onerel),
+								  onerel->rd_rel,
+								  options & VACOPT_ANALYZE))
 	{
-		/* No need for a WARNING if we already complained during VACUUM */
-		if (!(options & VACOPT_VACUUM))
-		{
-			if (onerel->rd_rel->relisshared)
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only superuser can analyze it",
-								RelationGetRelationName(onerel))));
-			else if (onerel->rd_rel->relnamespace == PG_CATALOG_NAMESPACE)
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only superuser or database owner can analyze it",
-								RelationGetRelationName(onerel))));
-			else
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only table or database owner can analyze it",
-								RelationGetRelationName(onerel))));
-		}
 		relation_close(onerel, ShareUpdateExclusiveLock);
 		return;
 	}
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index ee32fe8871..39816a742d 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -68,8 +68,8 @@ static BufferAccessStrategy vac_strategy;
 
 
 /* non-export function prototypes */
-static List *expand_vacuum_rel(VacuumRelation *vrel);
-static List *get_all_vacuum_rels(void);
+static List *expand_vacuum_rel(VacuumRelation *vrel, int options);
+static List *get_all_vacuum_rels(int options);
 static void vac_truncate_clog(TransactionId frozenXID,
 				  MultiXactId minMulti,
 				  TransactionId lastSaneFrozenXid,
@@ -257,7 +257,7 @@ vacuum(int options, List *relations, VacuumParams *params,
 			List	   *sublist;
 			MemoryContext old_context;
 
-			sublist = expand_vacuum_rel(vrel);
+			sublist = expand_vacuum_rel(vrel, options);
 			old_context = MemoryContextSwitchTo(vac_context);
 			newrels = list_concat(newrels, sublist);
 			MemoryContextSwitchTo(old_context);
@@ -265,7 +265,15 @@ vacuum(int options, List *relations, VacuumParams *params,
 		relations = newrels;
 	}
 	else
-		relations = get_all_vacuum_rels();
+		relations = get_all_vacuum_rels(options);
+
+	/*
+	 * Depending on the permission checks done while building the list
+	 * of relations to work on, it could be possible that the list is
+	 * empty, hence do nothing in this case.
+	 */
+	if (relations == NIL)
+		return;
 
 	/*
 	 * Decide whether we need to start/commit our own transactions.
@@ -408,6 +416,70 @@ vacuum(int options, List *relations, VacuumParams *params,
 	vac_context = NULL;
 }
 
+/*
+ * Check if a given relation can be safely vacuumed or not.  If the
+ * user is not the relation owner, issue a WARNING log message and return
+ * false to let the caller decide what to do with this relation.
+ */
+bool
+vacuum_is_relation_owner(Oid relid, Form_pg_class reltuple, int options)
+{
+	Assert((options & (VACOPT_VACUUM | VACOPT_ANALYZE)) != 0);
+
+	/*
+	 * Check permissions.
+	 *
+	 * We allow the user to vacuum a table if he is superuser, the table
+	 * owner, or the database owner (but in the latter case, only if it's not
+	 * a shared relation).  pg_class_ownercheck includes the superuser case.
+	 *
+	 * Note we choose to treat permissions failure as a WARNING and keep
+	 * trying to vacuum the rest of the DB --- is this appropriate?
+	 */
+	if (!(pg_class_ownercheck(relid, GetUserId()) ||
+		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !reltuple->relisshared)))
+	{
+		char		   *relname = NameStr(reltuple->relname);
+
+		if ((options & VACOPT_VACUUM) != 0)
+		{
+			if (reltuple->relisshared)
+				ereport(WARNING,
+						(errmsg("skipping \"%s\" --- only superuser can vacuum it",
+								relname)));
+			else if (reltuple->relnamespace == PG_CATALOG_NAMESPACE)
+				ereport(WARNING,
+						(errmsg("skipping \"%s\" --- only superuser or database owner can vacuum it",
+								relname)));
+			else
+				ereport(WARNING,
+						(errmsg("skipping \"%s\" --- only table or database owner can vacuum it",
+								relname)));
+		}
+
+		if ((options & VACOPT_ANALYZE) != 0)
+		{
+			if (reltuple->relisshared)
+				ereport(WARNING,
+						(errmsg("skipping \"%s\" --- only superuser can analyze it",
+								relname)));
+			else if (reltuple->relnamespace == PG_CATALOG_NAMESPACE)
+				ereport(WARNING,
+						(errmsg("skipping \"%s\" --- only superuser or database owner can analyze it",
+								relname)));
+			else
+				ereport(WARNING,
+						(errmsg("skipping \"%s\" --- only table or database owner can analyze it",
+								relname)));
+		}
+
+		return false;
+	}
+
+	return true;
+}
+
+
 /*
  * Given a VacuumRelation, fill in the table OID if it wasn't specified,
  * and optionally add VacuumRelations for partitions of the table.
@@ -423,7 +495,7 @@ vacuum(int options, List *relations, VacuumParams *params,
  * are made in vac_context.
  */
 static List *
-expand_vacuum_rel(VacuumRelation *vrel)
+expand_vacuum_rel(VacuumRelation *vrel, int options)
 {
 	List	   *vacrels = NIL;
 	MemoryContext oldcontext;
@@ -456,6 +528,28 @@ expand_vacuum_rel(VacuumRelation *vrel)
 		 */
 		relid = RangeVarGetRelid(vrel->relation, AccessShareLock, false);
 
+		/*
+		 * To check whether the relation is a partitioned table and its
+		 * ownership, fetch its syscache entry.
+		 */
+		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for relation %u", relid);
+		classForm = (Form_pg_class) GETSTRUCT(tuple);
+
+		/* check permissions of relation */
+		if (!vacuum_is_relation_owner(relid, classForm, options))
+		{
+			ReleaseSysCache(tuple);
+
+			/*
+			 * Release lock again with AccessShareLock -- see below for
+			 * the reason why this lock is released.
+			 */
+			UnlockRelationOid(relid, AccessShareLock);
+			return vacrels;
+		}
+
 		/*
 		 * Make a returnable VacuumRelation for this rel.
 		 */
@@ -465,14 +559,6 @@ expand_vacuum_rel(VacuumRelation *vrel)
 													  vrel->va_cols));
 		MemoryContextSwitchTo(oldcontext);
 
-		/*
-		 * To check whether the relation is a partitioned table, fetch its
-		 * syscache entry.
-		 */
-		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
-		if (!HeapTupleIsValid(tuple))
-			elog(ERROR, "cache lookup failed for relation %u", relid);
-		classForm = (Form_pg_class) GETSTRUCT(tuple);
 		include_parts = (classForm->relkind == RELKIND_PARTITIONED_TABLE);
 		ReleaseSysCache(tuple);
 
@@ -530,7 +616,7 @@ expand_vacuum_rel(VacuumRelation *vrel)
  * the current database.  The list is built in vac_context.
  */
 static List *
-get_all_vacuum_rels(void)
+get_all_vacuum_rels(int options)
 {
 	List	   *vacrels = NIL;
 	Relation	pgclass;
@@ -545,6 +631,11 @@ get_all_vacuum_rels(void)
 	{
 		Form_pg_class classForm = (Form_pg_class) GETSTRUCT(tuple);
 		MemoryContext oldcontext;
+		Oid			relid = HeapTupleGetOid(tuple);
+
+		/* check permissions of relation */
+		if (!vacuum_is_relation_owner(relid, classForm, options))
+			continue;
 
 		/*
 		 * We include partitioned tables here; depending on which operation is
@@ -563,7 +654,7 @@ get_all_vacuum_rels(void)
 		 */
 		oldcontext = MemoryContextSwitchTo(vac_context);
 		vacrels = lappend(vacrels, makeVacuumRelation(NULL,
-													  HeapTupleGetOid(tuple),
+													  relid,
 													  NIL));
 		MemoryContextSwitchTo(oldcontext);
 	}
@@ -1436,30 +1527,17 @@ vacuum_rel(Oid relid, RangeVar *relation, int options, VacuumParams *params)
 	}
 
 	/*
-	 * Check permissions.
-	 *
-	 * We allow the user to vacuum a table if he is superuser, the table
-	 * owner, or the database owner (but in the latter case, only if it's not
-	 * a shared relation).  pg_class_ownercheck includes the superuser case.
-	 *
-	 * Note we choose to treat permissions failure as a WARNING and keep
-	 * trying to vacuum the rest of the DB --- is this appropriate?
+	 * Check if relation needs to be skipped based on ownership.  This
+	 * check happens also when building the relation list to vacuum
+	 * for a manual operation, and needs to be done additionally here
+	 * as VACUUM could happen across multiple transactions where relation
+	 * ownership could have changed in-between.  Make sure to only generate
+	 * logs for VACUUM in this case.
 	 */
-	if (!(pg_class_ownercheck(RelationGetRelid(onerel), GetUserId()) ||
-		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !onerel->rd_rel->relisshared)))
+	if (!vacuum_is_relation_owner(RelationGetRelid(onerel),
+								  onerel->rd_rel,
+								  options & VACOPT_VACUUM))
 	{
-		if (onerel->rd_rel->relisshared)
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only superuser can vacuum it",
-							RelationGetRelationName(onerel))));
-		else if (onerel->rd_rel->relnamespace == PG_CATALOG_NAMESPACE)
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only superuser or database owner can vacuum it",
-							RelationGetRelationName(onerel))));
-		else
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only table or database owner can vacuum it",
-							RelationGetRelationName(onerel))));
 		relation_close(onerel, lmode);
 		PopActiveSnapshot();
 		CommitTransactionCommand();
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 85d472f0a5..85b181bf2f 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -15,6 +15,7 @@
 #define VACUUM_H
 
 #include "access/htup.h"
+#include "catalog/pg_class.h"
 #include "catalog/pg_statistic.h"
 #include "catalog/pg_type.h"
 #include "nodes/parsenodes.h"
@@ -185,6 +186,8 @@ extern void vacuum_set_xid_limits(Relation rel,
 					  MultiXactId *mxactFullScanLimit);
 extern void vac_update_datfrozenxid(void);
 extern void vacuum_delay_point(void);
+extern bool vacuum_is_relation_owner(Oid relid, Form_pg_class reltuple,
+									 int options);
 
 /* in commands/vacuumlazy.c */
 extern void lazy_vacuum_rel(Relation onerel, int options,
diff --git a/src/test/isolation/expected/vacuum-conflict.out b/src/test/isolation/expected/vacuum-conflict.out
new file mode 100644
index 0000000000..06ac75ef23
--- /dev/null
+++ b/src/test/isolation/expected/vacuum-conflict.out
@@ -0,0 +1,149 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1_begin s1_lock s2_auth s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s2_vacuum s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s1_lock s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_auth s2_vacuum s1_begin s1_lock s1_commit s2_reset
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s1_lock s2_auth s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s2_analyze s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s1_lock s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_auth s2_analyze s1_begin s1_lock s1_commit s2_reset
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s1_lock s2_auth s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_vacuum: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s2_vacuum s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s1_lock s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_vacuum: VACUUM vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_vacuum: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_grant s2_auth s2_vacuum s1_begin s1_lock s1_commit s2_reset
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s1_lock s2_auth s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_analyze: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s2_analyze s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s1_lock s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_analyze: ANALYZE vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_analyze: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_grant s2_auth s2_analyze s1_begin s1_lock s1_commit s2_reset
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 48ae740739..c23b401225 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -66,6 +66,7 @@ test: async-notify
 test: vacuum-reltuples
 test: timeouts
 test: vacuum-concurrent-drop
+test: vacuum-conflict
 test: predicate-hash
 test: predicate-gist
 test: predicate-gin
diff --git a/src/test/isolation/specs/vacuum-conflict.spec b/src/test/isolation/specs/vacuum-conflict.spec
new file mode 100644
index 0000000000..9b45d26c65
--- /dev/null
+++ b/src/test/isolation/specs/vacuum-conflict.spec
@@ -0,0 +1,51 @@
+# Tests for locking conflicts with VACUUM and ANALYZE commands.
+
+setup
+{
+	CREATE ROLE regress_vacuum_conflict;
+	CREATE TABLE vacuum_tab (a int);
+}
+
+teardown
+{
+	DROP TABLE vacuum_tab;
+	DROP ROLE regress_vacuum_conflict;
+}
+
+session "s1"
+step "s1_begin"          { BEGIN; }
+step "s1_lock"           { LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE; }
+step "s1_commit"         { COMMIT; }
+
+session "s2"
+step "s2_grant"          { ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict; }
+step "s2_auth"           { SET ROLE regress_vacuum_conflict; }
+step "s2_vacuum"         { VACUUM vacuum_tab; }
+step "s2_analyze"        { ANALYZE vacuum_tab; }
+step "s2_reset"          { RESET ROLE; }
+
+# The role doesn't have privileges to vacuum the table, so VACUUM should
+# immediately skip the table without waiting for a lock.
+permutation "s1_begin" "s1_lock" "s2_auth" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s2_vacuum" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s1_lock" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s2_auth" "s2_vacuum" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# Same as previously for ANALYZE
+permutation "s1_begin" "s1_lock" "s2_auth" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s2_analyze" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s1_lock" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s2_auth" "s2_analyze" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# The role has privileges to vacuum the table, VACUUM will block if
+# another session holds a lock on the table and succeed in all cases.
+permutation "s1_begin" "s2_grant" "s1_lock" "s2_auth" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s2_vacuum" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s1_lock" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s2_grant" "s2_auth" "s2_vacuum" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# Same as previously for ANALYZE
+permutation "s1_begin" "s2_grant" "s1_lock" "s2_auth" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s2_analyze" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s1_lock" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s2_grant" "s2_auth" "s2_analyze" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index d66e2aa3b7..6daee8e47d 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -122,3 +122,34 @@ LINE 1: ANALYZE (nonexistant-arg) does_not_exist;
 DROP TABLE vaccluster;
 DROP TABLE vactst;
 DROP TABLE vacparted;
+-- relation ownership, WARNING logs generated as all are skipped.
+CREATE TABLE vacowned (a int);
+CREATE ROLE regress_vacuum;
+SET ROLE regress_vacuum;
+-- Simple table
+VACUUM vacowned;
+WARNING:  skipping "vacowned" --- only table or database owner can vacuum it
+ANALYZE vacowned;
+WARNING:  skipping "vacowned" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacowned;
+WARNING:  skipping "vacowned" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned" --- only table or database owner can analyze it
+-- Catalog
+VACUUM pg_catalog.pg_class;
+WARNING:  skipping "pg_class" --- only superuser or database owner can vacuum it
+ANALYZE pg_catalog.pg_class;
+WARNING:  skipping "pg_class" --- only superuser or database owner can analyze it
+VACUUM (ANALYZE) pg_catalog.pg_class;
+WARNING:  skipping "pg_class" --- only superuser or database owner can vacuum it
+WARNING:  skipping "pg_class" --- only superuser or database owner can analyze it
+-- Shared catalog
+VACUUM pg_catalog.pg_authid;
+WARNING:  skipping "pg_authid" --- only superuser can vacuum it
+ANALYZE pg_catalog.pg_authid;
+WARNING:  skipping "pg_authid" --- only superuser can analyze it
+VACUUM (ANALYZE) pg_catalog.pg_authid;
+WARNING:  skipping "pg_authid" --- only superuser can vacuum it
+WARNING:  skipping "pg_authid" --- only superuser can analyze it
+RESET ROLE;
+DROP TABLE vacowned;
+DROP ROLE regress_vacuum;
diff --git a/src/test/regress/sql/vacuum.sql b/src/test/regress/sql/vacuum.sql
index 275ce2e270..0feff7c413 100644
--- a/src/test/regress/sql/vacuum.sql
+++ b/src/test/regress/sql/vacuum.sql
@@ -96,3 +96,23 @@ ANALYZE (nonexistant-arg) does_not_exist;
 DROP TABLE vaccluster;
 DROP TABLE vactst;
 DROP TABLE vacparted;
+
+-- relation ownership, WARNING logs generated as all are skipped.
+CREATE TABLE vacowned (a int);
+CREATE ROLE regress_vacuum;
+SET ROLE regress_vacuum;
+-- Simple table
+VACUUM vacowned;
+ANALYZE vacowned;
+VACUUM (ANALYZE) vacowned;
+-- Catalog
+VACUUM pg_catalog.pg_class;
+ANALYZE pg_catalog.pg_class;
+VACUUM (ANALYZE) pg_catalog.pg_class;
+-- Shared catalog
+VACUUM pg_catalog.pg_authid;
+ANALYZE pg_catalog.pg_authid;
+VACUUM (ANALYZE) pg_catalog.pg_authid;
+RESET ROLE;
+DROP TABLE vacowned;
+DROP ROLE regress_vacuum;
-- 
2.18.0

From 4215f41d38d8a3cbe1d52d97bf471d745783ac79 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Tue, 21 Aug 2018 10:11:45 +0900
Subject: [PATCH 1/2] Add regression tests for VACUUM and ANALYZE with relation
 skips

When a user does not have ownership on a relation, then specific log
messages are generated.  This new test suite adds coverage for all the
possible log messages generated.
---
 src/test/regress/expected/vacuum.out | 28 ++++++++++++++++++++++++++++
 src/test/regress/sql/vacuum.sql      | 20 ++++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index d66e2aa3b7..c9be71ef60 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -122,3 +122,31 @@ LINE 1: ANALYZE (nonexistant-arg) does_not_exist;
 DROP TABLE vaccluster;
 DROP TABLE vactst;
 DROP TABLE vacparted;
+-- relation ownership, WARNING logs generated as all are skipped.
+CREATE TABLE vacowned (a int);
+CREATE ROLE regress_vacuum;
+SET ROLE regress_vacuum;
+-- Simple table
+VACUUM vacowned;
+WARNING:  skipping "vacowned" --- only table or database owner can vacuum it
+ANALYZE vacowned;
+WARNING:  skipping "vacowned" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacowned;
+WARNING:  skipping "vacowned" --- only table or database owner can vacuum it
+-- Catalog
+VACUUM pg_catalog.pg_class;
+WARNING:  skipping "pg_class" --- only superuser or database owner can vacuum it
+ANALYZE pg_catalog.pg_class;
+WARNING:  skipping "pg_class" --- only superuser or database owner can analyze it
+VACUUM (ANALYZE) pg_catalog.pg_class;
+WARNING:  skipping "pg_class" --- only superuser or database owner can vacuum it
+-- Shared catalog
+VACUUM pg_catalog.pg_authid;
+WARNING:  skipping "pg_authid" --- only superuser can vacuum it
+ANALYZE pg_catalog.pg_authid;
+WARNING:  skipping "pg_authid" --- only superuser can analyze it
+VACUUM (ANALYZE) pg_catalog.pg_authid;
+WARNING:  skipping "pg_authid" --- only superuser can vacuum it
+RESET ROLE;
+DROP TABLE vacowned;
+DROP ROLE regress_vacuum;
diff --git a/src/test/regress/sql/vacuum.sql b/src/test/regress/sql/vacuum.sql
index 275ce2e270..0feff7c413 100644
--- a/src/test/regress/sql/vacuum.sql
+++ b/src/test/regress/sql/vacuum.sql
@@ -96,3 +96,23 @@ ANALYZE (nonexistant-arg) does_not_exist;
 DROP TABLE vaccluster;
 DROP TABLE vactst;
 DROP TABLE vacparted;
+
+-- relation ownership, WARNING logs generated as all are skipped.
+CREATE TABLE vacowned (a int);
+CREATE ROLE regress_vacuum;
+SET ROLE regress_vacuum;
+-- Simple table
+VACUUM vacowned;
+ANALYZE vacowned;
+VACUUM (ANALYZE) vacowned;
+-- Catalog
+VACUUM pg_catalog.pg_class;
+ANALYZE pg_catalog.pg_class;
+VACUUM (ANALYZE) pg_catalog.pg_class;
+-- Shared catalog
+VACUUM pg_catalog.pg_authid;
+ANALYZE pg_catalog.pg_authid;
+VACUUM (ANALYZE) pg_catalog.pg_authid;
+RESET ROLE;
+DROP TABLE vacowned;
+DROP ROLE regress_vacuum;
-- 
2.18.0

From f5b0f63c50cb8436bed093f280124fb2e2709af1 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Tue, 21 Aug 2018 10:25:05 +0900
Subject: [PATCH 2/2] Improve VACUUM and ANALYZE by avoiding early lock queue

A caller of VACUUM can perform early lookup obtention which can cause
other sessions to block on the request done, causing potentially DOS
attacks as even a non-privileged user can attempt a truncation of a
critical catalog table to block even all incoming connection attempts.

Contrary to TRUNCATE, a client could attempt a system-wide VACUUM after
building the list of relations to VACUUM, which can cause vacuum_rel()
to try to lock the relation but the thing would just lock.  When the
client specifies a list of relations and the relation needs to be
skipped, fail hard so as there is no conflict with any relation a user
has no rights to work on.

vacuum_rel() already had the sanity checks needed, except that those
were applied too late.  This commit refactors the code so as relation
skips are checked beforehand, making it safer to avoid too early lock,
for both manual VACUUM with and without a list of relations specified.

Reported-by: Lloyd Albin, Jeremy Schneider
Author: Michael Paquier
Reviewed by: Nathan Bossart, Kyotaro Horiguchi
Discussion: https://postgr.es/m/152512087100.19803.12733865831237526317@wrigleys.postgresql.org
---
 src/backend/commands/analyze.c                |  28 ++--
 src/backend/commands/vacuum.c                 | 153 +++++++++++++-----
 src/include/commands/vacuum.h                 |   3 +
 .../isolation/expected/vacuum-conflict.out    | 149 +++++++++++++++++
 src/test/isolation/isolation_schedule         |   1 +
 src/test/isolation/specs/vacuum-conflict.spec |  51 ++++++
 6 files changed, 329 insertions(+), 56 deletions(-)
 create mode 100644 src/test/isolation/expected/vacuum-conflict.out
 create mode 100644 src/test/isolation/specs/vacuum-conflict.spec

diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 3e148f03d0..4f0753e02a 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -196,27 +196,17 @@ analyze_rel(Oid relid, RangeVar *relation, int options,
 	}
 
 	/*
-	 * Check permissions --- this should match vacuum's check!
+	 * Check if relation needs to be skipped based on ownership.  This
+	 * check happens also when building the relation list to analyze
+	 * for a manual operation, and needs to be done additionally here
+	 * as ANALYZE could happen across multiple transactions where relation
+	 * ownership could have changed in-between.  Make sure to generate
+	 * only logs for ANALYZE in this case.
 	 */
-	if (!(pg_class_ownercheck(RelationGetRelid(onerel), GetUserId()) ||
-		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !onerel->rd_rel->relisshared)))
+	if (!vacuum_is_relation_owner(RelationGetRelid(onerel),
+								  onerel->rd_rel,
+								  options & VACOPT_ANALYZE))
 	{
-		/* No need for a WARNING if we already complained during VACUUM */
-		if (!(options & VACOPT_VACUUM))
-		{
-			if (onerel->rd_rel->relisshared)
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only superuser can analyze it",
-								RelationGetRelationName(onerel))));
-			else if (onerel->rd_rel->relnamespace == PG_CATALOG_NAMESPACE)
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only superuser or database owner can analyze it",
-								RelationGetRelationName(onerel))));
-			else
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only table or database owner can analyze it",
-								RelationGetRelationName(onerel))));
-		}
 		relation_close(onerel, ShareUpdateExclusiveLock);
 		return;
 	}
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index ee32fe8871..2cf2393ce1 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -68,8 +68,8 @@ static BufferAccessStrategy vac_strategy;
 
 
 /* non-export function prototypes */
-static List *expand_vacuum_rel(VacuumRelation *vrel);
-static List *get_all_vacuum_rels(void);
+static List *expand_vacuum_rel(VacuumRelation *vrel, int options);
+static List *get_all_vacuum_rels(int options);
 static void vac_truncate_clog(TransactionId frozenXID,
 				  MultiXactId minMulti,
 				  TransactionId lastSaneFrozenXid,
@@ -257,7 +257,7 @@ vacuum(int options, List *relations, VacuumParams *params,
 			List	   *sublist;
 			MemoryContext old_context;
 
-			sublist = expand_vacuum_rel(vrel);
+			sublist = expand_vacuum_rel(vrel, options);
 			old_context = MemoryContextSwitchTo(vac_context);
 			newrels = list_concat(newrels, sublist);
 			MemoryContextSwitchTo(old_context);
@@ -265,7 +265,7 @@ vacuum(int options, List *relations, VacuumParams *params,
 		relations = newrels;
 	}
 	else
-		relations = get_all_vacuum_rels();
+		relations = get_all_vacuum_rels(options);
 
 	/*
 	 * Decide whether we need to start/commit our own transactions.
@@ -408,6 +408,79 @@ vacuum(int options, List *relations, VacuumParams *params,
 	vac_context = NULL;
 }
 
+/*
+ * Check if a given relation can be safely vacuumed or analyzed.  If the
+ * user is not the relation owner, issue a WARNING log message and return
+ * false to let the caller decide what to do with this relation.  This
+ * routine is used for the decision-making of VACUUM and ANALYZE.
+ */
+bool
+vacuum_is_relation_owner(Oid relid, Form_pg_class reltuple, int options)
+{
+	char		   *relname;
+
+	Assert((options & (VACOPT_VACUUM | VACOPT_ANALYZE)) != 0);
+
+	/*
+	 * Check permissions.
+	 *
+	 * We allow the user to vacuum or analyze a table if he is superuser, the
+	 * table owner, or the database owner (but in the latter case, only if
+	 * it's not a shared relation).  pg_class_ownercheck includes the
+	 * superuser case.
+	 *
+	 * Note we choose to treat permissions failure as a WARNING and keep
+	 * trying to vacuum or analyze the rest of the DB --- is this appropriate?
+	 */
+	if (pg_class_ownercheck(relid, GetUserId()) ||
+		(pg_database_ownercheck(MyDatabaseId, GetUserId()) && !reltuple->relisshared))
+		return true;
+
+	relname = NameStr(reltuple->relname);
+
+	if ((options & VACOPT_VACUUM) != 0)
+	{
+		if (reltuple->relisshared)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser can vacuum it",
+							relname)));
+		else if (reltuple->relnamespace == PG_CATALOG_NAMESPACE)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser or database owner can vacuum it",
+							relname)));
+		else
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only table or database owner can vacuum it",
+							relname)));
+
+		/*
+		 * For VACUUM ANALYZE, both logs could show up, but just generate
+		 * information for VACUUM as that would be the first one to
+		 * process.
+		 */
+		return;
+	}
+
+	if ((options & VACOPT_ANALYZE) != 0)
+	{
+		if (reltuple->relisshared)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser can analyze it",
+							relname)));
+		else if (reltuple->relnamespace == PG_CATALOG_NAMESPACE)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser or database owner can analyze it",
+							relname)));
+		else
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only table or database owner can analyze it",
+							relname)));
+	}
+
+	return false;
+}
+
+
 /*
  * Given a VacuumRelation, fill in the table OID if it wasn't specified,
  * and optionally add VacuumRelations for partitions of the table.
@@ -423,7 +496,7 @@ vacuum(int options, List *relations, VacuumParams *params,
  * are made in vac_context.
  */
 static List *
-expand_vacuum_rel(VacuumRelation *vrel)
+expand_vacuum_rel(VacuumRelation *vrel, int options)
 {
 	List	   *vacrels = NIL;
 	MemoryContext oldcontext;
@@ -456,6 +529,28 @@ expand_vacuum_rel(VacuumRelation *vrel)
 		 */
 		relid = RangeVarGetRelid(vrel->relation, AccessShareLock, false);
 
+		/*
+		 * To check whether the relation is a partitioned table and its
+		 * ownership, fetch its syscache entry.
+		 */
+		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for relation %u", relid);
+		classForm = (Form_pg_class) GETSTRUCT(tuple);
+
+		/* check permissions of relation */
+		if (!vacuum_is_relation_owner(relid, classForm, options))
+		{
+			ReleaseSysCache(tuple);
+
+			/*
+			 * Release lock again with AccessShareLock -- see below for
+			 * the reason why this lock is released.
+			 */
+			UnlockRelationOid(relid, AccessShareLock);
+			return vacrels;
+		}
+
 		/*
 		 * Make a returnable VacuumRelation for this rel.
 		 */
@@ -465,14 +560,6 @@ expand_vacuum_rel(VacuumRelation *vrel)
 													  vrel->va_cols));
 		MemoryContextSwitchTo(oldcontext);
 
-		/*
-		 * To check whether the relation is a partitioned table, fetch its
-		 * syscache entry.
-		 */
-		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
-		if (!HeapTupleIsValid(tuple))
-			elog(ERROR, "cache lookup failed for relation %u", relid);
-		classForm = (Form_pg_class) GETSTRUCT(tuple);
 		include_parts = (classForm->relkind == RELKIND_PARTITIONED_TABLE);
 		ReleaseSysCache(tuple);
 
@@ -530,7 +617,7 @@ expand_vacuum_rel(VacuumRelation *vrel)
  * the current database.  The list is built in vac_context.
  */
 static List *
-get_all_vacuum_rels(void)
+get_all_vacuum_rels(int options)
 {
 	List	   *vacrels = NIL;
 	Relation	pgclass;
@@ -545,6 +632,11 @@ get_all_vacuum_rels(void)
 	{
 		Form_pg_class classForm = (Form_pg_class) GETSTRUCT(tuple);
 		MemoryContext oldcontext;
+		Oid			relid = HeapTupleGetOid(tuple);
+
+		/* check permissions of relation */
+		if (!vacuum_is_relation_owner(relid, classForm, options))
+			continue;
 
 		/*
 		 * We include partitioned tables here; depending on which operation is
@@ -563,7 +655,7 @@ get_all_vacuum_rels(void)
 		 */
 		oldcontext = MemoryContextSwitchTo(vac_context);
 		vacrels = lappend(vacrels, makeVacuumRelation(NULL,
-													  HeapTupleGetOid(tuple),
+													  relid,
 													  NIL));
 		MemoryContextSwitchTo(oldcontext);
 	}
@@ -1436,30 +1528,17 @@ vacuum_rel(Oid relid, RangeVar *relation, int options, VacuumParams *params)
 	}
 
 	/*
-	 * Check permissions.
-	 *
-	 * We allow the user to vacuum a table if he is superuser, the table
-	 * owner, or the database owner (but in the latter case, only if it's not
-	 * a shared relation).  pg_class_ownercheck includes the superuser case.
-	 *
-	 * Note we choose to treat permissions failure as a WARNING and keep
-	 * trying to vacuum the rest of the DB --- is this appropriate?
+	 * Check if relation needs to be skipped based on ownership.  This
+	 * check happens also when building the relation list to vacuum
+	 * for a manual operation, and needs to be done additionally here
+	 * as VACUUM could happen across multiple transactions where relation
+	 * ownership could have changed in-between.  Make sure to only generate
+	 * logs for VACUUM in this case.
 	 */
-	if (!(pg_class_ownercheck(RelationGetRelid(onerel), GetUserId()) ||
-		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !onerel->rd_rel->relisshared)))
+	if (!vacuum_is_relation_owner(RelationGetRelid(onerel),
+								  onerel->rd_rel,
+								  options & VACOPT_VACUUM))
 	{
-		if (onerel->rd_rel->relisshared)
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only superuser can vacuum it",
-							RelationGetRelationName(onerel))));
-		else if (onerel->rd_rel->relnamespace == PG_CATALOG_NAMESPACE)
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only superuser or database owner can vacuum it",
-							RelationGetRelationName(onerel))));
-		else
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only table or database owner can vacuum it",
-							RelationGetRelationName(onerel))));
 		relation_close(onerel, lmode);
 		PopActiveSnapshot();
 		CommitTransactionCommand();
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 85d472f0a5..85b181bf2f 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -15,6 +15,7 @@
 #define VACUUM_H
 
 #include "access/htup.h"
+#include "catalog/pg_class.h"
 #include "catalog/pg_statistic.h"
 #include "catalog/pg_type.h"
 #include "nodes/parsenodes.h"
@@ -185,6 +186,8 @@ extern void vacuum_set_xid_limits(Relation rel,
 					  MultiXactId *mxactFullScanLimit);
 extern void vac_update_datfrozenxid(void);
 extern void vacuum_delay_point(void);
+extern bool vacuum_is_relation_owner(Oid relid, Form_pg_class reltuple,
+									 int options);
 
 /* in commands/vacuumlazy.c */
 extern void lazy_vacuum_rel(Relation onerel, int options,
diff --git a/src/test/isolation/expected/vacuum-conflict.out b/src/test/isolation/expected/vacuum-conflict.out
new file mode 100644
index 0000000000..06ac75ef23
--- /dev/null
+++ b/src/test/isolation/expected/vacuum-conflict.out
@@ -0,0 +1,149 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1_begin s1_lock s2_auth s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s2_vacuum s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s1_lock s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_auth s2_vacuum s1_begin s1_lock s1_commit s2_reset
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s1_lock s2_auth s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s2_analyze s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s1_lock s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_auth s2_analyze s1_begin s1_lock s1_commit s2_reset
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s1_lock s2_auth s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_vacuum: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s2_vacuum s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s1_lock s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_vacuum: VACUUM vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_vacuum: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_grant s2_auth s2_vacuum s1_begin s1_lock s1_commit s2_reset
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s1_lock s2_auth s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_analyze: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s2_analyze s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s1_lock s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_analyze: ANALYZE vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_analyze: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_grant s2_auth s2_analyze s1_begin s1_lock s1_commit s2_reset
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 48ae740739..c23b401225 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -66,6 +66,7 @@ test: async-notify
 test: vacuum-reltuples
 test: timeouts
 test: vacuum-concurrent-drop
+test: vacuum-conflict
 test: predicate-hash
 test: predicate-gist
 test: predicate-gin
diff --git a/src/test/isolation/specs/vacuum-conflict.spec b/src/test/isolation/specs/vacuum-conflict.spec
new file mode 100644
index 0000000000..9b45d26c65
--- /dev/null
+++ b/src/test/isolation/specs/vacuum-conflict.spec
@@ -0,0 +1,51 @@
+# Tests for locking conflicts with VACUUM and ANALYZE commands.
+
+setup
+{
+	CREATE ROLE regress_vacuum_conflict;
+	CREATE TABLE vacuum_tab (a int);
+}
+
+teardown
+{
+	DROP TABLE vacuum_tab;
+	DROP ROLE regress_vacuum_conflict;
+}
+
+session "s1"
+step "s1_begin"          { BEGIN; }
+step "s1_lock"           { LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE; }
+step "s1_commit"         { COMMIT; }
+
+session "s2"
+step "s2_grant"          { ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict; }
+step "s2_auth"           { SET ROLE regress_vacuum_conflict; }
+step "s2_vacuum"         { VACUUM vacuum_tab; }
+step "s2_analyze"        { ANALYZE vacuum_tab; }
+step "s2_reset"          { RESET ROLE; }
+
+# The role doesn't have privileges to vacuum the table, so VACUUM should
+# immediately skip the table without waiting for a lock.
+permutation "s1_begin" "s1_lock" "s2_auth" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s2_vacuum" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s1_lock" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s2_auth" "s2_vacuum" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# Same as previously for ANALYZE
+permutation "s1_begin" "s1_lock" "s2_auth" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s2_analyze" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s1_lock" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s2_auth" "s2_analyze" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# The role has privileges to vacuum the table, VACUUM will block if
+# another session holds a lock on the table and succeed in all cases.
+permutation "s1_begin" "s2_grant" "s1_lock" "s2_auth" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s2_vacuum" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s1_lock" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s2_grant" "s2_auth" "s2_vacuum" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# Same as previously for ANALYZE
+permutation "s1_begin" "s2_grant" "s1_lock" "s2_auth" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s2_analyze" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s1_lock" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s2_grant" "s2_auth" "s2_analyze" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
-- 
2.18.0

diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index c9be71ef60..45e69573ce 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -119,9 +119,21 @@ ANALYZE (nonexistant-arg) does_not_exist;
 ERROR:  syntax error at or near "nonexistant"
 LINE 1: ANALYZE (nonexistant-arg) does_not_exist;
                  ^
+-- permission checks in vacuum_rel() and analyze_rel()
+CREATE ROLE vacparted_owner;
+ALTER TABLE vacparted OWNER TO vacparted_owner;
+SET ROLE vacparted_owner;
+VACUUM vacparted;
+WARNING:  skipping "vacparted1" --- only table or database owner can vacuum it
+ANALYZE vacparted;
+WARNING:  skipping "vacparted1" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacparted;
+WARNING:  skipping "vacparted1" --- only table or database owner can vacuum it
+RESET ROLE;
 DROP TABLE vaccluster;
 DROP TABLE vactst;
 DROP TABLE vacparted;
+DROP ROLE vacparted_owner;
 -- relation ownership, WARNING logs generated as all are skipped.
 CREATE TABLE vacowned (a int);
 CREATE ROLE regress_vacuum;
diff --git a/src/test/regress/sql/vacuum.sql b/src/test/regress/sql/vacuum.sql
index 0feff7c413..9e4d081da9 100644
--- a/src/test/regress/sql/vacuum.sql
+++ b/src/test/regress/sql/vacuum.sql
@@ -93,9 +93,19 @@ ANALYZE vactst (i), vacparted (does_not_exist);
 ANALYZE (VERBOSE) does_not_exist;
 ANALYZE (nonexistant-arg) does_not_exist;
 
+-- permission checks in vacuum_rel() and analyze_rel()
+CREATE ROLE vacparted_owner;
+ALTER TABLE vacparted OWNER TO vacparted_owner;
+SET ROLE vacparted_owner;
+VACUUM vacparted;
+ANALYZE vacparted;
+VACUUM (ANALYZE) vacparted;
+RESET ROLE;
+
 DROP TABLE vaccluster;
 DROP TABLE vactst;
 DROP TABLE vacparted;
+DROP ROLE vacparted_owner;
 
 -- relation ownership, WARNING logs generated as all are skipped.
 CREATE TABLE vacowned (a int);

diff --git a/src/test/regress/expected/vacuum.out b/src/test/regress/expected/vacuum.out
index c9be71ef60..57e451738c 100644
--- a/src/test/regress/expected/vacuum.out
+++ b/src/test/regress/expected/vacuum.out
@@ -124,6 +124,9 @@ DROP TABLE vactst;
 DROP TABLE vacparted;
 -- relation ownership, WARNING logs generated as all are skipped.
 CREATE TABLE vacowned (a int);
+CREATE TABLE vacowned_parted (a int) PARTITION BY LIST (a);
+CREATE TABLE vacowned_part1 PARTITION OF vacowned_parted FOR VALUES IN (1);
+CREATE TABLE vacowned_part2 PARTITION OF vacowned_parted FOR VALUES IN (2);
 CREATE ROLE regress_vacuum;
 SET ROLE regress_vacuum;
 -- Simple table
@@ -147,6 +150,102 @@ ANALYZE pg_catalog.pg_authid;
 WARNING:  skipping "pg_authid" --- only superuser can analyze it
 VACUUM (ANALYZE) pg_catalog.pg_authid;
 WARNING:  skipping "pg_authid" --- only superuser can vacuum it
+-- partitioned table and its partitions, no ownership.
+-- Relations are not listed in a single command to test ownership
+-- independently.
+VACUUM vacowned_parted;
+WARNING:  skipping "vacowned_parted" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM vacowned_part1;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+VACUUM vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+ANALYZE vacowned_parted;
+WARNING:  skipping "vacowned_parted" --- only table or database owner can analyze it
+WARNING:  skipping "vacowned_part1" --- only table or database owner can analyze it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+ANALYZE vacowned_part1;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can analyze it
+ANALYZE vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacowned_parted;
+WARNING:  skipping "vacowned_parted" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM (ANALYZE) vacowned_part1;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+VACUUM (ANALYZE) vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+RESET ROLE;
+-- Partitioned table and one partition owned by other user.
+ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
+ALTER TABLE vacowned_part1 OWNER TO regress_vacuum;
+SET ROLE regress_vacuum;
+VACUUM vacowned_parted;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM vacowned_part1;
+VACUUM vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+ANALYZE vacowned_parted;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+ANALYZE vacowned_part1;
+ANALYZE vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacowned_parted;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM (ANALYZE) vacowned_part1;
+VACUUM (ANALYZE) vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+RESET ROLE;
+-- Only one partition owned by other user
+ALTER TABLE vacowned_parted OWNER TO CURRENT_USER;
+SET ROLE regress_vacuum;
+VACUUM vacowned_parted;
+WARNING:  skipping "vacowned_parted" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM vacowned_part1;
+VACUUM vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+ANALYZE vacowned_parted;
+WARNING:  skipping "vacowned_parted" --- only table or database owner can analyze it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+ANALYZE vacowned_part1;
+ANALYZE vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacowned_parted;
+WARNING:  skipping "vacowned_parted" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM (ANALYZE) vacowned_part1;
+VACUUM (ANALYZE) vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+RESET ROLE;
+-- Only partitioned table owned by other user
+ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
+ALTER TABLE vacowned_part1 OWNER TO CURRENT_USER;
+SET ROLE regress_vacuum;
+VACUUM vacowned_parted;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM vacowned_part1;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+VACUUM vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+ANALYZE vacowned_parted;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can analyze it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+ANALYZE vacowned_part1;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can analyze it
+ANALYZE vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can analyze it
+VACUUM (ANALYZE) vacowned_parted;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
+VACUUM (ANALYZE) vacowned_part1;
+WARNING:  skipping "vacowned_part1" --- only table or database owner can vacuum it
+VACUUM (ANALYZE) vacowned_part2;
+WARNING:  skipping "vacowned_part2" --- only table or database owner can vacuum it
 RESET ROLE;
 DROP TABLE vacowned;
+DROP TABLE vacowned_parted;
 DROP ROLE regress_vacuum;
diff --git a/src/test/regress/sql/vacuum.sql b/src/test/regress/sql/vacuum.sql
index 0feff7c413..fc4c4a3bcc 100644
--- a/src/test/regress/sql/vacuum.sql
+++ b/src/test/regress/sql/vacuum.sql
@@ -99,6 +99,9 @@ DROP TABLE vacparted;
 
 -- relation ownership, WARNING logs generated as all are skipped.
 CREATE TABLE vacowned (a int);
+CREATE TABLE vacowned_parted (a int) PARTITION BY LIST (a);
+CREATE TABLE vacowned_part1 PARTITION OF vacowned_parted FOR VALUES IN (1);
+CREATE TABLE vacowned_part2 PARTITION OF vacowned_parted FOR VALUES IN (2);
 CREATE ROLE regress_vacuum;
 SET ROLE regress_vacuum;
 -- Simple table
@@ -113,6 +116,60 @@ VACUUM (ANALYZE) pg_catalog.pg_class;
 VACUUM pg_catalog.pg_authid;
 ANALYZE pg_catalog.pg_authid;
 VACUUM (ANALYZE) pg_catalog.pg_authid;
+-- partitioned table and its partitions, no ownership.
+-- Relations are not listed in a single command to test ownership
+-- independently.
+VACUUM vacowned_parted;
+VACUUM vacowned_part1;
+VACUUM vacowned_part2;
+ANALYZE vacowned_parted;
+ANALYZE vacowned_part1;
+ANALYZE vacowned_part2;
+VACUUM (ANALYZE) vacowned_parted;
+VACUUM (ANALYZE) vacowned_part1;
+VACUUM (ANALYZE) vacowned_part2;
+RESET ROLE;
+-- Partitioned table and one partition owned by other user.
+ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
+ALTER TABLE vacowned_part1 OWNER TO regress_vacuum;
+SET ROLE regress_vacuum;
+VACUUM vacowned_parted;
+VACUUM vacowned_part1;
+VACUUM vacowned_part2;
+ANALYZE vacowned_parted;
+ANALYZE vacowned_part1;
+ANALYZE vacowned_part2;
+VACUUM (ANALYZE) vacowned_parted;
+VACUUM (ANALYZE) vacowned_part1;
+VACUUM (ANALYZE) vacowned_part2;
+RESET ROLE;
+-- Only one partition owned by other user
+ALTER TABLE vacowned_parted OWNER TO CURRENT_USER;
+SET ROLE regress_vacuum;
+VACUUM vacowned_parted;
+VACUUM vacowned_part1;
+VACUUM vacowned_part2;
+ANALYZE vacowned_parted;
+ANALYZE vacowned_part1;
+ANALYZE vacowned_part2;
+VACUUM (ANALYZE) vacowned_parted;
+VACUUM (ANALYZE) vacowned_part1;
+VACUUM (ANALYZE) vacowned_part2;
+RESET ROLE;
+-- Only partitioned table owned by other user
+ALTER TABLE vacowned_parted OWNER TO regress_vacuum;
+ALTER TABLE vacowned_part1 OWNER TO CURRENT_USER;
+SET ROLE regress_vacuum;
+VACUUM vacowned_parted;
+VACUUM vacowned_part1;
+VACUUM vacowned_part2;
+ANALYZE vacowned_parted;
+ANALYZE vacowned_part1;
+ANALYZE vacowned_part2;
+VACUUM (ANALYZE) vacowned_parted;
+VACUUM (ANALYZE) vacowned_part1;
+VACUUM (ANALYZE) vacowned_part2;
 RESET ROLE;
 DROP TABLE vacowned;
+DROP TABLE vacowned_parted;
 DROP ROLE regress_vacuum;

From 4f0b6da8236286ab1487d2a537ae777aa3a6490b Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Fri, 24 Aug 2018 11:02:19 +0900
Subject: [PATCH] Improve VACUUM and ANALYZE by avoiding early lock queue

A caller of VACUUM can perform early lookup obtention which can cause
other sessions to block on the request done, causing potentially DOS
attacks as even a non-privileged user can attempt a truncation of a
critical catalog table to block even all incoming connection attempts.

Contrary to TRUNCATE, a client could attempt a system-wide VACUUM after
building the list of relations to VACUUM, which can cause vacuum_rel()
to try to lock the relation but the thing would just lock.  When the
client specifies a list of relations and the relation needs to be
skipped, fail hard so as there is no conflict with any relation a user
has no rights to work on.

vacuum_rel() already had the sanity checks needed, except that those
were applied too late.  This commit refactors the code so as relation
skips are checked beforehand, making it safer to avoid too early lock,
for both manual VACUUM with and without a list of relations specified.

An isolation test is added emulating the fact that early locks do not
happen anymore.

When a partitioned table is listed in a manual VACUUM or ANALYZE
command, its full list of partitions is fetched, all partition gets
added to the list to work on, and then each one of them is processed one
by one, with ownership checks happening at the later phase of
vacuum_rel() or analyze_rel().  Trying to do early ownership checks for
each partition is proving to be tedious as this would result in deadlock
risks with lock upgrades, and skipping all partitions if the listed
partitioned table is not owned would result in a behavior change
compared to how Postgres 10 has implemented vacuum for partitioned
tables.  The original problem reported related to early lock queue for
critical relations is fixed anyway, so priority is given to avoiding a
backward-incompatible behavior.

Reported-by: Lloyd Albin, Jeremy Schneider
Author: Michael Paquier
Reviewed by: Nathan Bossart, Kyotaro Horiguchi
Discussion: https://postgr.es/m/152512087100.19803.12733865831237526317@wrigleys.postgresql.org
---
 src/backend/commands/analyze.c                |  28 +---
 src/backend/commands/vacuum.c                 | 156 +++++++++++++-----
 src/include/commands/vacuum.h                 |   3 +
 .../isolation/expected/vacuum-conflict.out    | 149 +++++++++++++++++
 src/test/isolation/isolation_schedule         |   1 +
 src/test/isolation/specs/vacuum-conflict.spec |  51 ++++++
 6 files changed, 328 insertions(+), 60 deletions(-)
 create mode 100644 src/test/isolation/expected/vacuum-conflict.out
 create mode 100644 src/test/isolation/specs/vacuum-conflict.spec

diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 3e148f03d0..edbdce81f2 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -196,27 +196,17 @@ analyze_rel(Oid relid, RangeVar *relation, int options,
 	}
 
 	/*
-	 * Check permissions --- this should match vacuum's check!
+	 * Check if relation needs to be skipped based on ownership.  This check
+	 * happens also when building the relation list to analyze for a manual
+	 * operation, and needs to be done additionally here as ANALYZE could
+	 * happen across multiple transactions where relation ownership could have
+	 * changed in-between.  Make sure to generate only logs for ANALYZE in
+	 * this case.
 	 */
-	if (!(pg_class_ownercheck(RelationGetRelid(onerel), GetUserId()) ||
-		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !onerel->rd_rel->relisshared)))
+	if (!vacuum_is_relation_owner(RelationGetRelid(onerel),
+								  onerel->rd_rel,
+								  options & VACOPT_ANALYZE))
 	{
-		/* No need for a WARNING if we already complained during VACUUM */
-		if (!(options & VACOPT_VACUUM))
-		{
-			if (onerel->rd_rel->relisshared)
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only superuser can analyze it",
-								RelationGetRelationName(onerel))));
-			else if (onerel->rd_rel->relnamespace == PG_CATALOG_NAMESPACE)
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only superuser or database owner can analyze it",
-								RelationGetRelationName(onerel))));
-			else
-				ereport(WARNING,
-						(errmsg("skipping \"%s\" --- only table or database owner can analyze it",
-								RelationGetRelationName(onerel))));
-		}
 		relation_close(onerel, ShareUpdateExclusiveLock);
 		return;
 	}
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
index ee32fe8871..f166509734 100644
--- a/src/backend/commands/vacuum.c
+++ b/src/backend/commands/vacuum.c
@@ -68,8 +68,8 @@ static BufferAccessStrategy vac_strategy;
 
 
 /* non-export function prototypes */
-static List *expand_vacuum_rel(VacuumRelation *vrel);
-static List *get_all_vacuum_rels(void);
+static List *expand_vacuum_rel(VacuumRelation *vrel, int options);
+static List *get_all_vacuum_rels(int options);
 static void vac_truncate_clog(TransactionId frozenXID,
 				  MultiXactId minMulti,
 				  TransactionId lastSaneFrozenXid,
@@ -257,7 +257,7 @@ vacuum(int options, List *relations, VacuumParams *params,
 			List	   *sublist;
 			MemoryContext old_context;
 
-			sublist = expand_vacuum_rel(vrel);
+			sublist = expand_vacuum_rel(vrel, options);
 			old_context = MemoryContextSwitchTo(vac_context);
 			newrels = list_concat(newrels, sublist);
 			MemoryContextSwitchTo(old_context);
@@ -265,7 +265,7 @@ vacuum(int options, List *relations, VacuumParams *params,
 		relations = newrels;
 	}
 	else
-		relations = get_all_vacuum_rels();
+		relations = get_all_vacuum_rels(options);
 
 	/*
 	 * Decide whether we need to start/commit our own transactions.
@@ -408,6 +408,80 @@ vacuum(int options, List *relations, VacuumParams *params,
 	vac_context = NULL;
 }
 
+/*
+ * Check if a given relation can be safely vacuumed or analyzed.  If the
+ * user is not the relation owner, issue a WARNING log message and return
+ * false to let the caller decide what to do with this relation.  This
+ * routine is used to decide if a relation can be processed for VACUUM or
+ * ANALYZE.
+ */
+bool
+vacuum_is_relation_owner(Oid relid, Form_pg_class reltuple, int options)
+{
+	char	   *relname;
+
+	Assert((options & (VACOPT_VACUUM | VACOPT_ANALYZE)) != 0);
+
+	/*
+	 * Check permissions.
+	 *
+	 * We allow the user to vacuum or analyze a table if he is superuser, the
+	 * table owner, or the database owner (but in the latter case, only if
+	 * it's not a shared relation).  pg_class_ownercheck includes the
+	 * superuser case.
+	 *
+	 * Note we choose to treat permissions failure as a WARNING and keep
+	 * trying to vacuum or analyze the rest of the DB --- is this appropriate?
+	 */
+	if (pg_class_ownercheck(relid, GetUserId()) ||
+		(pg_database_ownercheck(MyDatabaseId, GetUserId()) && !reltuple->relisshared))
+		return true;
+
+	relname = NameStr(reltuple->relname);
+
+	if ((options & VACOPT_VACUUM) != 0)
+	{
+		if (reltuple->relisshared)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser can vacuum it",
+							relname)));
+		else if (reltuple->relnamespace == PG_CATALOG_NAMESPACE)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser or database owner can vacuum it",
+							relname)));
+		else
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only table or database owner can vacuum it",
+							relname)));
+
+		/*
+		 * For VACUUM ANALYZE, both logs could show up, but just generate
+		 * information for VACUUM as that would be the first one to be
+		 * processed.
+		 */
+		return false;
+	}
+
+	if ((options & VACOPT_ANALYZE) != 0)
+	{
+		if (reltuple->relisshared)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser can analyze it",
+							relname)));
+		else if (reltuple->relnamespace == PG_CATALOG_NAMESPACE)
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only superuser or database owner can analyze it",
+							relname)));
+		else
+			ereport(WARNING,
+					(errmsg("skipping \"%s\" --- only table or database owner can analyze it",
+							relname)));
+	}
+
+	return false;
+}
+
+
 /*
  * Given a VacuumRelation, fill in the table OID if it wasn't specified,
  * and optionally add VacuumRelations for partitions of the table.
@@ -423,7 +497,7 @@ vacuum(int options, List *relations, VacuumParams *params,
  * are made in vac_context.
  */
 static List *
-expand_vacuum_rel(VacuumRelation *vrel)
+expand_vacuum_rel(VacuumRelation *vrel, int options)
 {
 	List	   *vacrels = NIL;
 	MemoryContext oldcontext;
@@ -457,22 +531,28 @@ expand_vacuum_rel(VacuumRelation *vrel)
 		relid = RangeVarGetRelid(vrel->relation, AccessShareLock, false);
 
 		/*
-		 * Make a returnable VacuumRelation for this rel.
-		 */
-		oldcontext = MemoryContextSwitchTo(vac_context);
-		vacrels = lappend(vacrels, makeVacuumRelation(vrel->relation,
-													  relid,
-													  vrel->va_cols));
-		MemoryContextSwitchTo(oldcontext);
-
-		/*
-		 * To check whether the relation is a partitioned table, fetch its
-		 * syscache entry.
+		 * To check whether the relation is a partitioned table and its
+		 * ownership, fetch its syscache entry.
 		 */
 		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
 		if (!HeapTupleIsValid(tuple))
 			elog(ERROR, "cache lookup failed for relation %u", relid);
 		classForm = (Form_pg_class) GETSTRUCT(tuple);
+
+		/*
+		 * Make a returnable VacuumRelation for this rel if user is a proper
+		 * owner.
+		 */
+		if (vacuum_is_relation_owner(relid, classForm, options))
+		{
+			oldcontext = MemoryContextSwitchTo(vac_context);
+			vacrels = lappend(vacrels, makeVacuumRelation(vrel->relation,
+														  relid,
+														  vrel->va_cols));
+			MemoryContextSwitchTo(oldcontext);
+		}
+
+
 		include_parts = (classForm->relkind == RELKIND_PARTITIONED_TABLE);
 		ReleaseSysCache(tuple);
 
@@ -481,7 +561,9 @@ expand_vacuum_rel(VacuumRelation *vrel)
 		 * the list returned by find_all_inheritors() includes the passed-in
 		 * OID, so we have to skip that.  There's no point in taking locks on
 		 * the individual partitions yet, and doing so would just add
-		 * unnecessary deadlock risk.
+		 * unnecessary deadlock risk.  For this last reason we do not check
+		 * yet the ownership of the partitions, which get added to the list to
+		 * process.  Ownership will be checked later on anyway.
 		 */
 		if (include_parts)
 		{
@@ -530,7 +612,7 @@ expand_vacuum_rel(VacuumRelation *vrel)
  * the current database.  The list is built in vac_context.
  */
 static List *
-get_all_vacuum_rels(void)
+get_all_vacuum_rels(int options)
 {
 	List	   *vacrels = NIL;
 	Relation	pgclass;
@@ -545,6 +627,11 @@ get_all_vacuum_rels(void)
 	{
 		Form_pg_class classForm = (Form_pg_class) GETSTRUCT(tuple);
 		MemoryContext oldcontext;
+		Oid			relid = HeapTupleGetOid(tuple);
+
+		/* check permissions of relation */
+		if (!vacuum_is_relation_owner(relid, classForm, options))
+			continue;
 
 		/*
 		 * We include partitioned tables here; depending on which operation is
@@ -563,7 +650,7 @@ get_all_vacuum_rels(void)
 		 */
 		oldcontext = MemoryContextSwitchTo(vac_context);
 		vacrels = lappend(vacrels, makeVacuumRelation(NULL,
-													  HeapTupleGetOid(tuple),
+													  relid,
 													  NIL));
 		MemoryContextSwitchTo(oldcontext);
 	}
@@ -1436,30 +1523,17 @@ vacuum_rel(Oid relid, RangeVar *relation, int options, VacuumParams *params)
 	}
 
 	/*
-	 * Check permissions.
-	 *
-	 * We allow the user to vacuum a table if he is superuser, the table
-	 * owner, or the database owner (but in the latter case, only if it's not
-	 * a shared relation).  pg_class_ownercheck includes the superuser case.
-	 *
-	 * Note we choose to treat permissions failure as a WARNING and keep
-	 * trying to vacuum the rest of the DB --- is this appropriate?
+	 * Check if relation needs to be skipped based on ownership.  This check
+	 * happens also when building the relation list to vacuum for a manual
+	 * operation, and needs to be done additionally here as VACUUM could
+	 * happen across multiple transactions where relation ownership could have
+	 * changed in-between.  Make sure to only generate logs for VACUUM in this
+	 * case.
 	 */
-	if (!(pg_class_ownercheck(RelationGetRelid(onerel), GetUserId()) ||
-		  (pg_database_ownercheck(MyDatabaseId, GetUserId()) && !onerel->rd_rel->relisshared)))
+	if (!vacuum_is_relation_owner(RelationGetRelid(onerel),
+								  onerel->rd_rel,
+								  options & VACOPT_VACUUM))
 	{
-		if (onerel->rd_rel->relisshared)
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only superuser can vacuum it",
-							RelationGetRelationName(onerel))));
-		else if (onerel->rd_rel->relnamespace == PG_CATALOG_NAMESPACE)
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only superuser or database owner can vacuum it",
-							RelationGetRelationName(onerel))));
-		else
-			ereport(WARNING,
-					(errmsg("skipping \"%s\" --- only table or database owner can vacuum it",
-							RelationGetRelationName(onerel))));
 		relation_close(onerel, lmode);
 		PopActiveSnapshot();
 		CommitTransactionCommand();
diff --git a/src/include/commands/vacuum.h b/src/include/commands/vacuum.h
index 85d472f0a5..5af96fdc8a 100644
--- a/src/include/commands/vacuum.h
+++ b/src/include/commands/vacuum.h
@@ -15,6 +15,7 @@
 #define VACUUM_H
 
 #include "access/htup.h"
+#include "catalog/pg_class.h"
 #include "catalog/pg_statistic.h"
 #include "catalog/pg_type.h"
 #include "nodes/parsenodes.h"
@@ -185,6 +186,8 @@ extern void vacuum_set_xid_limits(Relation rel,
 					  MultiXactId *mxactFullScanLimit);
 extern void vac_update_datfrozenxid(void);
 extern void vacuum_delay_point(void);
+extern bool vacuum_is_relation_owner(Oid relid, Form_pg_class reltuple,
+						 int options);
 
 /* in commands/vacuumlazy.c */
 extern void lazy_vacuum_rel(Relation onerel, int options,
diff --git a/src/test/isolation/expected/vacuum-conflict.out b/src/test/isolation/expected/vacuum-conflict.out
new file mode 100644
index 0000000000..06ac75ef23
--- /dev/null
+++ b/src/test/isolation/expected/vacuum-conflict.out
@@ -0,0 +1,149 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s1_begin s1_lock s2_auth s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s2_vacuum s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s1_lock s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_auth s2_vacuum s1_begin s1_lock s1_commit s2_reset
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can vacuum it
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s1_lock s2_auth s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s2_analyze s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_auth s1_lock s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_auth s2_analyze s1_begin s1_lock s1_commit s2_reset
+step s2_auth: SET ROLE regress_vacuum_conflict;
+WARNING:  skipping "vacuum_tab" --- only table or database owner can analyze it
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s1_lock s2_auth s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_vacuum: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s2_vacuum s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s1_lock s2_vacuum s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_vacuum: VACUUM vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_vacuum: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_grant s2_auth s2_vacuum s1_begin s1_lock s1_commit s2_reset
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_vacuum: VACUUM vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s1_lock s2_auth s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_analyze: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s2_analyze s1_lock s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
+
+starting permutation: s1_begin s2_grant s2_auth s1_lock s2_analyze s1_commit s2_reset
+step s1_begin: BEGIN;
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s2_analyze: ANALYZE vacuum_tab; <waiting ...>
+step s1_commit: COMMIT;
+step s2_analyze: <... completed>
+step s2_reset: RESET ROLE;
+
+starting permutation: s2_grant s2_auth s2_analyze s1_begin s1_lock s1_commit s2_reset
+step s2_grant: ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict;
+step s2_auth: SET ROLE regress_vacuum_conflict;
+step s2_analyze: ANALYZE vacuum_tab;
+step s1_begin: BEGIN;
+step s1_lock: LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE;
+step s1_commit: COMMIT;
+step s2_reset: RESET ROLE;
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 48ae740739..c23b401225 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -66,6 +66,7 @@ test: async-notify
 test: vacuum-reltuples
 test: timeouts
 test: vacuum-concurrent-drop
+test: vacuum-conflict
 test: predicate-hash
 test: predicate-gist
 test: predicate-gin
diff --git a/src/test/isolation/specs/vacuum-conflict.spec b/src/test/isolation/specs/vacuum-conflict.spec
new file mode 100644
index 0000000000..9b45d26c65
--- /dev/null
+++ b/src/test/isolation/specs/vacuum-conflict.spec
@@ -0,0 +1,51 @@
+# Tests for locking conflicts with VACUUM and ANALYZE commands.
+
+setup
+{
+	CREATE ROLE regress_vacuum_conflict;
+	CREATE TABLE vacuum_tab (a int);
+}
+
+teardown
+{
+	DROP TABLE vacuum_tab;
+	DROP ROLE regress_vacuum_conflict;
+}
+
+session "s1"
+step "s1_begin"          { BEGIN; }
+step "s1_lock"           { LOCK vacuum_tab IN SHARE UPDATE EXCLUSIVE MODE; }
+step "s1_commit"         { COMMIT; }
+
+session "s2"
+step "s2_grant"          { ALTER TABLE vacuum_tab OWNER TO regress_vacuum_conflict; }
+step "s2_auth"           { SET ROLE regress_vacuum_conflict; }
+step "s2_vacuum"         { VACUUM vacuum_tab; }
+step "s2_analyze"        { ANALYZE vacuum_tab; }
+step "s2_reset"          { RESET ROLE; }
+
+# The role doesn't have privileges to vacuum the table, so VACUUM should
+# immediately skip the table without waiting for a lock.
+permutation "s1_begin" "s1_lock" "s2_auth" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s2_vacuum" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s1_lock" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s2_auth" "s2_vacuum" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# Same as previously for ANALYZE
+permutation "s1_begin" "s1_lock" "s2_auth" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s2_analyze" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_auth" "s1_lock" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s2_auth" "s2_analyze" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# The role has privileges to vacuum the table, VACUUM will block if
+# another session holds a lock on the table and succeed in all cases.
+permutation "s1_begin" "s2_grant" "s1_lock" "s2_auth" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s2_vacuum" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s1_lock" "s2_vacuum" "s1_commit" "s2_reset"
+permutation "s2_grant" "s2_auth" "s2_vacuum" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
+
+# Same as previously for ANALYZE
+permutation "s1_begin" "s2_grant" "s1_lock" "s2_auth" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s2_analyze" "s1_lock" "s1_commit" "s2_reset"
+permutation "s1_begin" "s2_grant" "s2_auth" "s1_lock" "s2_analyze" "s1_commit" "s2_reset"
+permutation "s2_grant" "s2_auth" "s2_analyze" "s1_begin" "s1_lock" "s1_commit" "s2_reset"
-- 
2.18.0