pg.ddx.io  pgsql-hackers@postgresql.org mailing list archive  

help / color / mirror / Atom feed

From: Mark Dilger <mark.dilger@enterprisedb.com>
To: Robert Haas <robertmhaas@gmail.com>
Cc: Thomas Munro <thomas.munro@gmail.com>
Cc: Tom Lane <tgl@sss.pgh.pa.us>
Cc: Peter Geoghegan <pg@bowt.ie>
Cc: Andrey M. Borodin <x4mmm@yandex-team.ru>
Cc: Stephen Frost <sfrost@snowman.net>
Cc: Michael Paquier <michael@paquier.xyz>
Cc: Amul Sul <sulamul@gmail.com>
Cc: Dilip Kumar <dilipbalaut@gmail.com>
Cc: Andres Freund <andres@anarazel.de>
Cc: PostgreSQL-development <pgsql-hackers@postgresql.org>
Subject: Re: new heapcheck contrib module
Date: Tue, 2 Feb 2021 15:10:34 -0800
Message-ID: <70655DF3-33CE-4527-9A4D-DDEB582B6BA0@enterprisedb.com> (raw)
In-Reply-To: <70A7BABE-EDC1-4217-8FEB-F446F2938CE1@enterprisedb.com>
References: <12ED3DA8-25F0-4B68-937D-D907CFBF08E7@enterprisedb.com>
	<20200420194210.sevnpi4lyghoanxi@alap3.anarazel.de>
	<23E60C3C-C20A-4508-84AC-E0CEA59E784A@enterprisedb.com>
	<CA+TgmobAm8vB1MNQ1taO1et+HxTN8ULnQvEG+grNDqNQTBjbtA@mail.gmail.com>
	<F845AC50-A479-43F3-ADFB-861A4D2AD287@enterprisedb.com>
	<5F743835-3399-419C-8324-2D424237E999@enterprisedb.com>
	<CAFiTN-syyHc3jZoou51v0SR8z0POoNfktqEO6MaGig4YS8mosA@mail.gmail.com>
	<E6F4258A-B661-4F2C-8A35-278C9436930A@enterprisedb.com>
	<CAFiTN-s-VX3DLfqv7OOo0zGCTOZ3xVU9UZ6GgzBZdJTPNnYFNg@mail.gmail.com>
	<885D2EC9-7D06-4903-A17D-72C184A1E1C5@enterprisedb.com>
	<CAFiTN-uYg_PiZ854=5R-D9skratapAHKhaDAG7fhJv-Vn5UhaQ@mail.gmail.com>
	<5CC85F28-F210-47AC-8DA8-EC4E82AF3FD5@enterprisedb.com>
	<CAFiTN-sP5iFJGERTq4a4OLTCzjZKhLC5P6_k2bxQKXnuvdmzGA@mail.gmail.com>
	<CAFiTN-tTg0R5AHvXC1ro9XphroX1wUtc5t91hU_+7DeGTpE5VQ@mail.gmail.com>
	<0C0349EB-DD85-4DA8-82C7-DF70AB714592@enterprisedb.com>
	<BA6E870F-241C-4DDB-B92D-DB39E679C7D4@enterprisedb.com>
	<C4A3E5AE-CBB2-4E3F-8BAE-29EE216762C2@enterprisedb.com>
	<CAAJ_b96_Xr2a+D0j8xKbOnCzURcVQm0V2DS+T_pfc+wt3BpL8g@mail.gmail.com>
	<9DF93B83-FE46-4EC4-A99D-77A60729FD2D@enterprisedb.com>
	<CA+TgmoZT3kCOaKq6=m5wHGKh_9XQDbCC1LZXL6w_+NMB-n1+mw@mail.gmail.com>
	<CAAJ_b94hsrF4x7UKQj03416fAEXJorUpZ9WhsZZjqgv-4+Wg9A@mail.gmail.com>
	<50F6468F-5A5E-412A-86EC-DF24B549D733@enterprisedb.com>
	<CAAJ_b94+A1Gk5r5sBYbYM2K5hH_gyp_o=+gwJuJhn5ZEm5_rqw@mail.gmail.com>
	<CAAJ_b94Wh_7kM9JuJk7wnNsYSNPdSzz35cOEaE_ZRnPcqvupCg@mail.gmail.com>
	<7D603B53-8078-4AF0-851B-803FF4416501@enterprisedb.com>
	<CA+TgmoZbPtTgkQBGoco39uqKUWuv=0+TOkiNcCuqWWvUkJVuxQ@mail.gmail.com>
	<A4C9E0AE-46BA-457F-A445-4527E5AA5E3C@enterprisedb.com>
	<98265AA9-7BDC-4708-9348-CC1B173BBE1D@enterprisedb.com>
	<CA+Tgmoam+pnGKhWS+RAYCk1eqjhbAPX7Vm69g1VF56xwyxXw+g@mail.gmail.com>
	<209783.1603313248@sss.pgh.pa.us>
	<77A60E99-6054-40FD-A5D6-769C287D4A0F@enterprisedb.com>
	<CA+TgmobGxdm8Tsxawt7LW-qigHRHxD+b17PC+ft9EE+6pMhC9w@mail.gmail.com>
	<721BFA83-EA6F-463D-A9DF-CE78A318B73A@enterprisedb.com>
	<2A7DA1A8-C4AA-43DF-A985-3CA52F4DC775@enterprisedb.com>
	<146814.1603463293@sss.pgh.pa.us>
	<221029.1603476265@sss.pgh.pa.us>
	<CA+TgmoZsuhZroXdwCNUb5+9QEWSFQMH9=aA1hx-SDeArppH76Q@mail.gmail.com>
	<C001DE91-CF4A-44A7-85B6-8BBAB1881196@enterprisedb.com>
	<CA+TgmoZJ53=joSyJuR7WmXEOfhbBK+0hQBrOO4mWftwekaHS6w@mail.gmail.com>
	<153EEBAD-2673-4367-B36D-4A6AE8EE7A9F@enterprisedb.com>
	<85BE72E5-A926-4F23-9FD5-64239C0C6BA5@enterprisedb.com>
	<CA+Tgmoa7sNcw15=soG1pLBgByMnn-w70YdfKZmhmYOb43KzzRw@mail.gmail.com>
	<B5CF5BFF-0A92-4792-868E-3DB4E8F1C389@enterprisedb.com>
	<10677BFE-3B45-4EF3-82E5-E45CA1445CA6@enterprisedb.com>
	<CA+hUKG+mX5sb6=ZG9=zHaYofPeLOxO910XHkahcDgP-XzEMLxQ@mail.gmail.com>
	<A4C2DFB8-436F-4802-BE8D-7A30290D52FA@enterprisedb.com>
	<CA+TgmoY=+bKef-MUb1nKnyL8nV_26cn7NJ8q50AkGg0fV_v_XA@mail.gmail.com>
	<411C72E5-23A6-485F-A101-FBF7CFA6D2D5@enterprisedb.com>
	<CA+Tgmoa_SfoePsmXfj9FkDFqOCdJyb8XpXUzDo7GahuyPrPnpA@mail.gmail.com>
	<70A7BABE-EDC1-4217-8FEB-F446F2938CE1@enterprisedb.com>



> On Jan 31, 2021, at 4:05 PM, Mark Dilger <mark.dilger@enterprisedb.com> wrote:
> 
> Attached is patch set 35.

I found some things to improve in the v35 patch set.  Please find attached the v36 patch set, which differs from v35 in the following ways:

0001 -- no changes

0002 -- fixing omissions in @pgfeutilsfiles in file src/tools/msvc/Mkvcbuild.pm

0003 -- no changes

0004:
  -- Fixes handling of amcheck contrib module installed in non-default schema.
  -- Adds database name to corruption messages to make identifying the relation being complained about unambiguous in multi-database checks
  -- Fixes an instance where pg_amcheck was querying pg_database without schema-qualifying it
  -- Simplifying some functions in pg_amcheck.c
  -- Updating a comment to reflect the renaming of a variable that the comment mentioned by name

0005 -- fixes =pod added in PostgresNode.pm.  The =pod was grammatically correct so far I can tell, but rendered strangely in perldoc.


—
Mark Dilger
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Attachments: download all (5 files, .tar.gz)

v36-0001-Refactoring-processSQLNamePattern.patch (application/octet-stream, 10.7K) download
From 5375f952570ff75a56e916de13e63c43c3ba2604 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Tue, 2 Feb 2021 12:29:11 -0800
Subject: [PATCH v36 1/5] Refactoring processSQLNamePattern.

Factoring out logic which transforms shell-style patterns into SQL
style regexp format from inside processSQLNamePattern into a
separate new function "patternToSQLRegex".  The interface and
semantics of processSQLNamePattern are unchanged.

The motivation for the refactoring is that processSQLNamePattern
mixes the job of transforming the pattern with the job of
constructing a where-clause based on a single pattern, which makes
the code hard to reuse from other places.

The new helper function patternToSQLRegex can handle parsing of
patterns of the form "database.schema.relation", "schema.relation",
and "relation".  The three-part form is unused in this commit, as
the pre-existing patternToSQLRegex function ignores the dbname
functionality, and there are not yet any other callers.  The
three-part form will be used by pg_amcheck, not yet committed, to
allow specifying on the command line the inclusion and exclusion of
relations spanning multiple databases.
---
 src/fe_utils/string_utils.c         | 260 +++++++++++++++++-----------
 src/include/fe_utils/string_utils.h |   4 +
 2 files changed, 167 insertions(+), 97 deletions(-)

diff --git a/src/fe_utils/string_utils.c b/src/fe_utils/string_utils.c
index a1a9d691d5..9a1ea9ab98 100644
--- a/src/fe_utils/string_utils.c
+++ b/src/fe_utils/string_utils.c
@@ -831,10 +831,6 @@ processSQLNamePattern(PGconn *conn, PQExpBuffer buf, const char *pattern,
 {
 	PQExpBufferData schemabuf;
 	PQExpBufferData namebuf;
-	int			encoding = PQclientEncoding(conn);
-	bool		inquotes;
-	const char *cp;
-	int			i;
 	bool		added_clause = false;
 
 #define WHEREAND() \
@@ -856,98 +852,12 @@ processSQLNamePattern(PGconn *conn, PQExpBuffer buf, const char *pattern,
 	initPQExpBuffer(&namebuf);
 
 	/*
-	 * Parse the pattern, converting quotes and lower-casing unquoted letters.
-	 * Also, adjust shell-style wildcard characters into regexp notation.
-	 *
-	 * We surround the pattern with "^(...)$" to force it to match the whole
-	 * string, as per SQL practice.  We have to have parens in case the string
-	 * contains "|", else the "^" and "$" will be bound into the first and
-	 * last alternatives which is not what we want.
-	 *
-	 * Note: the result of this pass is the actual regexp pattern(s) we want
-	 * to execute.  Quoting/escaping into SQL literal format will be done
-	 * below using appendStringLiteralConn().
+	 * Convert shell-style 'pattern' into the regular expression(s) we want to
+	 * execute.  Quoting/escaping into SQL literal format will be done below
+	 * using appendStringLiteralConn().
 	 */
-	appendPQExpBufferStr(&namebuf, "^(");
-
-	inquotes = false;
-	cp = pattern;
-
-	while (*cp)
-	{
-		char		ch = *cp;
-
-		if (ch == '"')
-		{
-			if (inquotes && cp[1] == '"')
-			{
-				/* emit one quote, stay in inquotes mode */
-				appendPQExpBufferChar(&namebuf, '"');
-				cp++;
-			}
-			else
-				inquotes = !inquotes;
-			cp++;
-		}
-		else if (!inquotes && isupper((unsigned char) ch))
-		{
-			appendPQExpBufferChar(&namebuf,
-								  pg_tolower((unsigned char) ch));
-			cp++;
-		}
-		else if (!inquotes && ch == '*')
-		{
-			appendPQExpBufferStr(&namebuf, ".*");
-			cp++;
-		}
-		else if (!inquotes && ch == '?')
-		{
-			appendPQExpBufferChar(&namebuf, '.');
-			cp++;
-		}
-		else if (!inquotes && ch == '.')
-		{
-			/* Found schema/name separator, move current pattern to schema */
-			resetPQExpBuffer(&schemabuf);
-			appendPQExpBufferStr(&schemabuf, namebuf.data);
-			resetPQExpBuffer(&namebuf);
-			appendPQExpBufferStr(&namebuf, "^(");
-			cp++;
-		}
-		else if (ch == '$')
-		{
-			/*
-			 * Dollar is always quoted, whether inside quotes or not. The
-			 * reason is that it's allowed in SQL identifiers, so there's a
-			 * significant use-case for treating it literally, while because
-			 * we anchor the pattern automatically there is no use-case for
-			 * having it possess its regexp meaning.
-			 */
-			appendPQExpBufferStr(&namebuf, "\\$");
-			cp++;
-		}
-		else
-		{
-			/*
-			 * Ordinary data character, transfer to pattern
-			 *
-			 * Inside double quotes, or at all times if force_escape is true,
-			 * quote regexp special characters with a backslash to avoid
-			 * regexp errors.  Outside quotes, however, let them pass through
-			 * as-is; this lets knowledgeable users build regexp expressions
-			 * that are more powerful than shell-style patterns.
-			 */
-			if ((inquotes || force_escape) &&
-				strchr("|*+?()[]{}.^$\\", ch))
-				appendPQExpBufferChar(&namebuf, '\\');
-			i = PQmblen(cp, encoding);
-			while (i-- && *cp)
-			{
-				appendPQExpBufferChar(&namebuf, *cp);
-				cp++;
-			}
-		}
-	}
+	patternToSQLRegex(PQclientEncoding(conn), NULL, &schemabuf, &namebuf,
+					  pattern, force_escape);
 
 	/*
 	 * Now decide what we need to emit.  We may run under a hostile
@@ -964,7 +874,6 @@ processSQLNamePattern(PGconn *conn, PQExpBuffer buf, const char *pattern,
 	{
 		/* We have a name pattern, so constrain the namevar(s) */
 
-		appendPQExpBufferStr(&namebuf, ")$");
 		/* Optimize away a "*" pattern */
 		if (strcmp(namebuf.data, "^(.*)$") != 0)
 		{
@@ -999,7 +908,6 @@ processSQLNamePattern(PGconn *conn, PQExpBuffer buf, const char *pattern,
 	{
 		/* We have a schema pattern, so constrain the schemavar */
 
-		appendPQExpBufferStr(&schemabuf, ")$");
 		/* Optimize away a "*" pattern */
 		if (strcmp(schemabuf.data, "^(.*)$") != 0 && schemavar)
 		{
@@ -1027,3 +935,161 @@ processSQLNamePattern(PGconn *conn, PQExpBuffer buf, const char *pattern,
 	return added_clause;
 #undef WHEREAND
 }
+
+/*
+ * Transform a possibly qualified shell-style object name pattern into up to
+ * three SQL-style regular expressions, converting quotes, lower-casing
+ * unquoted letters, and adjusting shell-style wildcard characters into regexp
+ * notation.
+ *
+ * If the dbnamebuf and schemabuf arguments are non-NULL, and the pattern
+ * contains two or more dbname/schema/name separators, we parse the portions of
+ * the pattern prior to the first and second separators into dbnamebuf and
+ * schemabuf, and the rest into namebuf.  (Additional dots in the name portion
+ * are not treated as special.)
+ *
+ * If dbnamebuf is NULL and schemabuf is non-NULL, and the pattern contains at
+ * least one separator, we parse the first portion into schemabuf and the rest
+ * into namebuf.
+ *
+ * Otherwise, we parse all the pattern into namebuf.
+ *
+ * We surround the regexps with "^(...)$" to force them to match whole strings,
+ * as per SQL practice.  We have to have parens in case strings contain "|",
+ * else the "^" and "$" will be bound into the first and last alternatives
+ * which is not what we want.
+ *
+ * The regexps we parse into the buffers are appended to the data (if any)
+ * already present.  If we parse fewer fields than the number of buffers we
+ * were given, the extra buffers are unaltered.
+ */
+void
+patternToSQLRegex(int encoding, PQExpBuffer dbnamebuf, PQExpBuffer schemabuf,
+				  PQExpBuffer namebuf, const char *pattern, bool force_escape)
+{
+	PQExpBufferData buf[3];
+	PQExpBuffer curbuf;
+	PQExpBuffer maxbuf;
+	int			i;
+	bool		inquotes;
+	const char *cp;
+
+	Assert(pattern != NULL);
+	Assert(namebuf != NULL);
+
+	/* callers should never expect "dbname.relname" format */
+	Assert(dbnamebuf == NULL || schemabuf != NULL);
+
+	inquotes = false;
+	cp = pattern;
+
+	if (dbnamebuf != NULL)
+		maxbuf = &buf[2];
+	else if (schemabuf != NULL)
+		maxbuf = &buf[1];
+	else
+		maxbuf = &buf[0];
+
+	curbuf = &buf[0];
+	initPQExpBuffer(curbuf);
+	appendPQExpBufferStr(curbuf, "^(");
+	while (*cp)
+	{
+		char		ch = *cp;
+
+		if (ch == '"')
+		{
+			if (inquotes && cp[1] == '"')
+			{
+				/* emit one quote, stay in inquotes mode */
+				appendPQExpBufferChar(curbuf, '"');
+				cp++;
+			}
+			else
+				inquotes = !inquotes;
+			cp++;
+		}
+		else if (!inquotes && isupper((unsigned char) ch))
+		{
+			appendPQExpBufferChar(curbuf,
+								  pg_tolower((unsigned char) ch));
+			cp++;
+		}
+		else if (!inquotes && ch == '*')
+		{
+			appendPQExpBufferStr(curbuf, ".*");
+			cp++;
+		}
+		else if (!inquotes && ch == '?')
+		{
+			appendPQExpBufferChar(curbuf, '.');
+			cp++;
+		}
+
+		/*
+		 * When we find a dbname/schema/name separator, we treat it specially
+		 * only if the caller requested more patterns to be parsed than we
+		 * have already parsed from the pattern.  Otherwise, dot characters
+		 * are not special.
+		 */
+		else if (!inquotes && ch == '.' && curbuf < maxbuf)
+		{
+			appendPQExpBufferStr(curbuf, ")$");
+			curbuf++;
+			initPQExpBuffer(curbuf);
+			appendPQExpBufferStr(curbuf, "^(");
+			cp++;
+		}
+		else if (ch == '$')
+		{
+			/*
+			 * Dollar is always quoted, whether inside quotes or not. The
+			 * reason is that it's allowed in SQL identifiers, so there's a
+			 * significant use-case for treating it literally, while because
+			 * we anchor the pattern automatically there is no use-case for
+			 * having it possess its regexp meaning.
+			 */
+			appendPQExpBufferStr(curbuf, "\\$");
+			cp++;
+		}
+		else
+		{
+			/*
+			 * Ordinary data character, transfer to pattern
+			 *
+			 * Inside double quotes, or at all times if force_escape is true,
+			 * quote regexp special characters with a backslash to avoid
+			 * regexp errors.  Outside quotes, however, let them pass through
+			 * as-is; this lets knowledgeable users build regexp expressions
+			 * that are more powerful than shell-style patterns.
+			 */
+			if ((inquotes || force_escape) &&
+				strchr("|*+?()[]{}.^$\\", ch))
+				appendPQExpBufferChar(curbuf, '\\');
+			i = PQmblen(cp, encoding);
+			while (i-- && *cp)
+			{
+				appendPQExpBufferChar(curbuf, *cp);
+				cp++;
+			}
+		}
+	}
+	appendPQExpBufferStr(curbuf, ")$");
+
+	appendPQExpBufferStr(namebuf, curbuf->data);
+	termPQExpBuffer(curbuf);
+
+	if (curbuf > buf)
+	{
+		curbuf--;
+		appendPQExpBufferStr(schemabuf, curbuf->data);
+		termPQExpBuffer(curbuf);
+
+		if (curbuf > buf)
+		{
+			curbuf--;
+			appendPQExpBufferStr(dbnamebuf, curbuf->data);
+			termPQExpBuffer(curbuf);
+		}
+	}
+}
diff --git a/src/include/fe_utils/string_utils.h b/src/include/fe_utils/string_utils.h
index c290c302f5..caafb97d29 100644
--- a/src/include/fe_utils/string_utils.h
+++ b/src/include/fe_utils/string_utils.h
@@ -56,4 +56,8 @@ extern bool processSQLNamePattern(PGconn *conn, PQExpBuffer buf,
 								  const char *schemavar, const char *namevar,
 								  const char *altnamevar, const char *visibilityrule);
 
+extern void patternToSQLRegex(int encoding, PQExpBuffer dbnamebuf,
+							  PQExpBuffer schemabuf, PQExpBuffer namebuf,
+							  const char *pattern, bool force_escape);
+
 #endif							/* STRING_UTILS_H */
-- 
2.21.1 (Apple Git-122.3)

=

v36-0002-Moving-code-from-src-bin-scripts-to-fe_utils.patch (application/octet-stream, 35.4K) download
From 3c671447d64d68bd2e378aebc04db1e6dae8206c Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Tue, 2 Feb 2021 12:30:53 -0800
Subject: [PATCH v36 2/5] Moving code from src/bin/scripts to fe_utils

To make code useable from contrib/pg_amcheck, moving
scripts_parallel.[ch] and handle_help_version_opts() into
fe_utils.

Moving supporting code from src/bin/scripts/common into fe_utils.

Updating applications in src/bin/scripts with the new location.
---
 src/bin/scripts/Makefile                      |   6 +-
 src/bin/scripts/clusterdb.c                   |   2 +
 src/bin/scripts/common.c                      | 318 +-----------------
 src/bin/scripts/common.h                      |  49 +--
 src/bin/scripts/createdb.c                    |   1 +
 src/bin/scripts/createuser.c                  |   1 +
 src/bin/scripts/dropdb.c                      |   1 +
 src/bin/scripts/dropuser.c                    |   1 +
 src/bin/scripts/nls.mk                        |   2 +-
 src/bin/scripts/pg_isready.c                  |   1 +
 src/bin/scripts/reindexdb.c                   |   4 +-
 src/bin/scripts/vacuumdb.c                    |   4 +-
 src/fe_utils/Makefile                         |   4 +
 src/fe_utils/connect_utils.c                  | 170 ++++++++++
 src/fe_utils/option_utils.c                   |  38 +++
 .../parallel_slot.c}                          |  63 +++-
 src/fe_utils/query_utils.c                    |  92 +++++
 src/fe_utils/string_utils.c                   |  17 +-
 src/include/fe_utils/connect_utils.h          |  48 +++
 src/include/fe_utils/option_utils.h           |  23 ++
 .../fe_utils/parallel_slot.h}                 |  13 +-
 src/include/fe_utils/query_utils.h            |  26 ++
 src/tools/msvc/Mkvcbuild.pm                   |   4 +-
 src/tools/pgindent/typedefs.list              |   1 +
 24 files changed, 495 insertions(+), 394 deletions(-)
 create mode 100644 src/fe_utils/connect_utils.c
 create mode 100644 src/fe_utils/option_utils.c
 rename src/{bin/scripts/scripts_parallel.c => fe_utils/parallel_slot.c} (80%)
 create mode 100644 src/fe_utils/query_utils.c
 create mode 100644 src/include/fe_utils/connect_utils.h
 create mode 100644 src/include/fe_utils/option_utils.h
 rename src/{bin/scripts/scripts_parallel.h => include/fe_utils/parallel_slot.h} (82%)
 create mode 100644 src/include/fe_utils/query_utils.h

diff --git a/src/bin/scripts/Makefile b/src/bin/scripts/Makefile
index a02e4e430c..b8d7cf2f2d 100644
--- a/src/bin/scripts/Makefile
+++ b/src/bin/scripts/Makefile
@@ -28,8 +28,8 @@ createuser: createuser.o common.o $(WIN32RES) | submake-libpq submake-libpgport
 dropdb: dropdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 dropuser: dropuser.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 clusterdb: clusterdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
-vacuumdb: vacuumdb.o common.o scripts_parallel.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
-reindexdb: reindexdb.o common.o scripts_parallel.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
+vacuumdb: vacuumdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
+reindexdb: reindexdb.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 pg_isready: pg_isready.o common.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 
 install: all installdirs
@@ -50,7 +50,7 @@ uninstall:
 
 clean distclean maintainer-clean:
 	rm -f $(addsuffix $(X), $(PROGRAMS)) $(addsuffix .o, $(PROGRAMS))
-	rm -f common.o scripts_parallel.o $(WIN32RES)
+	rm -f common.o $(WIN32RES)
 	rm -rf tmp_check
 
 check:
diff --git a/src/bin/scripts/clusterdb.c b/src/bin/scripts/clusterdb.c
index 7d25bb31d4..fc771eed77 100644
--- a/src/bin/scripts/clusterdb.c
+++ b/src/bin/scripts/clusterdb.c
@@ -13,6 +13,8 @@
 #include "common.h"
 #include "common/logging.h"
 #include "fe_utils/cancel.h"
+#include "fe_utils/option_utils.h"
+#include "fe_utils/query_utils.h"
 #include "fe_utils/simple_list.h"
 #include "fe_utils/string_utils.h"
 
diff --git a/src/bin/scripts/common.c b/src/bin/scripts/common.c
index 21ef297e6e..c86c19eae2 100644
--- a/src/bin/scripts/common.c
+++ b/src/bin/scripts/common.c
@@ -22,325 +22,9 @@
 #include "common/logging.h"
 #include "common/string.h"
 #include "fe_utils/cancel.h"
+#include "fe_utils/query_utils.h"
 #include "fe_utils/string_utils.h"
 
-#define ERRCODE_UNDEFINED_TABLE  "42P01"
-
-/*
- * Provide strictly harmonized handling of --help and --version
- * options.
- */
-void
-handle_help_version_opts(int argc, char *argv[],
-						 const char *fixed_progname, help_handler hlp)
-{
-	if (argc > 1)
-	{
-		if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
-		{
-			hlp(get_progname(argv[0]));
-			exit(0);
-		}
-		if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
-		{
-			printf("%s (PostgreSQL) " PG_VERSION "\n", fixed_progname);
-			exit(0);
-		}
-	}
-}
-
-
-/*
- * Make a database connection with the given parameters.
- *
- * An interactive password prompt is automatically issued if needed and
- * allowed by cparams->prompt_password.
- *
- * If allow_password_reuse is true, we will try to re-use any password
- * given during previous calls to this routine.  (Callers should not pass
- * allow_password_reuse=true unless reconnecting to the same database+user
- * as before, else we might create password exposure hazards.)
- */
-PGconn *
-connectDatabase(const ConnParams *cparams, const char *progname,
-				bool echo, bool fail_ok, bool allow_password_reuse)
-{
-	PGconn	   *conn;
-	bool		new_pass;
-	static char *password = NULL;
-
-	/* Callers must supply at least dbname; other params can be NULL */
-	Assert(cparams->dbname);
-
-	if (!allow_password_reuse && password)
-	{
-		free(password);
-		password = NULL;
-	}
-
-	if (cparams->prompt_password == TRI_YES && password == NULL)
-		password = simple_prompt("Password: ", false);
-
-	/*
-	 * Start the connection.  Loop until we have a password if requested by
-	 * backend.
-	 */
-	do
-	{
-		const char *keywords[8];
-		const char *values[8];
-		int			i = 0;
-
-		/*
-		 * If dbname is a connstring, its entries can override the other
-		 * values obtained from cparams; but in turn, override_dbname can
-		 * override the dbname component of it.
-		 */
-		keywords[i] = "host";
-		values[i++] = cparams->pghost;
-		keywords[i] = "port";
-		values[i++] = cparams->pgport;
-		keywords[i] = "user";
-		values[i++] = cparams->pguser;
-		keywords[i] = "password";
-		values[i++] = password;
-		keywords[i] = "dbname";
-		values[i++] = cparams->dbname;
-		if (cparams->override_dbname)
-		{
-			keywords[i] = "dbname";
-			values[i++] = cparams->override_dbname;
-		}
-		keywords[i] = "fallback_application_name";
-		values[i++] = progname;
-		keywords[i] = NULL;
-		values[i++] = NULL;
-		Assert(i <= lengthof(keywords));
-
-		new_pass = false;
-		conn = PQconnectdbParams(keywords, values, true);
-
-		if (!conn)
-		{
-			pg_log_error("could not connect to database %s: out of memory",
-						 cparams->dbname);
-			exit(1);
-		}
-
-		/*
-		 * No luck?  Trying asking (again) for a password.
-		 */
-		if (PQstatus(conn) == CONNECTION_BAD &&
-			PQconnectionNeedsPassword(conn) &&
-			cparams->prompt_password != TRI_NO)
-		{
-			PQfinish(conn);
-			if (password)
-				free(password);
-			password = simple_prompt("Password: ", false);
-			new_pass = true;
-		}
-	} while (new_pass);
-
-	/* check to see that the backend connection was successfully made */
-	if (PQstatus(conn) == CONNECTION_BAD)
-	{
-		if (fail_ok)
-		{
-			PQfinish(conn);
-			return NULL;
-		}
-		pg_log_error("%s", PQerrorMessage(conn));
-		exit(1);
-	}
-
-	/* Start strict; callers may override this. */
-	PQclear(executeQuery(conn, ALWAYS_SECURE_SEARCH_PATH_SQL, echo));
-
-	return conn;
-}
-
-/*
- * Try to connect to the appropriate maintenance database.
- *
- * This differs from connectDatabase only in that it has a rule for
- * inserting a default "dbname" if none was given (which is why cparams
- * is not const).  Note that cparams->dbname should typically come from
- * a --maintenance-db command line parameter.
- */
-PGconn *
-connectMaintenanceDatabase(ConnParams *cparams,
-						   const char *progname, bool echo)
-{
-	PGconn	   *conn;
-
-	/* If a maintenance database name was specified, just connect to it. */
-	if (cparams->dbname)
-		return connectDatabase(cparams, progname, echo, false, false);
-
-	/* Otherwise, try postgres first and then template1. */
-	cparams->dbname = "postgres";
-	conn = connectDatabase(cparams, progname, echo, true, false);
-	if (!conn)
-	{
-		cparams->dbname = "template1";
-		conn = connectDatabase(cparams, progname, echo, false, false);
-	}
-	return conn;
-}
-
-/*
- * Disconnect the given connection, canceling any statement if one is active.
- */
-void
-disconnectDatabase(PGconn *conn)
-{
-	char		errbuf[256];
-
-	Assert(conn != NULL);
-
-	if (PQtransactionStatus(conn) == PQTRANS_ACTIVE)
-	{
-		PGcancel   *cancel;
-
-		if ((cancel = PQgetCancel(conn)))
-		{
-			(void) PQcancel(cancel, errbuf, sizeof(errbuf));
-			PQfreeCancel(cancel);
-		}
-	}
-
-	PQfinish(conn);
-}
-
-/*
- * Run a query, return the results, exit program on failure.
- */
-PGresult *
-executeQuery(PGconn *conn, const char *query, bool echo)
-{
-	PGresult   *res;
-
-	if (echo)
-		printf("%s\n", query);
-
-	res = PQexec(conn, query);
-	if (!res ||
-		PQresultStatus(res) != PGRES_TUPLES_OK)
-	{
-		pg_log_error("query failed: %s", PQerrorMessage(conn));
-		pg_log_info("query was: %s", query);
-		PQfinish(conn);
-		exit(1);
-	}
-
-	return res;
-}
-
-
-/*
- * As above for a SQL command (which returns nothing).
- */
-void
-executeCommand(PGconn *conn, const char *query, bool echo)
-{
-	PGresult   *res;
-
-	if (echo)
-		printf("%s\n", query);
-
-	res = PQexec(conn, query);
-	if (!res ||
-		PQresultStatus(res) != PGRES_COMMAND_OK)
-	{
-		pg_log_error("query failed: %s", PQerrorMessage(conn));
-		pg_log_info("query was: %s", query);
-		PQfinish(conn);
-		exit(1);
-	}
-
-	PQclear(res);
-}
-
-
-/*
- * As above for a SQL maintenance command (returns command success).
- * Command is executed with a cancel handler set, so Ctrl-C can
- * interrupt it.
- */
-bool
-executeMaintenanceCommand(PGconn *conn, const char *query, bool echo)
-{
-	PGresult   *res;
-	bool		r;
-
-	if (echo)
-		printf("%s\n", query);
-
-	SetCancelConn(conn);
-	res = PQexec(conn, query);
-	ResetCancelConn();
-
-	r = (res && PQresultStatus(res) == PGRES_COMMAND_OK);
-
-	if (res)
-		PQclear(res);
-
-	return r;
-}
-
-/*
- * Consume all the results generated for the given connection until
- * nothing remains.  If at least one error is encountered, return false.
- * Note that this will block if the connection is busy.
- */
-bool
-consumeQueryResult(PGconn *conn)
-{
-	bool		ok = true;
-	PGresult   *result;
-
-	SetCancelConn(conn);
-	while ((result = PQgetResult(conn)) != NULL)
-	{
-		if (!processQueryResult(conn, result))
-			ok = false;
-	}
-	ResetCancelConn();
-	return ok;
-}
-
-/*
- * Process (and delete) a query result.  Returns true if there's no error,
- * false otherwise -- but errors about trying to work on a missing relation
- * are reported and subsequently ignored.
- */
-bool
-processQueryResult(PGconn *conn, PGresult *result)
-{
-	/*
-	 * If it's an error, report it.  Errors about a missing table are harmless
-	 * so we continue processing; but die for other errors.
-	 */
-	if (PQresultStatus(result) != PGRES_COMMAND_OK)
-	{
-		char	   *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
-
-		pg_log_error("processing of database \"%s\" failed: %s",
-					 PQdb(conn), PQerrorMessage(conn));
-
-		if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
-		{
-			PQclear(result);
-			return false;
-		}
-	}
-
-	PQclear(result);
-	return true;
-}
-
-
 /*
  * Split TABLE[(COLUMNS)] into TABLE and [(COLUMNS)] portions.  When you
  * finish using them, pg_free(*table).  *columns is a pointer into "spec",
diff --git a/src/bin/scripts/common.h b/src/bin/scripts/common.h
index 5630975712..ddd8f35274 100644
--- a/src/bin/scripts/common.h
+++ b/src/bin/scripts/common.h
@@ -10,58 +10,11 @@
 #define COMMON_H
 
 #include "common/username.h"
+#include "fe_utils/connect_utils.h"
 #include "getopt_long.h"		/* pgrminclude ignore */
 #include "libpq-fe.h"
 #include "pqexpbuffer.h"		/* pgrminclude ignore */
 
-enum trivalue
-{
-	TRI_DEFAULT,
-	TRI_NO,
-	TRI_YES
-};
-
-/* Parameters needed by connectDatabase/connectMaintenanceDatabase */
-typedef struct _connParams
-{
-	/* These fields record the actual command line parameters */
-	const char *dbname;			/* this may be a connstring! */
-	const char *pghost;
-	const char *pgport;
-	const char *pguser;
-	enum trivalue prompt_password;
-	/* If not NULL, this overrides the dbname obtained from command line */
-	/* (but *only* the DB name, not anything else in the connstring) */
-	const char *override_dbname;
-} ConnParams;
-
-typedef void (*help_handler) (const char *progname);
-
-extern void handle_help_version_opts(int argc, char *argv[],
-									 const char *fixed_progname,
-									 help_handler hlp);
-
-extern PGconn *connectDatabase(const ConnParams *cparams,
-							   const char *progname,
-							   bool echo, bool fail_ok,
-							   bool allow_password_reuse);
-
-extern PGconn *connectMaintenanceDatabase(ConnParams *cparams,
-										  const char *progname, bool echo);
-
-extern void disconnectDatabase(PGconn *conn);
-
-extern PGresult *executeQuery(PGconn *conn, const char *query, bool echo);
-
-extern void executeCommand(PGconn *conn, const char *query, bool echo);
-
-extern bool executeMaintenanceCommand(PGconn *conn, const char *query,
-									  bool echo);
-
-extern bool consumeQueryResult(PGconn *conn);
-
-extern bool processQueryResult(PGconn *conn, PGresult *result);
-
 extern void splitTableColumnsSpec(const char *spec, int encoding,
 								  char **table, const char **columns);
 
diff --git a/src/bin/scripts/createdb.c b/src/bin/scripts/createdb.c
index abf21d4942..041454f075 100644
--- a/src/bin/scripts/createdb.c
+++ b/src/bin/scripts/createdb.c
@@ -13,6 +13,7 @@
 
 #include "common.h"
 #include "common/logging.h"
+#include "fe_utils/option_utils.h"
 #include "fe_utils/string_utils.h"
 
 
diff --git a/src/bin/scripts/createuser.c b/src/bin/scripts/createuser.c
index 47b0e28bc6..ef7e0e549f 100644
--- a/src/bin/scripts/createuser.c
+++ b/src/bin/scripts/createuser.c
@@ -14,6 +14,7 @@
 #include "common.h"
 #include "common/logging.h"
 #include "common/string.h"
+#include "fe_utils/option_utils.h"
 #include "fe_utils/simple_list.h"
 #include "fe_utils/string_utils.h"
 
diff --git a/src/bin/scripts/dropdb.c b/src/bin/scripts/dropdb.c
index ba0dcdecb9..b154ed1bb6 100644
--- a/src/bin/scripts/dropdb.c
+++ b/src/bin/scripts/dropdb.c
@@ -13,6 +13,7 @@
 #include "postgres_fe.h"
 #include "common.h"
 #include "common/logging.h"
+#include "fe_utils/option_utils.h"
 #include "fe_utils/string_utils.h"
 
 
diff --git a/src/bin/scripts/dropuser.c b/src/bin/scripts/dropuser.c
index ff5b455ae5..61b8557bc7 100644
--- a/src/bin/scripts/dropuser.c
+++ b/src/bin/scripts/dropuser.c
@@ -14,6 +14,7 @@
 #include "common.h"
 #include "common/logging.h"
 #include "common/string.h"
+#include "fe_utils/option_utils.h"
 #include "fe_utils/string_utils.h"
 
 
diff --git a/src/bin/scripts/nls.mk b/src/bin/scripts/nls.mk
index 5d5dd11b7b..7fc716092e 100644
--- a/src/bin/scripts/nls.mk
+++ b/src/bin/scripts/nls.mk
@@ -7,7 +7,7 @@ GETTEXT_FILES    = $(FRONTEND_COMMON_GETTEXT_FILES) \
                    clusterdb.c vacuumdb.c reindexdb.c \
                    pg_isready.c \
                    common.c \
-                   scripts_parallel.c \
+                   ../../fe_utils/parallel_slot.c \
                    ../../fe_utils/cancel.c ../../fe_utils/print.c \
                    ../../common/fe_memutils.c ../../common/username.c
 GETTEXT_TRIGGERS = $(FRONTEND_COMMON_GETTEXT_TRIGGERS) simple_prompt yesno_prompt
diff --git a/src/bin/scripts/pg_isready.c b/src/bin/scripts/pg_isready.c
index ceb8a09b4c..fc6f7b0a93 100644
--- a/src/bin/scripts/pg_isready.c
+++ b/src/bin/scripts/pg_isready.c
@@ -12,6 +12,7 @@
 #include "postgres_fe.h"
 #include "common.h"
 #include "common/logging.h"
+#include "fe_utils/option_utils.h"
 
 #define DEFAULT_CONNECT_TIMEOUT "3"
 
diff --git a/src/bin/scripts/reindexdb.c b/src/bin/scripts/reindexdb.c
index dece8200fa..7781fb1151 100644
--- a/src/bin/scripts/reindexdb.c
+++ b/src/bin/scripts/reindexdb.c
@@ -16,9 +16,11 @@
 #include "common/connect.h"
 #include "common/logging.h"
 #include "fe_utils/cancel.h"
+#include "fe_utils/option_utils.h"
+#include "fe_utils/parallel_slot.h"
+#include "fe_utils/query_utils.h"
 #include "fe_utils/simple_list.h"
 #include "fe_utils/string_utils.h"
-#include "scripts_parallel.h"
 
 typedef enum ReindexType
 {
diff --git a/src/bin/scripts/vacuumdb.c b/src/bin/scripts/vacuumdb.c
index 8246327770..ed320817bc 100644
--- a/src/bin/scripts/vacuumdb.c
+++ b/src/bin/scripts/vacuumdb.c
@@ -18,9 +18,11 @@
 #include "common/connect.h"
 #include "common/logging.h"
 #include "fe_utils/cancel.h"
+#include "fe_utils/option_utils.h"
+#include "fe_utils/parallel_slot.h"
+#include "fe_utils/query_utils.h"
 #include "fe_utils/simple_list.h"
 #include "fe_utils/string_utils.h"
-#include "scripts_parallel.h"
 
 
 /* vacuum options controlled by user flags */
diff --git a/src/fe_utils/Makefile b/src/fe_utils/Makefile
index 10d6838cf9..456c441a33 100644
--- a/src/fe_utils/Makefile
+++ b/src/fe_utils/Makefile
@@ -23,9 +23,13 @@ OBJS = \
 	archive.o \
 	cancel.o \
 	conditional.o \
+	connect_utils.o \
 	mbprint.o \
+	option_utils.o \
+	parallel_slot.o \
 	print.o \
 	psqlscan.o \
+	query_utils.o \
 	recovery_gen.o \
 	simple_list.o \
 	string_utils.o
diff --git a/src/fe_utils/connect_utils.c b/src/fe_utils/connect_utils.c
new file mode 100644
index 0000000000..7475e2f366
--- /dev/null
+++ b/src/fe_utils/connect_utils.c
@@ -0,0 +1,170 @@
+#include "postgres_fe.h"
+
+#include "common/connect.h"
+#include "common/logging.h"
+#include "common/string.h"
+#include "fe_utils/connect_utils.h"
+#include "fe_utils/query_utils.h"
+
+/*
+ * Make a database connection with the given parameters.
+ *
+ * An interactive password prompt is automatically issued if needed and
+ * allowed by cparams->prompt_password.
+ *
+ * If allow_password_reuse is true, we will try to re-use any password
+ * given during previous calls to this routine.  (Callers should not pass
+ * allow_password_reuse=true unless reconnecting to the same database+user
+ * as before, else we might create password exposure hazards.)
+ */
+PGconn *
+connectDatabase(const ConnParams *cparams, const char *progname,
+				bool echo, bool fail_ok, bool allow_password_reuse)
+{
+	PGconn	   *conn;
+	bool		new_pass;
+	static char *password = NULL;
+
+	/* Callers must supply at least dbname; other params can be NULL */
+	Assert(cparams->dbname);
+
+	if (!allow_password_reuse && password)
+	{
+		free(password);
+		password = NULL;
+	}
+
+	if (cparams->prompt_password == TRI_YES && password == NULL)
+		password = simple_prompt("Password: ", false);
+
+	/*
+	 * Start the connection.  Loop until we have a password if requested by
+	 * backend.
+	 */
+	do
+	{
+		const char *keywords[8];
+		const char *values[8];
+		int			i = 0;
+
+		/*
+		 * If dbname is a connstring, its entries can override the other
+		 * values obtained from cparams; but in turn, override_dbname can
+		 * override the dbname component of it.
+		 */
+		keywords[i] = "host";
+		values[i++] = cparams->pghost;
+		keywords[i] = "port";
+		values[i++] = cparams->pgport;
+		keywords[i] = "user";
+		values[i++] = cparams->pguser;
+		keywords[i] = "password";
+		values[i++] = password;
+		keywords[i] = "dbname";
+		values[i++] = cparams->dbname;
+		if (cparams->override_dbname)
+		{
+			keywords[i] = "dbname";
+			values[i++] = cparams->override_dbname;
+		}
+		keywords[i] = "fallback_application_name";
+		values[i++] = progname;
+		keywords[i] = NULL;
+		values[i++] = NULL;
+		Assert(i <= lengthof(keywords));
+
+		new_pass = false;
+		conn = PQconnectdbParams(keywords, values, true);
+
+		if (!conn)
+		{
+			pg_log_error("could not connect to database %s: out of memory",
+						 cparams->dbname);
+			exit(1);
+		}
+
+		/*
+		 * No luck?  Trying asking (again) for a password.
+		 */
+		if (PQstatus(conn) == CONNECTION_BAD &&
+			PQconnectionNeedsPassword(conn) &&
+			cparams->prompt_password != TRI_NO)
+		{
+			PQfinish(conn);
+			if (password)
+				free(password);
+			password = simple_prompt("Password: ", false);
+			new_pass = true;
+		}
+	} while (new_pass);
+
+	/* check to see that the backend connection was successfully made */
+	if (PQstatus(conn) == CONNECTION_BAD)
+	{
+		if (fail_ok)
+		{
+			PQfinish(conn);
+			return NULL;
+		}
+		pg_log_error("%s", PQerrorMessage(conn));
+		exit(1);
+	}
+
+	/* Start strict; callers may override this. */
+	PQclear(executeQuery(conn, ALWAYS_SECURE_SEARCH_PATH_SQL, echo));
+
+	return conn;
+}
+
+/*
+ * Try to connect to the appropriate maintenance database.
+ *
+ * This differs from connectDatabase only in that it has a rule for
+ * inserting a default "dbname" if none was given (which is why cparams
+ * is not const).  Note that cparams->dbname should typically come from
+ * a --maintenance-db command line parameter.
+ */
+PGconn *
+connectMaintenanceDatabase(ConnParams *cparams,
+						   const char *progname, bool echo)
+{
+	PGconn	   *conn;
+
+	/* If a maintenance database name was specified, just connect to it. */
+	if (cparams->dbname)
+		return connectDatabase(cparams, progname, echo, false, false);
+
+	/* Otherwise, try postgres first and then template1. */
+	cparams->dbname = "postgres";
+	conn = connectDatabase(cparams, progname, echo, true, false);
+	if (!conn)
+	{
+		cparams->dbname = "template1";
+		conn = connectDatabase(cparams, progname, echo, false, false);
+	}
+	return conn;
+}
+
+/*
+ * Disconnect the given connection, canceling any statement if one is active.
+ */
+void
+disconnectDatabase(PGconn *conn)
+{
+	char		errbuf[256];
+
+	Assert(conn != NULL);
+
+	if (PQtransactionStatus(conn) == PQTRANS_ACTIVE)
+	{
+		PGcancel   *cancel;
+
+		if ((cancel = PQgetCancel(conn)))
+		{
+			(void) PQcancel(cancel, errbuf, sizeof(errbuf));
+			PQfreeCancel(cancel);
+		}
+	}
+
+	PQfinish(conn);
+}
diff --git a/src/fe_utils/option_utils.c b/src/fe_utils/option_utils.c
new file mode 100644
index 0000000000..97aca1f02b
--- /dev/null
+++ b/src/fe_utils/option_utils.c
@@ -0,0 +1,38 @@
+/*-------------------------------------------------------------------------
+ *
+ * Command line option processing facilities for frontend code
+ *
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/fe_utils/option_utils.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres_fe.h"
+
+#include "fe_utils/option_utils.h"
+
+/*
+ * Provide strictly harmonized handling of --help and --version
+ * options.
+ */
+void
+handle_help_version_opts(int argc, char *argv[],
+						 const char *fixed_progname, help_handler hlp)
+{
+	if (argc > 1)
+	{
+		if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
+		{
+			hlp(get_progname(argv[0]));
+			exit(0);
+		}
+		if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
+		{
+			printf("%s (PostgreSQL) " PG_VERSION "\n", fixed_progname);
+			exit(0);
+		}
+	}
+}
diff --git a/src/bin/scripts/scripts_parallel.c b/src/fe_utils/parallel_slot.c
similarity index 80%
rename from src/bin/scripts/scripts_parallel.c
rename to src/fe_utils/parallel_slot.c
index 1f863a1bb4..3987a4702b 100644
--- a/src/bin/scripts/scripts_parallel.c
+++ b/src/fe_utils/parallel_slot.c
@@ -1,13 +1,13 @@
 /*-------------------------------------------------------------------------
  *
- *	scripts_parallel.c
- *		Parallel support for bin/scripts/
+ *	parallel_slot.c
+ *		Parallel support for front-end parallel database connections
  *
  *
  * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * src/bin/scripts/scripts_parallel.c
+ * src/fe_utils/parallel_slot.c
  *
  *-------------------------------------------------------------------------
  */
@@ -22,13 +22,15 @@
 #include <sys/select.h>
 #endif
 
-#include "common.h"
 #include "common/logging.h"
 #include "fe_utils/cancel.h"
-#include "scripts_parallel.h"
+#include "fe_utils/parallel_slot.h"
+
+#define ERRCODE_UNDEFINED_TABLE  "42P01"
 
 static void init_slot(ParallelSlot *slot, PGconn *conn);
 static int	select_loop(int maxFd, fd_set *workerset);
+static bool processQueryResult(PGconn *conn, PGresult *result);
 
 static void
 init_slot(ParallelSlot *slot, PGconn *conn)
@@ -38,6 +40,57 @@ init_slot(ParallelSlot *slot, PGconn *conn)
 	slot->isFree = true;
 }
 
+/*
+ * Process (and delete) a query result.  Returns true if there's no error,
+ * false otherwise -- but errors about trying to work on a missing relation
+ * are reported and subsequently ignored.
+ */
+static bool
+processQueryResult(PGconn *conn, PGresult *result)
+{
+	/*
+	 * If it's an error, report it.  Errors about a missing table are harmless
+	 * so we continue processing; but die for other errors.
+	 */
+	if (PQresultStatus(result) != PGRES_COMMAND_OK)
+	{
+		char	   *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
+
+		pg_log_error("processing of database \"%s\" failed: %s",
+					 PQdb(conn), PQerrorMessage(conn));
+
+		if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
+		{
+			PQclear(result);
+			return false;
+		}
+	}
+
+	PQclear(result);
+	return true;
+}
+
+/*
+ * Consume all the results generated for the given connection until
+ * nothing remains.  If at least one error is encountered, return false.
+ * Note that this will block if the connection is busy.
+ */
+static bool
+consumeQueryResult(PGconn *conn)
+{
+	bool		ok = true;
+	PGresult   *result;
+
+	SetCancelConn(conn);
+	while ((result = PQgetResult(conn)) != NULL)
+	{
+		if (!processQueryResult(conn, result))
+			ok = false;
+	}
+	ResetCancelConn();
+	return ok;
+}
+
 /*
  * Wait until a file descriptor from the given set becomes readable.
  *
diff --git a/src/fe_utils/query_utils.c b/src/fe_utils/query_utils.c
new file mode 100644
index 0000000000..a70ae3c082
--- /dev/null
+++ b/src/fe_utils/query_utils.c
@@ -0,0 +1,92 @@
+/*-------------------------------------------------------------------------
+ *
+ * Facilities for frontend code to query a databases.
+ *
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/fe_utils/query_utils.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres_fe.h"
+
+#include "common/logging.h"
+#include "fe_utils/cancel.h"
+#include "fe_utils/query_utils.h"
+
+/*
+ * Run a query, return the results, exit program on failure.
+ */
+PGresult *
+executeQuery(PGconn *conn, const char *query, bool echo)
+{
+	PGresult   *res;
+
+	if (echo)
+		printf("%s\n", query);
+
+	res = PQexec(conn, query);
+	if (!res ||
+		PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("query failed: %s", PQerrorMessage(conn));
+		pg_log_info("query was: %s", query);
+		PQfinish(conn);
+		exit(1);
+	}
+
+	return res;
+}
+
+
+/*
+ * As above for a SQL command (which returns nothing).
+ */
+void
+executeCommand(PGconn *conn, const char *query, bool echo)
+{
+	PGresult   *res;
+
+	if (echo)
+		printf("%s\n", query);
+
+	res = PQexec(conn, query);
+	if (!res ||
+		PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		pg_log_error("query failed: %s", PQerrorMessage(conn));
+		pg_log_info("query was: %s", query);
+		PQfinish(conn);
+		exit(1);
+	}
+
+	PQclear(res);
+}
+
+
+/*
+ * As above for a SQL maintenance command (returns command success).
+ * Command is executed with a cancel handler set, so Ctrl-C can
+ * interrupt it.
+ */
+bool
+executeMaintenanceCommand(PGconn *conn, const char *query, bool echo)
+{
+	PGresult   *res;
+	bool		r;
+
+	if (echo)
+		printf("%s\n", query);
+
+	SetCancelConn(conn);
+	res = PQexec(conn, query);
+	ResetCancelConn();
+
+	r = (res && PQresultStatus(res) == PGRES_COMMAND_OK);
+
+	if (res)
+		PQclear(res);
+
+	return r;
+}
diff --git a/src/fe_utils/string_utils.c b/src/fe_utils/string_utils.c
index 9a1ea9ab98..94941132ac 100644
--- a/src/fe_utils/string_utils.c
+++ b/src/fe_utils/string_utils.c
@@ -852,9 +852,9 @@ processSQLNamePattern(PGconn *conn, PQExpBuffer buf, const char *pattern,
 	initPQExpBuffer(&namebuf);
 
 	/*
-	 * Convert shell-style 'pattern' into the regular expression(s) we want to
-	 * execute.  Quoting/escaping into SQL literal format will be done below
-	 * using appendStringLiteralConn().
+	 * Convert shell-style 'pattern' into the regular expression(s) we want
+	 * to execute.  Quoting/escaping into SQL literal format will be done
+	 * below using appendStringLiteralConn().
 	 */
 	patternToSQLRegex(PQclientEncoding(conn), NULL, &schemabuf, &namebuf,
 					  pattern, force_escape);
@@ -968,8 +968,8 @@ patternToSQLRegex(int encoding, PQExpBuffer dbnamebuf, PQExpBuffer schemabuf,
 				  PQExpBuffer namebuf, const char *pattern, bool force_escape)
 {
 	PQExpBufferData buf[3];
-	PQExpBuffer curbuf;
-	PQExpBuffer maxbuf;
+	PQExpBuffer	curbuf;
+	PQExpBuffer	maxbuf;
 	int			i;
 	bool		inquotes;
 	const char *cp;
@@ -1025,12 +1025,11 @@ patternToSQLRegex(int encoding, PQExpBuffer dbnamebuf, PQExpBuffer schemabuf,
 			appendPQExpBufferChar(curbuf, '.');
 			cp++;
 		}
-
 		/*
 		 * When we find a dbname/schema/name separator, we treat it specially
-		 * only if the caller requested more patterns to be parsed than we
-		 * have already parsed from the pattern.  Otherwise, dot characters
-		 * are not special.
+		 * only if the caller requested more patterns to be parsed than we have
+		 * already parsed from the pattern.  Otherwise, dot characters are not
+		 * special.
 		 */
 		else if (!inquotes && ch == '.' && curbuf < maxbuf)
 		{
diff --git a/src/include/fe_utils/connect_utils.h b/src/include/fe_utils/connect_utils.h
new file mode 100644
index 0000000000..8fde0ea2a0
--- /dev/null
+++ b/src/include/fe_utils/connect_utils.h
@@ -0,0 +1,48 @@
+/*-------------------------------------------------------------------------
+ *
+ * Facilities for frontend code to connect to and disconnect from databases.
+ *
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/fe_utils/connect_utils.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef CONNECT_UTILS_H
+#define CONNECT_UTILS_H
+
+#include "libpq-fe.h"
+
+enum trivalue
+{
+	TRI_DEFAULT,
+	TRI_NO,
+	TRI_YES
+};
+
+/* Parameters needed by connectDatabase/connectMaintenanceDatabase */
+typedef struct _connParams
+{
+	/* These fields record the actual command line parameters */
+	const char *dbname;			/* this may be a connstring! */
+	const char *pghost;
+	const char *pgport;
+	const char *pguser;
+	enum trivalue prompt_password;
+	/* If not NULL, this overrides the dbname obtained from command line */
+	/* (but *only* the DB name, not anything else in the connstring) */
+	const char *override_dbname;
+} ConnParams;
+
+extern PGconn *connectDatabase(const ConnParams *cparams,
+							   const char *progname,
+							   bool echo, bool fail_ok,
+							   bool allow_password_reuse);
+
+extern PGconn *connectMaintenanceDatabase(ConnParams *cparams,
+										  const char *progname, bool echo);
+
+extern void disconnectDatabase(PGconn *conn);
+
+#endif							/* CONNECT_UTILS_H */
diff --git a/src/include/fe_utils/option_utils.h b/src/include/fe_utils/option_utils.h
new file mode 100644
index 0000000000..ef6eb24ae0
--- /dev/null
+++ b/src/include/fe_utils/option_utils.h
@@ -0,0 +1,23 @@
+/*-------------------------------------------------------------------------
+ *
+ * Command line option processing facilities for frontend code
+ *
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/fe_utils/option_utils.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef OPTION_UTILS_H
+#define OPTION_UTILS_H
+
+#include "postgres_fe.h"
+
+typedef void (*help_handler) (const char *progname);
+
+extern void handle_help_version_opts(int argc, char *argv[],
+									 const char *fixed_progname,
+									 help_handler hlp);
+
+#endif							/* OPTION_UTILS_H */
diff --git a/src/bin/scripts/scripts_parallel.h b/src/include/fe_utils/parallel_slot.h
similarity index 82%
rename from src/bin/scripts/scripts_parallel.h
rename to src/include/fe_utils/parallel_slot.h
index f62692510a..99eeb3328d 100644
--- a/src/bin/scripts/scripts_parallel.h
+++ b/src/include/fe_utils/parallel_slot.h
@@ -1,21 +1,20 @@
 /*-------------------------------------------------------------------------
  *
- *	scripts_parallel.h
+ *	parallel_slot.h
  *		Parallel support for bin/scripts/
  *
  *	Copyright (c) 2003-2021, PostgreSQL Global Development Group
  *
- *	src/bin/scripts/scripts_parallel.h
+ *	src/include/fe_utils/parallel_slot.h
  *
  *-------------------------------------------------------------------------
  */
-#ifndef SCRIPTS_PARALLEL_H
-#define SCRIPTS_PARALLEL_H
+#ifndef PARALLEL_SLOT_H
+#define PARALLEL_SLOT_H
 
-#include "common.h"
+#include "fe_utils/connect_utils.h"
 #include "libpq-fe.h"
 
-
 typedef struct ParallelSlot
 {
 	PGconn	   *connection;		/* One connection */
@@ -33,4 +32,4 @@ extern void ParallelSlotsTerminate(ParallelSlot *slots, int numslots);
 extern bool ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots);
 
 
-#endif							/* SCRIPTS_PARALLEL_H */
+#endif							/* PARALLEL_SLOT_H */
diff --git a/src/include/fe_utils/query_utils.h b/src/include/fe_utils/query_utils.h
new file mode 100644
index 0000000000..1f5812bbf6
--- /dev/null
+++ b/src/include/fe_utils/query_utils.h
@@ -0,0 +1,26 @@
+/*-------------------------------------------------------------------------
+ *
+ * Facilities for frontend code to query a databases.
+ *
+ * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/fe_utils/query_utils.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef QUERY_UTILS_H
+#define QUERY_UTILS_H
+
+#include "postgres_fe.h"
+
+#include "libpq-fe.h"
+
+extern PGresult *executeQuery(PGconn *conn, const char *query, bool echo);
+
+extern void executeCommand(PGconn *conn, const char *query, bool echo);
+
+extern bool executeMaintenanceCommand(PGconn *conn, const char *query,
+									  bool echo);
+
+#endif							/* QUERY_UTILS_H */
diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm
index 90328db04e..7be6e6c9e5 100644
--- a/src/tools/msvc/Mkvcbuild.pm
+++ b/src/tools/msvc/Mkvcbuild.pm
@@ -147,8 +147,8 @@ sub mkvcbuild
 	our @pgcommonbkndfiles = @pgcommonallfiles;
 
 	our @pgfeutilsfiles = qw(
-	  archive.c cancel.c conditional.c mbprint.c print.c psqlscan.l
-	  psqlscan.c simple_list.c string_utils.c recovery_gen.c);
+	  archive.c cancel.c conditional.c connect_utils.c mbprint.c option_utils.c parallel_slot.c print.c psqlscan.l
+	  psqlscan.c query_utils.c simple_list.c string_utils.c recovery_gen.c);
 
 	$libpgport = $solution->AddProject('libpgport', 'lib', 'misc');
 	$libpgport->AddDefine('FRONTEND');
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 1d540fe489..4d0d09a5dd 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -403,6 +403,7 @@ ConfigData
 ConfigVariable
 ConnCacheEntry
 ConnCacheKey
+ConnParams
 ConnStatusType
 ConnType
 ConnectionStateEnum
-- 
2.21.1 (Apple Git-122.3)

=

v36-0003-Parameterizing-parallel-slot-result-handling.patch (application/octet-stream, 11.3K) download
From 968f9bcb440580f912d34a74ed2928e8658fb710 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Tue, 2 Feb 2021 12:35:56 -0800
Subject: [PATCH v36 3/5] Parameterizing parallel slot result handling

The function consumeQueryResult was being used to handle all results
returned by queries executed through the parallel slot interface,
but this hardcodes knowledge about the expectations of reindexdb and
vacuumdb such as the expected result status being PGRES_COMMAND_OK
(as opposed to, say, PGRES_TUPLES_OK).

Reworking the slot interface to optionally include a PGresultHandler
and related fields per slot.  The idea is that a caller who executes
a command or query through the slot can set the handler to be called
when the query completes.

The old logic of consumeQueryResults is moved into a new callback
function, TableCommandSlotHandler(), which gets registered as the
slot handler explicitly from vacuumdb and reindexdb.  This is
defined in fe_utils/parallel_slot.c rather than somewhere in
src/bin/scripts where its only callers reside, partly to keep it
close to the rest of the shared parallel slot handling code and
partly in anticipation that other utility programs will eventually
want to use it also.

Adding a default handler which is used to handle results for slots
which have no handler explicitly registered.  The default simply
checks the status of the result and makes a judgement about whether
the status is ok, similarly to psql's AcceptResult().  I also
considered whether to just have a missing handler always be an
error, but decided against requiring users of the parallel slot
infrastructure to pedantically specify the default handler.  Both
designs seem reasonable, but the tie-breaker for me is that edge
cases that do not come up in testing will be better handled in
production with this design than with pedantically erroring out.

The expectation of this commit is that pg_amcheck will have handlers
for table and index checks which will process the PGresults of calls
to the amcheck functions.  This commit sets up the infrastructure
necessary to support those handlers being different from the one
used by vacuumdb and reindexdb.
---
 src/bin/scripts/reindexdb.c          |   1 +
 src/bin/scripts/vacuumdb.c           |   1 +
 src/fe_utils/parallel_slot.c         | 142 +++++++++++++++++++++------
 src/include/fe_utils/parallel_slot.h |  32 ++++++
 4 files changed, 148 insertions(+), 28 deletions(-)

diff --git a/src/bin/scripts/reindexdb.c b/src/bin/scripts/reindexdb.c
index 7781fb1151..29394d4a4a 100644
--- a/src/bin/scripts/reindexdb.c
+++ b/src/bin/scripts/reindexdb.c
@@ -466,6 +466,7 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 			goto finish;
 		}
 
+		ParallelSlotSetHandler(free_slot, TableCommandSlotHandler, NULL);
 		run_reindex_command(free_slot->connection, process_type, objname,
 							echo, verbose, concurrently, true);
 
diff --git a/src/bin/scripts/vacuumdb.c b/src/bin/scripts/vacuumdb.c
index ed320817bc..1158f7b776 100644
--- a/src/bin/scripts/vacuumdb.c
+++ b/src/bin/scripts/vacuumdb.c
@@ -713,6 +713,7 @@ vacuum_one_database(const ConnParams *cparams,
 		 * Execute the vacuum.  All errors are handled in processQueryResult
 		 * through ParallelSlotsGetIdle.
 		 */
+		ParallelSlotSetHandler(free_slot, TableCommandSlotHandler, NULL);
 		run_vacuum_command(free_slot->connection, sql.data,
 						   echo, tabname);
 
diff --git a/src/fe_utils/parallel_slot.c b/src/fe_utils/parallel_slot.c
index 3987a4702b..8e0c65988d 100644
--- a/src/fe_utils/parallel_slot.c
+++ b/src/fe_utils/parallel_slot.c
@@ -30,7 +30,7 @@
 
 static void init_slot(ParallelSlot *slot, PGconn *conn);
 static int	select_loop(int maxFd, fd_set *workerset);
-static bool processQueryResult(PGconn *conn, PGresult *result);
+static bool handleOneQueryResult(ParallelSlot *slot, PGresult *result);
 
 static void
 init_slot(ParallelSlot *slot, PGconn *conn)
@@ -38,53 +38,46 @@ init_slot(ParallelSlot *slot, PGconn *conn)
 	slot->connection = conn;
 	/* Initially assume connection is idle */
 	slot->isFree = true;
+	ParallelSlotClearHandler(slot);
 }
 
 /*
- * Process (and delete) a query result.  Returns true if there's no error,
- * false otherwise -- but errors about trying to work on a missing relation
- * are reported and subsequently ignored.
+ * Invoke the slot's handler for a single query result, or fall back to the
+ * default handler if none is defined for the slot.  Returns true if the
+ * handler reports that there's no error, false otherwise.
  */
 static bool
-processQueryResult(PGconn *conn, PGresult *result)
+handleOneQueryResult(ParallelSlot *slot, PGresult *result)
 {
-	/*
-	 * If it's an error, report it.  Errors about a missing table are harmless
-	 * so we continue processing; but die for other errors.
-	 */
-	if (PQresultStatus(result) != PGRES_COMMAND_OK)
-	{
-		char	   *sqlState = PQresultErrorField(result, PG_DIAG_SQLSTATE);
+	PGresultHandler handler = slot->handler;
 
-		pg_log_error("processing of database \"%s\" failed: %s",
-					 PQdb(conn), PQerrorMessage(conn));
+	if (!handler)
+		handler = DefaultSlotHandler;
 
-		if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
-		{
-			PQclear(result);
-			return false;
-		}
-	}
+	/* On failure, the handler should return NULL after freeing the result */
+	if (!handler(result, slot->connection, slot->handler_context))
+		return false;
 
+	/* Ok, we have to free it ourself */
 	PQclear(result);
 	return true;
 }
 
 /*
- * Consume all the results generated for the given connection until
+ * Handle all the results generated for the given connection until
  * nothing remains.  If at least one error is encountered, return false.
  * Note that this will block if the connection is busy.
  */
 static bool
-consumeQueryResult(PGconn *conn)
+handleQueryResults(ParallelSlot *slot)
 {
 	bool		ok = true;
 	PGresult   *result;
 
-	SetCancelConn(conn);
-	while ((result = PQgetResult(conn)) != NULL)
+	SetCancelConn(slot->connection);
+	while ((result = PQgetResult(slot->connection)) != NULL)
 	{
-		if (!processQueryResult(conn, result))
+		if (!handleOneQueryResult(slot, result))
 			ok = false;
 	}
 	ResetCancelConn();
@@ -227,14 +220,15 @@ ParallelSlotsGetIdle(ParallelSlot *slots, int numslots)
 
 				if (result != NULL)
 				{
-					/* Check and discard the command result */
-					if (!processQueryResult(slots[i].connection, result))
+					/* Handle and discard the command result */
+					if (!handleOneQueryResult(slots + i, result))
 						return NULL;
 				}
 				else
 				{
 					/* This connection has become idle */
 					slots[i].isFree = true;
+					ParallelSlotClearHandler(slots + i);
 					if (firstFree < 0)
 						firstFree = i;
 					break;
@@ -329,9 +323,101 @@ ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots)
 
 	for (i = 0; i < numslots; i++)
 	{
-		if (!consumeQueryResult((slots + i)->connection))
+		if (!handleQueryResults(slots + i))
 			return false;
 	}
 
 	return true;
 }
+
+/*
+ * DefaultSlotHandler
+ *
+ * PGresultHandler for query results from slots with no handler registered.
+ * Success or failure is determined entirely by examining the status of the
+ * query result.  This is very basic, but users who need better can register a
+ * custom handler.
+ *
+ * res: PGresult from the query executed on the slot's connection
+ * conn: connection belonging to the slot
+ * context: unused
+ */
+PGresult *
+DefaultSlotHandler(PGresult *res, PGconn *conn, void *context)
+{
+	switch (PQresultStatus(res))
+	{
+			/* Success codes */
+		case PGRES_EMPTY_QUERY:
+		case PGRES_COMMAND_OK:
+		case PGRES_TUPLES_OK:
+		case PGRES_COPY_OUT:
+		case PGRES_COPY_IN:
+		case PGRES_COPY_BOTH:
+		case PGRES_SINGLE_TUPLE:
+			/* Ok */
+			return res;
+
+			/* Error codes */
+		case PGRES_BAD_RESPONSE:
+		case PGRES_NONFATAL_ERROR:
+		case PGRES_FATAL_ERROR:
+			break;
+
+			/* Intentionally no default here */
+	}
+
+	/*
+	 * Handle all error cases here, including anything not matched in the
+	 * switch (though that should not happen.)  The 'query' argument may be
+	 * NULL or garbage left over from a prior usage of the lot.  Don't include
+	 * it in the error message!
+	 */
+	pg_log_error("processing in database \"%s\" failed: %s", PQdb(conn),
+				 PQerrorMessage(conn));
+	PQclear(res);
+	return NULL;
+}
+
+/*
+ * TableCommandSlotHandler
+ *
+ * PGresultHandler for results of commands (not queries) against tables.
+ *
+ * Requires that the result status is either PGRES_COMMAND_OK or an error about
+ * a missing table.  This is useful for utilities that compile a list of tables
+ * to process and then run commands (vacuum, reindex, or whatever) against
+ * those tables, as there is a race condition between the time the list is
+ * compiled and the time the command attempts to open the table.
+ *
+ * For missing tables, logs an error but allows processing to continue.
+ *
+ * For all other errors, logs an error and terminates further processing.
+ *
+ * res: PGresult from the query executed on the slot's connection
+ * conn: connection belonging to the slot
+ * context: unused
+ */
+PGresult *
+TableCommandSlotHandler(PGresult *res, PGconn *conn, void *context)
+{
+	/*
+	 * If it's an error, report it.  Errors about a missing table are harmless
+	 * so we continue processing; but die for other errors.
+	 */
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		char	   *sqlState = PQresultErrorField(res, PG_DIAG_SQLSTATE);
+
+		pg_log_error("processing of database \"%s\" failed: %s",
+					 PQdb(conn), PQerrorMessage(conn));
+
+		if (sqlState && strcmp(sqlState, ERRCODE_UNDEFINED_TABLE) != 0)
+		{
+			PQclear(res);
+			return NULL;
+		}
+	}
+
+	return res;
+}
diff --git a/src/include/fe_utils/parallel_slot.h b/src/include/fe_utils/parallel_slot.h
index 99eeb3328d..524d62306d 100644
--- a/src/include/fe_utils/parallel_slot.h
+++ b/src/include/fe_utils/parallel_slot.h
@@ -15,12 +15,39 @@
 #include "fe_utils/connect_utils.h"
 #include "libpq-fe.h"
 
+typedef PGresult *(*PGresultHandler) (PGresult *res, PGconn *conn,
+									  void *context);
+
 typedef struct ParallelSlot
 {
 	PGconn	   *connection;		/* One connection */
 	bool		isFree;			/* Is it known to be idle? */
+
+	/*
+	 * Prior to issuing a command or query on 'connection', a handler callback
+	 * function may optionally be registered to be invoked to process the
+	 * results, and context information may optionally be registered for use
+	 * by the handler.  If unset, these fields should be NULL.
+	 */
+	PGresultHandler handler;
+	void	   *handler_context;
 } ParallelSlot;
 
+static inline void
+ParallelSlotSetHandler(ParallelSlot *slot, PGresultHandler handler,
+					   void *context)
+{
+	slot->handler = handler;
+	slot->handler_context = context;
+}
+
+static inline void
+ParallelSlotClearHandler(ParallelSlot *slot)
+{
+	slot->handler = NULL;
+	slot->handler_context = NULL;
+}
+
 extern ParallelSlot *ParallelSlotsGetIdle(ParallelSlot *slots, int numslots);
 
 extern ParallelSlot *ParallelSlotsSetup(const ConnParams *cparams,
@@ -31,5 +58,10 @@ extern void ParallelSlotsTerminate(ParallelSlot *slots, int numslots);
 
 extern bool ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots);
 
+extern PGresult *DefaultSlotHandler(PGresult *res, PGconn *conn,
+									void *context);
+
+extern PGresult *TableCommandSlotHandler(PGresult *res, PGconn *conn,
+										 void *context);
 
 #endif							/* PARALLEL_SLOT_H */
-- 
2.21.1 (Apple Git-122.3)

=

v36-0004-Adding-contrib-module-pg_amcheck.patch (application/octet-stream, 132.6K) download
From 1fbb9903eaef4d5e30f986f77226fdd7fdf7c680 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Tue, 2 Feb 2021 12:36:59 -0800
Subject: [PATCH v36 4/5] Adding contrib module pg_amcheck

Adding new contrib module pg_amcheck, which is a command line
interface for running amcheck's verifications against tables and
indexes.
---
 contrib/Makefile                           |    1 +
 contrib/pg_amcheck/.gitignore              |    3 +
 contrib/pg_amcheck/Makefile                |   29 +
 contrib/pg_amcheck/pg_amcheck.c            | 1519 ++++++++++++++++++++
 contrib/pg_amcheck/pg_amcheck.h            |  135 ++
 contrib/pg_amcheck/t/001_basic.pl          |    9 +
 contrib/pg_amcheck/t/002_nonesuch.pl       |   78 +
 contrib/pg_amcheck/t/003_check.pl          |  475 ++++++
 contrib/pg_amcheck/t/004_verify_heapam.pl  |  496 +++++++
 contrib/pg_amcheck/t/005_opclass_damage.pl |   52 +
 doc/src/sgml/contrib.sgml                  |    1 +
 doc/src/sgml/filelist.sgml                 |    1 +
 doc/src/sgml/pgamcheck.sgml                | 1004 +++++++++++++
 src/tools/msvc/Install.pm                  |    2 +-
 src/tools/msvc/Mkvcbuild.pm                |    6 +-
 src/tools/pgindent/typedefs.list           |    4 +
 16 files changed, 3811 insertions(+), 4 deletions(-)
 create mode 100644 contrib/pg_amcheck/.gitignore
 create mode 100644 contrib/pg_amcheck/Makefile
 create mode 100644 contrib/pg_amcheck/pg_amcheck.c
 create mode 100644 contrib/pg_amcheck/pg_amcheck.h
 create mode 100644 contrib/pg_amcheck/t/001_basic.pl
 create mode 100644 contrib/pg_amcheck/t/002_nonesuch.pl
 create mode 100644 contrib/pg_amcheck/t/003_check.pl
 create mode 100644 contrib/pg_amcheck/t/004_verify_heapam.pl
 create mode 100644 contrib/pg_amcheck/t/005_opclass_damage.pl
 create mode 100644 doc/src/sgml/pgamcheck.sgml

diff --git a/contrib/Makefile b/contrib/Makefile
index f27e458482..a72dcf7304 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -30,6 +30,7 @@ SUBDIRS = \
 		old_snapshot	\
 		pageinspect	\
 		passwordcheck	\
+		pg_amcheck	\
 		pg_buffercache	\
 		pg_freespacemap \
 		pg_prewarm	\
diff --git a/contrib/pg_amcheck/.gitignore b/contrib/pg_amcheck/.gitignore
new file mode 100644
index 0000000000..c21a14de31
--- /dev/null
+++ b/contrib/pg_amcheck/.gitignore
@@ -0,0 +1,3 @@
+pg_amcheck
+
+/tmp_check/
diff --git a/contrib/pg_amcheck/Makefile b/contrib/pg_amcheck/Makefile
new file mode 100644
index 0000000000..bc61ee7970
--- /dev/null
+++ b/contrib/pg_amcheck/Makefile
@@ -0,0 +1,29 @@
+# contrib/pg_amcheck/Makefile
+
+PGFILEDESC = "pg_amcheck - detects corruption within database relations"
+PGAPPICON = win32
+
+PROGRAM = pg_amcheck
+OBJS = \
+	$(WIN32RES) \
+	pg_amcheck.o
+
+REGRESS_OPTS += --load-extension=amcheck --load-extension=pageinspect
+EXTRA_INSTALL += contrib/amcheck contrib/pageinspect
+
+TAP_TESTS = 1
+
+PG_CPPFLAGS = -I$(libpq_srcdir)
+PG_LIBS_INTERNAL = -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_pgport)
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+SHLIB_PREREQS = submake-libpq
+subdir = contrib/pg_amcheck
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/pg_amcheck/pg_amcheck.c b/contrib/pg_amcheck/pg_amcheck.c
new file mode 100644
index 0000000000..4209b5ec50
--- /dev/null
+++ b/contrib/pg_amcheck/pg_amcheck.c
@@ -0,0 +1,1519 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_amcheck.c
+ *		Detects corruption within database relations.
+ *
+ * Copyright (c) 2017-2020, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  contrib/pg_amcheck/pg_amcheck.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres_fe.h"
+
+#include "catalog/pg_class.h"
+#include "common/connect.h"
+#include "common/logging.h"
+#include "common/username.h"
+#include "fe_utils/cancel.h"
+#include "fe_utils/connect_utils.h"
+#include "fe_utils/option_utils.h"
+#include "fe_utils/parallel_slot.h"
+#include "fe_utils/query_utils.h"
+#include "fe_utils/simple_list.h"
+#include "fe_utils/string_utils.h"
+#include "getopt_long.h"		/* pgrminclude ignore */
+#include "libpq-fe.h"
+#include "pg_amcheck.h"
+#include "pqexpbuffer.h"		/* pgrminclude ignore */
+#include "storage/block.h"
+
+/* Keep this in order by CheckType */
+static const CheckTypeFilter ctfilter[] = {
+	{
+		.relam = HEAP_TABLE_AM_OID,
+		.relkinds = CppAsString2(RELKIND_RELATION) ","
+		CppAsString2(RELKIND_MATVIEW) ","
+		CppAsString2(RELKIND_TOASTVALUE),
+		.typname = "heap"
+	},
+	{
+		.relam = BTREE_AM_OID,
+		.relkinds = CppAsString2(RELKIND_INDEX),
+		.typname = "btree index"
+	}
+};
+
+/*
+ * Query for determining if contrib's amcheck is installed.  If so, selects the
+ * namespace name where amcheck's functions can be found.
+ */
+static const char *amcheck_sql =
+"SELECT n.nspname, x.extversion"
+"\nFROM pg_catalog.pg_extension x"
+"\nJOIN pg_catalog.pg_namespace n"
+"\nON x.extnamespace OPERATOR(pg_catalog.=) n.oid"
+"\nWHERE x.extname OPERATOR(pg_catalog.=) 'amcheck'";
+
+int
+main(int argc, char *argv[])
+{
+	static struct option long_options[] = {
+		/* Connection options */
+		{"host", required_argument, NULL, 'h'},
+		{"port", required_argument, NULL, 'p'},
+		{"username", required_argument, NULL, 'U'},
+		{"no-password", no_argument, NULL, 'w'},
+		{"password", no_argument, NULL, 'W'},
+		{"maintenance-db", required_argument, NULL, 1},
+
+		/* check options */
+		{"all", no_argument, NULL, 'a'},
+		{"dbname", required_argument, NULL, 'd'},
+		{"exclude-dbname", required_argument, NULL, 'D'},
+		{"echo", no_argument, NULL, 'e'},
+		{"heapallindexed", no_argument, NULL, 'H'},
+		{"index", required_argument, NULL, 'i'},
+		{"exclude-index", required_argument, NULL, 'I'},
+		{"jobs", required_argument, NULL, 'j'},
+		{"parent-check", no_argument, NULL, 'P'},
+		{"quiet", no_argument, NULL, 'q'},
+		{"relation", required_argument, NULL, 'r'},
+		{"exclude-relation", required_argument, NULL, 'R'},
+		{"schema", required_argument, NULL, 's'},
+		{"exclude-schema", required_argument, NULL, 'S'},
+		{"table", required_argument, NULL, 't'},
+		{"exclude-table", required_argument, NULL, 'T'},
+		{"verbose", no_argument, NULL, 'v'},
+		{"exclude-indexes", no_argument, NULL, 2},
+		{"exclude-toast", no_argument, NULL, 3},
+		{"exclude-toast-pointers", no_argument, NULL, 4},
+		{"on-error-stop", no_argument, NULL, 5},
+		{"skip", required_argument, NULL, 6},
+		{"startblock", required_argument, NULL, 7},
+		{"endblock", required_argument, NULL, 8},
+		{"rootdescend", no_argument, NULL, 9},
+		{"no-dependents", no_argument, NULL, 10},
+
+		{NULL, 0, NULL, 0}
+	};
+
+	const char *progname;
+	int			optindex;
+	int			c;
+
+	const char *maintenance_db = NULL;
+	const char *connect_db = NULL;
+	const char *host = NULL;
+	const char *port = NULL;
+	const char *username = NULL;
+	enum trivalue prompt_password = TRI_DEFAULT;
+	ConnParams	cparams;
+
+	amcheckOptions checkopts = {
+		.alldb = false,
+		.echo = false,
+		.quiet = false,
+		.verbose = false,
+		.dependents = true,
+		.no_indexes = false,
+		.on_error_stop = false,
+		.parent_check = false,
+		.rootdescend = false,
+		.heapallindexed = false,
+		.exclude_toast = false,
+		.reconcile_toast = true,
+		.skip = "none",
+		.jobs = -1,
+		.startblock = -1,
+		.endblock = -1
+	};
+
+	amcheckObjects objects = {
+		.databases = {NULL, NULL},
+		.schemas = {NULL, NULL},
+		.tables = {NULL, NULL},
+		.indexes = {NULL, NULL},
+		.exclude_databases = {NULL, NULL},
+		.exclude_schemas = {NULL, NULL},
+		.exclude_tables = {NULL, NULL},
+		.exclude_indexes = {NULL, NULL}
+	};
+
+	pg_logging_init(argv[0]);
+	progname = get_progname(argv[0]);
+	set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("contrib"));
+
+	handle_help_version_opts(argc, argv, progname, help);
+
+	/* process command-line options */
+	while ((c = getopt_long(argc, argv, "ad:D:eh:Hi:I:j:p:Pqr:R:s:S:t:T:U:wWv",
+							long_options, &optindex)) != -1)
+	{
+		char	   *endptr;
+
+		switch (c)
+		{
+			case 'a':
+				checkopts.alldb = true;
+				break;
+			case 'd':
+				simple_string_list_append(&objects.databases, optarg);
+				break;
+			case 'D':
+				simple_string_list_append(&objects.exclude_databases, optarg);
+				break;
+			case 'e':
+				checkopts.echo = true;
+				break;
+			case 'h':
+				host = pg_strdup(optarg);
+				break;
+			case 'H':
+				checkopts.heapallindexed = true;
+				break;
+			case 'i':
+				simple_string_list_append(&objects.indexes, optarg);
+				break;
+			case 'I':
+				simple_string_list_append(&objects.exclude_indexes, optarg);
+				break;
+			case 'j':
+				checkopts.jobs = atoi(optarg);
+				if (checkopts.jobs <= 0)
+				{
+					pg_log_error("number of parallel jobs must be at least 1");
+					exit(1);
+				}
+				break;
+			case 'p':
+				port = pg_strdup(optarg);
+				break;
+			case 'P':
+				checkopts.parent_check = true;
+				break;
+			case 'q':
+				checkopts.quiet = true;
+				break;
+			case 'r':
+				simple_string_list_append(&objects.indexes, optarg);
+				simple_string_list_append(&objects.tables, optarg);
+				break;
+			case 'R':
+				simple_string_list_append(&objects.exclude_tables, optarg);
+				simple_string_list_append(&objects.exclude_indexes, optarg);
+				break;
+			case 's':
+				simple_string_list_append(&objects.schemas, optarg);
+				break;
+			case 'S':
+				simple_string_list_append(&objects.exclude_schemas, optarg);
+				break;
+			case 't':
+				simple_string_list_append(&objects.tables, optarg);
+				break;
+			case 'T':
+				simple_string_list_append(&objects.exclude_tables, optarg);
+				break;
+			case 'U':
+				username = pg_strdup(optarg);
+				break;
+			case 'w':
+				prompt_password = TRI_NO;
+				break;
+			case 'W':
+				prompt_password = TRI_YES;
+				break;
+			case 'v':
+				checkopts.verbose = true;
+				pg_logging_increase_verbosity();
+				break;
+			case 1:
+				maintenance_db = pg_strdup(optarg);
+				break;
+			case 2:
+				checkopts.no_indexes = true;
+				break;
+			case 3:
+				checkopts.exclude_toast = true;
+				break;
+			case 4:
+				checkopts.reconcile_toast = false;
+				break;
+			case 5:
+				checkopts.on_error_stop = true;
+				break;
+			case 6:
+				if (pg_strcasecmp(optarg, "all-visible") == 0)
+					checkopts.skip = "all visible";
+				else if (pg_strcasecmp(optarg, "all-frozen") == 0)
+					checkopts.skip = "all frozen";
+				else
+				{
+					fprintf(stderr, _("invalid skip options"));
+					exit(1);
+				}
+				break;
+			case 7:
+				checkopts.startblock = strtol(optarg, &endptr, 10);
+				if (*endptr != '\0')
+				{
+					fprintf(stderr,
+							_("relation starting block argument contains garbage characters"));
+					exit(1);
+				}
+				if (checkopts.startblock > (long) MaxBlockNumber)
+				{
+					fprintf(stderr,
+							_("relation starting block argument out of bounds"));
+					exit(1);
+				}
+				break;
+			case 8:
+				checkopts.endblock = strtol(optarg, &endptr, 10);
+				if (*endptr != '\0')
+				{
+					fprintf(stderr,
+							_("relation ending block argument contains garbage characters"));
+					exit(1);
+				}
+				if (checkopts.startblock > (long) MaxBlockNumber)
+				{
+					fprintf(stderr,
+							_("relation ending block argument out of bounds"));
+					exit(1);
+				}
+				break;
+			case 9:
+				checkopts.rootdescend = true;
+				checkopts.parent_check = true;
+				break;
+			case 10:
+				checkopts.dependents = false;
+				break;
+			default:
+				fprintf(stderr,
+						_("Try \"%s --help\" for more information.\n"),
+						progname);
+				exit(1);
+		}
+	}
+
+	if (checkopts.endblock >= 0 && checkopts.endblock < checkopts.startblock)
+	{
+		pg_log_error("relation ending block argument precedes starting block argument");
+		exit(1);
+	}
+
+	/* non-option arguments specify database names */
+	while (optind < argc)
+	{
+		if (connect_db == NULL)
+			connect_db = argv[optind];
+		simple_string_list_append(&objects.databases, argv[optind]);
+		optind++;
+	}
+
+	/* fill cparams except for dbname, which is set below */
+	cparams.pghost = host;
+	cparams.pgport = port;
+	cparams.pguser = username;
+	cparams.prompt_password = prompt_password;
+	cparams.override_dbname = NULL;
+
+	setup_cancel_handler(NULL);
+
+	/* choose the database for our initial connection */
+	if (maintenance_db)
+		cparams.dbname = maintenance_db;
+	else if (connect_db != NULL)
+		cparams.dbname = connect_db;
+	else if (objects.databases.head != NULL)
+		cparams.dbname = objects.databases.head->val;
+	else
+	{
+		const char *default_db;
+
+		if (getenv("PGDATABASE"))
+			default_db = getenv("PGDATABASE");
+		else if (getenv("PGUSER"))
+			default_db = getenv("PGUSER");
+		else
+			default_db = get_user_name_or_exit(progname);
+
+		if (objects.databases.head == NULL)
+			simple_string_list_append(&objects.databases, default_db);
+
+		cparams.dbname = default_db;
+	}
+
+	check_each_database(&cparams, &objects, &checkopts, progname);
+
+	exit(0);
+}
+
+/*
+ * check_each_database
+ *
+ * Connects to the initial database and resolves a list of all databases that
+ * should be checked per the user supplied options.  Sequentially checks each
+ * database in the list.
+ *
+ * The user supplied options may include zero databases, or only one database,
+ * in which case we could skip the step of resolving a list of databases, but
+ * it seems not worth optimizing, especially considering that there are
+ * multiple ways in which no databases or just one database might be specified,
+ * including a pattern that happens to match no entries or to match only one
+ * entry in pg_database.
+ *
+ * cparams: parameters for the initial database connection
+ * objects: lists of include and exclude patterns for filtering objects
+ * checkopts: user supplied program options
+ * progname: name of this program, such as "pg_amcheck"
+ */
+static void
+check_each_database(ConnParams *cparams, const amcheckObjects *objects,
+					const amcheckOptions *checkopts, const char *progname)
+{
+	PGconn	   *conn;
+	PGresult   *databases;
+	PQExpBufferData sql;
+	int			ntups;
+	int			i;
+	SimpleStringList dbregex = {NULL, NULL};
+	SimpleStringList exclude = {NULL, NULL};
+
+	/*
+	 * Get a list of all database SQL regexes to use for selecting database
+	 * names.  We assemble these regexes from fully-qualified relation
+	 * patterns and database patterns.  This process may result in the same
+	 * database regex in the list multiple times, but the query against
+	 * pg_database will deduplice, so we don't care.
+	 */
+	get_db_regexes_from_fqrps(&dbregex, &objects->tables);
+	get_db_regexes_from_fqrps(&dbregex, &objects->indexes);
+	get_db_regexes_from_patterns(&dbregex, &objects->databases);
+
+	/*
+	 * Assemble SQL regexes for databases to be excluded.  Note that excluded
+	 * relations are not considered here, as excluding relation x.y.z does not
+	 * imply excluding database x.  Excluding x.*.* would imply excluding
+	 * database x, but we do not check for that here.
+	 */
+	get_db_regexes_from_patterns(&exclude, &objects->exclude_databases);
+
+	conn = connectMaintenanceDatabase(cparams, progname, checkopts->echo);
+
+	initPQExpBuffer(&sql);
+	dbname_select(conn, &sql, &dbregex, checkopts->alldb);
+	appendPQExpBufferStr(&sql, "\nEXCEPT");
+	dbname_select(conn, &sql, &exclude, false);
+	executeCommand(conn, "RESET search_path;", checkopts->echo);
+	databases = executeQuery(conn, sql.data, checkopts->echo);
+	if (PQresultStatus(databases) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("query failed: %s", PQerrorMessage(conn));
+		pg_log_error("query was: %s", sql.data);
+		PQfinish(conn);
+		exit(1);
+	}
+
+	termPQExpBuffer(&sql);
+	PQclear(executeQuery(conn, ALWAYS_SECURE_SEARCH_PATH_SQL, checkopts->echo));
+	PQfinish(conn);
+
+	ntups = PQntuples(databases);
+	if (ntups == 0 && !checkopts->quiet)
+		printf(_("%s: no databases to check\n"), progname);
+
+	for (i = 0; i < ntups; i++)
+	{
+		cparams->override_dbname = PQgetvalue(databases, i, 0);
+		check_one_database(cparams, objects, checkopts, progname);
+	}
+
+	PQclear(databases);
+}
+
+/*
+ * string_in_list
+ *
+ * Returns whether a given string is in the list of strings.
+ */
+static bool
+string_in_list(const SimpleStringList *list, const char *str)
+{
+	const SimpleStringListCell *cell;
+
+	for (cell = list->head; cell; cell = cell->next)
+		if (strcmp(cell->val, str) == 0)
+			return true;
+	return false;
+}
+
+/*
+ * check_one_database
+ *
+ * Connects to this next database and checks all relations that match the
+ * supplied objects list.  Patterns in the object lists are matched to the
+ * relations that exit in this next database.
+ *
+ * cparams: parameters for this next database connection
+ * objects: lists of include and exclude patterns for filtering objects
+ * checkopts: user supplied program options
+ * progname: name of this program, such as "pg_amcheck"
+ */
+static void
+check_one_database(const ConnParams *cparams, const amcheckObjects *objects,
+				   const amcheckOptions *checkopts, const char *progname)
+{
+	PQExpBufferData sql;
+	PGconn	   *conn;
+	PGresult   *result;
+	ParallelSlot *slots;
+	int			ntups;
+	int			i;
+	int			parallel_workers;
+	bool		inclusive;
+	bool		failed = false;
+	char	   *amcheck_schema = NULL;
+
+	conn = connectDatabase(cparams, progname, checkopts->echo, false, true);
+
+	if (!checkopts->quiet)
+	{
+		printf(_("%s: checking database \"%s\"\n"),
+			   progname, PQdb(conn));
+		fflush(stdout);
+	}
+
+	/*
+	 * Verify that amcheck is installed for this next database.  User error
+	 * could result in a database not having amcheck that should have it, but
+	 * we also could be iterating over multiple databases where not all of
+	 * them have amcheck installed (for example, 'template1').
+	 */
+	result = executeQuery(conn, amcheck_sql, checkopts->echo);
+	if (PQresultStatus(result) != PGRES_TUPLES_OK)
+	{
+		/* Querying the catalog failed. */
+		pg_log_error(_("%s: database \"%s\": %s\n"),
+					 progname, PQdb(conn), PQerrorMessage(conn));
+		pg_log_error(_("%s: query was: %s"), progname, amcheck_sql);
+		PQclear(result);
+		PQfinish(conn);
+		return;
+	}
+	ntups = PQntuples(result);
+	if (ntups == 0)
+	{
+		/* Querying the catalog succeeded, but amcheck is missing. */
+		if (!checkopts->quiet &&
+			(checkopts->verbose ||
+			 string_in_list(&objects->databases, PQdb(conn))))
+		{
+			printf(_("%s: skipping database \"%s\": amcheck is not installed"),
+				   progname, PQdb(conn));
+		}
+		PQfinish(conn);
+		return;
+	}
+	amcheck_schema = PQgetvalue(result, 0, 0);
+	if (checkopts->verbose)
+		printf(_("%s: in database \"%s\": using amcheck version \"%s\" in schema \"%s\""),
+			   progname, PQdb(conn), PQgetvalue(result, 0, 1), amcheck_schema);
+	amcheck_schema = PQescapeIdentifier(conn, amcheck_schema, strlen(amcheck_schema));
+	PQclear(result);
+
+	/*
+	 * If we were given no tables nor indexes to check, then we select all
+	 * targets not excluded.  Otherwise, we select only the targets that we
+	 * were given.
+	 */
+	inclusive = objects->tables.head == NULL &&
+		objects->indexes.head == NULL;
+
+	initPQExpBuffer(&sql);
+	target_select(conn, &sql, objects, checkopts, progname, inclusive);
+	executeCommand(conn, "RESET search_path;", checkopts->echo);
+	result = executeQuery(conn, sql.data, checkopts->echo);
+	if (PQresultStatus(result) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("query failed: %s", PQerrorMessage(conn));
+		pg_log_error("query was: %s", sql.data);
+		PQfinish(conn);
+		exit(1);
+	}
+	termPQExpBuffer(&sql);
+	PQclear(executeQuery(conn, ALWAYS_SECURE_SEARCH_PATH_SQL, checkopts->echo));
+
+	/*
+	 * If no rows are returned, there are no matching relations, so we are
+	 * done.
+	 */
+	ntups = PQntuples(result);
+	if (ntups == 0)
+	{
+		PQclear(result);
+		PQfinish(conn);
+		PQfreemem(amcheck_schema);
+		return;
+	}
+
+	/*
+	 * Ensure parallel_workers is sane.  If there are more connections than
+	 * relations to be checked, we don't need to use them all.
+	 */
+	parallel_workers = checkopts->jobs;
+	if (parallel_workers > ntups)
+		parallel_workers = ntups;
+	if (parallel_workers <= 0)
+		parallel_workers = 1;
+
+	/*
+	 * Setup the database connections.  We reuse the connection we already
+	 * have for the first slot.  If not in parallel mode, the first slot in
+	 * the array contains the connection.
+	 */
+	slots = ParallelSlotsSetup(cparams, progname, checkopts->echo, conn,
+							   parallel_workers);
+
+	initPQExpBuffer(&sql);
+
+	/*
+	 * Loop over all objects to be checked, and execute amcheck checking
+	 * commands for each.  We do not wait for the checks to complete, nor do
+	 * we handle the results of those checks in the loop.  We register
+	 * handlers for doing all that.
+	 */
+	for (i = 0; i < ntups; i++)
+	{
+		ParallelSlot *free_slot;
+
+		CheckType	checktype = atoi(PQgetvalue(result, i, 0));
+		Oid			reloid = atooid(PQgetvalue(result, i, 1));
+
+		if (CancelRequested)
+		{
+			failed = true;
+			goto finish;
+		}
+
+		/*
+		 * Get a parallel slot for the next amcheck command, blocking if
+		 * necessary until one is available, or until a previously issued slot
+		 * command fails, indicating that we should abort checking the
+		 * remaining objects.
+		 */
+		free_slot = ParallelSlotsGetIdle(slots, parallel_workers);
+		if (!free_slot)
+		{
+			/*
+			 * Something failed.  We don't need to know what it was, because
+			 * the handler should already have emitted the necessary error
+			 * messages.
+			 */
+			failed = true;
+			goto finish;
+		}
+
+		/* Execute the amcheck command for the given relation type. */
+		switch (checktype)
+		{
+				/* heapam types */
+			case CT_TABLE:
+				prepare_table_command(&sql, checkopts, reloid, amcheck_schema);
+				ParallelSlotSetHandler(free_slot, VerifyHeapamSlotHandler, sql.data);
+				run_command(free_slot->connection, sql.data, checkopts, reloid,
+							ctfilter[checktype].typname);
+				break;
+
+				/* btreeam types */
+			case CT_BTREE:
+				prepare_btree_command(&sql, checkopts, reloid, amcheck_schema);
+				ParallelSlotSetHandler(free_slot, VerifyBtreeSlotHandler, NULL);
+				run_command(free_slot->connection, sql.data, checkopts, reloid,
+							ctfilter[checktype].typname);
+				break;
+
+				/* intentionally no default here */
+		}
+	}
+
+	/*
+	 * Wait for all slots to complete, or for one to indicate that an error
+	 * occurred.  Like above, we rely on the handler emitting the necessary
+	 * error messages.
+	 */
+	if (!ParallelSlotsWaitCompletion(slots, parallel_workers))
+		failed = true;
+
+finish:
+	ParallelSlotsTerminate(slots, parallel_workers);
+	pg_free(slots);
+
+	termPQExpBuffer(&sql);
+
+	if (amcheck_schema != NULL)
+		PQfreemem(amcheck_schema);
+
+	if (failed)
+		exit(1);
+}
+
+/*
+ * prepare_table_command
+ *
+ * Creates a SQL command for running amcheck checking on the given heap
+ * relation.  The command is phrased as a SQL query, with column order and
+ * names matching the expectations of VerifyHeapamSlotHandler, which will
+ * receive and handle each row returned from the verify_heapam() function.
+ *
+ * sql: buffer into which the table checking command will be written
+ * checkopts: user supplied program options
+ * reloid: relation of the table to be checked
+ */
+static void
+prepare_table_command(PQExpBuffer sql, const amcheckOptions *checkopts,
+					  Oid reloid, const char *amcheck_schema)
+{
+	resetPQExpBuffer(sql);
+	appendPQExpBuffer(sql,
+					  "SELECT n.nspname, c.relname, v.blkno, v.offnum, v.attnum, v.msg"
+					  "\nFROM %s.verify_heapam("
+					  "\nrelation := %u,"
+					  "\non_error_stop := %s,"
+					  "\ncheck_toast := %s,"
+					  "\nskip := '%s'",
+					  amcheck_schema,
+					  reloid,
+					  checkopts->on_error_stop ? "true" : "false",
+					  checkopts->reconcile_toast ? "true" : "false",
+					  checkopts->skip);
+	if (checkopts->startblock >= 0)
+		appendPQExpBuffer(sql, ",\nstartblock := %ld", checkopts->startblock);
+	if (checkopts->endblock >= 0)
+		appendPQExpBuffer(sql, ",\nendblock := %ld", checkopts->endblock);
+	appendPQExpBuffer(sql, "\n) v,"
+					  "\npg_catalog.pg_class c"
+					  "\nJOIN pg_catalog.pg_namespace n"
+					  "\nON c.relnamespace OPERATOR(pg_catalog.=) n.oid"
+					  "\nWHERE c.oid OPERATOR(pg_catalog.=) %u",
+					  reloid);
+}
+
+/*
+ * prepare_btree_command
+ *
+ * Creates a SQL command for running amcheck checking on the given btree index
+ * relation.  The command does not select any columns, as btree checking
+ * functions do not return any, but rather return corruption information by
+ * raising errors, which VerifyBtreeSlotHandler expects.
+ *
+ * sql: buffer into which the table checking command will be written
+ * checkopts: user supplied program options
+ * reloid: relation of the table to be checked
+ */
+static void
+prepare_btree_command(PQExpBuffer sql, const amcheckOptions *checkopts,
+					  Oid reloid, const char *amcheck_schema)
+{
+	resetPQExpBuffer(sql);
+	if (checkopts->parent_check)
+		appendPQExpBuffer(sql,
+						  "SELECT %s.bt_index_parent_check("
+						  "\nindex := '%u'::regclass,"
+						  "\nheapallindexed := %s,"
+						  "\nrootdescend := %s)",
+						  amcheck_schema,
+						  reloid,
+						  (checkopts->heapallindexed ? "true" : "false"),
+						  (checkopts->rootdescend ? "true" : "false"));
+	else
+		appendPQExpBuffer(sql,
+						  "SELECT %s.bt_index_check("
+						  "\nindex := '%u'::regclass,"
+						  "\nheapallindexed := %s)",
+						  amcheck_schema,
+						  reloid,
+						  (checkopts->heapallindexed ? "true" : "false"));
+}
+
+/*
+ * run_command
+ *
+ * Sends a command to the server without waiting for the command to complete.
+ * Logs an error if the command cannot be sent, but otherwise any errors are
+ * expected to be handled by a ParallelSlotHandler.
+ *
+ * conn: connection to the server associated with the slot to use
+ * sql: query to send
+ * checkopts: user supplied program options
+ * reloid: oid of the object being checked, for error reporting
+ * typ: type of object being checked, for error reporting
+ */
+static void
+run_command(PGconn *conn, const char *sql, const amcheckOptions *checkopts,
+			Oid reloid, const char *typ)
+{
+	bool		status;
+
+	if (checkopts->echo)
+		printf("%s\n", sql);
+
+	status = PQsendQuery(conn, sql) == 1;
+
+	if (!status)
+	{
+		pg_log_error("check of %s with id %u in database \"%s\" failed: %s",
+					 typ, reloid, PQdb(conn), PQerrorMessage(conn));
+		pg_log_error("command was: %s", sql);
+	}
+}
+
+/*
+ * VerifyHeapamSlotHandler
+ *
+ * ParallelSlotHandler that receives results from a table checking command
+ * created by prepare_table_command and outputs the results for the user.
+ *
+ * res: result from an executed sql query
+ * conn: connection on which the sql query was executed
+ * context: the sql query being handled, as a cstring
+ */
+static PGresult *
+VerifyHeapamSlotHandler(PGresult *res, PGconn *conn, void *context)
+{
+	int			ntups = PQntuples(res);
+
+	if (PQresultStatus(res) == PGRES_TUPLES_OK)
+	{
+		int			i;
+
+		for (i = 0; i < ntups; i++)
+		{
+			if (!PQgetisnull(res, i, 4))
+				printf("relation %s.%s.%s, block %s, offset %s, attribute %s\n    %s\n",
+					   PQdb(conn),
+					   PQgetvalue(res, i, 0),	/* schema */
+					   PQgetvalue(res, i, 1),	/* relname */
+					   PQgetvalue(res, i, 2),	/* blkno */
+					   PQgetvalue(res, i, 3),	/* offnum */
+					   PQgetvalue(res, i, 4),	/* attnum */
+					   PQgetvalue(res, i, 5));	/* msg */
+
+			else if (!PQgetisnull(res, i, 3))
+				printf("relation %s.%s.%s, block %s, offset %s\n    %s\n",
+					   PQdb(conn),
+					   PQgetvalue(res, i, 0),	/* schema */
+					   PQgetvalue(res, i, 1),	/* relname */
+					   PQgetvalue(res, i, 2),	/* blkno */
+					   PQgetvalue(res, i, 3),	/* offnum */
+				/* attnum is null: 4 */
+					   PQgetvalue(res, i, 5));	/* msg */
+
+			else if (!PQgetisnull(res, i, 2))
+				printf("relation %s.%s.%s, block %s\n    %s\n",
+					   PQdb(conn),
+					   PQgetvalue(res, i, 0),	/* schema */
+					   PQgetvalue(res, i, 1),	/* relname */
+					   PQgetvalue(res, i, 2),	/* blkno */
+				/* offnum is null: 3 */
+				/* attnum is null: 4 */
+					   PQgetvalue(res, i, 5));	/* msg */
+
+			else if (!PQgetisnull(res, i, 1))
+				printf("relation %s.%s.%s\n    %s\n",
+					   PQdb(conn),
+					   PQgetvalue(res, i, 0),	/* schema */
+					   PQgetvalue(res, i, 1),	/* relname */
+				/* blkno is null:  2 */
+				/* offnum is null: 3 */
+				/* attnum is null: 4 */
+					   PQgetvalue(res, i, 5));	/* msg */
+
+			else
+				printf("%s.%s\n",
+					   PQdb(conn),
+					   PQgetvalue(res, i, 5));	/* msg */
+		}
+	}
+	else if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		printf("%s: %s\n", PQdb(conn), PQerrorMessage(conn));
+		printf(_("query was: %s\n"), (const char *) context);
+	}
+
+	return res;
+}
+
+/*
+ * VerifyBtreeSlotHandler
+ *
+ * ParallelSlotHandler that receives results from a btree checking command
+ * created by prepare_btree_command and outputs them for the user.  The results
+ * from the btree checking command is assumed to be empty, but when the results
+ * are an error code, the useful information about the corruption is expected
+ * in the connection's error message.
+ *
+ * res: result from an executed sql query
+ * conn: connection on which the sql query was executed
+ * context: unused
+ */
+static PGresult *
+VerifyBtreeSlotHandler(PGresult *res, PGconn *conn, void *context)
+{
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		printf("%s: %s\n", PQdb(conn), PQerrorMessage(conn));
+	return res;
+}
+
+/*
+ * help
+ *
+ * Prints help page for the program
+ *
+ * progname: the name of the executed program, such as "pg_amcheck"
+ */
+static void
+help(const char *progname)
+{
+	printf(_("%s checks objects in a PostgreSQL database for corruption.\n\n"), progname);
+	printf(_("Usage:\n"));
+	printf(_("  %s [OPTION]... [DBNAME]\n"), progname);
+	printf(_("\nTarget Options:\n"));
+	printf(_("  -a, --all                 check all databases\n"));
+	printf(_("  -d, --dbname=DBNAME       check specific database(s)\n"));
+	printf(_("  -D, --exclude-dbname=DBNAME do NOT check specific database(s)\n"));
+	printf(_("  -i, --index=INDEX         check specific index(es)\n"));
+	printf(_("  -I, --exclude-index=INDEX do NOT check specific index(es)\n"));
+	printf(_("  -r, --relation=RELNAME    check specific relation(s)\n"));
+	printf(_("  -R, --exclude-relation=RELNAME do NOT check specific relation(s)\n"));
+	printf(_("  -s, --schema=SCHEMA       check specific schema(s)\n"));
+	printf(_("  -S, --exclude-schema=SCHEMA do NOT check specific schema(s)\n"));
+	printf(_("  -t, --table=TABLE         check specific table(s)\n"));
+	printf(_("  -T, --exclude-table=TABLE do NOT check specific table(s)\n"));
+	printf(_("      --exclude-indexes     do NOT perform any index checking\n"));
+	printf(_("      --exclude-toast       do NOT check any toast tables or indexes\n"));
+	printf(_("      --no-dependents       do NOT automatically check dependent objects\n"));
+	printf(_("\nIndex Checking Options:\n"));
+	printf(_("  -H, --heapallindexed      check all heap tuples are found within indexes\n"));
+	printf(_("  -P, --parent-check        check parent/child relationships during index checking\n"));
+	printf(_("      --rootdescend         search from root page to refind tuples at the leaf level\n"));
+	printf(_("\nTable Checking Options:\n"));
+	printf(_("      --exclude-toast-pointers do NOT check relation toast pointers against toast\n"));
+	printf(_("      --on-error-stop       stop checking a relation at end of first corrupt page\n"));
+	printf(_("      --skip=OPTION         do NOT check \"all-frozen\" or \"all-visible\" blocks\n"));
+	printf(_("      --startblock          begin checking table(s) at the given starting block number\n"));
+	printf(_("      --endblock            check table(s) only up to the given ending block number\n"));
+	printf(_("\nConnection options:\n"));
+	printf(_("  -h, --host=HOSTNAME       database server host or socket directory\n"));
+	printf(_("  -p, --port=PORT           database server port\n"));
+	printf(_("  -U, --username=USERNAME   user name to connect as\n"));
+	printf(_("  -w, --no-password         never prompt for password\n"));
+	printf(_("  -W, --password            force password prompt\n"));
+	printf(_("  --maintenance-db=DBNAME   alternate maintenance database\n"));
+	printf(_("\nOther Options:\n"));
+	printf(_("  -e, --echo                show the commands being sent to the server\n"));
+	printf(_("  -j, --jobs=NUM            use this many concurrent connections to the server\n"));
+	printf(_("  -q, --quiet               don't write any messages\n"));
+	printf(_("  -v, --verbose             write a lot of output\n"));
+	printf(_("  -V, --version             output version information, then exit\n"));
+	printf(_("  -?, --help                show this help, then exit\n"));
+
+	printf(_("\nRead the description of the amcheck contrib module for details.\n"));
+	printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
+	printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
+}
+
+/*
+ * get_db_regexes_from_fqrps
+ *
+ * For each pattern in the patterns list, if it is in fully-qualified
+ * database.schema.name format (fully-qualifed relation pattern (fqrp)), parse
+ * the database portion of the pattern, convert it to SQL regex format, and
+ * append it to the databases list.  Patterns that are not fully-qualified are
+ * skipped over.  No deduplication of regexes is performed.
+ *
+ * regexes: list to which parsed and converted database regexes are appended
+ * patterns: list of all patterns to parse
+ */
+static void
+get_db_regexes_from_fqrps(SimpleStringList *regexes,
+						  const SimpleStringList *patterns)
+{
+	const SimpleStringListCell *cell;
+	PQExpBufferData dbnamebuf;
+	PQExpBufferData schemabuf;
+	PQExpBufferData namebuf;
+	int			encoding = pg_get_encoding_from_locale(NULL, false);
+
+	initPQExpBuffer(&dbnamebuf);
+	initPQExpBuffer(&schemabuf);
+	initPQExpBuffer(&namebuf);
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		/* parse the pattern as dbname.schema.relname, if possible */
+		patternToSQLRegex(encoding, &dbnamebuf, &schemabuf, &namebuf,
+						  cell->val, false);
+
+		/* add the database name (or pattern), if any, to the list */
+		if (dbnamebuf.data[0])
+			simple_string_list_append(regexes, dbnamebuf.data);
+
+		/* we do not use the schema or relname portions */
+
+		/* we may have dirtied the buffers */
+		resetPQExpBuffer(&dbnamebuf);
+		resetPQExpBuffer(&schemabuf);
+		resetPQExpBuffer(&namebuf);
+	}
+	termPQExpBuffer(&dbnamebuf);
+	termPQExpBuffer(&schemabuf);
+	termPQExpBuffer(&namebuf);
+}
+
+/*
+ * get_db_regexes_from_patterns
+ *
+ * Convert each unqualified pattern in the list to SQL regex format and append
+ * it to the regexes list.
+ *
+ * regexes: list to which converted regexes are appended
+ * patterns: list of patterns to be converted
+ */
+static void
+get_db_regexes_from_patterns(SimpleStringList *regexes,
+							 const SimpleStringList *patterns)
+{
+	const SimpleStringListCell *cell;
+	PQExpBufferData buf;
+	int			encoding = pg_get_encoding_from_locale(NULL, false);
+
+	initPQExpBuffer(&buf);
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		patternToSQLRegex(encoding, NULL, NULL, &buf, cell->val, false);
+		if (buf.data[0])
+			simple_string_list_append(regexes, buf.data);
+		resetPQExpBuffer(&buf);
+	}
+	termPQExpBuffer(&buf);
+}
+
+/*
+ * dbname_select
+ *
+ * Appends a statement which selects the names of all databases matching the
+ * given SQL regular expressions.
+ *
+ * conn: connection to the initial database
+ * sql: buffer into which the constructed sql statement is appended
+ * regexes: list of database name regular expressions to match
+ * alldb: when true, select all databases which allow connections
+ */
+static void
+dbname_select(PGconn *conn, PQExpBuffer sql, const SimpleStringList *regexes,
+			  bool alldb)
+{
+	SimpleStringListCell *cell;
+	const char *comma;
+
+	if (alldb)
+	{
+		appendPQExpBufferStr(sql, "\nSELECT datname::TEXT AS datname"
+							 "\nFROM pg_catalog.pg_database"
+							 "\nWHERE datallowconn");
+		return;
+	}
+	else if (regexes->head == NULL)
+	{
+		appendPQExpBufferStr(sql, "\nSELECT ''::TEXT AS datname"
+							 "\nWHERE false");
+		return;
+	}
+
+	appendPQExpBufferStr(sql, "\nSELECT datname::TEXT AS datname"
+						 "\nFROM pg_catalog.pg_database"
+						 "\nWHERE datallowconn"
+						 "\nAND datname::TEXT OPERATOR(pg_catalog.~) ANY(ARRAY[\n");
+	for (cell = regexes->head, comma = ""; cell; cell = cell->next, comma = ",\n")
+	{
+		appendPQExpBufferStr(sql, comma);
+		appendStringLiteralConn(sql, cell->val, conn);
+		appendPQExpBufferStr(sql, "::TEXT COLLATE pg_catalog.default");
+	}
+	appendPQExpBufferStr(sql, "\n]::TEXT[])");
+}
+
+/*
+ * schema_select
+ *
+ * Appends a statement which selects all schemas matching the given patterns
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql statement is appended
+ * fieldname: alias to use for the oid field within the created SELECT
+ * statement
+ * patterns: list of schema name patterns to match
+ * inclusive: when patterns is an empty list, whether the select statement
+ * should match all non-system schemas
+ */
+static void
+schema_select(PGconn *conn, PQExpBuffer sql, const SimpleStringList *patterns,
+			  bool inclusive)
+{
+	SimpleStringListCell *cell;
+	const char *comma;
+	int			encoding = PQclientEncoding(conn);
+
+	if (patterns->head == NULL)
+	{
+		if (!inclusive)
+			appendPQExpBufferStr(sql,
+								 "\nSELECT 0::pg_catalog.oid AS nspoid WHERE false");
+		else
+			appendPQExpBufferStr(sql,
+								 "\nSELECT oid AS nspoid"
+								 "\nFROM pg_catalog.pg_namespace"
+								 "\nWHERE oid OPERATOR(pg_catalog.!=) pg_catalog.regnamespace('pg_catalog')"
+								 "\nAND oid OPERATOR(pg_catalog.!=) pg_catalog.regnamespace('pg_toast')");
+		return;
+	}
+
+	appendPQExpBufferStr(sql,
+						 "\nSELECT oid AS nspoid"
+						 "\nFROM pg_catalog.pg_namespace"
+						 "\nWHERE nspname OPERATOR(pg_catalog.~) ANY(ARRAY[\n");
+	for (cell = patterns->head, comma = ""; cell; cell = cell->next, comma = ",\n")
+	{
+		PQExpBufferData regexbuf;
+
+		initPQExpBuffer(&regexbuf);
+		patternToSQLRegex(encoding, NULL, NULL, &regexbuf, cell->val, false);
+		appendPQExpBufferStr(sql, comma);
+		appendStringLiteralConn(sql, regexbuf.data, conn);
+		appendPQExpBufferStr(sql, "::TEXT COLLATE pg_catalog.default");
+		termPQExpBuffer(&regexbuf);
+	}
+	appendPQExpBufferStr(sql, "\n]::TEXT[])");
+}
+
+/*
+ * schema_cte
+ *
+ * Appends a Common Table Expression (CTE) which selects all schemas to be
+ * checked, with the CTE and oid field named as requested.  The CTE will select
+ * all schemas matching the include list except any schemas matching the
+ * exclude list.
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql statement is appended
+ * ctename: name of the schema CTE to be created
+ * fieldname: name of the oid field within the schema CTE to be created
+ * include: list of schema name patterns for inclusion
+ * exclude: list of schema name patterns for exclusion
+ * inclusive: when 'include' is an empty list, whether to use all schemas in
+ * the database in lieu of the include list.
+ */
+static void
+schema_cte(PGconn *conn, PQExpBuffer sql, const char *ctename,
+		   const SimpleStringList *include, const SimpleStringList *exclude,
+		   bool inclusive)
+{
+	appendPQExpBuffer(sql, "\n%s (nspoid) AS (", ctename);
+	schema_select(conn, sql, include, inclusive);
+	appendPQExpBufferStr(sql, "\nEXCEPT");
+	schema_select(conn, sql, exclude, false);
+	appendPQExpBufferStr(sql, "\n)");
+}
+
+/*
+ * append_ctfilter_quals
+ *
+ * Appends quals to a buffer that restrict the rows selected from pg_class to
+ * only those which match the given checktype.  No initial "WHERE" or "AND" is
+ * appended, nor do we surround our appended clauses in parens.  The caller is
+ * assumed to take care of such matters.
+ *
+ * sql: buffer into which the constructed sql quals are appended
+ * relname: name (or alias) of pg_class in the surrounding query
+ * checktype: struct containing filter info
+ */
+static void
+append_ctfilter_quals(PQExpBuffer sql, const char *relname, CheckType checktype)
+{
+	appendPQExpBuffer(sql,
+					  "%s.relam OPERATOR(pg_catalog.=) %u"
+					  "\nAND %s.relkind OPERATOR(pg_catalog.=) ANY(ARRAY[%s])",
+					  relname, ctfilter[checktype].relam,
+					  relname, ctfilter[checktype].relkinds);
+}
+
+/*
+ * relation_select
+ *
+ * Appends a statement which selects the oid of all relations matching the
+ * given parameters.  Expects a mixture of qualified and unqualified relation
+ * name patterns.
+ *
+ * For unqualified relation patterns, selects relations that match the relation
+ * name portion of the pattern which are in namespaces that are in the given
+ * namespace CTE.
+ *
+ * For qualified relation patterns, ignores the given namespace CTE and selects
+ * relations that match the relation name portion of the pattern which are in
+ * namespaces that match the schema portion of the pattern.
+ *
+ * For fully qualified relation patterns (database.schema.name), the pattern
+ * will be ignored unless the database portion of the pattern matches the name
+ * of the current database, as retrieved from conn.
+ *
+ * Only relations of the specified checktype will be selected.
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql statement is appended
+ * schemacte: name of the CTE which selects all schemas to be checked
+ * fieldname: alias to use for the oid field within the created SELECT
+ * statement
+ * patterns: list of (possibly qualified) relation name patterns to match
+ * checktype: the type of relation to select
+ * inclusive: when patterns is an empty list, whether the select statement
+ * should match all relations of the given type
+ */
+static void
+relation_select(PGconn *conn, PQExpBuffer sql, const char *schemacte,
+				const char *fieldname,
+				const SimpleStringList *patterns, CheckType checktype,
+				bool inclusive)
+{
+	SimpleStringListCell *cell;
+	const char *comma = "";
+	const char *qor = "";
+	PQExpBufferData qualified;
+	PQExpBufferData unqualified;
+	PQExpBufferData dbnamebuf;
+	PQExpBufferData schemabuf;
+	PQExpBufferData namebuf;
+	int			encoding = PQclientEncoding(conn);
+
+	if (patterns->head == NULL)
+	{
+		if (!inclusive)
+			appendPQExpBuffer(sql,
+							  "\nSELECT 0::pg_catalog.oid AS %s WHERE false",
+							  fieldname);
+		else
+		{
+			appendPQExpBuffer(sql,
+							  "\nSELECT oid AS %s"
+							  "\nFROM pg_catalog.pg_class c"
+							  "\nJOIN %s n"
+							  "\nON n.nspoid OPERATOR(pg_catalog.=) c.relnamespace"
+							  "\nWHERE ",
+							  fieldname, schemacte);
+			append_ctfilter_quals(sql, "c", checktype);
+		}
+		return;
+	}
+
+	/*
+	 * We have to distinguish between schema-qualified and unqualified
+	 * relation patterns.  The unqualified patterns need to be restricted by
+	 * the list of schemas returned by the schema CTE, but not so for the
+	 * qualified patterns.
+	 *
+	 * We treat fully-qualified relation patterns (database.schema.relation)
+	 * like schema-qualified patterns except that we also require the database
+	 * portion to match the current database name.
+	 */
+	initPQExpBuffer(&qualified);
+	initPQExpBuffer(&unqualified);
+	initPQExpBuffer(&dbnamebuf);
+	initPQExpBuffer(&schemabuf);
+	initPQExpBuffer(&namebuf);
+
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		patternToSQLRegex(encoding, &dbnamebuf, &schemabuf, &namebuf,
+						  cell->val, false);
+
+		if (schemabuf.data[0])
+		{
+			/* Qualified relation pattern */
+			appendPQExpBuffer(&qualified, "%s\n(", qor);
+
+			if (dbnamebuf.data[0])
+			{
+				/*
+				 * Fully-qualified relation pattern.  Require the database
+				 * name of our connection to match the database portion of the
+				 * relation pattern.
+				 */
+				appendPQExpBufferStr(&qualified, "\n");
+				appendStringLiteralConn(&qualified, PQdb(conn), conn);
+				appendPQExpBufferStr(&qualified,
+									 "::TEXT OPERATOR(pg_catalog.~) ");
+				appendStringLiteralConn(&qualified, dbnamebuf.data, conn);
+				appendPQExpBufferStr(&qualified,
+									 "::TEXT COLLATE pg_catalog.default AND");
+			}
+
+			/*
+			 * Require the namespace name to match the schema portion of the
+			 * relation pattern and the relation name to match the relname
+			 * portion of the relation pattern.
+			 */
+			appendPQExpBufferStr(&qualified,
+								 "\nn.nspname OPERATOR(pg_catalog.~) ");
+			appendStringLiteralConn(&qualified, schemabuf.data, conn);
+			appendPQExpBufferStr(&qualified,
+								 "::TEXT COLLATE pg_catalog.default AND"
+								 "\nc.relname OPERATOR(pg_catalog.~) ");
+			appendStringLiteralConn(&qualified, namebuf.data, conn);
+			appendPQExpBufferStr(&qualified,
+								 "::TEXT COLLATE pg_catalog.default)");
+			qor = "\nOR";
+		}
+		else
+		{
+			/* Unqualified relation pattern */
+			appendPQExpBufferStr(&unqualified, comma);
+			appendStringLiteralConn(&unqualified, namebuf.data, conn);
+			appendPQExpBufferStr(&unqualified,
+								 "::TEXT COLLATE pg_catalog.default");
+			comma = "\n, ";
+		}
+
+		resetPQExpBuffer(&dbnamebuf);
+		resetPQExpBuffer(&schemabuf);
+		resetPQExpBuffer(&namebuf);
+	}
+
+	if (qualified.data[0])
+	{
+		appendPQExpBuffer(sql,
+						  "\nSELECT c.oid AS %s"
+						  "\nFROM pg_catalog.pg_class c"
+						  "\nJOIN pg_catalog.pg_namespace n"
+						  "\nON c.relnamespace OPERATOR(pg_catalog.=) n.oid"
+						  "\nWHERE (",
+						  fieldname);
+		appendPQExpBufferStr(sql, qualified.data);
+		appendPQExpBufferStr(sql, ")\nAND ");
+		append_ctfilter_quals(sql, "c", checktype);
+		if (unqualified.data[0])
+			appendPQExpBufferStr(sql, "\nUNION ALL");
+	}
+	if (unqualified.data[0])
+	{
+		appendPQExpBuffer(sql,
+						  "\nSELECT c.oid AS %s"
+						  "\nFROM pg_catalog.pg_class c"
+						  "\nJOIN %s ls"
+						  "\nON c.relnamespace OPERATOR(pg_catalog.=) ls.nspoid"
+						  "\nWHERE c.relname OPERATOR(pg_catalog.~) ANY(ARRAY[",
+						  fieldname, schemacte);
+		appendPQExpBufferStr(sql, unqualified.data);
+		appendPQExpBufferStr(sql, "\n]::TEXT[])\nAND ");
+		append_ctfilter_quals(sql, "c", checktype);
+	}
+}
+
+/*
+ * table_cte
+ *
+ * Appends to the buffer 'sql' a Common Table Expression (CTE) which selects
+ * all table relations matching the given filters.
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql statement is appended
+ * schemacte: name of the CTE which selects all schemas to be checked
+ * ctename: name of the table CTE to be created
+ * fieldname: name of the oid field within the table CTE to be created
+ * include: list of table name patterns for inclusion
+ * exclude: list of table name patterns for exclusion
+ * inclusive: when 'include' is an empty list, whether the select statement
+ * should match all relations
+ * toast: whether to also select the associated toast tables
+ */
+static void
+table_cte(PGconn *conn, PQExpBuffer sql, const char *schemacte,
+		  const char *ctename, const char *fieldname,
+		  const SimpleStringList *include, const SimpleStringList *exclude,
+		  bool inclusive, bool toast)
+{
+	appendPQExpBuffer(sql, "\n%s (%s) AS (", ctename, fieldname);
+
+	if (toast)
+	{
+		/*
+		 * Compute the primary tables, then union on all associated toast
+		 * tables.  We depend on left to right evaluation of the UNION before
+		 * the EXCEPT which gets added below.   UNION and EXCEPT have equal
+		 * precedence, so be careful if you rearrange this query.
+		 */
+		appendPQExpBuffer(sql, "\nWITH primary_table AS (");
+		relation_select(conn, sql, schemacte, fieldname, include,
+						CT_TABLE, inclusive);
+		appendPQExpBuffer(sql, "\n)"
+						  "\nSELECT %s"
+						  "\nFROM primary_table"
+						  "\nUNION"
+						  "\nSELECT c.reltoastrelid AS %s"
+						  "\nFROM pg_catalog.pg_class c"
+						  "\nJOIN primary_table pt"
+						  "\nON pt.%s OPERATOR(pg_catalog.=) c.oid"
+						  "\nWHERE c.reltoastrelid OPERATOR(pg_catalog.!=) 0",
+						  fieldname, fieldname, fieldname);
+	}
+	else
+		relation_select(conn, sql, schemacte, fieldname, include,
+						CT_TABLE, inclusive);
+
+	appendPQExpBufferStr(sql, "\nEXCEPT");
+	relation_select(conn, sql, schemacte, fieldname, exclude,
+					CT_TABLE, false);
+	appendPQExpBufferStr(sql, "\n)");
+}
+
+/*
+ * exclude_index_cte
+ *		Appends a CTE which selects all indexes to be excluded
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql CTE is appended
+ * schemacte: name of the CTE which selects all schemas to be checked
+ * ctename: name of the index CTE to be created
+ * fieldname: name of the oid field within the index CTE to be created
+ * patterns: list of index name patterns to match
+ */
+static void
+exclude_index_cte(PGconn *conn, PQExpBuffer sql, const char *schemacte,
+				  const char *ctename,
+				  const char *fieldname, const SimpleStringList *patterns)
+{
+	appendPQExpBuffer(sql, "\n%s (%s) AS (", ctename, fieldname);
+	relation_select(conn, sql, schemacte, fieldname, patterns,
+					CT_BTREE, false);
+	appendPQExpBufferStr(sql, "\n)");
+}
+
+/*
+ * index_cte
+ *		Appends a CTE which selects all indexes to be checked
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql CTE is appended
+ * schemacte: name of the CTE which selects all schemas to be checked
+ * ctename: name of the index CTE to be created
+ * fieldname: name of the oid field within the index CTE to be created
+ * excludecte: name of the CTE which contains all indexes to be excluded
+ * tablescte: optional; if automatically including indexes for checked tables,
+ *		the name of the CTE which contains all tables to be checked
+ * tablesfield: if tablescte is not NULL, the name of the oid field in the
+ *		tables CTE
+ * patterns: list of index name patterns to match
+ * inclusive: when 'include' is an empty list, whether the select statement should match all relations
+ */
+static void
+index_cte(PGconn *conn, PQExpBuffer sql, const char *schemacte,
+		  const char *ctename, const char *fieldname,
+		  const char *excludecte, const char *tablescte,
+		  const char *tablesfield, const SimpleStringList *patterns,
+		  bool inclusive)
+{
+	appendPQExpBuffer(sql, "\n%s (%s) AS (", ctename, fieldname);
+	appendPQExpBuffer(sql, "\nSELECT %s FROM (", fieldname);
+	relation_select(conn, sql, schemacte, fieldname, patterns,
+					CT_BTREE, inclusive);
+	if (tablescte)
+	{
+		appendPQExpBuffer(sql,
+						  "\nUNION"
+						  "\nSELECT i.indexrelid AS %s"
+						  "\nFROM pg_catalog.pg_index i"
+						  "\nJOIN %s t ON t.%s OPERATOR(pg_catalog.=) i.indrelid",
+						  fieldname, tablescte, tablesfield);
+	}
+	appendPQExpBuffer(sql,
+					  "\n) AS included_indexes"
+					  "\nEXCEPT"
+					  "\nSELECT %s FROM %s",
+					  fieldname, excludecte);
+	appendPQExpBufferStr(sql, "\n)");
+}
+
+/*
+ * target_select
+ *
+ * Construct a query that will return a list of all tables and indexes in
+ * the database matching the user specified options, sorted by size.  We
+ * want the largest tables and indexes first, so that the parallel
+ * processing of the larger database objects gets started sooner.
+ *
+ * If 'inclusive' is true, include all tables and indexes not otherwise
+ * excluded; if false, include only tables and indexes explicitly included.
+ *
+ * conn: connection to the current database
+ * sql: buffer into which the constructed sql select statement is appended
+ * objects: lists of include and exclude patterns for filtering objects
+ * checkopts: user supplied program options
+ * progname: name of this program, such as "pg_amcheck"
+ * inclusive: when list of objects to include is empty, whether the select
+ * statement should match all objects not otherwise excluded
+ */
+static void
+target_select(PGconn *conn, PQExpBuffer sql, const amcheckObjects *objects,
+			  const amcheckOptions *checkopts, const char *progname,
+			  bool inclusive)
+{
+	appendPQExpBufferStr(sql, "WITH");
+	schema_cte(conn, sql, "namespaces", &objects->schemas,
+			   &objects->exclude_schemas, inclusive);
+	appendPQExpBufferStr(sql, ",");
+	table_cte(conn, sql, "namespaces", "tables", "tbloid",
+			  &objects->tables, &objects->exclude_tables, inclusive,
+			  !checkopts->exclude_toast);
+	if (!checkopts->no_indexes)
+	{
+		appendPQExpBufferStr(sql, ",");
+		exclude_index_cte(conn, sql, "namespaces",
+						  "excluded_indexes", "idxoid",
+						  &objects->exclude_indexes);
+		appendPQExpBufferStr(sql, ",");
+		if (checkopts->dependents)
+			index_cte(conn, sql, "namespaces", "indexes", "idxoid",
+					  "excluded_indexes", "tables", "tbloid",
+					  &objects->indexes, inclusive);
+		else
+			index_cte(conn, sql, "namespaces", "indexes", "idxoid",
+					  "excluded_indexes", NULL, NULL, &objects->indexes,
+					  inclusive);
+	}
+	appendPQExpBuffer(sql,
+					  "\nSELECT checktype, oid FROM ("
+					  "\nSELECT %u AS checktype, tables.tbloid AS oid, c.relpages"
+					  "\nFROM pg_catalog.pg_class c"
+					  "\nJOIN tables"
+					  "\nON tables.tbloid OPERATOR(pg_catalog.=) c.oid"
+					  "\nWHERE ",
+					  CT_TABLE);
+	append_ctfilter_quals(sql, "c", CT_TABLE);
+	if (!checkopts->no_indexes)
+	{
+		appendPQExpBuffer(sql,
+						  "\nUNION ALL"
+						  "\nSELECT %u AS checktype, indexes.idxoid AS oid, c.relpages"
+						  "\nFROM pg_catalog.pg_class c"
+						  "\nJOIN indexes"
+						  "\nON indexes.idxoid OPERATOR(pg_catalog.=) c.oid"
+						  "\nWHERE ",
+						  CT_BTREE);
+		append_ctfilter_quals(sql, "c", CT_BTREE);
+	}
+	appendPQExpBufferStr(sql,
+						 "\n) AS ss"
+						 "\nORDER BY relpages DESC, checktype, oid");
+}
diff --git a/contrib/pg_amcheck/pg_amcheck.h b/contrib/pg_amcheck/pg_amcheck.h
new file mode 100644
index 0000000000..b5ca276033
--- /dev/null
+++ b/contrib/pg_amcheck/pg_amcheck.h
@@ -0,0 +1,135 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_amcheck.h
+ *		Detects corruption within database relations.
+ *
+ * Copyright (c) 2020-2021, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  contrib/pg_amcheck/pg_amcheck.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_AMCHECK_H
+#define PG_AMCHECK_H
+
+#include "fe_utils/simple_list.h"
+#include "fe_utils/string_utils.h"
+#include "libpq-fe.h"
+#include "pqexpbuffer.h"		/* pgrminclude ignore */
+
+/* amcheck options controlled by user flags */
+typedef struct amcheckOptions
+{
+	bool		alldb;
+	bool		echo;
+	bool		quiet;
+	bool		verbose;
+	bool		dependents;
+	bool		no_indexes;
+	bool		exclude_toast;
+	bool		reconcile_toast;
+	bool		on_error_stop;
+	bool		parent_check;
+	bool		rootdescend;
+	bool		heapallindexed;
+	const char *skip;
+	int			jobs;			/* >= 0 indicates user specified the parallel
+								 * degree, otherwise -1 */
+	long		startblock;
+	long		endblock;
+} amcheckOptions;
+
+/* names of database objects to include or exclude controlled by user flags */
+typedef struct amcheckObjects
+{
+	SimpleStringList databases;
+	SimpleStringList schemas;
+	SimpleStringList tables;
+	SimpleStringList indexes;
+	SimpleStringList exclude_databases;
+	SimpleStringList exclude_schemas;
+	SimpleStringList exclude_tables;
+	SimpleStringList exclude_indexes;
+} amcheckObjects;
+
+/*
+ * We cannot launch the same amcheck function for all checked objects.  For
+ * btree indexes, we must use either bt_index_check() or
+ * bt_index_parent_check().  For heap relations, we must use verify_heapam().
+ * We silently ignore all other object types.
+ *
+ * The following CheckType enum and corresponding ctfilter array track which
+ * which kinds of relations get which treatment.
+ */
+typedef enum
+{
+	CT_TABLE = 0,
+	CT_BTREE
+} CheckType;
+
+/*
+ * This struct is used for filtering relations in pg_catalog.pg_class to just
+ * those of a given CheckType.  The relam field should equal pg_class.relam,
+ * and the pg_class.relkind should be contained in the relkinds comma separated
+ * list.
+ *
+ * The 'typname' field is not strictly for filtering, but for printing messages
+ * about relations that matched the filter.
+ */
+typedef struct
+{
+	Oid			relam;
+	const char *relkinds;
+	const char *typname;
+} CheckTypeFilter;
+
+/* Constants taken from pg_catalog/pg_am.dat */
+#define HEAP_TABLE_AM_OID 2
+#define BTREE_AM_OID 403
+
+static void check_each_database(ConnParams *cparams,
+								const amcheckObjects *objects,
+								const amcheckOptions *checkopts,
+								const char *progname);
+
+static void check_one_database(const ConnParams *cparams,
+							   const amcheckObjects *objects,
+							   const amcheckOptions *checkopts,
+							   const char *progname);
+static void prepare_table_command(PQExpBuffer sql,
+								  const amcheckOptions *checkopts, Oid reloid,
+								  const char *nspname);
+
+static void prepare_btree_command(PQExpBuffer sql,
+								  const amcheckOptions *checkopts, Oid reloid,
+								  const char *nspname);
+
+static void run_command(PGconn *conn, const char *sql,
+						const amcheckOptions *checkopts, Oid reloid,
+						const char *typ);
+
+static PGresult *VerifyHeapamSlotHandler(PGresult *res, PGconn *conn,
+										 void *context);
+
+static PGresult *VerifyBtreeSlotHandler(PGresult *res, PGconn *conn,
+										void *context);
+
+static void help(const char *progname);
+
+
+static void get_db_regexes_from_fqrps(SimpleStringList *regexes,
+									  const SimpleStringList *patterns);
+
+static void get_db_regexes_from_patterns(SimpleStringList *regexes,
+										 const SimpleStringList *patterns);
+
+static void dbname_select(PGconn *conn, PQExpBuffer sql,
+						  const SimpleStringList *regexes, bool alldb);
+
+static void target_select(PGconn *conn, PQExpBuffer sql,
+						  const amcheckObjects *objects,
+						  const amcheckOptions *options, const char *progname,
+						  bool inclusive);
+
+#endif							/* PG_AMCHECK_H */
diff --git a/contrib/pg_amcheck/t/001_basic.pl b/contrib/pg_amcheck/t/001_basic.pl
new file mode 100644
index 0000000000..dfa0ae9e06
--- /dev/null
+++ b/contrib/pg_amcheck/t/001_basic.pl
@@ -0,0 +1,9 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 8;
+
+program_help_ok('pg_amcheck');
+program_version_ok('pg_amcheck');
+program_options_handling_ok('pg_amcheck');
diff --git a/contrib/pg_amcheck/t/002_nonesuch.pl b/contrib/pg_amcheck/t/002_nonesuch.pl
new file mode 100644
index 0000000000..b52039c79b
--- /dev/null
+++ b/contrib/pg_amcheck/t/002_nonesuch.pl
@@ -0,0 +1,78 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 16;
+
+# Test set-up
+my ($node, $port);
+$node = get_new_node('test');
+$node->init;
+$node->start;
+$port = $node->port;
+
+# Load the amcheck extension, upon which pg_amcheck depends
+$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
+
+#########################################
+# Test connecting to a non-existent database
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'qqq' ],
+	qr/database "qqq" does not exist/,
+	'checking a non-existent database');
+
+#########################################
+# Test connecting with a non-existent user
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-U=no_such_user', 'postgres' ],
+	qr/role "=no_such_user" does not exist/,
+	'checking with a non-existent user');
+
+#########################################
+# Test checking a database without amcheck installed, by name.  We should see a
+# message about missing amcheck
+
+$node->command_like(
+	[ 'pg_amcheck', '-p', $port, 'template1' ],
+	qr/pg_amcheck: skipping database "template1": amcheck is not installed/,
+	'checking a database by name without amcheck installed');
+
+#########################################
+# Test checking a database without amcheck installed, by only indirectly using
+# a dbname pattern.  In verbose mode, we should see a message about missing
+# amcheck
+
+$node->command_like(
+	[ 'pg_amcheck', '-p', $port, '-v', '-d', '*', 'postgres' ],
+	qr/pg_amcheck: skipping database "template1": amcheck is not installed/,
+	'checking a database by dbname implication without amcheck installed');
+
+#########################################
+# Test checking non-existent schemas, tables, and indexes
+
+$node->command_ok(
+	[ 'pg_amcheck', '-p', $port, '-s', 'no_such_schema' ],
+	'checking a non-existent schema');
+
+$node->command_ok(
+	[ 'pg_amcheck', '-p', $port, '-t', 'no_such_table' ],
+	'checking a non-existent table');
+
+$node->command_ok(
+	[ 'pg_amcheck', '-p', $port, '-i', 'no_such_index' ],
+	'checking a non-existent schema');
+
+$node->command_ok(
+	[ 'pg_amcheck', '-p', $port, '-s', 'no*such*schema*' ],
+	'no matching schemas');
+
+$node->command_ok(
+	[ 'pg_amcheck', '-p', $port, '-t', 'no*such*table*' ],
+	'no matching tables');
+
+$node->command_ok(
+	[ 'pg_amcheck', '-p', $port, '-i', 'no*such*index' ],
+	'no matching indexes');
diff --git a/contrib/pg_amcheck/t/003_check.pl b/contrib/pg_amcheck/t/003_check.pl
new file mode 100644
index 0000000000..957094fcdd
--- /dev/null
+++ b/contrib/pg_amcheck/t/003_check.pl
@@ -0,0 +1,475 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 70;
+
+my ($node, $port);
+
+# Returns the filesystem path for the named relation.
+#
+# Assumes the test node is running
+sub relation_filepath($$)
+{
+	my ($dbname, $relname) = @_;
+
+	my $pgdata = $node->data_dir;
+	my $rel = $node->safe_psql($dbname,
+							   qq(SELECT pg_relation_filepath('$relname')));
+	die "path not found for relation $relname" unless defined $rel;
+	return "$pgdata/$rel";
+}
+
+# Returns the name of the toast relation associated with the named relation.
+#
+# Assumes the test node is running
+sub relation_toast($$)
+{
+	my ($dbname, $relname) = @_;
+
+	my $rel = $node->safe_psql($dbname, qq(
+		SELECT ct.relname
+			FROM pg_catalog.pg_class cr, pg_catalog.pg_class ct
+			WHERE cr.oid = '$relname'::regclass
+			  AND cr.reltoastrelid = ct.oid
+			));
+	return undef unless defined $rel;
+	return "pg_toast.$rel";
+}
+
+# Stops the test node, corrupts the first page of the named relation, and
+# restarts the node.
+#
+# Assumes the test node is running.
+sub corrupt_first_page($$)
+{
+	my ($dbname, $relname) = @_;
+	my $relpath = relation_filepath($dbname, $relname);
+
+	$node->stop;
+	my $fh;
+	open($fh, '+<', $relpath);
+	binmode $fh;
+	seek($fh, 32, 0);
+	syswrite($fh, '\x77\x77\x77\x77', 500);
+	close($fh);
+	$node->start;
+}
+
+# Stops the test node, unlinks the file from the filesystem that backs the
+# relation, and restarts the node.
+#
+# Assumes the test node is running
+sub remove_relation_file($$)
+{
+	my ($dbname, $relname) = @_;
+	my $relpath = relation_filepath($dbname, $relname);
+
+	$node->stop();
+	unlink($relpath);
+	$node->start;
+}
+
+# Stops the test node, unlinks the file from the filesystem that backs the
+# toast table (if any) corresponding to the given main table relation, and
+# restarts the node.
+#
+# Assumes the test node is running
+sub remove_toast_file($$)
+{
+	my ($dbname, $relname) = @_;
+	my $toastname = relation_toast($dbname, $relname);
+	remove_relation_file($dbname, $toastname) if ($toastname);
+}
+
+# Test set-up
+$node = get_new_node('test');
+$node->init;
+$node->start;
+$port = $node->port;
+
+for my $dbname (qw(db1 db2 db3))
+{
+	# Create the database
+	$node->safe_psql('postgres', qq(CREATE DATABASE $dbname));
+
+	# Load the amcheck extension, upon which pg_amcheck depends.  Put the
+	# extension in an unexpected location to test that pg_amcheck finds it
+	# correctly.  Create tables with names that look like pg_catalog names to
+	# check that pg_amcheck does not get confused by them.  Create functions in
+	# schema public that look like amcheck functions to check that pg_amcheck
+	# does not use them.
+	$node->safe_psql($dbname, q(
+		CREATE SCHEMA amcheck_schema;
+		CREATE EXTENSION amcheck WITH SCHEMA amcheck_schema;
+		CREATE TABLE amcheck_schema.pg_database (junk text);
+		CREATE TABLE amcheck_schema.pg_namespace (junk text);
+		CREATE TABLE amcheck_schema.pg_class (junk text);
+		CREATE TABLE amcheck_schema.pg_operator (junk text);
+		CREATE TABLE amcheck_schema.pg_proc (junk text);
+		CREATE TABLE amcheck_schema.pg_tablespace (junk text);
+
+		CREATE FUNCTION public.bt_index_check(index regclass,
+											  heapallindexed boolean default false)
+		RETURNS VOID AS $$
+		BEGIN
+			RAISE EXCEPTION 'Invoked wrong bt_index_check!';
+		END;
+		$$ LANGUAGE plpgsql;
+
+		CREATE FUNCTION public.bt_index_parent_check(index regclass,
+													 heapallindexed boolean default false,
+													 rootdescend boolean default false)
+		RETURNS VOID AS $$
+		BEGIN
+			RAISE EXCEPTION 'Invoked wrong bt_index_parent_check!';
+		END;
+		$$ LANGUAGE plpgsql;
+
+		CREATE FUNCTION public.verify_heapam(relation regclass,
+											 on_error_stop boolean default false,
+											 check_toast boolean default false,
+											 skip text default 'none',
+											 startblock bigint default null,
+											 endblock bigint default null,
+											 blkno OUT bigint,
+											 offnum OUT integer,
+											 attnum OUT integer,
+											 msg OUT text)
+		RETURNS SETOF record AS $$
+		BEGIN
+			RAISE EXCEPTION 'Invoked wrong verify_heapam!';
+		END;
+		$$ LANGUAGE plpgsql;
+	));
+
+	# Create schemas, tables and indexes in five separate
+	# schemas.  The schemas are all identical to start, but
+	# we will corrupt them differently later.
+	#
+	for my $schema (qw(s1 s2 s3 s4 s5))
+	{
+		$node->safe_psql($dbname, qq(
+			CREATE SCHEMA $schema;
+			CREATE SEQUENCE $schema.seq1;
+			CREATE SEQUENCE $schema.seq2;
+			CREATE TABLE $schema.t1 (
+				i INTEGER,
+				b BOX,
+				ia int4[],
+				ir int4range,
+				t TEXT
+			);
+			CREATE TABLE $schema.t2 (
+				i INTEGER,
+				b BOX,
+				ia int4[],
+				ir int4range,
+				t TEXT
+			);
+			CREATE VIEW $schema.t2_view AS (
+				SELECT i*2, t FROM $schema.t2
+			);
+			ALTER TABLE $schema.t2
+				ALTER COLUMN t
+				SET STORAGE EXTERNAL;
+
+			INSERT INTO $schema.t1 (i, b, ia, ir, t)
+				(SELECT gs::INTEGER AS i,
+						box(point(gs,gs+5),point(gs*2,gs*3)) AS b,
+						array[gs, gs + 1]::int4[] AS ia,
+						int4range(gs, gs+100) AS ir,
+						repeat('foo', gs) AS t
+					 FROM generate_series(1,10000,3000) AS gs);
+
+			INSERT INTO $schema.t2 (i, b, ia, ir, t)
+				(SELECT gs::INTEGER AS i,
+						box(point(gs,gs+5),point(gs*2,gs*3)) AS b,
+						array[gs, gs + 1]::int4[] AS ia,
+						int4range(gs, gs+100) AS ir,
+						repeat('foo', gs) AS t
+					 FROM generate_series(1,10000,3000) AS gs);
+
+			CREATE MATERIALIZED VIEW $schema.t1_mv AS SELECT * FROM $schema.t1;
+			CREATE MATERIALIZED VIEW $schema.t2_mv AS SELECT * FROM $schema.t2;
+
+			create table $schema.p1 (a int, b int) PARTITION BY list (a);
+			create table $schema.p2 (a int, b int) PARTITION BY list (a);
+
+			create table $schema.p1_1 partition of $schema.p1 for values in (1, 2, 3);
+			create table $schema.p1_2 partition of $schema.p1 for values in (4, 5, 6);
+			create table $schema.p2_1 partition of $schema.p2 for values in (1, 2, 3);
+			create table $schema.p2_2 partition of $schema.p2 for values in (4, 5, 6);
+
+			CREATE INDEX t1_btree ON $schema.t1 USING BTREE (i);
+			CREATE INDEX t2_btree ON $schema.t2 USING BTREE (i);
+
+			CREATE INDEX t1_hash ON $schema.t1 USING HASH (i);
+			CREATE INDEX t2_hash ON $schema.t2 USING HASH (i);
+
+			CREATE INDEX t1_brin ON $schema.t1 USING BRIN (i);
+			CREATE INDEX t2_brin ON $schema.t2 USING BRIN (i);
+
+			CREATE INDEX t1_gist ON $schema.t1 USING GIST (b);
+			CREATE INDEX t2_gist ON $schema.t2 USING GIST (b);
+
+			CREATE INDEX t1_gin ON $schema.t1 USING GIN (ia);
+			CREATE INDEX t2_gin ON $schema.t2 USING GIN (ia);
+
+			CREATE INDEX t1_spgist ON $schema.t1 USING SPGIST (ir);
+			CREATE INDEX t2_spgist ON $schema.t2 USING SPGIST (ir);
+		));
+	}
+}
+
+# Database 'db1' corruptions
+#
+
+# Corrupt indexes in schema "s1"
+remove_relation_file('db1', 's1.t1_btree');
+corrupt_first_page('db1', 's1.t2_btree');
+
+# Corrupt tables in schema "s2"
+remove_relation_file('db1', 's2.t1');
+corrupt_first_page('db1', 's2.t2');
+
+# Corrupt tables, partitions, matviews, and btrees in schema "s3"
+remove_relation_file('db1', 's3.t1');
+corrupt_first_page('db1', 's3.t2');
+
+remove_relation_file('db1', 's3.t1_mv');
+remove_relation_file('db1', 's3.p1_1');
+
+corrupt_first_page('db1', 's3.t2_mv');
+corrupt_first_page('db1', 's3.p2_1');
+
+remove_relation_file('db1', 's3.t1_btree');
+corrupt_first_page('db1', 's3.t2_btree');
+
+# Corrupt toast table, partitions, and materialized views in schema "s4"
+remove_toast_file('db1', 's4.t2');
+
+# Corrupt all other object types in schema "s5".  We don't have amcheck support
+# for these types, but we check that their corruption does not trigger any
+# errors in pg_amcheck
+remove_relation_file('db1', 's5.seq1');
+remove_relation_file('db1', 's5.t1_hash');
+remove_relation_file('db1', 's5.t1_gist');
+remove_relation_file('db1', 's5.t1_gin');
+remove_relation_file('db1', 's5.t1_brin');
+remove_relation_file('db1', 's5.t1_spgist');
+
+corrupt_first_page('db1', 's5.seq2');
+corrupt_first_page('db1', 's5.t2_hash');
+corrupt_first_page('db1', 's5.t2_gist');
+corrupt_first_page('db1', 's5.t2_gin');
+corrupt_first_page('db1', 's5.t2_brin');
+corrupt_first_page('db1', 's5.t2_spgist');
+
+
+# Database 'db2' corruptions
+#
+remove_relation_file('db2', 's1.t1');
+remove_relation_file('db2', 's1.t1_btree');
+
+
+# Leave 'db3' uncorrupted
+#
+
+
+# Standard first arguments to TestLib functions
+my @cmd = ('pg_amcheck', '--quiet', '-p', $port);
+
+# The pg_amcheck command itself should return a success exit status, even
+# though tables and indexes are corrupt.  An error code returned would mean the
+# pg_amcheck command itself failed, for example because a connection to the
+# database could not be established.
+#
+# For these checks, we're ignoring any corruption reported and focusing
+# exclusively on the exit code from pg_amcheck.
+#
+$node->command_ok(
+	[ @cmd,, 'db1' ],
+	'pg_amcheck all schemas, tables and indexes in database db1');
+
+$node->command_ok(
+	[ @cmd,, 'db1', 'db2', 'db3' ],
+	'pg_amcheck all schemas, tables and indexes in databases db1, db2 and db3');
+
+$node->command_ok(
+	[ @cmd, '--all' ],
+	'pg_amcheck all schemas, tables and indexes in all databases');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-s', 's1' ],
+	'pg_amcheck all objects in schema s1');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-r', 's*.t1' ],
+	'pg_amcheck all tables named t1 and their indexes');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-i', 'i*.idx', '-i', 'idx.i*' ],
+	'pg_amcheck all indexes with qualified names matching /i*.idx/ or /idx.i*/');
+
+$node->command_ok(
+	[ @cmd, '--no-dependents', 'db1', '-r', 's*.t1' ],
+	'pg_amcheck all tables with qualified names matching /s*.t1/');
+
+$node->command_ok(
+	[ @cmd, '--no-dependents', 'db1', '-t', 's*.t1', '-t', 'foo*.bar*' ],
+	'pg_amcheck all tables with qualified names matching /s*.t1/ or /foo*.bar*/');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-T', 't1' ],
+	'pg_amcheck everything except tables named t1');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-S', 's1', '-R', 't1' ],
+	'pg_amcheck everything not named t1 nor in schema s1');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-t', '*.*.*' ],
+	'pg_amcheck all tables across all databases and schemas');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-t', '*.*.t1' ],
+	'pg_amcheck all tables named t1 across all databases and schemas');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-t', '*.s1.*' ],
+	'pg_amcheck all tables across all databases in schemas named s1');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-t', 'db2.*.*' ],
+	'pg_amcheck all tables across all schemas in database db2');
+
+$node->command_ok(
+	[ @cmd, 'db1', '-t', 'db2.*.*', '-t', 'db3.*.*' ],
+	'pg_amcheck all tables across all schemas in databases db2 and db3');
+
+# Scans of indexes in s1 should detect the specific corruption that we created
+# above.  For missing relation forks, we know what the error message looks
+# like.  For corrupted index pages, the error might vary depending on how the
+# page was formatted on disk, including variations due to alignment differences
+# between platforms, so we accept any non-empty error message.
+#
+$node->command_like(
+	[ @cmd, '--all', '-s', 's1', '-i', 't1_btree' ],
+	qr/index "t1_btree" lacks a main relation fork/,
+	'pg_amcheck index s1.t1_btree reports missing main relation fork');
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's1', '-i', 't2_btree' ],
+	qr/.+/,			# Any non-empty error message is acceptable
+	'pg_amcheck index s1.s2 reports index corruption');
+
+# Checking db1.s1 should show no corruptions if indexes are excluded
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's1', '--exclude-indexes' ],
+	qr/^$/,
+	'pg_amcheck of db1.s1 excluding indexes');
+
+# But checking across all databases in schema s1 should show corruptions
+# messages for tables in db2
+$node->command_like(
+	[ @cmd, '--all', '-s', 's1', '--exclude-indexes' ],
+	qr/could not open file/,
+	'pg_amcheck of schema s1 across all databases but excluding indexes');
+
+# Checking across a list of databases should also work
+$node->command_like(
+	[ @cmd, '-d', 'db2', '-d', 'db1', '-s', 's1', '--exclude-indexes' ],
+	qr/could not open file/,
+	'pg_amcheck of schema s1 across db1 and db2 but excluding indexes');
+
+# In schema s3, the tables and indexes are both corrupt.  We should see
+# corruption messages on stdout, nothing on stderr, and an exit
+# status of zero.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's3' ],
+	0,
+	[ qr/index "t1_btree" lacks a main relation fork/,
+	  qr/could not open file/ ],
+	[ qr/^$/ ],
+	'pg_amcheck schema s3 reports table and index errors');
+
+# In schema s2, only tables are corrupt.  Check that table corruption is
+# reported as expected.
+#
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's2', '-t', 't1' ],
+	qr/could not open file/,
+	'pg_amcheck in schema s2 reports table corruption');
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's2', '-t', 't2' ],
+	qr/.+/,			# Any non-empty error message is acceptable
+	'pg_amcheck in schema s2 reports table corruption');
+
+# In schema s4, only toast tables are corrupt.  Check that under default
+# options the toast corruption is reported, but when excluding toast we get no
+# error reports.
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's4' ],
+	qr/could not open file/,
+	'pg_amcheck in schema s4 reports toast corruption');
+
+$node->command_like(
+	[ @cmd, '--exclude-toast', '--exclude-toast-pointers', 'db1', '-s', 's4' ],
+	qr/^$/,			# Empty
+	'pg_amcheck in schema s4 excluding toast reports no corruption');
+
+# Check that no corruption is reported in schema s5
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's5' ],
+	qr/^$/,			# Empty
+	'pg_amcheck over schema s5 reports no corruption');
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's1', '-I', 't1_btree', '-I', 't2_btree' ],
+	qr/^$/,			# Empty
+	'pg_amcheck over schema s1 with corrupt indexes excluded reports no corruption');
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's1', '--exclude-indexes' ],
+	qr/^$/,			# Empty
+	'pg_amcheck over schema s1 with all indexes excluded reports no corruption');
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's2', '-T', 't1', '-T', 't2' ],
+	qr/^$/,			# Empty
+	'pg_amcheck over schema s2 with corrupt tables excluded reports no corruption');
+
+# Check errors about bad block range command line arguments.  We use schema s5
+# to avoid getting messages about corrupt tables or indexes.
+command_fails_like(
+	[ @cmd, 'db1', '-s', 's5', '--startblock', 'junk' ],
+	qr/relation starting block argument contains garbage characters/,
+	'pg_amcheck rejects garbage startblock');
+
+command_fails_like(
+	[ @cmd, 'db1', '-s', 's5', '--endblock', '1234junk' ],
+	qr/relation ending block argument contains garbage characters/,
+	'pg_amcheck rejects garbage endblock');
+
+command_fails_like(
+	[ @cmd, 'db1', '-s', 's5', '--startblock', '5', '--endblock', '4' ],
+	qr/relation ending block argument precedes starting block argument/,
+	'pg_amcheck rejects invalid block range');
+
+# Check bt_index_parent_check alternates.  We don't create any index corruption
+# that would behave differently under these modes, so just smoke test that the
+# arguments are handled sensibly.
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's1', '-i', 't1_btree', '--parent-check' ],
+	qr/index "t1_btree" lacks a main relation fork/,
+	'pg_amcheck smoke test --parent-check');
+
+$node->command_like(
+	[ @cmd, 'db1', '-s', 's1', '-i', 't1_btree', '--heapallindexed', '--rootdescend' ],
+	qr/index "t1_btree" lacks a main relation fork/,
+	'pg_amcheck smoke test --heapallindexed --rootdescend');
diff --git a/contrib/pg_amcheck/t/004_verify_heapam.pl b/contrib/pg_amcheck/t/004_verify_heapam.pl
new file mode 100644
index 0000000000..7e71d612fc
--- /dev/null
+++ b/contrib/pg_amcheck/t/004_verify_heapam.pl
@@ -0,0 +1,496 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+
+use Test::More tests => 22;
+
+# This regression test demonstrates that the pg_amcheck binary supplied with
+# the pg_amcheck contrib module correctly identifies specific kinds of
+# corruption within pages.  To test this, we need a mechanism to create corrupt
+# pages with predictable, repeatable corruption.  The postgres backend cannot
+# be expected to help us with this, as its design is not consistent with the
+# goal of intentionally corrupting pages.
+#
+# Instead, we create a table to corrupt, and with careful consideration of how
+# postgresql lays out heap pages, we seek to offsets within the page and
+# overwrite deliberately chosen bytes with specific values calculated to
+# corrupt the page in expected ways.  We then verify that pg_amcheck reports
+# the corruption, and that it runs without crashing.  Note that the backend
+# cannot simply be started to run queries against the corrupt table, as the
+# backend will crash, at least for some of the corruption types we generate.
+#
+# Autovacuum potentially touching the table in the background makes the exact
+# behavior of this test harder to reason about.  We turn it off to keep things
+# simpler.  We use a "belt and suspenders" approach, turning it off for the
+# system generally in postgresql.conf, and turning it off specifically for the
+# test table.
+#
+# This test depends on the table being written to the heap file exactly as we
+# expect it to be, so we take care to arrange the columns of the table, and
+# insert rows of the table, that give predictable sizes and locations within
+# the table page.
+#
+# The HeapTupleHeaderData has 23 bytes of fixed size fields before the variable
+# length t_bits[] array.  We have exactly 3 columns in the table, so natts = 3,
+# t_bits is 1 byte long, and t_hoff = MAXALIGN(23 + 1) = 24.
+#
+# We're not too fussy about which datatypes we use for the test, but we do care
+# about some specific properties.  We'd like to test both fixed size and
+# varlena types.  We'd like some varlena data inline and some toasted.  And
+# we'd like the layout of the table such that the datums land at predictable
+# offsets within the tuple.  We choose a structure without padding on all
+# supported architectures:
+#
+# 	a BIGINT
+#	b TEXT
+#	c TEXT
+#
+# We always insert a 7-ascii character string into field 'b', which with a
+# 1-byte varlena header gives an 8 byte inline value.  We always insert a long
+# text string in field 'c', long enough to force toast storage.
+#
+# We choose to read and write binary copies of our table's tuples, using perl's
+# pack() and unpack() functions.  Perl uses a packing code system in which:
+#
+#	L = "Unsigned 32-bit Long",
+#	S = "Unsigned 16-bit Short",
+#	C = "Unsigned 8-bit Octet",
+#	c = "signed 8-bit octet",
+#	q = "signed 64-bit quadword"
+#
+# Each tuple in our table has a layout as follows:
+#
+#    xx xx xx xx            t_xmin: xxxx		offset = 0		L
+#    xx xx xx xx            t_xmax: xxxx		offset = 4		L
+#    xx xx xx xx          t_field3: xxxx		offset = 8		L
+#    xx xx                   bi_hi: xx			offset = 12		S
+#    xx xx                   bi_lo: xx			offset = 14		S
+#    xx xx                ip_posid: xx			offset = 16		S
+#    xx xx             t_infomask2: xx			offset = 18		S
+#    xx xx              t_infomask: xx			offset = 20		S
+#    xx                     t_hoff: x			offset = 22		C
+#    xx                     t_bits: x			offset = 23		C
+#    xx xx xx xx xx xx xx xx   'a': xxxxxxxx	offset = 24		q
+#    xx xx xx xx xx xx xx xx   'b': xxxxxxxx	offset = 32		Cccccccc
+#    xx xx xx xx xx xx xx xx   'c': xxxxxxxx	offset = 40		SSSS
+#    xx xx xx xx xx xx xx xx      : xxxxxxxx	 ...continued	SSSS
+#    xx xx                        : xx      	 ...continued	S
+#
+# We could choose to read and write columns 'b' and 'c' in other ways, but
+# it is convenient enough to do it this way.  We define packing code
+# constants here, where they can be compared easily against the layout.
+
+use constant HEAPTUPLE_PACK_CODE => 'LLLSSSSSCCqCcccccccSSSSSSSSS';
+use constant HEAPTUPLE_PACK_LENGTH => 58;     # Total size
+
+# Read a tuple of our table from a heap page.
+#
+# Takes an open filehandle to the heap file, and the offset of the tuple.
+#
+# Rather than returning the binary data from the file, unpacks the data into a
+# perl hash with named fields.  These fields exactly match the ones understood
+# by write_tuple(), below.  Returns a reference to this hash.
+#
+sub read_tuple ($$)
+{
+	my ($fh, $offset) = @_;
+	my ($buffer, %tup);
+	seek($fh, $offset, 0);
+	sysread($fh, $buffer, HEAPTUPLE_PACK_LENGTH);
+
+	@_ = unpack(HEAPTUPLE_PACK_CODE, $buffer);
+	%tup = (t_xmin => shift,
+			t_xmax => shift,
+			t_field3 => shift,
+			bi_hi => shift,
+			bi_lo => shift,
+			ip_posid => shift,
+			t_infomask2 => shift,
+			t_infomask => shift,
+			t_hoff => shift,
+			t_bits => shift,
+			a => shift,
+			b_header => shift,
+			b_body1 => shift,
+			b_body2 => shift,
+			b_body3 => shift,
+			b_body4 => shift,
+			b_body5 => shift,
+			b_body6 => shift,
+			b_body7 => shift,
+			c1 => shift,
+			c2 => shift,
+			c3 => shift,
+			c4 => shift,
+			c5 => shift,
+			c6 => shift,
+			c7 => shift,
+			c8 => shift,
+			c9 => shift);
+	# Stitch together the text for column 'b'
+	$tup{b} = join('', map { chr($tup{"b_body$_"}) } (1..7));
+	return \%tup;
+}
+
+# Write a tuple of our table to a heap page.
+#
+# Takes an open filehandle to the heap file, the offset of the tuple, and a
+# reference to a hash with the tuple values, as returned by read_tuple().
+# Writes the tuple fields from the hash into the heap file.
+#
+# The purpose of this function is to write a tuple back to disk with some
+# subset of fields modified.  The function does no error checking.  Use
+# cautiously.
+#
+sub write_tuple($$$)
+{
+	my ($fh, $offset, $tup) = @_;
+	my $buffer = pack(HEAPTUPLE_PACK_CODE,
+					$tup->{t_xmin},
+					$tup->{t_xmax},
+					$tup->{t_field3},
+					$tup->{bi_hi},
+					$tup->{bi_lo},
+					$tup->{ip_posid},
+					$tup->{t_infomask2},
+					$tup->{t_infomask},
+					$tup->{t_hoff},
+					$tup->{t_bits},
+					$tup->{a},
+					$tup->{b_header},
+					$tup->{b_body1},
+					$tup->{b_body2},
+					$tup->{b_body3},
+					$tup->{b_body4},
+					$tup->{b_body5},
+					$tup->{b_body6},
+					$tup->{b_body7},
+					$tup->{c1},
+					$tup->{c2},
+					$tup->{c3},
+					$tup->{c4},
+					$tup->{c5},
+					$tup->{c6},
+					$tup->{c7},
+					$tup->{c8},
+					$tup->{c9});
+	seek($fh, $offset, 0);
+	syswrite($fh, $buffer, HEAPTUPLE_PACK_LENGTH);
+	return;
+}
+
+# Set umask so test directories and files are created with default permissions
+umask(0077);
+
+# Set up the node.  Once we create and corrupt the table,
+# autovacuum workers visiting the table could crash the backend.
+# Disable autovacuum so that won't happen.
+my $node = get_new_node('test');
+$node->init;
+$node->append_conf('postgresql.conf', 'autovacuum=off');
+
+# Start the node and load the extensions.  We depend on both
+# amcheck and pageinspect for this test.
+$node->start;
+my $port = $node->port;
+my $pgdata = $node->data_dir;
+$node->safe_psql('postgres', "CREATE EXTENSION amcheck");
+$node->safe_psql('postgres', "CREATE EXTENSION pageinspect");
+
+# Get a non-zero datfrozenxid
+$node->safe_psql('postgres', qq(VACUUM FREEZE));
+
+# Create the test table with precisely the schema that our corruption function
+# expects.
+$node->safe_psql(
+	'postgres', qq(
+		CREATE TABLE public.test (a BIGINT, b TEXT, c TEXT);
+		ALTER TABLE public.test SET (autovacuum_enabled=false);
+		ALTER TABLE public.test ALTER COLUMN c SET STORAGE EXTERNAL;
+		CREATE INDEX test_idx ON public.test(a, b);
+	));
+
+# We want (0 < datfrozenxid < test.relfrozenxid).  To achieve this, we freeze
+# an otherwise unused table, public.junk, prior to inserting data and freezing
+# public.test
+$node->safe_psql(
+	'postgres', qq(
+		CREATE TABLE public.junk AS SELECT 'junk'::TEXT AS junk_column;
+		ALTER TABLE public.junk SET (autovacuum_enabled=false);
+		VACUUM FREEZE public.junk
+	));
+
+my $rel = $node->safe_psql('postgres', qq(SELECT pg_relation_filepath('public.test')));
+my $relpath = "$pgdata/$rel";
+
+# Insert data and freeze public.test
+use constant ROWCOUNT => 16;
+$node->safe_psql('postgres', qq(
+	INSERT INTO public.test (a, b, c)
+		VALUES (
+			12345678,
+			'abcdefg',
+			repeat('w', 10000)
+		);
+	VACUUM FREEZE public.test
+	)) for (1..ROWCOUNT);
+
+my $relfrozenxid = $node->safe_psql('postgres',
+	q(select relfrozenxid from pg_class where relname = 'test'));
+my $datfrozenxid = $node->safe_psql('postgres',
+	q(select datfrozenxid from pg_database where datname = 'postgres'));
+
+# Find where each of the tuples is located on the page.
+my @lp_off;
+for my $tup (0..ROWCOUNT-1)
+{
+	push (@lp_off, $node->safe_psql('postgres', qq(
+select lp_off from heap_page_items(get_raw_page('test', 'main', 0))
+	offset $tup limit 1)));
+}
+
+# Check that pg_amcheck runs against the uncorrupted table without error.
+$node->command_ok(['pg_amcheck', '-p', $port, 'postgres'],
+				  'pg_amcheck test table, prior to corruption');
+
+# Check that pg_amcheck runs against the uncorrupted table and index without error.
+$node->command_ok(['pg_amcheck', '-p', $port, 'postgres'],
+				  'pg_amcheck test table and index, prior to corruption');
+
+$node->stop;
+
+# Sanity check that our 'test' table has a relfrozenxid newer than the
+# datfrozenxid for the database, and that the datfrozenxid is greater than the
+# first normal xid.  We rely on these invariants in some of our tests.
+if ($datfrozenxid <= 3 || $datfrozenxid >= $relfrozenxid)
+{
+	fail('Xid thresholds not as expected');
+	$node->clean_node;
+	exit;
+}
+
+# Some #define constants from access/htup_details.h for use while corrupting.
+use constant HEAP_HASNULL            => 0x0001;
+use constant HEAP_XMAX_LOCK_ONLY     => 0x0080;
+use constant HEAP_XMIN_COMMITTED     => 0x0100;
+use constant HEAP_XMIN_INVALID       => 0x0200;
+use constant HEAP_XMAX_COMMITTED     => 0x0400;
+use constant HEAP_XMAX_INVALID       => 0x0800;
+use constant HEAP_NATTS_MASK         => 0x07FF;
+use constant HEAP_XMAX_IS_MULTI      => 0x1000;
+use constant HEAP_KEYS_UPDATED       => 0x2000;
+
+# Helper function to generate a regular expression matching the header we
+# expect verify_heapam() to return given which fields we expect to be non-null.
+sub header
+{
+	my ($blkno, $offnum, $attnum) = @_;
+	return qr/relation postgres\.public\.test, block $blkno, offset $offnum, attribute $attnum\s+/ms
+		if (defined $attnum);
+	return qr/relation postgres\.public\.test, block $blkno, offset $offnum\s+/ms
+		if (defined $offnum);
+	return qr/relation postgres\.public\.test\s+/ms
+		if (defined $blkno);
+	return qr/relation postgres\.public\.test\s+/ms;
+}
+
+# Corrupt the tuples, one type of corruption per tuple.  Some types of
+# corruption cause verify_heapam to skip to the next tuple without
+# performing any remaining checks, so we can't exercise the system properly if
+# we focus all our corruption on a single tuple.
+#
+my @expected;
+my $file;
+open($file, '+<', $relpath);
+binmode $file;
+
+for (my $tupidx = 0; $tupidx < ROWCOUNT; $tupidx++)
+{
+	my $offnum = $tupidx + 1;  # offnum is 1-based, not zero-based
+	my $offset = $lp_off[$tupidx];
+	my $tup = read_tuple($file, $offset);
+
+	# Sanity-check that the data appears on the page where we expect.
+	if ($tup->{a} ne '12345678' || $tup->{b} ne 'abcdefg')
+	{
+		fail('Page layout differs from our expectations');
+		$node->clean_node;
+		exit;
+	}
+
+	my $header = header(0, $offnum, undef);
+	if ($offnum == 1)
+	{
+		# Corruptly set xmin < relfrozenxid
+		my $xmin = $relfrozenxid - 1;
+		$tup->{t_xmin} = $xmin;
+		$tup->{t_infomask} &= ~HEAP_XMIN_COMMITTED;
+		$tup->{t_infomask} &= ~HEAP_XMIN_INVALID;
+
+		# Expected corruption report
+		push @expected,
+			qr/${header}xmin $xmin precedes relation freeze threshold 0:\d+/;
+	}
+	if ($offnum == 2)
+	{
+		# Corruptly set xmin < datfrozenxid
+		my $xmin = 3;
+		$tup->{t_xmin} = $xmin;
+		$tup->{t_infomask} &= ~HEAP_XMIN_COMMITTED;
+		$tup->{t_infomask} &= ~HEAP_XMIN_INVALID;
+
+		push @expected,
+			qr/${$header}xmin $xmin precedes oldest valid transaction ID 0:\d+/;
+	}
+	elsif ($offnum == 3)
+	{
+		# Corruptly set xmin < datfrozenxid, further back, noting circularity
+		# of xid comparison.  For a new cluster with epoch = 0, the corrupt
+		# xmin will be interpreted as in the future
+		$tup->{t_xmin} = 4026531839;
+		$tup->{t_infomask} &= ~HEAP_XMIN_COMMITTED;
+		$tup->{t_infomask} &= ~HEAP_XMIN_INVALID;
+
+		push @expected,
+			qr/${$header}xmin 4026531839 equals or exceeds next valid transaction ID 0:\d+/;
+	}
+	elsif ($offnum == 4)
+	{
+		# Corruptly set xmax < relminmxid;
+		$tup->{t_xmax} = 4026531839;
+		$tup->{t_infomask} &= ~HEAP_XMAX_INVALID;
+
+		push @expected,
+			qr/${$header}xmax 4026531839 equals or exceeds next valid transaction ID 0:\d+/;
+	}
+	elsif ($offnum == 5)
+	{
+		# Corrupt the tuple t_hoff, but keep it aligned properly
+		$tup->{t_hoff} += 128;
+
+		push @expected,
+			qr/${$header}data begins at offset 152 beyond the tuple length 58/,
+			qr/${$header}tuple data should begin at byte 24, but actually begins at byte 152 \(3 attributes, no nulls\)/;
+	}
+	elsif ($offnum == 6)
+	{
+		# Corrupt the tuple t_hoff, wrong alignment
+		$tup->{t_hoff} += 3;
+
+		push @expected,
+			qr/${$header}tuple data should begin at byte 24, but actually begins at byte 27 \(3 attributes, no nulls\)/;
+	}
+	elsif ($offnum == 7)
+	{
+		# Corrupt the tuple t_hoff, underflow but correct alignment
+		$tup->{t_hoff} -= 8;
+
+		push @expected,
+			qr/${$header}tuple data should begin at byte 24, but actually begins at byte 16 \(3 attributes, no nulls\)/;
+	}
+	elsif ($offnum == 8)
+	{
+		# Corrupt the tuple t_hoff, underflow and wrong alignment
+		$tup->{t_hoff} -= 3;
+
+		push @expected,
+			qr/${$header}tuple data should begin at byte 24, but actually begins at byte 21 \(3 attributes, no nulls\)/;
+	}
+	elsif ($offnum == 9)
+	{
+		# Corrupt the tuple to look like it has lots of attributes, not just 3
+		$tup->{t_infomask2} |= HEAP_NATTS_MASK;
+
+		push @expected,
+			qr/${$header}number of attributes 2047 exceeds maximum expected for table 3/;
+	}
+	elsif ($offnum == 10)
+	{
+		# Corrupt the tuple to look like it has lots of attributes, some of
+		# them null.  This falsely creates the impression that the t_bits
+		# array is longer than just one byte, but t_hoff still says otherwise.
+		$tup->{t_infomask} |= HEAP_HASNULL;
+		$tup->{t_infomask2} |= HEAP_NATTS_MASK;
+		$tup->{t_bits} = 0xAA;
+
+		push @expected,
+			qr/${$header}tuple data should begin at byte 280, but actually begins at byte 24 \(2047 attributes, has nulls\)/;
+	}
+	elsif ($offnum == 11)
+	{
+		# Same as above, but this time t_hoff plays along
+		$tup->{t_infomask} |= HEAP_HASNULL;
+		$tup->{t_infomask2} |= (HEAP_NATTS_MASK & 0x40);
+		$tup->{t_bits} = 0xAA;
+		$tup->{t_hoff} = 32;
+
+		push @expected,
+			qr/${$header}number of attributes 67 exceeds maximum expected for table 3/;
+	}
+	elsif ($offnum == 12)
+	{
+		# Corrupt the bits in column 'b' 1-byte varlena header
+		$tup->{b_header} = 0x80;
+
+		$header = header(0, $offnum, 1);
+		push @expected,
+			qr/${header}attribute 1 with length 4294967295 ends at offset 416848000 beyond total tuple length 58/;
+	}
+	elsif ($offnum == 13)
+	{
+		# Corrupt the bits in column 'c' toast pointer
+		$tup->{c6} = 41;
+		$tup->{c7} = 41;
+
+		$header = header(0, $offnum, 2);
+		push @expected,
+			qr/${header}final toast chunk number 0 differs from expected value 6/,
+			qr/${header}toasted value for attribute 2 missing from toast table/;
+	}
+	elsif ($offnum == 14)
+	{
+		# Set both HEAP_XMAX_LOCK_ONLY and HEAP_KEYS_UPDATED
+		$tup->{t_infomask} |= HEAP_XMAX_LOCK_ONLY;
+		$tup->{t_infomask2} |= HEAP_KEYS_UPDATED;
+
+		push @expected,
+			qr/${header}tuple is marked as only locked, but also claims key columns were updated/;
+	}
+	elsif ($offnum == 15)
+	{
+		# Set both HEAP_XMAX_COMMITTED and HEAP_XMAX_IS_MULTI
+		$tup->{t_infomask} |= HEAP_XMAX_COMMITTED;
+		$tup->{t_infomask} |= HEAP_XMAX_IS_MULTI;
+		$tup->{t_xmax} = 4;
+
+		push @expected,
+			qr/${header}multitransaction ID 4 equals or exceeds next valid multitransaction ID 1/;
+	}
+	elsif ($offnum == 16)	# Last offnum must equal ROWCOUNT
+	{
+		# Set both HEAP_XMAX_COMMITTED and HEAP_XMAX_IS_MULTI
+		$tup->{t_infomask} |= HEAP_XMAX_COMMITTED;
+		$tup->{t_infomask} |= HEAP_XMAX_IS_MULTI;
+		$tup->{t_xmax} = 4000000000;
+
+		push @expected,
+			qr/${header}multitransaction ID 4000000000 precedes relation minimum multitransaction ID threshold 1/;
+	}
+	write_tuple($file, $offset, $tup);
+}
+close($file);
+$node->start;
+
+# Run pg_amcheck against the corrupt table with epoch=0, comparing actual
+# corruption messages against the expected messages
+$node->command_checks_all(
+	['pg_amcheck', '--exclude-indexes', '-p', $port, 'postgres'],
+	0,
+	[ @expected ],
+	[ qr/^$/ ],
+	'Expected corruption message output');
+
+$node->teardown_node;
+$node->clean_node;
diff --git a/contrib/pg_amcheck/t/005_opclass_damage.pl b/contrib/pg_amcheck/t/005_opclass_damage.pl
new file mode 100644
index 0000000000..379225cbf8
--- /dev/null
+++ b/contrib/pg_amcheck/t/005_opclass_damage.pl
@@ -0,0 +1,52 @@
+# This regression test checks the behavior of the btree validation in the
+# presence of breaking sort order changes.
+#
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 6;
+
+my $node = get_new_node('test');
+$node->init;
+$node->start;
+
+# Create a custom operator class and an index which uses it.
+$node->safe_psql('postgres', q(
+	CREATE EXTENSION amcheck;
+
+	CREATE FUNCTION int4_asc_cmp (a int4, b int4) RETURNS int LANGUAGE sql AS $$
+		SELECT CASE WHEN $1 = $2 THEN 0 WHEN $1 > $2 THEN 1 ELSE -1 END; $$;
+
+	CREATE OPERATOR CLASS int4_fickle_ops FOR TYPE int4 USING btree AS
+	    OPERATOR 1 < (int4, int4), OPERATOR 2 <= (int4, int4),
+	    OPERATOR 3 = (int4, int4), OPERATOR 4 >= (int4, int4),
+	    OPERATOR 5 > (int4, int4), FUNCTION 1 int4_asc_cmp(int4, int4);
+
+	CREATE TABLE int4tbl (i int4);
+	INSERT INTO int4tbl (SELECT * FROM generate_series(1,1000) gs);
+	CREATE INDEX fickleidx ON int4tbl USING btree (i int4_fickle_ops);
+));
+
+# We have not yet broken the index, so we should get no corruption
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $node->port, 'postgres' ],
+	qr/^$/,
+	'pg_amcheck all schemas, tables and indexes reports no corruption');
+
+# Change the operator class to use a function which sorts in a different
+# order to corrupt the btree index
+$node->safe_psql('postgres', q(
+	CREATE FUNCTION int4_desc_cmp (int4, int4) RETURNS int LANGUAGE sql AS $$
+		SELECT CASE WHEN $1 = $2 THEN 0 WHEN $1 > $2 THEN -1 ELSE 1 END; $$;
+	UPDATE pg_catalog.pg_amproc
+		SET amproc = 'int4_desc_cmp'::regproc
+		WHERE amproc = 'int4_asc_cmp'::regproc
+));
+
+# Index corruption should now be reported
+$node->command_like(
+	[ 'pg_amcheck', '-p', $node->port, 'postgres' ],
+	qr/item order invariant violated for index "fickleidx"/,
+	'pg_amcheck all schemas, tables and indexes reports fickleidx corruption'
+);
diff --git a/doc/src/sgml/contrib.sgml b/doc/src/sgml/contrib.sgml
index d3ca4b6932..7e101f7c11 100644
--- a/doc/src/sgml/contrib.sgml
+++ b/doc/src/sgml/contrib.sgml
@@ -185,6 +185,7 @@ pages.
   </para>
 
  &oid2name;
+ &pgamcheck;
  &vacuumlo;
  </sect1>
 
diff --git a/doc/src/sgml/filelist.sgml b/doc/src/sgml/filelist.sgml
index db1d369743..5115cb03d0 100644
--- a/doc/src/sgml/filelist.sgml
+++ b/doc/src/sgml/filelist.sgml
@@ -133,6 +133,7 @@
 <!ENTITY oldsnapshot     SYSTEM "oldsnapshot.sgml">
 <!ENTITY pageinspect     SYSTEM "pageinspect.sgml">
 <!ENTITY passwordcheck   SYSTEM "passwordcheck.sgml">
+<!ENTITY pgamcheck       SYSTEM "pgamcheck.sgml">
 <!ENTITY pgbuffercache   SYSTEM "pgbuffercache.sgml">
 <!ENTITY pgcrypto        SYSTEM "pgcrypto.sgml">
 <!ENTITY pgfreespacemap  SYSTEM "pgfreespacemap.sgml">
diff --git a/doc/src/sgml/pgamcheck.sgml b/doc/src/sgml/pgamcheck.sgml
new file mode 100644
index 0000000000..2b2c73ca8b
--- /dev/null
+++ b/doc/src/sgml/pgamcheck.sgml
@@ -0,0 +1,1004 @@
+<!-- doc/src/sgml/pgamcheck.sgml -->
+
+<refentry id="pgamcheck">
+ <indexterm zone="pgamcheck">
+  <primary>pg_amcheck</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle><application>pg_amcheck</application></refentrytitle>
+  <manvolnum>1</manvolnum>
+  <refmiscinfo>Application</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>pg_amcheck</refname>
+  <refpurpose>checks for corruption in one or more <productname>PostgreSQL</productname> databases</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+  <cmdsynopsis>
+   <command>pg_amcheck</command>
+   <arg rep="repeat"><replaceable>option</replaceable></arg>
+   <arg rep="repeat"><replaceable>dbname</replaceable></arg>
+  </cmdsynopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <application>pg_amcheck</application> supports running
+   <xref linkend="amcheck"/>'s corruption checking functions against one or more
+   databases, with options to select which schemas, tables and indexes to check,
+   which kinds of checking to perform, and whether to perform the checks in
+   parallel, and if so, the number of parallel connections to establish and use.
+  </para>
+
+  <para>
+   Only table relations and btree indexes are currently supported.  Other
+   relation types are silently skipped.
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Usage</title>
+
+  <refsect2>
+   <title>Parallelism Options</title>
+
+   <para>
+    <variablelist>
+     <varlistentry>
+      <term><literal>pg_amcheck --jobs=20 --all</literal></term>
+      <listitem>
+       <para>
+        Check all databases one after another, but for each database checked,
+        use up to 20 simultaneous connections to check relations in parallel.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --jobs=8 mydb yourdb</literal></term>
+      <listitem>
+       <para>
+        Check databases <literal>mydb</literal> and <literal>yourdb</literal>
+        one after another, using up to 8 simultaneous connections to check
+        relations in parallel.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </para>
+  </refsect2>
+    
+  <refsect2>
+   <title>Checking Option Specification</title>
+
+   <para>
+    If no checking options are specified, by default all table relation checks
+    and default level btree index checks are performed.  A variety of options
+    exist to change the set of checks performed on whichever relations are
+    being checked.  They are briefly mentioned here in the following examples,
+    but see their full descriptions below.
+   </para>
+   
+   <para>
+    <variablelist>
+     <varlistentry>
+      <term><literal>pg_amcheck --parent-check --heapallindexed</literal></term>
+      <listitem>
+       <para>
+        For each btree index checked, performs more extensive checks.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --exclude-toast-pointers</literal></term>
+      <listitem>
+       <para>
+        For each table relation checked, do not check toast pointers against
+        the toast relation.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --on-error-stop</literal></term>
+      <listitem>
+       <para>
+        For each table relation checked, do not continue checking pages after
+        the first page where corruption is encountered.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --skip="all-frozen"</literal></term>
+      <listitem>
+       <para>
+        For each table relation checked, skips over blocks marked as all
+        frozen.  Note that "all-visible" may also be specified.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --startblock=3000 --endblock=4000</literal></term>
+      <listitem>
+       <para>
+        For each table relation checked, check only blocks in the given block
+        range.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </para>
+  </refsect2>
+
+  <refsect2>
+   <title>Relation Specification</title>
+
+   <para>
+    If no relations are explicitly listed, by default all relations will be
+    checked, but there are options to specify which relations to check.
+   </para>
+   <para>
+    <variablelist>
+     <varlistentry>
+      <term><literal>pg_amcheck -r mytable -r yourtable</literal></term>
+      <listitem>
+       <para>
+        If one or more relations are explicitly given, they are interpreted as
+        an exhaustive list of all relations to be checked, with one caveat:
+        for all such relations, associated toast relations and indexes are by
+        default included in the list of relations to check.
+       </para>
+       <para>
+        Assuming <literal>mytable</literal> is an ordinary table, and that it
+        is indexed by <literal>mytable_idx</literal> and has an associated
+        toast table <literal>pg_toast_12345</literal>, checking will be
+        performed on <literal>mytable</literal>,
+        <literal>mytable_idx</literal>, and <literal>pg_toast_12345</literal>.
+       </para>
+       <para>
+        Likewise for <literal>yourtable</literal>.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -r mytable --no-dependents</literal></term>
+      <listitem>
+       <para>
+        This restricts the list of relations checked to just
+        <literal>mytable</literal>, without pulling in the corresponding
+        indexes or toast, but see also
+        <option>--exclude-toast-pointers</option>.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -t mytable -i myindex</literal></term>
+      <listitem>
+       <para>
+        The <option>-r</option> (<option>--relation</option>) will match any
+        relation, but <option>-t</option> (<option>--table</option>) and
+        <option>-i</option> (<option>--index</option>) may be used to avoid
+        matching objects of the other type.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -r="my*" -R="mytemp*"</literal></term>
+      <listitem>
+       <para>
+        Relations may be included (<option>-r</option>) or excluded
+        (<option>-R</option>) using shell-style patterns.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -r="my*" -I="myanmar"</literal></term>
+      <listitem>
+       <para>
+        Table and index inclusion and exclusion patterns may be used
+        equivalently with <option>-t</option>, <option>-T</option>,
+        <option>-i</option> and <option>-I</option>.  The above example checks
+        all tables and indexes starting with <literal>my</literal> except for
+        indexes starting with <literal>myanmar</literal>.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -R="india" -T="laos" -I="myanmar"</literal></term>
+      <listitem>
+       <para>
+        Unlike specifying one ore more <option>--relation</option> options, which
+        disables the default behavior of checking all relations, specifying one or
+        more of <option>-R</option>, <option>-T</option> or <option>-I</option> does not.
+        The above command will check all relations not named
+        <literal>india</literal>, not a table named
+        <literal>laos</literal>, nor an index named <literal>myanmar</literal>.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </para>
+  </refsect2>
+
+  <refsect2>
+   <title>Schema Specification</title>
+
+   <para>
+    If no schemas are explicitly listed, by default all schemas except
+    <literal>pg_catalog</literal> and <literal>pg_toast</literal> will be
+    checked.
+   </para>
+   <para>
+    <variablelist>
+     <varlistentry>
+      <term><literal>pg_amcheck -s s1 -s s2 -r mytable</literal></term>
+      <listitem>
+       <para>
+        If one or more schemas are listed with <option>-s</option>, unqualified
+        relation names will be checked only in the given schemas.  The above
+        command will check tables <literal>s1.mytable</literal> and
+        <literal>s2.mytable</literal> but not tables named
+        <literal>mytable</literal> in other schemas.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -S s1 -S s2 -r mytable</literal></term>
+      <listitem>
+       <para>
+        As with relations, schemas may be excluded.  The above command will
+        check any table named <literal>mytable</literal> not in schemas
+        <literal>s1</literal> and <literal>s2</literal>.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -S s1 -S s2 -r mytable -t s1.stuff</literal></term>
+      <listitem>
+       <para>
+        Relations may be included or excluded with a schema-qualified name
+        without interference from the <option>-s</option> or
+        <option>-S</option> options.  Even though schema <literal>s1</literal>
+        has been excluded, the table <literal>s1.stuff</literal> will be
+        checked.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </para>
+  </refsect2>
+
+  <refsect2>
+   <title>Database Specification</title>
+
+   <para>
+    If no databases are explicitly listed, the database to check is obtained
+    from environment variables in the usual way.  Otherwise, when one or more
+    databases are explicitly given, they are interpreted as an exhaustive list
+    of all databases to be checked.  This list of databases to check may
+    contain patterns, but because any such patterns need to be reconciled
+    against a list of all databases to find the matching database names, at
+    least one database specified must be a literal database name and not merely
+    a pattern, and the one so specified must be in a location where
+    <application>pg_amcheck</application> expects to find it.
+   </para>
+   <para>
+    For example:
+   </para>
+   <para>
+    <variablelist>
+     <varlistentry>
+      <term><literal>pg_amcheck --all --maintenance-db=foo</literal></term>
+      <listitem>
+       <para>
+        If the <option>--maintenance-db</option> option is given, it will be
+        used to look up the matching databases, though it will not itself be
+        added to the list of databases for checking.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck foo bar baz</literal></term>
+      <listitem>
+       <para>
+        Otherwise, if one or more plain database name arguments not preceded by
+        <option>-d</option> or <option>--dbname</option> are given, the first
+        one will be used for this purpose, and it will also be included in the
+        list of databases to check.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck -d foo -d bar baz</literal></term>
+      <listitem>
+       <para>
+        If a mixture of plain database names and databases preceded with
+        <option>-d</option> or <option>--dbname</option> are given, the first
+        plain database name will be used for this purpose.  In the above
+        example, <literal>baz</literal> will be used.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --dbname=foo --dbname="bar*"</literal></term>
+      <listitem>
+       <para>
+        Otherwise, if one or more databases are given with the
+        <option>-d</option> or <option>--dbname</option> option, the first one
+        will be used and must be a literal database name.  In this example,
+        <literal>foo</literal> will be used.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry>
+      <term><literal>pg_amcheck --relation="accounts_*.*.*"</literal></term>
+      <listitem>
+       <para>
+        Otherwise, the environment will be consulted for the database to be
+        used.  In the example above, the default database will be queried to
+        find all databases with names that begin with
+        <literal>accounts_</literal>.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </para>
+
+   <para>
+    As discussed above for schema-qualified relations, a database-qualified
+    relation name or pattern may also be given.
+<programlisting>
+pg_amcheck mydb \
+           --schema="t*" \
+           --exclude-schema="tmp*" \
+           --relation=baz \
+           --relation=bar.baz \
+           --relation=foo.bar.baz \
+           --relation="f*".a.b \
+           --exclude-relation=foo.a.b
+</programlisting>
+    will check relations in database <literal>mydb</literal> using the schema
+    resolution rules discussed above, but additionally will check all relations
+    named <literal>a.b</literal> in all databases with names starting with
+    <literal>f</literal> except database <literal>foo</literal>.
+   </para>
+
+  </refsect2>
+ </refsect1>
+
+ <refsect1>
+  <title>Options</title>
+
+  <para>
+   <application>pg_amcheck</application> accepts the following command-line arguments:
+  </para>
+
+  <refsect2>
+   <title>Help and Version Information Options</title>
+
+   <variablelist>
+    <varlistentry>
+     <term><option>-?</option></term>
+     <term><option>--help</option></term>
+     <listitem>
+      <para>
+       Show help about <application>pg_amcheck</application> command line
+       arguments, and exit.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-V</option></term>
+     <term><option>--version</option></term>
+     <listitem>
+      <para>
+       Print the <application>pg_amcheck</application> version and exit.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-e</option></term>
+     <term><option>--echo</option></term>
+     <listitem>
+      <para>
+       Print to stdout all commands and queries being executed against the
+       server.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-q</option></term>
+     <term><option>--quiet</option></term>
+     <listitem>
+      <para>
+       Do not write additional messages beyond those about corruption.
+      </para>
+      <para>
+       This option does not quiet any output specifically due to the use of
+       the <option>-e</option> <option>--echo</option> option.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-v</option></term>
+     <term><option>--verbose</option></term>
+     <listitem>
+      <para>
+       Increases the log level verbosity.  This option may be given more than
+       once.
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+  </refsect2>
+
+  <refsect2>
+   <title>Database Connection and Concurrent Connection Options</title>
+
+   <variablelist>
+    <varlistentry>
+     <term><option>-h</option></term>
+     <term><option>--host=HOSTNAME</option></term>
+     <listitem>
+      <para>
+       Specifies the host name of the machine on which the server is running.
+       If the value begins with a slash, it is used as the directory for the
+       Unix domain socket.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-p</option></term>
+     <term><option>--port=PORT</option></term>
+     <listitem>
+      <para>
+       Specifies the TCP port or local Unix domain socket file extension on
+       which the server is listening for connections.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-U</option></term>
+     <term><option>--username=USERNAME</option></term>
+     <listitem>
+      <para>
+       User name to connect as.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-w</option></term>
+     <term><option>--no-password</option></term>
+     <listitem>
+      <para>
+       Never issue a password prompt.  If the server requires password
+       authentication and a password is not available by other means such as
+       a <filename>.pgpass</filename> file, the connection attempt will fail.
+       This option can be useful in batch jobs and scripts where no user is
+       present to enter a password.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-W</option></term>
+     <term><option>--password</option></term>
+     <listitem>
+      <para>
+       Force <application>pg_amcheck</application> to prompt for a password
+       before connecting to a database.
+      </para>
+      <para>
+       This option is never essential, since
+       <application>pg_amcheck</application> will automatically prompt for a
+       password if the server demands password authentication.  However,
+       <application>pg_amcheck</application> will waste a connection attempt
+       finding out that the server wants a password.  In some cases it is
+       worth typing <option>-W</option> to avoid the extra connection attempt.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>--maintenance-db=DBNAME</option></term>
+     <listitem>
+      <para>
+       Specifies the name of the database to connect to when querying the
+       list of all databases.  If not specified, the
+       <literal>postgres</literal> database will be used; if that does not
+       exist <literal>template1</literal> will be used.  This can be a
+       <link linkend="libpq-connstring">connection string</link>.  If so,
+       connection string parameters will override any conflicting command
+       line options.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-j</option></term>
+     <term><option>--jobs=NUM</option></term>
+     <listitem>
+      <para>
+       Use the specified number of concurrent connections to the server, or
+       one per object to be checked, whichever number is smaller.
+      </para>
+      <para>
+       When used in conjunction with the <option>-a</option>
+       <option>--all</option> option, the total number of objects to check,
+       and correspondingly the number of concurrent connections to use, is
+       recalculated per database.  If the number of objects to check differs
+       from one database to the next and is less than the concurrency level
+       specified, the number of concurrent connections open to the server
+       will fluctuate to meet the needs of each database processed.
+      </para>
+      <para>
+       The default is to use a single connection.
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+  </refsect2>
+
+  <refsect2>
+   <title>Options Controlling Index Checking Functions</title>
+
+   <variablelist>
+    <varlistentry>
+     <term><option>-P</option></term>
+     <term><option>--parent-check</option></term>
+     <listitem>
+      <para>
+       For each btree index checked, use <xref linkend="amcheck"/>'s
+       <function>bt_index_parent_check</function> function, which performs
+       additional checks of parent/child relationships during index checking.
+      </para>
+      <para>
+       The default is to use <application>amcheck</application>'s
+       <function>bt_index_check</function> function, but note that use of the
+       <option>--rootdescend</option> option implicitly
+       selects <function>bt_index_parent_check</function>.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>-H</option></term>
+     <term><option>--heapallindexed</option></term>
+     <listitem>
+      <para>
+       For each index checked, verify the presence of all heap tuples as index
+       tuples in the index using <application>amcheck</application>'s
+       <option>heapallindexed</option> option.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>--rootdescend</option></term>
+     <listitem>
+      <para>
+       For each index checked, re-find tuples on the leaf level by performing
+       a new search from the root page for each tuple using
+       <xref linkend="amcheck"/>'s <option>rootdescend</option> option.
+      </para>
+      <para>
+       Use of this option implicitly also selects the <option>-P</option>
+       <option>--parent-check</option> option.
+      </para>
+      <para>
+       This form of verification was originally written to help in the
+       development of btree index features.  It may be of limited use or even
+       of no use in helping detect the kinds of corruption that occur in
+       practice.  It may also cause corruption checking to take considerably
+       longer and consume considerably more resources on the server.
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+  </refsect2>
+
+  <refsect2>
+   <title>Options Controlling Table Checking Functions</title>
+
+   <variablelist>
+    <varlistentry>
+     <term><option>--exclude-toast-pointers</option></term>
+     <listitem>
+      <para>
+       When checking main relations, do not look up entries in toast tables
+       corresponding to toast pointers in the main releation.
+      </para>
+      <para>
+       The default behavior checks each toast pointer encountered in the main
+       table to verify, as much as possible, that the pointer points at
+       something in the toast table that is reasonable.  Toast pointers which
+       point beyond the end of the toast table, or to the middle (rather than
+       the beginning) of a toast entry, are identified as corrupt.
+      </para>
+      <para>
+       The process by which <xref linkend="amcheck"/>'s
+       <function>verify_heapam</function> function checks each toast pointer
+       is slow and may be improved in a future release.  Some users may wish
+       to disable this check to save time.
+      </para>
+      <para>
+       Note that, despite their similar names, this option is unrelated to the
+       <option>--exclude-toast</option> option.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>--on-error-stop</option></term>
+     <listitem>
+      <para>
+       After reporting all corruptions on the first page of a table where
+       corruptions are found, stop processing that table relation and move on
+       to the next table or index.
+      </para>
+      <para>
+       Note that index checking always stops after the first corrupt page.
+       This option only has meaning relative to table relations.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>--skip=OPTION</option></term>
+     <listitem>
+      <para>
+       If <literal>"all-frozen"</literal> is given, table corruption checks
+       will skip over pages in all tables that are marked as all frozen.
+      </para>
+      <para>
+       If <literal>"all-visible"</literal> is given, table corruption checks
+       will skip over pages in all tables that are marked as all visible.
+      </para>
+      <para>
+       By default, no pages are skipped.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>--startblock=BLOCK</option></term>
+     <listitem>
+      <para>
+       Skip (do not check) pages prior to the given starting block.
+      </para>
+      <para>
+       By default, no pages are skipped.  This option will be applied to all
+       table relations that are checked, including toast tables.
+      </para>
+     </listitem>
+    </varlistentry>
+    <varlistentry>
+     <term><option>--endblock=BLOCK</option></term>
+     <listitem>
+      <para>
+       Skip (do not check) all pages after the given ending block.
+      </para>
+      <para>
+       By default, no pages are skipped.  This option will be applied to all
+       table relations that are checked, including toast tables.
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+  </refsect2>
+
+  <refsect2>
+   <title>Corruption Checking Target Options</title>
+
+   <para>
+    Objects to be checked may span schemas in more than one database.  Options
+    for restricting the list of databases, schemas, tables and indexes are
+    described below.  In each place where a name may be specified, a
+    <link linkend="app-psql-patterns"><replaceable class="parameter">pattern</replaceable></link>
+    may also be used.
+   </para>
+
+   <para>
+    <variablelist>
+     <varlistentry>
+      <term><option>--all</option></term>
+        <listitem>
+       <para>
+        Perform checking in all databases.
+       </para>
+       <para>
+        In the absence of any other options, selects all objects across all
+        schemas and databases.
+       </para>
+       <para>
+        Option <option>-D</option> <option>--exclude-db</option> takes
+        precedence over <option>--all</option>.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-d</option></term>
+      <term><option>--dbname</option></term>
+      <listitem>
+       <para>
+        Perform checking in the specified database.
+       </para>
+       <para>
+        This option may be specified multiple times to list more than one
+        database (or database pattern) for checking.  By default, all objects in
+        the matching database(s) will be checked.
+       </para>
+       <para>
+        If no <option>maintenance-db</option> argument is given nor is any
+        database name given as a command line argument, the first argument
+        specified with <option>-d</option> <option>--dbname</option> will be
+        used for the initial connection.  If that argument is not a literal
+        database name, the attempt to connect will fail.
+       </para>
+       <para>
+        If <option>--all</option> is also specified, <option>-d</option>
+        <option>--dbname</option> does not affect which databases are checked,
+        but may be used to specify the database for the initial connection.
+       </para>
+       <para>
+        Option <option>-D</option> <option>--exclude-db</option> takes
+        precedence over <option>-d</option> <option>--dbname</option>.
+       </para>
+       <para>
+        Examples:
+        <simplelist>
+         <member><literal>--dbname=africa</literal></member>
+         <member><literal>--dbname="a*"</literal></member>
+         <member><literal>--dbname="africa|asia|europe"</literal></member>
+        </simplelist>
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-D</option></term>
+      <term><option>--exclude-db</option></term>
+      <listitem>
+       <para>
+        Do not perform checking in the specified database.
+       </para>
+       <para>
+        This option may be specified multiple times to list more than one
+        database (or database pattern) for exclusion.
+       </para>
+       <para>
+        If a database which is included using <option>--all</option> or
+        <option>-d</option> <option>--dbname</option> is also excluded using
+        <option>-D</option> <option>--exclude-db</option>, the database will be
+        excluded.
+       </para>
+       <para>
+        Examples:
+        <simplelist>
+         <member><literal>--exclude-db=america</literal></member>
+         <member><literal>--exclude-db="*pacific*"</literal></member>
+        </simplelist>
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-s</option></term>
+      <term><option>--schema</option></term>
+      <listitem>
+       <para>
+        Perform checking in the specified schema(s).
+       </para>
+       <para>
+        This option may be specified multiple times to list more than one
+        schema (or schema pattern) for checking.  By default, all objects in
+        the matching schema(s) will be checked.
+       </para>
+       <para>
+        Option <option>-S</option> <option>--exclude-schema</option> takes
+        precedence over <option>-s</option> <option>--schema</option>.
+       </para>
+       <para>
+        Examples:
+        <simplelist>
+         <member><literal>--schema=corp</literal></member>
+         <member><literal>--schema="corp|llc|npo"</literal></member>
+        </simplelist>
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-S</option></term>
+      <term><option>--exclude-schema</option></term>
+      <listitem>
+       <para>
+        Do not perform checking in the specified schema.
+       </para>
+       <para>
+        This option may be specified multiple times to list more than one
+        schema (or schema pattern) for exclusion.
+       </para>
+       <para>
+        If a schema which is included using
+        <option>-s</option> <option>--schema</option> is also excluded using
+        <option>-S</option> <option>--exclude-schema</option>, the schema will be
+        excluded.
+       </para>
+       <para>
+        Examples:
+        <simplelist>
+         <member><literal>-S corp -S llc</literal></member>
+         <member><literal>--exclude-schema="*c*"</literal></member>
+        </simplelist>
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-r</option></term>
+      <term><option>--relation</option></term>
+      <listitem>
+       <para>
+        Perform checking on the specified relation(s).
+       </para>
+       <para>
+        This option may be specified multiple times to list more than one
+        relation (or relation pattern) for checking.
+       </para>
+       <para>
+        Option <option>-R</option> <option>--exclude-relation</option> takes
+        precedence over <option>-r</option> <option>--relation</option>.
+       </para>
+       <para>
+        If the relation is not schema qualified, database and schema
+        inclusion/exclusion lists will determine in which databases or schemas
+        matching relations will be checked.
+       </para>
+       <para>
+        Examples:
+        <simplelist>
+         <member><literal>--relation=accounts_idx</literal></member>
+         <member><literal>--relation="llc.accounts_idx"</literal></member>
+         <member><literal>--relation="asia|africa.corp|llc.accounts_idx"</literal></member>
+        </simplelist>
+       </para>
+       <para>
+        The first example, <literal>--relation=accounts_idx</literal>, checks
+        relations named <literal>accounts_idx</literal> in all selected schemas
+        and databases.
+       </para>
+       <para>
+        The second example, <literal>--relation="llc.accounts_idx"</literal>,
+        checks relations named <literal>accounts_idx</literal> in schema
+        <literal>llc</literal> in all selected databases. 
+       </para>
+       <para>
+        The third example,
+        <literal>--relation="asia|africa.corp|llc.accounts_idx"</literal>,
+        checks relations named <literal>accounts_idx</literal> in
+        schemas <literal>corp</literal> and <literal>llc</literal> in databases
+        <literal>asia</literal> and <literal>africa</literal>.
+       </para>
+       <para>
+        Note that if a database is implicated in a relation pattern, such as
+        <literal>asia</literal> and <literal>africa</literal> in the third
+        example above, the database need not be otherwise given in the command
+        arguments for the relation to be checked.  As an extreme example of
+        this:
+        <simplelist>
+         <member><literal>pg_amcheck --relation="*.*.*" mydb</literal></member>
+        </simplelist>
+        will check all relations in all databases.  The <literal>mydb</literal>
+        argument only serves to tell <application>pg_amcheck</application> the
+        name of the database to use for querying the list of all databases.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-R</option></term>
+      <term><option>--exclude-relation</option></term>
+      <listitem>
+       <para>
+        Exclude checks on the specified relation(s).
+       </para>
+       <para>
+        Option <option>-R</option> <option>--exclude-relation</option> takes
+        precedence over <option>-r</option> <option>--relation</option>,
+        <option>-t</option> <option>--table</option> and <option>-i</option>
+        <option>--index</option>.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-t</option></term>
+      <term><option>--table</option></term>
+      <listitem>
+       <para>
+        Perform checks on the specified tables(s).  This is an alias for the
+        <option>-r</option> <option>--relation</option> option, except that it
+        applies only to tables, not indexes.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-T</option></term>
+      <term><option>--exclude-table</option></term>
+      <listitem>
+       <para>
+        Exclude checks on the specified tables(s).  This is an alias for the
+        <option>-R</option> <option>--exclude-relation</option> option, except
+        that it applies only to tables, not indexes.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-i</option></term>
+      <term><option>--index</option></term>
+      <listitem>
+       <para>
+        Perform checks on the specified index(es).  This is an alias for the
+        <option>-r</option> <option>--relation</option> option, except that it
+        applies only to indexes, not tables.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-I</option></term>
+      <term><option>--exclude-index</option></term>
+      <listitem>
+       <para>
+        Exclude checks on the specified index(es).  This is an alias for the
+        <option>-R</option> <option>--exclude-relation</option> option, except
+        that it applies only to indexes, not tables.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>--no-dependents</option></term>
+      <listitem>
+       <para>
+        When calculating the list of objects to be checked, do not automatically
+        expand the list to include associated indexes and toast tables of
+        elements otherwise in the list.
+       </para>
+       <para>
+        By default, for each main table relation checked, any associated toast
+        table and all associated indexes are also checked, unless explicitly
+        excluded.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </para>
+  </refsect2>
+ </refsect1>
+
+ <refsect1>
+  <title>Notes</title>
+
+  <para>
+   <application>pg_amcheck</application> is designed to work with
+   <productname>PostgreSQL</productname> 14.0 and later.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Author</title>
+
+  <para>
+   Mark Dilger <email>mark.dilger@enterprisedb.com</email>
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="amcheck"/></member>
+  </simplelist>
+ </refsect1>
+</refentry>
diff --git a/src/tools/msvc/Install.pm b/src/tools/msvc/Install.pm
index ea3af48777..49ad558b74 100644
--- a/src/tools/msvc/Install.pm
+++ b/src/tools/msvc/Install.pm
@@ -18,7 +18,7 @@ our (@ISA, @EXPORT_OK);
 @EXPORT_OK = qw(Install);
 
 my $insttype;
-my @client_contribs = ('oid2name', 'pgbench', 'vacuumlo');
+my @client_contribs = ('oid2name', 'pg_amcheck', 'pgbench', 'vacuumlo');
 my @client_program_files = (
 	'clusterdb',      'createdb',   'createuser',    'dropdb',
 	'dropuser',       'ecpg',       'libecpg',       'libecpg_compat',
diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm
index 7be6e6c9e5..53fbfa012e 100644
--- a/src/tools/msvc/Mkvcbuild.pm
+++ b/src/tools/msvc/Mkvcbuild.pm
@@ -33,9 +33,9 @@ my @unlink_on_exit;
 
 # Set of variables for modules in contrib/ and src/test/modules/
 my $contrib_defines = { 'refint' => 'REFINT_VERBOSE' };
-my @contrib_uselibpq = ('dblink', 'oid2name', 'postgres_fdw', 'vacuumlo');
-my @contrib_uselibpgport   = ('oid2name', 'vacuumlo');
-my @contrib_uselibpgcommon = ('oid2name', 'vacuumlo');
+my @contrib_uselibpq = ('dblink', 'oid2name', 'pg_amcheck', 'postgres_fdw', 'vacuumlo');
+my @contrib_uselibpgport   = ('oid2name', 'pg_amcheck', 'vacuumlo');
+my @contrib_uselibpgcommon = ('oid2name', 'pg_amcheck', 'vacuumlo');
 my $contrib_extralibs      = undef;
 my $contrib_extraincludes = { 'dblink' => ['src/backend'] };
 my $contrib_extrasource = {
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 4d0d09a5dd..26920cc512 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -336,6 +336,8 @@ CheckPointStmt
 CheckpointStatsData
 CheckpointerRequest
 CheckpointerShmemStruct
+CheckType
+CheckTypeFilter
 Chromosome
 CkptSortItem
 CkptTsStatus
@@ -2847,6 +2849,8 @@ ambuildempty_function
 ambuildphasename_function
 ambulkdelete_function
 amcanreturn_function
+amcheckObjects
+amcheckOptions
 amcostestimate_function
 amendscan_function
 amestimateparallelscan_function
-- 
2.21.1 (Apple Git-122.3)

=

v36-0005-Extending-PostgresNode-to-test-corruption.patch (application/octet-stream, 16.6K) download
From 30808fc59c1a1eb6f9a238b70e103846de44b834 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Tue, 2 Feb 2021 12:37:58 -0800
Subject: [PATCH v36 5/5] Extending PostgresNode to test corruption.

PostgresNode now has functions for overwriting relation files
with full or partial prior versions of those files, creating
corruption beyond merely twiddling the bits of a heap relation
file.

Adding a regression test for pg_amcheck based on this new
functionality.
---
 contrib/pg_amcheck/t/006_relfile_damage.pl    | 135 +++++++++
 src/test/modules/Makefile                     |   1 +
 src/test/modules/corruption/Makefile          |  16 ++
 .../modules/corruption/t/001_corruption.pl    |  83 ++++++
 src/test/perl/PostgresNode.pm                 | 261 ++++++++++++++++++
 5 files changed, 496 insertions(+)
 create mode 100644 contrib/pg_amcheck/t/006_relfile_damage.pl
 create mode 100644 src/test/modules/corruption/Makefile
 create mode 100644 src/test/modules/corruption/t/001_corruption.pl

diff --git a/contrib/pg_amcheck/t/006_relfile_damage.pl b/contrib/pg_amcheck/t/006_relfile_damage.pl
new file mode 100644
index 0000000000..d997db5b63
--- /dev/null
+++ b/contrib/pg_amcheck/t/006_relfile_damage.pl
@@ -0,0 +1,135 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 27;
+use PostgresNode;
+
+my ($node, $port);
+
+# Returns the name of the toast relation associated with the named relation.
+#
+# Assumes the test node is running
+sub relation_toast($$)
+{
+	my ($dbname, $relname) = @_;
+
+	my $rel = $node->safe_psql($dbname, qq(
+		SELECT ct.relname
+			FROM pg_catalog.pg_class cr, pg_catalog.pg_class ct
+			WHERE cr.oid = '$relname'::regclass
+			  AND cr.reltoastrelid = ct.oid
+			));
+	return undef unless defined $rel;
+	return "pg_toast.$rel";
+}
+
+# Test set-up
+$node = get_new_node('test');
+$node->init;
+$node->start;
+$port = $node->port;
+
+# Load the amcheck extension, upon which pg_amcheck depends
+$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
+
+# Create a table with a btree index.  Use a fillfactor for the table and index
+# that will allow some fraction of updates to be on the original pages and some
+# on new pages.
+#
+$node->safe_psql('postgres', qq(
+create schema t;
+create table t.t1 (id integer, t text) with (fillfactor=75);
+alter table t.t1 alter column t set storage external;
+insert into t.t1 select gs, repeat('x',gs) from generate_series(9990,10000) gs;
+create index t1_idx on t.t1 (id) with (fillfactor=75);
+));
+
+my $toastrel = relation_toast('postgres', 't.t1');
+
+# Flush relation files to disk and take snapshots of the toast and index
+#
+$node->restart;
+$node->take_relfile_snapshot_minimal('postgres', 'idx', 't.t1_idx');
+$node->take_relfile_snapshot_minimal('postgres', 'toast', $toastrel);
+
+# Insert new data into the table and index
+#
+$node->safe_psql('postgres', qq(
+insert into t.t1 select gs, repeat('y',gs) from generate_series(10001,10100) gs;
+));
+
+# Revert index.  The reverted snapshot file is not corrupt, but it also
+# does not match the current contents of the table.
+#
+$node->stop;
+$node->revert_to_snapshot('idx');
+
+# Restart the node and check table and index with varying options.
+#
+$node->start;
+
+# Checks which do not reconcile the index and table via --heapallindexed will
+# not notice any problems
+#
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*' ],
+	qr/^$/,
+	'pg_amcheck reverted index at default checking level');
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*' ],
+	qr/^$/,
+	'pg_amcheck reverted index at default checking level');
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--parent-check' ],
+	qr/^$/,
+	'pg_amcheck reverted index with --parent-check');
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--rootdescend' ],
+	qr/^$/,
+	'pg_amcheck reverted index with --rootdescend');
+
+# Checks which do reconcile the index and table via --heapallindexed will
+# notice the mismatch in their contents
+#
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--heapallindexed' ],
+	qr/heap tuple .* from table "t1" lacks matching index tuple within index "t1_idx"/,
+	'pg_amcheck reverted index with --heapallindexed');
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--heapallindexed', '--rootdescend' ],
+	qr/heap tuple .* from table "t1" lacks matching index tuple within index "t1_idx"/,
+	'pg_amcheck reverted index with --heapallindexed --rootdescend');
+
+# Revert the toast.  The reverted toast table is not corrupt, but it does not
+# have entries for all toast pointers in the main table
+#
+$node->stop;
+$node->revert_to_snapshot('toast');
+
+# Restart the node and check table and toast with varying options.  When
+# checking the toast pointers, we may get errors produced by verify_heapam, but
+# we may also get errors from failure to read toast blocks that are beyond the
+# end of the toast table, of the form /ERROR:  could not read block/.  To avoid
+# having a brittle test, we accept any error message.
+#
+$node->start;
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', $toastrel ],
+	qr/^$/,
+	'pg_amcheck reverted toast table');
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--exclude-toast-pointers' ],
+	qr/^$/,
+	'pg_amcheck with reverted toast using --exclude-toast-pointers');
+
+$node->command_like(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*' ],
+	qr/.+/,			# Any non-empty error message is acceptable
+	'pg_amcheck with reverted toast and default checking');
diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile
index 5391f461a2..c92d1702b4 100644
--- a/src/test/modules/Makefile
+++ b/src/test/modules/Makefile
@@ -7,6 +7,7 @@ include $(top_builddir)/src/Makefile.global
 SUBDIRS = \
 		  brin \
 		  commit_ts \
+		  corruption \
 		  delay_execution \
 		  dummy_index_am \
 		  dummy_seclabel \
diff --git a/src/test/modules/corruption/Makefile b/src/test/modules/corruption/Makefile
new file mode 100644
index 0000000000..ba461c645d
--- /dev/null
+++ b/src/test/modules/corruption/Makefile
@@ -0,0 +1,16 @@
+# src/test/modules/corruption/Makefile
+
+# EXTRA_INSTALL = contrib/pg_amcheck
+
+TAP_TESTS = 1
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = src/test/modules/corruption
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/src/test/modules/corruption/t/001_corruption.pl b/src/test/modules/corruption/t/001_corruption.pl
new file mode 100644
index 0000000000..ae4a262e06
--- /dev/null
+++ b/src/test/modules/corruption/t/001_corruption.pl
@@ -0,0 +1,83 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 10;
+use PostgresNode;
+
+my $node = get_new_node('test');
+$node->init;
+$node->start;
+
+# Create something non-trivial for the first snapshot
+$node->safe_psql('postgres', qq(
+create table t1 (id integer, short_text text, long_text text);
+insert into t1 (id, short_text, long_text)
+	(select gs, 'foo', repeat('x', gs)
+		from generate_series(1,10000) gs);
+create unique index idx1 on t1 (id, short_text);
+vacuum freeze;
+));
+
+# Flush relation files to disk and take snapshot of them
+$node->restart;
+$node->take_relfile_snapshot('postgres', 'snap1', 'public.t1');
+
+# Update data in the table, toast table, and index
+$node->safe_psql('postgres', qq(
+update t1 set
+	short_text = 'bar',
+	long_text = repeat('y', id);
+));
+
+# Flush relation files to disk and take second snapshot
+$node->restart;
+$node->take_relfile_snapshot('postgres', 'snap2', 'public.t1');
+
+# Revert the first page of t1 using a torn snapshot.  This should be a partial
+# and corrupt reverting of the update.
+$node->stop;
+$node->revert_to_torn_relfile_snapshot('snap1', 8192);
+
+# Restart the node and count the number of rows in t1 with the original
+# (pre-update) values.  It should not be zero, but nor will it be the full
+# 10000.
+$node->start;
+my ($old, $new, $oldtoast, $newtoast) = counts();
+ok($old > 0 && $old < 10000, "Torn snapshot reverts some of the main updates");
+ok($new > 0 && $new <= 10000, "Torn snapshot retains some of the main updates");
+
+# Revert t1 fully to the first snapshot.  This should fully restore the
+# original (pre-update) values.
+$node->stop;
+$node->revert_to_snapshot('snap1');
+
+# Restart the node and verify only old values remain
+$node->start;
+($old, $new, $oldtoast, $newtoast) = counts();
+is($old, 10000, "Full snapshot restores all the old main values");
+is($oldtoast, 10000, "Full snapshot restores all the old toast values");
+is($new, 0, "Full snapshot reverts all the new main values");
+is($newtoast, 0, "Full snapshot reverts all the new toast values");
+
+# Restore t1 fully to the second snapshot.  This should fully restore the
+# new (post-update) values.
+$node->stop;
+$node->revert_to_snapshot('snap2');
+
+# Restart the node and verify only new values remain
+$node->start;
+($old, $new, $oldtoast, $newtoast) = counts();
+is($old, 0, "Full snapshot reverts all the old main values");
+is($oldtoast, 0, "Full snapshot reverts all the old toast values");
+is($new, 10000, "Full snapshot restores all the new main values");
+is($newtoast, 10000, "Full snapshot restores all the new toast values");
+
+sub counts {
+	return map {
+		$node->safe_psql('postgres', qq(select count(*) from t1 where $_))
+	} ("short_text = 'foo'",
+	   "short_text = 'bar'",
+	   "long_text ~ 'x'",
+	   "long_text ~ 'y'");
+}
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 9667f7667e..d470af93c5 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -2225,6 +2225,267 @@ sub pg_recvlogical_upto
 
 =back
 
+=head1 DATABASE CORRUPTION METHODS
+
+=over
+
+=item $node->relfile_snapshot_repository()
+
+The path to the parent directory of all directories storing snapshots of
+relation backing files.
+
+=cut
+
+sub relfile_snapshot_repository
+{
+	my ($self) = @_;
+	my $snaprepo = join('/', $self->basedir, 'snapshot');
+	unless (-d $snaprepo)
+	{
+		mkdir $snaprepo
+			or $!{EEXIST}
+			or BAIL_OUT("could not create snapshot repository directory \"$snaprepo\": $!");
+	}
+	return $snaprepo;
+}
+
+=pod
+
+=item $node->relfile_snapshot_directory(snapname)
+
+The path to the directory for storing the named snapshot.
+
+=cut
+
+sub relfile_snapshot_directory
+{
+	my ($self, $snapname) = @_;
+
+	join("/", $self->relfile_snapshot_repository(), $snapname);
+}
+
+=pod
+
+=item $node->take_relfile_snapshot($self, $dbname, $snapname, @relnames)
+
+Makes a copy of the files backing the relations B<@relname>, the associated
+toast relations (if any), and all associated indexes (if any).  No attempt is
+made to flush these files to disk, meaning the snapshot taken could be stale
+unless the caller ensures these files have been flushed prior to calling.
+
+Dies on failure to invoke psql.
+
+Dies on missing relations.
+
+Dies if the given B<$snapname> is already in use.
+
+=cut
+
+=pod
+
+=item $node->take_relfile_snapshot_minimal($self, $dbname, $snapname, @relnames)
+
+Makes a copy of the files backing the relations B<@relnames>.  No attempt is made
+to flush these files to disk, meaning the snapshot taken could be stale unless the
+caller ensures these files have been flushed prior to calling.
+
+Dies on failure to invoke psql.
+
+Dies on missing relation.
+
+Dies if the given B<$snapname> is already in use.
+
+=cut
+
+sub take_relfile_snapshot
+{
+	my ($self, $dbname, $snapname, @relnames) = @_;
+	$self->take_relfile_snapshot_helper($dbname, $snapname, 1, @relnames);
+}
+
+sub take_relfile_snapshot_minimal
+{
+	my ($self, $dbname, $snapname, @relnames) = @_;
+	$self->take_relfile_snapshot_helper($dbname, $snapname, 0, @relnames);
+}
+
+sub take_relfile_snapshot_helper
+{
+	my ($self, $dbname, $snapname, $extended, @relnames) = @_;
+
+	croak "dbname must be specified" unless defined $dbname;
+	croak "relnames must be defined" unless scalar(grep { defined $_ } @relnames);
+	croak "snapname must be specified" unless defined $snapname;
+	croak "snapname must be unique" if exists $self->{snapshot}->{$snapname};
+
+	my $pgdata = $self->data_dir;
+	my $snapdir = $self->relfile_snapshot_directory($snapname);
+	croak "snapname directory name already in use: $snapdir" if (-e $snapdir);
+	mkdir $snapdir
+		or BAIL_OUT("could not create snapshot directory \"$snapdir\": $!");
+
+	my @relpaths = map {
+		$self->safe_psql($dbname,
+			qq(SELECT pg_relation_filepath('$_')));
+	} @relnames;
+
+	my (@toastpaths, @idxpaths);
+	if ($extended)
+	{
+		for my $relname (@relnames)
+		{
+			push (@toastpaths, grep /\w/, split(/(?:\s*\r?\n\s*)+/, $self->safe_psql($dbname,
+				qq(SELECT pg_relation_filepath(c.reltoastrelid)
+					FROM pg_catalog.pg_class c
+					WHERE c.oid = '$relname'::regclass
+					AND c.reltoastrelid != 0::oid))));
+			push (@idxpaths, grep /\w/, split(/(?:\s*\r?\n\s*)+/, $self->safe_psql($dbname,
+				qq(SELECT pg_relation_filepath(i.indexrelid)
+					FROM pg_catalog.pg_index i
+					WHERE i.indrelid = '$relname'::regclass))));
+		}
+	}
+
+	$self->{snapshot}->{$snapname} = {};
+	for my $path (@relpaths, grep { defined($_) } @toastpaths, @idxpaths)
+	{
+		croak "file backing relation is missing: $pgdata/$path" unless -f "$pgdata/$path";
+		copy_file($snapdir, $pgdata, 0, $path);
+		$self->{snapshot}->{$snapname}->{$path} = 1;
+	}
+}
+
+=pod
+
+=item $node->revert_to_snapshot($self, $snapname)
+
+Overwrites the database's relation files with files previously saved in
+B<$snapname>.
+
+Dies if the given B<$snapname> does not exist.
+
+=cut
+
+=pod
+
+=item $node->revert_to_torn_relfile_snapshot($self, $snapname, $bytes)
+
+Partially overwrites the database's relation files using prefixes of the given
+number of bytes from the files saved in B<$snapname>.  If B<$bytes> is
+negative, uses suffixes of the given byte length rather than prefixes.
+
+If B<$bytes> is null, replaces the database's relation files using the saved
+files in the B<$snapname>, which unlike for non-undef values, means the file
+may become shorter if the saved file is shorter than the current file.
+
+=cut
+
+sub revert_to_snapshot
+{
+	my ($self, $snapname) = @_;
+	$self->revert_to_torn_relfile_snapshot($snapname, undef);
+}
+
+sub revert_to_torn_relfile_snapshot
+{
+	my ($self, $snapname, $bytes) = @_;
+
+	croak "no such snapshot" unless exists $self->{snapshot}->{$snapname};
+
+	my $pgdata = $self->data_dir;
+	my $snaprepo = join('/', $self->relfile_snapshot_repository, $snapname);
+	croak "snapname directory missing: $snaprepo" unless (-d $snaprepo);
+
+	if (defined $bytes)
+	{
+		tear_file($pgdata, $snaprepo, $bytes, $_)
+			for (keys %{$self->{snapshot}->{$snapname}});
+	}
+	else
+	{
+		copy_file($pgdata, $snaprepo, 1, $_)
+			for (keys %{$self->{snapshot}->{$snapname}});
+	}
+}
+
+sub copy_file
+{
+	my ($dstdir, $srcdir, $overwrite, $path) = @_;
+
+	croak "No such directory: $dstdir" unless -d $dstdir;
+	croak "No such directory: $srcdir" unless -d $srcdir;
+
+	foreach my $part (split(m{/}, $path))
+	{
+		my $srcpart = "$srcdir/$part";
+		my $dstpart = "$dstdir/$part";
+
+		if (-d $srcpart)
+		{
+			$srcdir = $srcpart;
+			$dstdir = $dstpart;
+			die "$dstdir is in the way" if (-e $dstdir && ! -d $dstdir);
+			unless (-d $dstdir)
+			{
+				mkdir $dstdir
+					or BAIL_OUT("could not create directory \"$dstdir\": $!");
+			}
+		}
+		elsif (-f $srcpart)
+		{
+			die "$dstdir/$part is in the way" if (!$overwrite && -e "$dstdir/$part");
+
+			File::Copy::copy($srcpart, "$dstdir/$part");
+		}
+	}
+}
+
+sub tear_file
+{
+	my ($dstdir, $srcdir, $bytes, $path) = @_;
+
+	croak "No such directory: $dstdir" unless -d $dstdir;
+	croak "No such directory: $srcdir" unless -d $srcdir;
+
+	my $srcfile = "$srcdir/$path";
+	my $dstfile = "$dstdir/$path";
+
+	croak "No such file: $srcfile" unless -f $srcfile;
+	croak "No such file: $dstfile" unless -f $dstfile;
+
+	my ($srcfh, $dstfh);
+	open($srcfh, '<', $srcfile) or die "Cannot read $srcfile: $!";
+	open($dstfh, '+<', $dstfile) or die "Cannot modify $dstfile: $!";
+	binmode($srcfh);
+	binmode($dstfh);
+
+	my $buffer;
+	if ($bytes < 0)
+	{
+		$bytes *= -1;		# Easier to use positive value
+		my $srcsize = (stat($srcfh))[7];
+		my $offset = $srcsize - $bytes;
+		seek($srcfh, $offset, 0);
+		seek($dstfh, $offset, 0);
+		sysread($srcfh, $buffer, $bytes);
+		syswrite($dstfh, $buffer, $bytes);
+	}
+	else
+	{
+		seek($srcfh, 0, 0);
+		seek($dstfh, 0, 0);
+		sysread($srcfh, $buffer, $bytes);
+		syswrite($dstfh, $buffer, $bytes);
+	}
+
+	close($srcfh);
+	close($dstfh);
+}
+
+=pod
+
+=back
+
 =cut
 
 1;
-- 
2.21.1 (Apple Git-122.3)

=

view thread (218+ messages)  latest in thread

Message-ID: <70655DF3-33CE-4527-9A4D-DDEB582B6BA0@enterprisedb.com>
Permalink:  ../70655DF3-33CE-4527-9A4D-DDEB582B6BA0@enterprisedb.com/
Also on:    postgresql.org/message-id/70655DF3-33CE-4527-9A4D-DDEB582B6BA0@enterprisedb.com

copy link · copy postgr.es · copy git am

reply

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Reply to all the recipients using the --to and --cc options:
  reply via email

  To: pgsql-hackers@postgresql.org
  Cc: mark.dilger@enterprisedb.com, robertmhaas@gmail.com, thomas.munro@gmail.com, tgl@sss.pgh.pa.us, pg@bowt.ie, x4mmm@yandex-team.ru, sfrost@snowman.net, michael@paquier.xyz, sulamul@gmail.com, dilipbalaut@gmail.com, andres@anarazel.de
  Subject: Re: new heapcheck contrib module
  In-Reply-To: <70655DF3-33CE-4527-9A4D-DDEB582B6BA0@enterprisedb.com>

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox