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: pg_amcheck contrib application
Date: Wed, 3 Mar 2021 07:22:28 -0800
Message-ID: <BA592F2D-F928-46FF-9516-2B827F067F57@enterprisedb.com> (raw)

New thread, was "Re: new heapcheck contrib module"


> On Mar 2, 2021, at 10:24 AM, Robert Haas <robertmhaas@gmail.com> wrote:
> 
> On Tue, Mar 2, 2021 at 12:10 PM Mark Dilger
> <mark.dilger@enterprisedb.com> wrote:
>> On further reflection, I decided to implement these changes and not worry about the behavioral change.
> 
> Thanks.
> 
>> I skipped this part.  The initcmd argument is only handed to ParallelSlotsGetIdle().  Doing as you suggest would not really be simpler, it would just move that argument to ParallelSlotsSetup().  But I don't feel strongly about it, so I can move this, too, if you like.
>> 
>> I didn't do this either, and for the same reason.  It's just a parameter to ParallelSlotsGetIdle(), so nothing is really gained by moving it to ParallelSlotsSetup().
> 
> OK. I thought it was more natural to pass a bunch of arguments at
> setup time rather than passing a bunch of arguments at get-idle time,
> but I don't feel strongly enough about it to insist, and somebody else
> can always change it later if they decide I had the right idea.

When you originally proposed the idea, I thought that it would work out as a simpler interface to have it your way, but in terms of the interface it came out about the same.  Internally it is still simpler to do it your way, so since you seem to still like your way better, this next version has it that way.

>> Rather than the slots user tweak the slot's ConnParams, ParallelSlotsGetIdle() takes a dbname argument, and uses it as ConnParams->override_dbname.
> 
> OK, but you forgot to update the comments. ParallelSlotsGetIdle()
> still talks about a cparams argument that it no longer has.

Fixed.

> The usual idiom for sizing a memory allocation involving
> FLEXIBLE_ARRAY_MEMBER is something like offsetof(ParallelSlotArray,
> slots) + numslots * sizeof(ParallelSlot). Your version uses sizeof();
> don't.

Fixed.

> Other than that 0001 looks to me to be in pretty good shape now.




And your other review email, also moved to this new thread....

> On Mar 2, 2021, at 12:39 PM, Robert Haas <robertmhaas@gmail.com> wrote:
> 
> On Tue, Mar 2, 2021 at 1:24 PM Robert Haas <robertmhaas@gmail.com> wrote:
>> Other than that 0001 looks to me to be in pretty good shape now.
> 
> Incidentally, we might want to move this to a new thread with a better
> subject line, since the current subject line really doesn't describe
> the uncommitted portion of the work. And create a new CF entry, too.

Moved here.

> Moving onto 0002:
> 
> The index checking options should really be called btree index
> checking options. I think I'd put the table options first, and the
> btree options second. Other kinds of indexes could follow some day. I
> would personally omit the short forms of --heapallindexed and
> --parent-check; I think we'll run out of option names too quickly if
> people add more kinds of checks.

Done.

While doing this, I also renamed some of the variables to more closely match the option name.  I think the code is clearer now.

> 
> Perhaps VerifyBtreeSlotHandler should emit a warning of some kind if
> PQntuples(res) != 0.

The functions bt_index_check and bt_index_parent_check are defined to return VOID, which results in PQntuples(res) == 1.  I added code to verify this condition, but it only serves to alert the user if the amcheck version is behaving in an unexpected way, perhaps due to a amcheck/pg_amcheck version mismatch.

> +               /*
> +                * Test that this function works, but for now we're
> not using the list
> +                * 'relations' that it builds.
> +                */
> +               conn = connectDatabase(&cparams, progname, opts.echo,
> false, true);
> 
> This comment appears to have nothing to do with the code, since
> connectDatabase() does not build a list of 'relations'.

True.  Removed.

> amcheck_sql seems to include paranoia, but do we need that if we're
> using a secure search path? Similarly for other SQL queries, e.g. in
> prepare_table_command.

I removed the OPERATOR(pg_catalog.=) paranoia.

> It might not be strictly necessary for the static functions in
> pg_amcheck.c to use_three completelyDifferent NamingConventions for
> its static functions.

The idea is that the functions that interoperate with parallel slots would follow its NamingConvention; those interoperating with patternToSQLRegex and PQExpBuffers would follow their namingConvention; and those not so interoperating would follow a less obnoxious naming_convention.  To my eye, that color codes the function names in a useful way.  To your eye, it just looks awful.  I've changed it to use just one naming_convention.

> should_processing_continue() is one semicolon over budget.

That's not the first time I've done that recently.  Removed.

> The initializer for opts puts a comma even after the last member
> initializer. Is that going to be portable to all compilers?

I don't know.  I learned to put commas at the end of lists back when I did mostly perl programming, as you get cleaner diffs when you add more stuff to the list later.  Whether I can get away with that in C using initializers I don't know.  I don't have a multiplicity of compilers to check.

I have removed the extra comma.

> +   for (failed = false, cell = opts.include.head; cell; cell = cell->next)
> 
> I think failed has to be false here, because it gets initialized at
> the top of the function. If we need to reinitialize it for some
> reason, I would prefer you do that on the previous line, separate from
> the for loop stuff.

It does have to be false there.  There is no need to reinitialize it.

> +   char       *dbrgx;          /* Database regexp parsed from pattern, or
> +                                * NULL */
> +   char       *nsprgx;         /* Schema regexp parsed from pattern, or NULL */
> +   char       *relrgx;         /* Relation regexp parsed from pattern, or
> +                                * NULL */
> +   bool        tblonly;        /* true if relrgx should only match tables */
> +   bool        idxonly;        /* true if relrgx should only match indexes */
> 
> Maybe: db_regex, nsp_regex, rel_regex, table_only, index_only?
> 
> Just because it seems theoretically possible that someone will see
> nsprgx and not immediately understand what it's supposed to mean, even
> if they know that nsp is a common abbreviation for namespace in
> PostgreSQL code, and even if they also know what a regular expression
> is.

Changed.  Along the way, I noticed that "tbl" and "idx" were being used in C/SQL both to mean ("table_only", "index_only") in some contexts and ("is_table", "is_index') in others, so I replaced all instances of "tbl" and "idx" with the unambiguous labels.

> Your four messages about there being nothing to check seem like they
> could be consolidated down to one: "nothing to check for pattern
> \"%s\"".

I anticipated your review comment, but I'm worried about the case that somebody runs

  pg_amcheck -t "foo" -i "foo"

and one of those matches and the other does not.  The message 'nothing to check for pattern "foo"' will be wrong (because there was something to check for it) and unhelpful (because it doesn't say which failed to match.)

> I would favor changing things so that once argument parsing is
> complete, we switch to reporting all errors that way. So in other
> words here, and everything that follows:
> 
> +               fprintf(stderr, "%s: no databases to check\n", progname);

Same concern about the output for

  pg_amcheck -t "foo" -i "foo" -d "foo"

You might think I'm being silly here, as database names, table names, and index names should in normal usage not be hard for the user to distinguish.  But consider

  pg_amcheck "mydb.myschema.mytable"

If it says, 'nothing to check for pattern "mydb.myschema.mytable"', you don't know if the database doesn't exist or if the table doesn't exist.

> 
> +        * ParallelSlots based event loop follows.
> 
> "Main event loop."

Changed.

> To me it would read slightly better to change each reference to
> "relations list" to "list of relations", but perhaps that is too
> nitpicky.

No harm picking those nits.  Changed.

> I think the two instances of goto finish could be avoided with not
> much work. At most a few things need to happen only if !failed, and
> maybe not even that, if you just said "break;" instead.

Good point.  Changed.

> + * Note: Heap relation corruption is returned by verify_heapam() without the
> + * use of raising errors, but running verify_heapam() on a corrupted table may
> 
> How about "Heap relation corruption() is reported by verify_heapam()
> via the result set, rather than an ERROR, ..."

Changed, though I assumed your parens for corruption() were not intended.



Ok, so now you've moved on to reviewing the regression tests....

> It seems mighty inefficient to have a whole bunch of consecutive calls
> to remove_relation_file() or corrupt_first_page() when every such call
> stops and restarts the database. I would guess these tests will run
> noticeably faster if you don't do that. Either the functions need to
> take a list of arguments, or the stop/start needs to be pulled up and
> done in the caller.

Changed.

> corrupt_first_page() could use a comment explaining what exactly we're
> overwriting, and in particular noting that we don't want to just
> clobber the LSN, but rather something where we can detect a wrong
> value.

Added comments that we're skipping past the PageHeader and overwriting garbage starting in the line pointers.

> There's a long list of calls to command_checks_all() in 003_check.pl
> that don't actually check anything but that the command failed, but
> run it with a bunch of different options. I don't understand the value
> of that, and suggest reducing the number of cases tested. If you want,
> you can have tests elsewhere that focus -- perhaps by using verbose
> mode -- on checking that the right tables are being checked.

This should be better in this next patch series.

> 
> This is not yet a full review of everything in this patch -- I haven't
> sorted through all of the tests yet, or all of the new query
> construction logic -- but to me this looks pretty close to
> committable.

Thanks for the review!


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

Attachments:

v42-0001-Reworking-ParallelSlots-for-mutliple-DB-use.patch (application/octet-stream, 24.6K) download
From 79c357606f06c6422ed2039b6395a59050a4cff1 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Wed, 3 Mar 2021 07:16:55 -0800
Subject: [PATCH v42 1/3] Reworking ParallelSlots for mutliple DB use

The existing implementation of ParallelSlots is used by reindexdb
and vacuumdb to process tables in parallel in only one database at
a time.  The ParallelSlots interface reflects this usage pattern.
The function to set up the slots assumes all slots should be
connected to the same database, and the function for getting the
next idle slot pays no attention to which database the slot may be
connected to.

In anticipation of pg_amcheck using parallel slots to process
multiple databases in parallel, reworking the interface while
trying to remain reasonably simple for reindexdb and vacuumdb to
use:

ParallelSlotsSetup() no longer creates or receives database
connections.  It takes arguments that it stores for use in
subsequent operations when a connection needs to be formed.

Callers who already have a connection and want to reuse it can give
it to the parallel slots using a new function,
ParallelSlotsAdoptConn().  Both reindexdb and vacuumdb use this.

ParallelSlotsGetIdle() is extended to take a dbname argument
indicating the database to which a connection is desired, and to
manage a heterogeneous set of slots potentially connected to varying
databases and some perhaps not yet connected.  The function will
reuse an existing connection or form a new connection as necessary.

The logic for determining whether a slot's connection is suitable
for reuse is based on the database the slot's connection is
connected to, and whether that matches the database desired.  Other
connection parameters (user, host, port, etc.) are assumed not to
change from slot to slot.
---
 src/bin/scripts/reindexdb.c          |  17 +-
 src/bin/scripts/vacuumdb.c           |  46 +--
 src/fe_utils/parallel_slot.c         | 411 +++++++++++++++++++--------
 src/include/fe_utils/parallel_slot.h |  27 +-
 src/tools/pgindent/typedefs.list     |   2 +
 5 files changed, 342 insertions(+), 161 deletions(-)

diff --git a/src/bin/scripts/reindexdb.c b/src/bin/scripts/reindexdb.c
index cf28176243..fc0681538a 100644
--- a/src/bin/scripts/reindexdb.c
+++ b/src/bin/scripts/reindexdb.c
@@ -36,7 +36,7 @@ static SimpleStringList *get_parallel_object_list(PGconn *conn,
 												  ReindexType type,
 												  SimpleStringList *user_list,
 												  bool echo);
-static void reindex_one_database(const ConnParams *cparams, ReindexType type,
+static void reindex_one_database(ConnParams *cparams, ReindexType type,
 								 SimpleStringList *user_list,
 								 const char *progname,
 								 bool echo, bool verbose, bool concurrently,
@@ -330,7 +330,7 @@ main(int argc, char *argv[])
 }
 
 static void
-reindex_one_database(const ConnParams *cparams, ReindexType type,
+reindex_one_database(ConnParams *cparams, ReindexType type,
 					 SimpleStringList *user_list,
 					 const char *progname, bool echo,
 					 bool verbose, bool concurrently, int concurrentCons,
@@ -341,7 +341,7 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 	bool		parallel = concurrentCons > 1;
 	SimpleStringList *process_list = user_list;
 	ReindexType process_type = type;
-	ParallelSlot *slots;
+	ParallelSlotArray *sa;
 	bool		failed = false;
 	int			items_count = 0;
 
@@ -461,7 +461,8 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 
 	Assert(process_list != NULL);
 
-	slots = ParallelSlotsSetup(cparams, progname, echo, conn, concurrentCons);
+	sa = ParallelSlotsSetup(concurrentCons, cparams, progname, echo, NULL);
+	ParallelSlotsAdoptConn(sa, conn);
 
 	cell = process_list->head;
 	do
@@ -475,7 +476,7 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 			goto finish;
 		}
 
-		free_slot = ParallelSlotsGetIdle(slots, concurrentCons);
+		free_slot = ParallelSlotsGetIdle(sa, NULL);
 		if (!free_slot)
 		{
 			failed = true;
@@ -489,7 +490,7 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 		cell = cell->next;
 	} while (cell != NULL);
 
-	if (!ParallelSlotsWaitCompletion(slots, concurrentCons))
+	if (!ParallelSlotsWaitCompletion(sa))
 		failed = true;
 
 finish:
@@ -499,8 +500,8 @@ finish:
 		pg_free(process_list);
 	}
 
-	ParallelSlotsTerminate(slots, concurrentCons);
-	pfree(slots);
+	ParallelSlotsTerminate(sa);
+	pfree(sa);
 
 	if (failed)
 		exit(1);
diff --git a/src/bin/scripts/vacuumdb.c b/src/bin/scripts/vacuumdb.c
index 602fd45c42..7901c41f16 100644
--- a/src/bin/scripts/vacuumdb.c
+++ b/src/bin/scripts/vacuumdb.c
@@ -45,7 +45,7 @@ typedef struct vacuumingOptions
 } vacuumingOptions;
 
 
-static void vacuum_one_database(const ConnParams *cparams,
+static void vacuum_one_database(ConnParams *cparams,
 								vacuumingOptions *vacopts,
 								int stage,
 								SimpleStringList *tables,
@@ -408,7 +408,7 @@ main(int argc, char *argv[])
  * a list of tables from the database.
  */
 static void
-vacuum_one_database(const ConnParams *cparams,
+vacuum_one_database(ConnParams *cparams,
 					vacuumingOptions *vacopts,
 					int stage,
 					SimpleStringList *tables,
@@ -421,13 +421,14 @@ vacuum_one_database(const ConnParams *cparams,
 	PGresult   *res;
 	PGconn	   *conn;
 	SimpleStringListCell *cell;
-	ParallelSlot *slots;
+	ParallelSlotArray *sa;
 	SimpleStringList dbtables = {NULL, NULL};
 	int			i;
 	int			ntups;
 	bool		failed = false;
 	bool		tables_listed = false;
 	bool		has_where = false;
+	const char *initcmd;
 	const char *stage_commands[] = {
 		"SET default_statistics_target=1; SET vacuum_cost_delay=0;",
 		"SET default_statistics_target=10; RESET vacuum_cost_delay;",
@@ -684,26 +685,25 @@ vacuum_one_database(const ConnParams *cparams,
 		concurrentCons = 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.
+	 * All slots need to be prepared to run the appropriate analyze stage, if
+	 * caller requested that mode.  We have to prepare the initial connection
+	 * ourselves before setting up the slots.
 	 */
-	slots = ParallelSlotsSetup(cparams, progname, echo, conn, concurrentCons);
+	if (stage == ANALYZE_NO_STAGE)
+		initcmd = NULL;
+	else
+	{
+		initcmd = stage_commands[stage];
+		executeCommand(conn, initcmd, echo);
+	}
 
 	/*
-	 * Prepare all the connections to run the appropriate analyze stage, if
-	 * caller requested that mode.
+	 * 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.
 	 */
-	if (stage != ANALYZE_NO_STAGE)
-	{
-		int			j;
-
-		/* We already emitted the message above */
-
-		for (j = 0; j < concurrentCons; j++)
-			executeCommand((slots + j)->connection,
-						   stage_commands[stage], echo);
-	}
+	sa = ParallelSlotsSetup(concurrentCons, cparams, progname, echo, initcmd);
+	ParallelSlotsAdoptConn(sa, conn);
 
 	initPQExpBuffer(&sql);
 
@@ -719,7 +719,7 @@ vacuum_one_database(const ConnParams *cparams,
 			goto finish;
 		}
 
-		free_slot = ParallelSlotsGetIdle(slots, concurrentCons);
+		free_slot = ParallelSlotsGetIdle(sa, NULL);
 		if (!free_slot)
 		{
 			failed = true;
@@ -740,12 +740,12 @@ vacuum_one_database(const ConnParams *cparams,
 		cell = cell->next;
 	} while (cell != NULL);
 
-	if (!ParallelSlotsWaitCompletion(slots, concurrentCons))
+	if (!ParallelSlotsWaitCompletion(sa))
 		failed = true;
 
 finish:
-	ParallelSlotsTerminate(slots, concurrentCons);
-	pg_free(slots);
+	ParallelSlotsTerminate(sa);
+	pg_free(sa);
 
 	termPQExpBuffer(&sql);
 
diff --git a/src/fe_utils/parallel_slot.c b/src/fe_utils/parallel_slot.c
index b625deb254..a09e5460e5 100644
--- a/src/fe_utils/parallel_slot.c
+++ b/src/fe_utils/parallel_slot.c
@@ -25,31 +25,23 @@
 #include "common/logging.h"
 #include "fe_utils/cancel.h"
 #include "fe_utils/parallel_slot.h"
+#include "fe_utils/query_utils.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(ParallelSlot *slot, PGresult *result);
 
-static void
-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 problem,
- * false otherwise. It's up to the handler to decide what cosntitutes a
+ * false otherwise. It's up to the handler to decide what constitutes a
  * problem.
  */
 static bool
 processQueryResult(ParallelSlot *slot, PGresult *result)
 {
 	Assert(slot->handler != NULL);
+	Assert(slot->connection != NULL);
 
 	/* On failure, the handler should return NULL after freeing the result */
 	if (!slot->handler(result, slot->connection, slot->handler_context))
@@ -71,6 +63,9 @@ consumeQueryResult(ParallelSlot *slot)
 	bool		ok = true;
 	PGresult   *result;
 
+	Assert(slot != NULL);
+	Assert(slot->connection != NULL);
+
 	SetCancelConn(slot->connection);
 	while ((result = PQgetResult(slot->connection)) != NULL)
 	{
@@ -137,151 +132,316 @@ select_loop(int maxFd, fd_set *workerset)
 }
 
 /*
- * ParallelSlotsGetIdle
- *		Return a connection slot that is ready to execute a command.
- *
- * This returns the first slot we find that is marked isFree, if one is;
- * otherwise, we loop on select() until one socket becomes available.  When
- * this happens, we read the whole set and mark as free all sockets that
- * become available.  If an error occurs, NULL is returned.
+ * Return the offset of a suitable idle slot, or -1 if none are available.  If
+ * the given dbname is not null, only idle slots connected to the given
+ * database are considered suitable, otherwise all idle connected slots are
+ * considered suitable.
  */
-ParallelSlot *
-ParallelSlotsGetIdle(ParallelSlot *slots, int numslots)
+static int
+find_matching_idle_slot(const ParallelSlotArray *sa, const char *dbname)
 {
 	int			i;
-	int			firstFree = -1;
 
-	/*
-	 * Look for any connection currently free.  If there is one, mark it as
-	 * taken and let the caller know the slot to use.
-	 */
-	for (i = 0; i < numslots; i++)
+	for (i = 0; i < sa->numslots; i++)
 	{
-		if (slots[i].isFree)
-		{
-			slots[i].isFree = false;
-			return slots + i;
-		}
+		if (sa->slots[i].inUse)
+			continue;
+
+		if (sa->slots[i].connection == NULL)
+			continue;
+
+		if (dbname == NULL ||
+			strcmp(PQdb(sa->slots[i].connection), dbname) == 0)
+			return i;
+	}
+	return -1;
+}
+
+/*
+ * Return the offset of the first slot without a database connection, or -1 if
+ * all slots are connected.
+ */
+static int
+find_unconnected_slot(const ParallelSlotArray *sa)
+{
+	int			i;
+
+	for (i = 0; i < sa->numslots; i++)
+	{
+		if (sa->slots[i].inUse)
+			continue;
+
+		if (sa->slots[i].connection == NULL)
+			return i;
+	}
+
+	return -1;
+}
+
+/*
+ * Return the offset of the first idle slot, or -1 if all slots are busy.
+ */
+static int
+find_any_idle_slot(const ParallelSlotArray *sa)
+{
+	int			i;
+
+	for (i = 0; i < sa->numslots; i++)
+		if (!sa->slots[i].inUse)
+			return i;
+
+	return -1;
+}
+
+/*
+ * Wait for any slot's connection to have query results, consume the results,
+ * and update the slot's status as appropriate.  Returns true on success,
+ * false on cancellation, on error, or if no slots are connected.
+ */
+static bool
+wait_on_slots(ParallelSlotArray *sa)
+{
+	int			i;
+	fd_set		slotset;
+	int			maxFd = 0;
+	PGconn	   *cancelconn = NULL;
+
+	/* We must reconstruct the fd_set for each call to select_loop */
+	FD_ZERO(&slotset);
+
+	for (i = 0; i < sa->numslots; i++)
+	{
+		int			sock;
+
+		/* We shouldn't get here if we still have slots without connections */
+		Assert(sa->slots[i].connection != NULL);
+
+		sock = PQsocket(sa->slots[i].connection);
+
+		/*
+		 * We don't really expect any connections to lose their sockets after
+		 * startup, but just in case, cope by ignoring them.
+		 */
+		if (sock < 0)
+			continue;
+
+		/* Keep track of the first valid connection we see. */
+		if (cancelconn == NULL)
+			cancelconn = sa->slots[i].connection;
+
+		FD_SET(sock, &slotset);
+		if (sock > maxFd)
+			maxFd = sock;
 	}
 
 	/*
-	 * No free slot found, so wait until one of the connections has finished
-	 * its task and return the available slot.
+	 * If we get this far with no valid connections, processing cannot
+	 * continue.
 	 */
-	while (firstFree < 0)
+	if (cancelconn == NULL)
+		return false;
+
+	SetCancelConn(sa->slots->connection);
+	i = select_loop(maxFd, &slotset);
+	ResetCancelConn();
+
+	/* failure? */
+	if (i < 0)
+		return false;
+
+	for (i = 0; i < sa->numslots; i++)
 	{
-		fd_set		slotset;
-		int			maxFd = 0;
+		int			sock;
 
-		/* We must reconstruct the fd_set for each call to select_loop */
-		FD_ZERO(&slotset);
+		sock = PQsocket(sa->slots[i].connection);
 
-		for (i = 0; i < numslots; i++)
+		if (sock >= 0 && FD_ISSET(sock, &slotset))
 		{
-			int			sock = PQsocket(slots[i].connection);
-
-			/*
-			 * We don't really expect any connections to lose their sockets
-			 * after startup, but just in case, cope by ignoring them.
-			 */
-			if (sock < 0)
-				continue;
-
-			FD_SET(sock, &slotset);
-			if (sock > maxFd)
-				maxFd = sock;
+			/* select() says input is available, so consume it */
+			PQconsumeInput(sa->slots[i].connection);
 		}
 
-		SetCancelConn(slots->connection);
-		i = select_loop(maxFd, &slotset);
-		ResetCancelConn();
-
-		/* failure? */
-		if (i < 0)
-			return NULL;
-
-		for (i = 0; i < numslots; i++)
+		/* Collect result(s) as long as any are available */
+		while (!PQisBusy(sa->slots[i].connection))
 		{
-			int			sock = PQsocket(slots[i].connection);
+			PGresult   *result = PQgetResult(sa->slots[i].connection);
 
-			if (sock >= 0 && FD_ISSET(sock, &slotset))
+			if (result != NULL)
 			{
-				/* select() says input is available, so consume it */
-				PQconsumeInput(slots[i].connection);
+				/* Handle and discard the command result */
+				if (!processQueryResult(&sa->slots[i], result))
+					return false;
 			}
-
-			/* Collect result(s) as long as any are available */
-			while (!PQisBusy(slots[i].connection))
+			else
 			{
-				PGresult   *result = PQgetResult(slots[i].connection);
-
-				if (result != NULL)
-				{
-					/* Handle and discard the command result */
-					if (!processQueryResult(slots + i, result))
-						return NULL;
-				}
-				else
-				{
-					/* This connection has become idle */
-					slots[i].isFree = true;
-					ParallelSlotClearHandler(slots + i);
-					if (firstFree < 0)
-						firstFree = i;
-					break;
-				}
+				/* This connection has become idle */
+				sa->slots[i].inUse = false;
+				ParallelSlotClearHandler(&sa->slots[i]);
+				break;
 			}
 		}
 	}
+	return true;
+}
 
-	slots[firstFree].isFree = false;
-	return slots + firstFree;
+/*
+ * Open a new database connection using the stored connection parameters and
+ * optionally a given dbname if not null, execute the stored initial command if
+ * any, and associate the new connection with the given slot.
+ */
+static void
+connect_slot(ParallelSlotArray *sa, int slotno, const char *dbname)
+{
+	const char *old_override;
+	ParallelSlot *slot = &sa->slots[slotno];
+
+	old_override = sa->cparams->override_dbname;
+	if (dbname)
+		sa->cparams->override_dbname = dbname;
+	slot->connection = connectDatabase(sa->cparams, sa->progname, sa->echo, false, true);
+	sa->cparams->override_dbname = old_override;
+
+	if (PQsocket(slot->connection) >= FD_SETSIZE)
+	{
+		pg_log_fatal("too many jobs for this platform");
+		exit(1);
+	}
+
+	/* Setup the connection using the supplied command, if any. */
+	if (sa->initcmd)
+		executeCommand(slot->connection, sa->initcmd, sa->echo);
 }
 
 /*
- * ParallelSlotsSetup
- *		Prepare a set of parallel slots to use on a given database.
+ * ParallelSlotsGetIdle
+ *		Return a connection slot that is ready to execute a command.
+ *
+ * The slot returned is chosen as follows:
+ *
+ * If any idle slot already has an open connection, and if either dbname is
+ * null or the existing connection is to the given database, that slot will be
+ * returned allowing the connection to be reused.
+ *
+ * Otherwise, if any idle slot is not yet connected to any database, the slot
+ * will be returned with it's connection opened using the stored cparams and
+ * optionally the given dbname if not null.
+ *
+ * Otherwise, if any idle slot exists, an idle slot will be chosen and returned
+ * after having it's connection disconnected and reconnected using the stored
+ * cparams and optionally the given dbname if not null.
  *
- * This creates and initializes a set of connections to the database
- * using the information given by the caller, marking all parallel slots
- * as free and ready to use.  "conn" is an initial connection set up
- * by the caller and is associated with the first slot in the parallel
- * set.
+ * Otherwise, if any slots have connections that are busy, we loop on select()
+ * until one socket becomes available.  When this happens, we read the whole
+ * set and mark as free all sockets that become available.  We then select a
+ * slot using the same rules as above.
+ *
+ * Otherwise, we cannot return a slot, which is an error, and NULL is returned.
+ *
+ * For any connection created, if the stored initcmd is not null, it will be
+ * executed as a command on the newly formed connection before the slot is
+ * returned.
+ *
+ * If an error occurs, NULL is returned.
  */
 ParallelSlot *
-ParallelSlotsSetup(const ConnParams *cparams,
-				   const char *progname, bool echo,
-				   PGconn *conn, int numslots)
+ParallelSlotsGetIdle(ParallelSlotArray *sa, const char *dbname)
 {
-	ParallelSlot *slots;
-	int			i;
+	int			offset;
 
-	Assert(conn != NULL);
+	Assert(sa);
+	Assert(sa->numslots > 0);
 
-	slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * numslots);
-	init_slot(slots, conn);
-	if (numslots > 1)
+	while (1)
 	{
-		for (i = 1; i < numslots; i++)
+		/* First choice: a slot already connected to the desired database. */
+		offset = find_matching_idle_slot(sa, dbname);
+		if (offset >= 0)
 		{
-			conn = connectDatabase(cparams, progname, echo, false, true);
-
-			/*
-			 * Fail and exit immediately if trying to use a socket in an
-			 * unsupported range.  POSIX requires open(2) to use the lowest
-			 * unused file descriptor and the hint given relies on that.
-			 */
-			if (PQsocket(conn) >= FD_SETSIZE)
-			{
-				pg_log_fatal("too many jobs for this platform -- try %d", i);
-				exit(1);
-			}
+			sa->slots[offset].inUse = true;
+			return &sa->slots[offset];
+		}
+
+		/* Second choice: a slot not connected to any database. */
+		offset = find_unconnected_slot(sa);
+		if (offset >= 0)
+		{
+			connect_slot(sa, offset, dbname);
+			sa->slots[offset].inUse = true;
+			return &sa->slots[offset];
+		}
 
-			init_slot(slots + i, conn);
+		/* Third choice: a slot connected to the wrong database. */
+		offset = find_any_idle_slot(sa);
+		if (offset >= 0)
+		{
+			disconnectDatabase(sa->slots[offset].connection);
+			sa->slots[offset].connection = NULL;
+			connect_slot(sa, offset, dbname);
+			sa->slots[offset].inUse = true;
+			return &sa->slots[offset];
 		}
+
+		/*
+		 * Fourth choice: block until one or more slots become available. If
+		 * any slot's hit a fatal error, we'll find out about that here and
+		 * return NULL.
+		 */
+		if (!wait_on_slots(sa))
+			return NULL;
 	}
+}
+
+/*
+ * ParallelSlotsSetup
+ *		Prepare a set of parallel slots but do not connect to any database.
+ *
+ * This creates and initializes a set of slots, marking all parallel slots as
+ * free and ready to use.  Establishing connections is delayed until requesting
+ * a free slot.  The cparams, progname, echo, and initcmd are stored for later
+ * use and must remain valid for the lifetime of the returned array.
+ */
+ParallelSlotArray *
+ParallelSlotsSetup(int numslots, ConnParams *cparams, const char *progname,
+				   bool echo, const char *initcmd)
+{
+	ParallelSlotArray *sa;
 
-	return slots;
+	Assert(numslots > 0);
+	Assert(cparams != NULL);
+	Assert(progname != NULL);
+
+	sa = (ParallelSlotArray *) palloc0(offsetof(ParallelSlotArray, slots) +
+									   numslots * sizeof(ParallelSlot));
+
+	sa->numslots = numslots;
+	sa->cparams = cparams;
+	sa->progname = progname;
+	sa->echo = echo;
+	sa->initcmd = initcmd;
+
+	return sa;
+}
+
+/*
+ * ParallelSlotsAdoptConn
+ *		Assign an open connection to the slots array for reuse.
+ *
+ * This turns over ownership of an open connection to a slots array.  The
+ * caller should not further use or close the connection.  All the connection's
+ * parameters (user, host, port, etc.) except possibly dbname should match
+ * those of the slots array's cparams, as given in ParallelSlotsSetup.  If
+ * these parameters differ, subsequent behavior is undefined.
+ */
+void
+ParallelSlotsAdoptConn(ParallelSlotArray *sa, PGconn *conn)
+{
+	int		offset;
+
+	offset = find_unconnected_slot(sa);
+	if (offset >= 0)
+		sa->slots[offset].connection = conn;
+	else
+		disconnectDatabase(conn);
 }
 
 /*
@@ -292,13 +452,13 @@ ParallelSlotsSetup(const ConnParams *cparams,
  * terminate all connections.
  */
 void
-ParallelSlotsTerminate(ParallelSlot *slots, int numslots)
+ParallelSlotsTerminate(ParallelSlotArray *sa)
 {
 	int			i;
 
-	for (i = 0; i < numslots; i++)
+	for (i = 0; i < sa->numslots; i++)
 	{
-		PGconn	   *conn = slots[i].connection;
+		PGconn	   *conn = sa->slots[i].connection;
 
 		if (conn == NULL)
 			continue;
@@ -314,13 +474,15 @@ ParallelSlotsTerminate(ParallelSlot *slots, int numslots)
  * error has been found on the way.
  */
 bool
-ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots)
+ParallelSlotsWaitCompletion(ParallelSlotArray *sa)
 {
 	int			i;
 
-	for (i = 0; i < numslots; i++)
+	for (i = 0; i < sa->numslots; i++)
 	{
-		if (!consumeQueryResult(slots + i))
+		if (sa->slots[i].connection == NULL)
+			continue;
+		if (!consumeQueryResult(&sa->slots[i]))
 			return false;
 	}
 
@@ -350,6 +512,9 @@ ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots)
 bool
 TableCommandResultHandler(PGresult *res, PGconn *conn, void *context)
 {
+	Assert(res != NULL);
+	Assert(conn != NULL);
+
 	/*
 	 * If it's an error, report it.  Errors about a missing table are harmless
 	 * so we continue processing; but die for other errors.
diff --git a/src/include/fe_utils/parallel_slot.h b/src/include/fe_utils/parallel_slot.h
index 8902f8d4f4..b7e2b0a29b 100644
--- a/src/include/fe_utils/parallel_slot.h
+++ b/src/include/fe_utils/parallel_slot.h
@@ -21,7 +21,7 @@ typedef bool (*ParallelSlotResultHandler) (PGresult *res, PGconn *conn,
 typedef struct ParallelSlot
 {
 	PGconn	   *connection;		/* One connection */
-	bool		isFree;			/* Is it known to be idle? */
+	bool		inUse;			/* Is the slot being used? */
 
 	/*
 	 * Prior to issuing a command or query on 'connection', a handler callback
@@ -33,6 +33,16 @@ typedef struct ParallelSlot
 	void	   *handler_context;
 } ParallelSlot;
 
+typedef struct ParallelSlotArray
+{
+	int			numslots;
+	ConnParams *cparams;
+	const char *progname;
+	bool		echo;
+	const char *initcmd;
+	ParallelSlot slots[FLEXIBLE_ARRAY_MEMBER];
+} ParallelSlotArray;
+
 static inline void
 ParallelSlotSetHandler(ParallelSlot *slot, ParallelSlotResultHandler handler,
 					   void *context)
@@ -48,15 +58,18 @@ ParallelSlotClearHandler(ParallelSlot *slot)
 	slot->handler_context = NULL;
 }
 
-extern ParallelSlot *ParallelSlotsGetIdle(ParallelSlot *slots, int numslots);
+extern ParallelSlot *ParallelSlotsGetIdle(ParallelSlotArray *slots,
+										  const char *dbname);
+
+extern ParallelSlotArray *ParallelSlotsSetup(int numslots, ConnParams *cparams,
+											 const char *progname, bool echo,
+											 const char *initcmd);
 
-extern ParallelSlot *ParallelSlotsSetup(const ConnParams *cparams,
-										const char *progname, bool echo,
-										PGconn *conn, int numslots);
+extern void ParallelSlotsAdoptConn(ParallelSlotArray *sa, PGconn *conn);
 
-extern void ParallelSlotsTerminate(ParallelSlot *slots, int numslots);
+extern void ParallelSlotsTerminate(ParallelSlotArray *sa);
 
-extern bool ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots);
+extern bool ParallelSlotsWaitCompletion(ParallelSlotArray *sa);
 
 extern bool TableCommandResultHandler(PGresult *res, PGconn *conn,
 									  void *context);
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8bd95aefa1..b1dec43f9d 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
@@ -1729,6 +1730,7 @@ ParallelHashJoinState
 ParallelIndexScanDesc
 ParallelReadyList
 ParallelSlot
+ParallelSlotArray
 ParallelState
 ParallelTableScanDesc
 ParallelTableScanDescData
-- 
2.21.1 (Apple Git-122.3)

=

v42-0002-Adding-contrib-module-pg_amcheck.patch (application/octet-stream, 131.7K) download
From b59f43cc27ae69c73d8c1209346166b9c13ef68e Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Tue, 2 Mar 2021 08:34:40 -0800
Subject: [PATCH v42 2/3] 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            | 1939 ++++++++++++++++++++
 contrib/pg_amcheck/t/001_basic.pl          |    9 +
 contrib/pg_amcheck/t/002_nonesuch.pl       |  213 +++
 contrib/pg_amcheck/t/003_check.pl          |  497 +++++
 contrib/pg_amcheck/t/004_verify_heapam.pl  |  487 +++++
 contrib/pg_amcheck/t/005_opclass_damage.pl |   54 +
 doc/src/sgml/contrib.sgml                  |    1 +
 doc/src/sgml/filelist.sgml                 |    1 +
 doc/src/sgml/pgamcheck.sgml                |  668 +++++++
 src/tools/msvc/Install.pm                  |    2 +-
 src/tools/msvc/Mkvcbuild.pm                |    6 +-
 src/tools/pgindent/typedefs.list           |    3 +
 15 files changed, 3909 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/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..63982fd66b
--- /dev/null
+++ b/contrib/pg_amcheck/pg_amcheck.c
@@ -0,0 +1,1939 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_amcheck.c
+ *		Detects corruption within database relations.
+ *
+ * Copyright (c) 2017-2021, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  contrib/pg_amcheck/pg_amcheck.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres_fe.h"
+
+#include <time.h>
+
+#include "catalog/pg_am_d.h"
+#include "catalog/pg_namespace_d.h"
+#include "common/logging.h"
+#include "common/username.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 "getopt_long.h"		/* pgrminclude ignore */
+#include "pgtime.h"
+#include "storage/block.h"
+
+/* pg_amcheck command line options controlled by user flags */
+typedef struct AmcheckOptions
+{
+	bool		alldb;
+	bool		echo;
+	bool		quiet;
+	bool		verbose;
+	bool		strict_names;
+	bool		show_progress;
+	int			jobs;
+
+	/*
+	 * Relations to check or not to check (both heap and btree together), as
+	 * lists of PatternInfo structs.
+	 */
+	SimplePtrList include;
+	SimplePtrList exclude;
+
+	/*
+	 * As an optimization, if any pattern in the exclude list applies to heap
+	 * tables, or similarly if any such pattern applies to btree indexes, then
+	 * these will be true, otherwise false.  These should always agree with
+	 * what you'd conclude by grep'ing through the exclude list.
+	 */
+	bool		excludetbl;
+	bool		excludeidx;
+
+	/*
+	 * If any inclusion pattern exists, then we should only be checking
+	 * matching relations rather than all relations, so this is true iff
+	 * include is empty.
+	 */
+	bool		allrel;
+
+	/* heap table checking options */
+	bool		no_toast_expansion;
+	bool		reconcile_toast;
+	bool		on_error_stop;
+	long		startblock;
+	long		endblock;
+	const char *skip;
+
+	/* btree index checking options */
+	bool		parent_check;
+	bool		rootdescend;
+	bool		heapallindexed;
+
+	/* heap and btree hybrid option */
+	bool		no_index_expansion;
+} AmcheckOptions;
+
+static AmcheckOptions opts = {
+	.alldb = false,
+	.echo = false,
+	.quiet = false,
+	.verbose = false,
+	.strict_names = true,
+	.show_progress = false,
+	.jobs = 1,
+	.include = {NULL, NULL},
+	.exclude = {NULL, NULL},
+	.excludetbl = false,
+	.excludeidx = false,
+	.allrel = true,
+	.no_toast_expansion = false,
+	.reconcile_toast = true,
+	.on_error_stop = false,
+	.startblock = -1,
+	.endblock = -1,
+	.skip = "none",
+	.parent_check = false,
+	.rootdescend = false,
+	.heapallindexed = false,
+	.no_index_expansion = false
+};
+
+static const char *progname = NULL;
+
+typedef struct PatternInfo
+{
+	int			pattern_id;		/* Unique ID of this pattern */
+	const char *pattern;		/* Unaltered pattern from the command line */
+	char	   *db_regex;		/* Database regexp parsed from pattern, or
+								 * NULL */
+	char	   *nsp_regex;		/* Schema regexp parsed from pattern, or NULL */
+	char	   *rel_regex;		/* Relation regexp parsed from pattern, or
+								 * NULL */
+	bool		table_only;		/* true if rel_regex should only match tables */
+	bool		index_only;		/* true if rel_regex should only match indexes */
+	bool		matched;		/* true if the pattern matched in any database */
+}			PatternInfo;
+
+/* Unique pattern id counter */
+static int	next_id = 1;
+
+/* Whether all relations have so far passed their corruption checks */
+static bool all_checks_pass = true;
+
+/* Time last progress report was displayed */
+static pg_time_t last_progress_report = 0;
+
+typedef struct DatabaseInfo
+{
+	char	   *datname;
+	char	   *amcheck_schema; /* escaped, quoted literal */
+} DatabaseInfo;
+
+typedef struct RelationInfo
+{
+	const DatabaseInfo *datinfo;	/* shared by other relinfos */
+	Oid			reloid;
+	bool		is_table;		/* true if heap, false if btree */
+} RelationInfo;
+
+/*
+ * 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 = n.oid"
+"\nWHERE x.extname = 'amcheck'";
+
+static void prepare_table_command(PQExpBuffer sql, Oid reloid,
+								  const char *nspname);
+static void prepare_btree_command(PQExpBuffer sql, Oid reloid,
+								  const char *nspname);
+static void run_command(ParallelSlot *slot, const char *sql,
+						ConnParams *cparams);
+static bool verify_heapam_slot_handler(PGresult *res, PGconn *conn,
+									   void *context);
+static bool verify_btree_slot_handler(PGresult *res, PGconn *conn, void *context);
+static void help(const char *progname);
+static void progress_report(uint64 relations_total, uint64 relations_checked,
+							const char *datname, bool force, bool finished);
+
+static void append_database_pattern(SimplePtrList *list, const char *pattern,
+									int encoding);
+static void append_schema_pattern(SimplePtrList *list, const char *pattern,
+								  int encoding);
+static void append_relation_pattern(SimplePtrList *list, const char *pattern,
+									int encoding);
+static void append_table_pattern(SimplePtrList *list, const char *pattern,
+								 int encoding);
+static void append_index_pattern(SimplePtrList *list, const char *pattern,
+								 int encoding);
+static void compile_database_list(PGconn *conn, SimplePtrList *databases);
+static void compile_relation_list_one_db(PGconn *conn, SimplePtrList *relations,
+										 const DatabaseInfo *datinfo);
+
+int
+main(int argc, char *argv[])
+{
+	PGconn	   *conn;
+	SimplePtrListCell *cell;
+	SimplePtrList databases = {NULL, NULL};
+	SimplePtrList relations = {NULL, NULL};
+	bool		failed = false;
+	const char *latest_datname;
+	int			parallel_workers;
+	ParallelSlotArray *sa;
+	PQExpBufferData sql;
+	long long int reltotal;
+	long long int relprogress;
+
+	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'},
+		{"index", required_argument, NULL, 'i'},
+		{"exclude-index", required_argument, NULL, 'I'},
+		{"jobs", required_argument, NULL, 'j'},
+		{"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'},
+		{"no-index-expansion", no_argument, NULL, 2},
+		{"no-toast-expansion", 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-strict-names", no_argument, NULL, 10},
+		{"progress", no_argument, NULL, 11},
+		{"heapallindexed", no_argument, NULL, 12},
+		{"parent-check", no_argument, NULL, 13},
+
+		{NULL, 0, NULL, 0}
+	};
+
+	int			optindex;
+	int			c;
+
+	/*
+	 * If a maintenance database is specified, that will be used for the
+	 * initial connection.  Failing that, the first plain argument (without a
+	 * flag) will be used.  If neither of those are given, the first database
+	 * specified with -d.
+	 */
+	const char *primary_db = NULL;
+	const char *secondary_db = NULL;
+	const char *tertiary_db = NULL;
+
+	const char *host = NULL;
+	const char *port = NULL;
+	const char *username = NULL;
+	enum trivalue prompt_password = TRI_DEFAULT;
+	int			encoding = pg_get_encoding_from_locale(NULL, false);
+	ConnParams	cparams;
+
+	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':
+				opts.alldb = true;
+				break;
+			case 'd':
+				if (tertiary_db == NULL)
+					tertiary_db = optarg;
+				append_database_pattern(&opts.include, optarg, encoding);
+				break;
+			case 'D':
+				append_database_pattern(&opts.exclude, optarg, encoding);
+				break;
+			case 'e':
+				opts.echo = true;
+				break;
+			case 'h':
+				host = pg_strdup(optarg);
+				break;
+			case 'i':
+				opts.allrel = false;
+				append_index_pattern(&opts.include, optarg, encoding);
+				break;
+			case 'I':
+				opts.excludeidx = true;
+				append_index_pattern(&opts.exclude, optarg, encoding);
+				break;
+			case 'j':
+				opts.jobs = atoi(optarg);
+				if (opts.jobs < 1)
+				{
+					pg_log_error("number of parallel jobs must be at least 1");
+					exit(1);
+				}
+				break;
+			case 'p':
+				port = pg_strdup(optarg);
+				break;
+			case 'q':
+				opts.quiet = true;
+				break;
+			case 'r':
+				opts.allrel = false;
+				append_relation_pattern(&opts.include, optarg, encoding);
+				break;
+			case 'R':
+				opts.excludeidx = true;
+				opts.excludetbl = true;
+				append_relation_pattern(&opts.exclude, optarg, encoding);
+				break;
+			case 's':
+				opts.allrel = false;
+				append_schema_pattern(&opts.include, optarg, encoding);
+				break;
+			case 'S':
+				append_schema_pattern(&opts.exclude, optarg, encoding);
+				break;
+			case 't':
+				opts.allrel = false;
+				append_table_pattern(&opts.include, optarg, encoding);
+				break;
+			case 'T':
+				opts.excludetbl = true;
+				append_table_pattern(&opts.exclude, optarg, encoding);
+				break;
+			case 'U':
+				username = pg_strdup(optarg);
+				break;
+			case 'w':
+				prompt_password = TRI_NO;
+				break;
+			case 'W':
+				prompt_password = TRI_YES;
+				break;
+			case 'v':
+				opts.verbose = true;
+				pg_logging_increase_verbosity();
+				break;
+			case 1:
+				primary_db = pg_strdup(optarg);
+				break;
+			case 2:
+				opts.no_index_expansion = true;
+				break;
+			case 3:
+				opts.no_toast_expansion = true;
+				break;
+			case 4:
+				opts.reconcile_toast = false;
+				break;
+			case 5:
+				opts.on_error_stop = true;
+				break;
+			case 6:
+				if (pg_strcasecmp(optarg, "all-visible") == 0)
+					opts.skip = "all visible";
+				else if (pg_strcasecmp(optarg, "all-frozen") == 0)
+					opts.skip = "all frozen";
+				else
+				{
+					fprintf(stderr, "invalid skip options");
+					exit(1);
+				}
+				break;
+			case 7:
+				opts.startblock = strtol(optarg, &endptr, 10);
+				if (*endptr != '\0')
+				{
+					fprintf(stderr,
+							"relation starting block argument contains garbage characters");
+					exit(1);
+				}
+				if (opts.startblock > (long) MaxBlockNumber)
+				{
+					fprintf(stderr,
+							"relation starting block argument out of bounds");
+					exit(1);
+				}
+				break;
+			case 8:
+				opts.endblock = strtol(optarg, &endptr, 10);
+				if (*endptr != '\0')
+				{
+					fprintf(stderr,
+							"relation ending block argument contains garbage characters");
+					exit(1);
+				}
+				if (opts.startblock > (long) MaxBlockNumber)
+				{
+					fprintf(stderr,
+							"relation ending block argument out of bounds");
+					exit(1);
+				}
+				break;
+			case 9:
+				opts.rootdescend = true;
+				opts.parent_check = true;
+				break;
+			case 10:
+				opts.strict_names = false;
+				break;
+			case 11:
+				opts.show_progress = true;
+				break;
+			case 12:
+				opts.heapallindexed = true;
+				break;
+			case 13:
+				opts.parent_check = true;
+				break;
+			default:
+				fprintf(stderr,
+						"Try \"%s --help\" for more information.\n",
+						progname);
+				exit(1);
+		}
+	}
+
+	if (opts.endblock >= 0 && opts.endblock < opts.startblock)
+	{
+		pg_log_error("relation ending block argument precedes starting block argument");
+		exit(1);
+	}
+
+	/* non-option arguments specify database names */
+	while (optind < argc)
+	{
+		if (secondary_db == NULL)
+			secondary_db = argv[optind];
+		append_database_pattern(&opts.include, argv[optind], encoding);
+		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 (primary_db)
+		cparams.dbname = primary_db;
+	else if (secondary_db != NULL)
+		cparams.dbname = secondary_db;
+	else if (tertiary_db != NULL)
+		cparams.dbname = tertiary_db;
+	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);
+
+		/*
+		 * Users expect the database name inferred from the environment to get
+		 * checked, not just get used for the initial connection.
+		 */
+		append_database_pattern(&opts.include, default_db, encoding);
+
+		cparams.dbname = default_db;
+	}
+
+	conn = connectMaintenanceDatabase(&cparams, progname, opts.echo);
+	compile_database_list(conn, &databases);
+	disconnectDatabase(conn);
+
+	if (databases.head == NULL)
+	{
+		fprintf(stderr, "%s: no databases to check\n", progname);
+		exit(0);
+	}
+
+	/*
+	 * Compile a list of all relations spanning all databases to be checked.
+	 */
+	for (cell = databases.head; cell; cell = cell->next)
+	{
+		PGresult   *result;
+		int			ntups;
+		const char *amcheck_schema = NULL;
+		DatabaseInfo *dat = (DatabaseInfo *) cell->ptr;
+
+		cparams.override_dbname = dat->datname;
+		conn = connectDatabase(&cparams, progname, opts.echo, false, true);
+
+		/*
+		 * 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, opts.echo);
+		if (PQresultStatus(result) != PGRES_TUPLES_OK)
+		{
+			/* Querying the catalog failed. */
+			pg_log_error("database \"%s\": %s\n",
+						 PQdb(conn), PQerrorMessage(conn));
+			pg_log_error("query was: %s", amcheck_sql);
+			PQclear(result);
+			disconnectDatabase(conn);
+			exit(1);
+		}
+		ntups = PQntuples(result);
+		if (ntups == 0)
+		{
+			/* Querying the catalog succeeded, but amcheck is missing. */
+			fprintf(stderr,
+					"%s: skipping database \"%s\": amcheck is not installed\n",
+					progname, PQdb(conn));
+			disconnectDatabase(conn);
+			continue;
+		}
+		amcheck_schema = PQgetvalue(result, 0, 0);
+		if (opts.verbose)
+			fprintf(stderr,
+					"%s: in database \"%s\": using amcheck version \"%s\" in schema \"%s\"\n",
+					progname, PQdb(conn), PQgetvalue(result, 0, 1),
+					amcheck_schema);
+		dat->amcheck_schema = PQescapeIdentifier(conn, amcheck_schema,
+												 strlen(amcheck_schema));
+		PQclear(result);
+
+		compile_relation_list_one_db(conn, &relations, dat);
+		disconnectDatabase(conn);
+	}
+
+	/*
+	 * Check that all inclusion patterns matched at least one schema or
+	 * relation that we can check.
+	 */
+	for (cell = opts.include.head; cell; cell = cell->next)
+	{
+		PatternInfo *pat = (PatternInfo *) cell->ptr;
+
+		if (!pat->matched && (pat->nsp_regex != NULL || pat->rel_regex != NULL))
+		{
+			failed = opts.strict_names;
+
+			if (!opts.quiet)
+			{
+				if (pat->table_only)
+					fprintf(stderr, "%s: no tables to check for \"%s\"\n",
+							progname, pat->pattern);
+				else if (pat->index_only)
+					fprintf(stderr, "%s: no btree indexes to check for \"%s\"\n",
+							progname, pat->pattern);
+				else if (pat->rel_regex == NULL)
+					fprintf(stderr, "%s: no relations to check in schemas for \"%s\"\n",
+							progname, pat->pattern);
+				else
+					fprintf(stderr, "%s: no relations to check for \"%s\"\n",
+							progname, pat->pattern);
+			}
+		}
+	}
+
+	if (failed)
+		exit(1);
+
+	/*
+	 * Set parallel_workers to the lesser of opts.jobs and the number of
+	 * relations.
+	 */
+	reltotal = 0;
+	parallel_workers = 0;
+	for (cell = relations.head; cell; cell = cell->next)
+	{
+		reltotal++;
+		if (parallel_workers < opts.jobs)
+			parallel_workers++;
+	}
+
+	if (reltotal == 0)
+	{
+		fprintf(stderr, "%s: no relations to check", progname);
+		exit(1);
+	}
+	progress_report(reltotal, 0, NULL, true, false);
+
+	/*
+	 * Main event loop.
+	 *
+	 * We use server-side parallelism to check up to parallel_workers
+	 * relations in parallel.  The list of relations was computed in database
+	 * order, which minimizes the number of connects and disconnects as we
+	 * process the list.
+	 */
+	latest_datname = NULL;
+	sa = ParallelSlotsSetup(parallel_workers, &cparams, progname, opts.echo,
+							NULL);
+
+	initPQExpBuffer(&sql);
+	for (relprogress = 0, cell = relations.head; cell; cell = cell->next)
+	{
+		ParallelSlot *free_slot;
+		RelationInfo *rel;
+
+		rel = (RelationInfo *) cell->ptr;
+
+		if (CancelRequested)
+		{
+			failed = true;
+			break;
+		}
+
+		/*
+		 * The list of relations is in database sorted order.  If this next
+		 * relation is in a different database than the last one seen, we are
+		 * about to start checking this database.  Note that other slots may
+		 * still be working on relations from prior databases.
+		 */
+		latest_datname = rel->datinfo->datname;
+
+		progress_report(reltotal, relprogress, latest_datname, false, false);
+
+		relprogress++;
+
+		/*
+		 * 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(sa, rel->datinfo->datname);
+		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;
+			break;
+		}
+
+		/*
+		 * Execute the appropriate amcheck command for this relation using our
+		 * slot's database connection.  We do not wait for the command to
+		 * complete, nor do we perform any error checking, as that is done by
+		 * the parallel slots and our handler callback functions.
+		 */
+		if (rel->is_table)
+		{
+			prepare_table_command(&sql, rel->reloid,
+								  rel->datinfo->amcheck_schema);
+			ParallelSlotSetHandler(free_slot, verify_heapam_slot_handler,
+								   sql.data);
+			run_command(free_slot, sql.data, &cparams);
+		}
+		else
+		{
+			prepare_btree_command(&sql, rel->reloid,
+								  rel->datinfo->amcheck_schema);
+			ParallelSlotSetHandler(free_slot, verify_btree_slot_handler, NULL);
+			run_command(free_slot, sql.data, &cparams);
+		}
+	}
+	termPQExpBuffer(&sql);
+
+	if (!failed)
+	{
+
+		/*
+		 * 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 (sa && !ParallelSlotsWaitCompletion(sa))
+			failed = true;
+
+		progress_report(reltotal, relprogress, NULL, true, true);
+	}
+
+	if (sa)
+	{
+		ParallelSlotsTerminate(sa);
+		pg_free(sa);
+	}
+
+	if (failed)
+		exit(1);
+
+	if (!all_checks_pass)
+		exit(2);
+}
+
+/*
+ * 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 verify_heapam_slot_handler, which will
+ * receive and handle each row returned from the verify_heapam() function.
+ *
+ * sql: buffer into which the table checking command will be written
+ * reloid: relation of the table to be checked
+ * amcheck_schema: escaped and quoted name of schema in which amcheck contrib
+ * module is installed
+ */
+static void
+prepare_table_command(PQExpBuffer sql, 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,
+					  opts.on_error_stop ? "true" : "false",
+					  opts.reconcile_toast ? "true" : "false",
+					  opts.skip);
+	if (opts.startblock >= 0)
+		appendPQExpBuffer(sql, ",\nstartblock := %ld", opts.startblock);
+	if (opts.endblock >= 0)
+		appendPQExpBuffer(sql, ",\nendblock := %ld", opts.endblock);
+	appendPQExpBuffer(sql, "\n) v,"
+					  "\npg_catalog.pg_class c"
+					  "\nJOIN pg_catalog.pg_namespace n"
+					  "\nON c.relnamespace = n.oid"
+					  "\nWHERE c.oid = %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 verify_btree_slot_handler expects.
+ *
+ * sql: buffer into which the table checking command will be written
+ * reloid: relation of the table to be checked
+ * amcheck_schema: escaped and quoted name of schema in which amcheck contrib
+ * module is installed
+ */
+static void
+prepare_btree_command(PQExpBuffer sql, Oid reloid, const char *amcheck_schema)
+{
+	resetPQExpBuffer(sql);
+	if (opts.parent_check)
+		appendPQExpBuffer(sql,
+						  "SELECT %s.bt_index_parent_check("
+						  "\nindex := '%u'::regclass,"
+						  "\nheapallindexed := %s,"
+						  "\nrootdescend := %s)",
+						  amcheck_schema,
+						  reloid,
+						  (opts.heapallindexed ? "true" : "false"),
+						  (opts.rootdescend ? "true" : "false"));
+	else
+		appendPQExpBuffer(sql,
+						  "SELECT %s.bt_index_check("
+						  "\nindex := '%u'::regclass,"
+						  "\nheapallindexed := %s)",
+						  amcheck_schema,
+						  reloid,
+						  (opts.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.
+ *
+ * If reconnecting to the database is necessary, the cparams argument may be
+ * modified.
+ *
+ * slot: slot with connection to the server we should use for the command
+ * sql: query to send
+ * cparams: connection parameters in case the slot needs to be reconnected
+ */
+static void
+run_command(ParallelSlot *slot, const char *sql, ConnParams *cparams)
+{
+	if (opts.echo)
+		printf("%s\n", sql);
+
+	if (PQsendQuery(slot->connection, sql) == 0)
+	{
+		pg_log_error("error sending command to database \"%s\": %s",
+					 PQdb(slot->connection),
+					 PQerrorMessage(slot->connection));
+		pg_log_error("command was: %s", sql);
+		exit(1);
+	}
+}
+
+/*
+ * should_processing_continue
+ *
+ * Checks a query result returned from a query (presumably issued on a slot's
+ * connection) to determine if parallel slots should continue issuing further
+ * commands.
+ *
+ * Note: Heap relation corruption is reported by verify_heapam() via the result
+ * set, rather than an ERROR, but running verify_heapam() on a corrupted table
+ * may still result in an error being returned from the server due to missing
+ * relation files, bad checksums, etc.  The btree corruption checking functions
+ * always use errors to communicate corruption messages.  We can't just abort
+ * processing because we got a mere ERROR.
+ *
+ * res: result from an executed sql query
+ */
+static bool
+should_processing_continue(PGresult *res)
+{
+	const char *severity;
+
+	switch (PQresultStatus(res))
+	{
+			/* These are expected and ok */
+		case PGRES_COMMAND_OK:
+		case PGRES_TUPLES_OK:
+		case PGRES_NONFATAL_ERROR:
+			break;
+
+			/* This is expected but requires closer scrutiny */
+		case PGRES_FATAL_ERROR:
+			severity = PQresultErrorField(res, PG_DIAG_SEVERITY_NONLOCALIZED);
+			if (strcmp(severity, "FATAL") == 0)
+				return false;
+			if (strcmp(severity, "PANIC") == 0)
+				return false;
+			break;
+
+			/* These are unexpected */
+		case PGRES_BAD_RESPONSE:
+		case PGRES_EMPTY_QUERY:
+		case PGRES_COPY_OUT:
+		case PGRES_COPY_IN:
+		case PGRES_COPY_BOTH:
+		case PGRES_SINGLE_TUPLE:
+			return false;
+	}
+	return true;
+}
+
+/*
+ * verify_heapam_slot_handler
+ *
+ * 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 bool
+verify_heapam_slot_handler(PGresult *res, PGconn *conn, void *context)
+{
+	if (PQresultStatus(res) == PGRES_TUPLES_OK)
+	{
+		int			i;
+		int			ntups = PQntuples(res);
+
+		if (ntups > 0)
+			all_checks_pass = false;
+
+		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)
+	{
+		all_checks_pass = false;
+		printf("%s: %s\n", PQdb(conn), PQerrorMessage(conn));
+		printf("%s: query was: %s\n", PQdb(conn), (const char *) context);
+	}
+
+	return should_processing_continue(res);
+}
+
+/*
+ * verify_btree_slot_handler
+ *
+ * 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 bool
+verify_btree_slot_handler(PGresult *res, PGconn *conn, void *context)
+{
+	if (PQresultStatus(res) == PGRES_TUPLES_OK)
+	{
+		int			ntups = PQntuples(res);
+
+		if (ntups != 1)
+		{
+			/*
+			 * We expect the btree checking functions to return one void row
+			 * each, so we should output some sort of warning if we get
+			 * anything else, not because it indicates corruption, but because
+			 * it suggests a mismatch between amcheck and pg_amcheck versions.
+			 *
+			 * In conjunction with --progress, anything written to stderr at
+			 * this time would present strangely to the user without an extra
+			 * newline, so we print one.  If we were multithreaded, we'd have
+			 * to avoid splitting this across multiple calls, but we're in an
+			 * event loop, so it doesn't matter.
+			 */
+			if (opts.show_progress)
+				fprintf(stderr, "\n");
+			fprintf(stderr, "%s: btree checking function returned unexpected number of rows: %d\n",
+					progname, ntups);
+			fprintf(stderr, "%s: are %s's and amcheck's versions compatible?\n",
+					progname, progname);
+		}
+	}
+	else
+	{
+		all_checks_pass = false;
+		printf("%s: %s\n", PQdb(conn), PQerrorMessage(conn));
+	}
+
+	return should_processing_continue(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("      --no-index-expansion  do NOT expand list of relations to include indexes\n");
+	printf("      --no-toast-expansion  do NOT expand list of relations to include toast\n");
+	printf("      --no-strict-names     do NOT require patterns to match objects\n");
+	printf("\nTable Checking Options:\n");
+	printf("      --exclude-toast-pointers do NOT follow relation toast pointers\n");
+	printf("      --on-error-stop       stop checking at end of first corrupt page\n");
+	printf("      --skip=OPTION         do NOT check \"all-frozen\" or \"all-visible\" blocks\n");
+	printf("      --startblock=BLOCK    begin checking table(s) at the given block number\n");
+	printf("      --endblock=BLOCK      check table(s) only up to the given block number\n");
+	printf("\nBtree Index Checking Options:\n");
+	printf("      --heapallindexed      check all heap tuples are found within indexes\n");
+	printf("      --parent-check        check index parent/child relationships\n");
+	printf("      --rootdescend         search from root page to refind tuples\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("      --progress            show progress information\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);
+}
+
+/*
+ * Print a progress report based on the global variables. If verbose output
+ * is enabled, also print the current file name.
+ *
+ * Progress report is written at maximum once per second, unless the force
+ * parameter is set to true.
+ *
+ * If finished is set to true, this is the last progress report. The cursor
+ * is moved to the next line.
+ */
+static void
+progress_report(uint64 relations_total, uint64 relations_checked,
+				const char *datname, bool force, bool finished)
+{
+	int			percent = 0;
+	char		checked_str[32];
+	char		total_str[32];
+	pg_time_t	now;
+
+	if (!opts.show_progress)
+		return;
+
+	now = time(NULL);
+	if (now == last_progress_report && !force && !finished)
+		return;					/* Max once per second */
+
+	last_progress_report = now;
+	if (relations_total)
+		percent = (int) (relations_checked * 100 / relations_total);
+
+	/*
+	 * Separate step to keep platform-dependent format code out of fprintf
+	 * calls.  We only test for INT64_FORMAT availability in snprintf, not
+	 * fprintf.
+	 */
+	snprintf(checked_str, sizeof(checked_str), INT64_FORMAT, relations_checked);
+	snprintf(total_str, sizeof(total_str), INT64_FORMAT, relations_total);
+
+#define VERBOSE_DATNAME_LENGTH 35
+	if (opts.verbose)
+	{
+		if (!datname)
+
+			/*
+			 * No datname given, so clear the status line (used for first and
+			 * last call)
+			 */
+			fprintf(stderr,
+					"%*s/%s (%d%%) %*s",
+					(int) strlen(total_str),
+					checked_str, total_str, percent,
+					VERBOSE_DATNAME_LENGTH + 2, "");
+		else
+		{
+			bool		truncate = (strlen(datname) > VERBOSE_DATNAME_LENGTH);
+
+			fprintf(stderr,
+					"%*s/%s (%d%%), (%s%-*.*s)",
+					(int) strlen(total_str),
+					checked_str, total_str, percent,
+			/* Prefix with "..." if we do leading truncation */
+					truncate ? "..." : "",
+					truncate ? VERBOSE_DATNAME_LENGTH - 3 : VERBOSE_DATNAME_LENGTH,
+					truncate ? VERBOSE_DATNAME_LENGTH - 3 : VERBOSE_DATNAME_LENGTH,
+			/* Truncate datname at beginning if it's too long */
+					truncate ? datname + strlen(datname) - VERBOSE_DATNAME_LENGTH + 3 : datname);
+		}
+	}
+	else
+		fprintf(stderr,
+				"%*s/%s (%d%%)",
+				(int) strlen(total_str),
+				checked_str, total_str, percent);
+
+	/*
+	 * Stay on the same line if reporting to a terminal and we're not done
+	 * yet.
+	 */
+	fputc((!finished && isatty(fileno(stderr))) ? '\r' : '\n', stderr);
+}
+
+/*
+ * append_database_pattern
+ *
+ * Adds to a list the given pattern interpreted as a database name pattern.
+ *
+ * list: the list to be appended
+ * pattern: the database name pattern
+ * encoding: client encoding for parsing the pattern
+ */
+static void
+append_database_pattern(SimplePtrList *list, const char *pattern, int encoding)
+{
+	PQExpBufferData buf;
+	PatternInfo *info = (PatternInfo *) palloc0(sizeof(PatternInfo));
+
+	info->pattern_id = next_id++;
+
+	initPQExpBuffer(&buf);
+	patternToSQLRegex(encoding, NULL, NULL, &buf, pattern, false);
+	info->pattern = pattern;
+	info->db_regex = pstrdup(buf.data);
+
+	termPQExpBuffer(&buf);
+
+	simple_ptr_list_append(list, info);
+}
+
+/*
+ * append_schema_pattern
+ *
+ * Adds to a list the given pattern interpreted as a schema name pattern.
+ *
+ * list: the list to be appended
+ * pattern: the schema name pattern
+ * encoding: client encoding for parsing the pattern
+ */
+static void
+append_schema_pattern(SimplePtrList *list, const char *pattern, int encoding)
+{
+	PQExpBufferData buf;
+	PatternInfo *info = (PatternInfo *) palloc0(sizeof(PatternInfo));
+
+	info->pattern_id = next_id++;
+
+	initPQExpBuffer(&buf);
+	patternToSQLRegex(encoding, NULL, NULL, &buf, pattern, false);
+	info->pattern = pattern;
+	info->nsp_regex = pstrdup(buf.data);
+	termPQExpBuffer(&buf);
+
+	simple_ptr_list_append(list, info);
+}
+
+/*
+ * append_relation_pattern_helper
+ *
+ * Adds to a list the given pattern interpreted as a relation pattern.
+ *
+ * list: the list to be appended
+ * pattern: the relation name pattern
+ * encoding: client encoding for parsing the pattern
+ * table_only: whether the pattern should only be matched against heap tables
+ * index_only: whether the pattern should only be matched against btree indexes
+ */
+static void
+append_relation_pattern_helper(SimplePtrList *list, const char *pattern,
+							   int encoding, bool table_only, bool index_only)
+{
+	PQExpBufferData dbbuf;
+	PQExpBufferData nspbuf;
+	PQExpBufferData relbuf;
+	PatternInfo *info = (PatternInfo *) palloc0(sizeof(PatternInfo));
+
+	info->pattern_id = next_id++;
+
+	initPQExpBuffer(&dbbuf);
+	initPQExpBuffer(&nspbuf);
+	initPQExpBuffer(&relbuf);
+
+	patternToSQLRegex(encoding, &dbbuf, &nspbuf, &relbuf, pattern, false);
+	info->pattern = pattern;
+	if (dbbuf.data[0])
+		info->db_regex = pstrdup(dbbuf.data);
+	if (nspbuf.data[0])
+		info->nsp_regex = pstrdup(nspbuf.data);
+	if (relbuf.data[0])
+		info->rel_regex = pstrdup(relbuf.data);
+
+	termPQExpBuffer(&dbbuf);
+	termPQExpBuffer(&nspbuf);
+	termPQExpBuffer(&relbuf);
+
+	info->table_only = table_only;
+	info->index_only = index_only;
+
+	simple_ptr_list_append(list, info);
+}
+
+/*
+ * append_relation_pattern
+ *
+ * Adds to a list the given pattern interpreted as a relation pattern, to be
+ * matched against both tables and indexes.
+ *
+ * list: the list to be appended
+ * pattern: the relation name pattern
+ * encoding: client encoding for parsing the pattern
+ */
+static void
+append_relation_pattern(SimplePtrList *list, const char *pattern, int encoding)
+{
+	append_relation_pattern_helper(list, pattern, encoding, false, false);
+}
+
+/*
+ * append_table_pattern
+ *
+ * Adds to a list the given pattern interpreted as a relation pattern, to be
+ * matched only against tables.
+ *
+ * list: the list to be appended
+ * pattern: the relation name pattern
+ * encoding: client encoding for parsing the pattern
+ */
+static void
+append_table_pattern(SimplePtrList *list, const char *pattern, int encoding)
+{
+	append_relation_pattern_helper(list, pattern, encoding, true, false);
+}
+
+/*
+ * append_index_pattern
+ *
+ * Adds to a list the given pattern interpreted as a relation pattern, to be
+ * matched only against indexes.
+ *
+ * list: the list to be appended
+ * pattern: the relation name pattern
+ * encoding: client encoding for parsing the pattern
+ */
+static void
+append_index_pattern(SimplePtrList *list, const char *pattern, int encoding)
+{
+	append_relation_pattern_helper(list, pattern, encoding, false, true);
+}
+
+/*
+ * append_db_pattern_cte
+ *
+ * Appends to the buffer the body of a Common Table Expression (CTE) containing
+ * the database portions filtered from the list of patterns expressed as three
+ * columns:
+ *
+ *     id: the unique pattern ID
+ *     pat: the full user specified pattern from the command line
+ *     rgx: the database regular expression parsed from the pattern
+ *
+ * Patterns without a database portion are skipped.  Patterns with more than
+ * just a database portion are optionally skipped, depending on argument
+ * 'inclusive'.
+ *
+ * buf: the buffer to be appended
+ * patterns: the list of patterns to be inserted into the CTE
+ * conn: the database connection
+ * inclusive: whether to include patterns with schema and/or relation parts
+ */
+static void
+append_db_pattern_cte(PQExpBuffer buf, const SimplePtrList *patterns,
+					  PGconn *conn, bool inclusive)
+{
+	SimplePtrListCell *cell;
+	const char *comma;
+	bool		have_values;
+
+	comma = "";
+	have_values = false;
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		PatternInfo *info = (PatternInfo *) cell->ptr;
+
+		if (info->db_regex != NULL &&
+			(inclusive || (info->nsp_regex == NULL && info->rel_regex == NULL)))
+		{
+			if (!have_values)
+				appendPQExpBufferStr(buf, "\nVALUES");
+			have_values = true;
+			appendPQExpBuffer(buf, "%s\n(%d, ", comma, info->pattern_id);
+			appendStringLiteralConn(buf, info->pattern, conn);
+			appendPQExpBufferStr(buf, ", ");
+			appendStringLiteralConn(buf, info->db_regex, conn);
+			appendPQExpBufferStr(buf, ")");
+			comma = ",";
+		}
+	}
+
+	if (!have_values)
+		appendPQExpBufferStr(buf, "\nSELECT NULL, NULL, NULL WHERE false");
+}
+
+/*
+ * compile_database_list
+ *
+ * Compiles a list of databases to check based on the user supplied options,
+ * sorted to preserve the order they were specified on the command line.  In
+ * the event that multiple databases match a single command line pattern, they
+ * are secondarily sorted by name.
+ *
+ * conn: connection to the initial database
+ * databases: the list onto which databases should be appended
+ */
+static void
+compile_database_list(PGconn *conn, SimplePtrList *databases)
+{
+	PGresult   *res;
+	PQExpBufferData sql;
+	int			ntups;
+	int			i;
+	bool		fatal;
+
+	initPQExpBuffer(&sql);
+
+	/* Append the include patterns CTE. */
+	appendPQExpBufferStr(&sql, "WITH include_raw (id, pat, rgx) AS (");
+	append_db_pattern_cte(&sql, &opts.include, conn, true);
+
+	/* Append the exclude patterns CTE. */
+	appendPQExpBufferStr(&sql, "\n),\nexclude_raw (id, pat, rgx) AS (");
+	append_db_pattern_cte(&sql, &opts.exclude, conn, false);
+	appendPQExpBufferStr(&sql, "\n),");
+
+	/*
+	 * Append the database CTE, which includes whether each database is
+	 * connectable and also joins against exclude_raw to determine whether
+	 * each database is excluded.
+	 */
+	appendPQExpBufferStr(&sql,
+						 "\ndatabase (datname) AS ("
+						 "\nSELECT d.datname"
+						 "\nFROM pg_catalog.pg_database d"
+						 "\nLEFT OUTER JOIN exclude_raw e"
+						 "\nON d.datname ~ e.rgx"
+						 "\nWHERE d.datallowconn"
+						 "\nAND e.id IS NULL"
+						 "\n),"
+
+	/*
+	 * Append the include_pat CTE, which joins the include_raw CTE against the
+	 * databases CTE to determine if all the inclusion patterns had matches,
+	 * and whether each matched pattern had the misfortune of only matching
+	 * excluded or unconnectable databases.
+	 */
+						 "\ninclude_pat (id, pat, checkable) AS ("
+						 "\nSELECT i.id, i.pat,"
+						 "\nCOUNT(*) FILTER ("
+						 "\nWHERE d IS NOT NULL"
+						 "\n) AS checkable"
+						 "\nFROM include_raw i"
+						 "\nLEFT OUTER JOIN database d"
+						 "\nON d.datname ~ i.rgx"
+						 "\nGROUP BY i.id, i.pat"
+						 "\n),"
+
+	/*
+	 * Append the filtered_databases CTE, which selects from the database CTE
+	 * optionally joined against the include_raw CTE to only select databases
+	 * that match an inclusion pattern.  This appears to duplicate what the
+	 * include_pat CTE already did above, but here we want only databses, and
+	 * there we wanted patterns.
+	 */
+						 "\nfiltered_databases (datname) AS ("
+						 "\nSELECT DISTINCT d.datname"
+						 "\nFROM database d");
+	if (!opts.alldb)
+		appendPQExpBufferStr(&sql,
+							 "\nINNER JOIN include_raw i"
+							 "\nON d.datname ~ i.rgx");
+	appendPQExpBufferStr(&sql,
+						 "\n)"
+
+	/*
+	 * Select the checkable databases and the unmatched inclusion patterns.
+	 */
+						 "\nSELECT pat, datname"
+						 "\nFROM ("
+						 "\nSELECT id, pat, NULL::TEXT AS datname"
+						 "\nFROM include_pat"
+						 "\nWHERE checkable = 0"
+						 "\nUNION ALL"
+						 "\nSELECT NULL, NULL, datname"
+						 "\nFROM filtered_databases"
+						 "\n) AS combined_records"
+						 "\nORDER BY id NULLS LAST, datname");
+
+	res = executeQuery(conn, sql.data, opts.echo);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("query failed: %s", PQerrorMessage(conn));
+		pg_log_error("query was: %s", sql.data);
+		disconnectDatabase(conn);
+		exit(1);
+	}
+	termPQExpBuffer(&sql);
+
+	ntups = PQntuples(res);
+	for (fatal = false, i = 0; i < ntups; i++)
+	{
+		const char *pat = NULL;
+		const char *datname = NULL;
+
+		if (!PQgetisnull(res, i, 0))
+			pat = PQgetvalue(res, i, 0);
+		if (!PQgetisnull(res, i, 1))
+			datname = PQgetvalue(res, i, 1);
+
+		if (pat != NULL)
+		{
+			/*
+			 * Current record pertains to an inclusion pattern that matched no
+			 * checkable databases.
+			 */
+			fatal = opts.strict_names;
+			fprintf(stderr, "%s: no checkable database: \"%s\"\n",
+					progname, pat);
+		}
+		else
+		{
+			/* Current record pertains to a database */
+			Assert(datname != NULL);
+
+			DatabaseInfo *dat = (DatabaseInfo *) palloc0(sizeof(DatabaseInfo));
+
+			/* This database is included.  Add to list */
+			if (opts.verbose)
+				fprintf(stderr, "%s: including database: \"%s\"\n", progname,
+						datname);
+
+			dat->datname = pstrdup(datname);
+			simple_ptr_list_append(databases, dat);
+		}
+	}
+	PQclear(res);
+
+	if (fatal)
+	{
+		disconnectDatabase(conn);
+		exit(1);
+	}
+}
+
+/*
+ * append_rel_pattern_raw_cte
+ *
+ * Appends to the buffer the body of a Common Table Expression (CTE) containing
+ * the patterns from the given list as seven columns:
+ *
+ *     id: the unique pattern ID
+ *     pat: the full user specified pattern from the command line
+ *     db_regex: the database regexp parsed from the pattern, or NULL if the
+ *               pattern had no database part
+ *     nsp_regex: the namespace regexp parsed from the pattern, or NULL if the
+ *                pattern had no namespace part
+ *     rel_regex: the relname regexp parsed from the pattern, or NULL if the
+ *                pattern had no relname part
+ *     table_only: true if the pattern applies only to tables (not indexes)
+ *     index_only: true if the pattern applies only to indexes (not tables)
+ *
+ * buf: the buffer to be appended
+ * patterns: the list of patterns to be inserted into the CTE
+ * conn: the database connection
+ */
+static void
+append_rel_pattern_raw_cte(PQExpBuffer buf, const SimplePtrList *patterns,
+						   PGconn *conn)
+{
+	SimplePtrListCell *cell;
+	const char *comma;
+	bool		have_values;
+
+	comma = "";
+	have_values = false;
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		PatternInfo *info = (PatternInfo *) cell->ptr;
+
+		if (!have_values)
+			appendPQExpBufferStr(buf, "\nVALUES");
+		have_values = true;
+		appendPQExpBuffer(buf, "%s\n(%d::INTEGER, ", comma, info->pattern_id);
+		appendStringLiteralConn(buf, info->pattern, conn);
+		appendPQExpBufferStr(buf, "::TEXT, ");
+		if (info->db_regex == NULL)
+			appendPQExpBufferStr(buf, "NULL");
+		else
+			appendStringLiteralConn(buf, info->db_regex, conn);
+		appendPQExpBufferStr(buf, "::TEXT, ");
+		if (info->nsp_regex == NULL)
+			appendPQExpBufferStr(buf, "NULL");
+		else
+			appendStringLiteralConn(buf, info->nsp_regex, conn);
+		appendPQExpBufferStr(buf, "::TEXT, ");
+		if (info->rel_regex == NULL)
+			appendPQExpBufferStr(buf, "NULL");
+		else
+			appendStringLiteralConn(buf, info->rel_regex, conn);
+		if (info->table_only)
+			appendPQExpBufferStr(buf, "::TEXT, true::BOOLEAN");
+		else
+			appendPQExpBufferStr(buf, "::TEXT, false::BOOLEAN");
+		if (info->index_only)
+			appendPQExpBufferStr(buf, ", true::BOOLEAN");
+		else
+			appendPQExpBufferStr(buf, ", false::BOOLEAN");
+		appendPQExpBufferStr(buf, ")");
+		comma = ",";
+	}
+
+	if (!have_values)
+		appendPQExpBufferStr(buf,
+							 "\nSELECT NULL::INTEGER, NULL::TEXT, NULL::TEXT,"
+							 "\nNULL::TEXT, NULL::TEXT, NULL::BOOLEAN,"
+							 "\nNULL::BOOLEAN"
+							 "\nWHERE false");
+}
+
+/*
+ * append_rel_pattern_filtered_cte
+ *
+ * Appends to the buffer a Common Table Expression (CTE) which selects
+ * all patterns from the named raw CTE, filtered by database.  All patterns
+ * which have no database portion or whose database portion matches our
+ * connection's database name are selected, with other patterns excluded.
+ *
+ * The basic idea here is that if we're connected to database "foo" and we have
+ * patterns "foo.bar.baz", "alpha.beta" and "one.two.three", we only want to
+ * use the first two while processing relations in this database, as the third
+ * one is not relevant.
+ *
+ * buf: the buffer to be appended
+ * raw: the name of the CTE to select from
+ * filtered: the name of the CTE to create
+ * conn: the database connection
+ */
+static void
+append_rel_pattern_filtered_cte(PQExpBuffer buf, const char *raw,
+								const char *filtered, PGconn *conn)
+{
+	appendPQExpBuffer(buf,
+					  "\n%s (id, pat, nsp_regex, rel_regex, table_only, index_only) AS ("
+					  "\nSELECT id, pat, nsp_regex, rel_regex, table_only, index_only"
+					  "\nFROM %s r"
+					  "\nWHERE (r.db_regex IS NULL"
+					  "\nOR ",
+					  filtered, raw);
+	appendStringLiteralConn(buf, PQdb(conn), conn);
+	appendPQExpBufferStr(buf, " ~ r.db_regex)");
+	appendPQExpBufferStr(buf,
+						 "\nAND (r.nsp_regex IS NOT NULL"
+						 "\nOR r.rel_regex IS NOT NULL)"
+						 "\n),");
+}
+
+/*
+ * compile_relation_list_one_db
+ *
+ * Compiles a list of relations to check within the currently connected
+ * database based on the user supplied options, sorted by descending size,
+ * and appends them to the given list of relations.
+ *
+ * The cells of the constructed list contain all information about the relation
+ * necessary to connect to the database and check the object, including which
+ * database to connect to, where contrib/amcheck is installed, and the Oid and
+ * type of object (table vs. index).  Rather than duplicating the database
+ * details per relation, the relation structs use references to the same
+ * database object, provided by the caller.
+ *
+ * conn: connection to this next database, which should be the same as in 'dat'
+ * relations: list onto which the relations information should be appended
+ * dat: the database info struct for use by each relation
+ */
+static void
+compile_relation_list_one_db(PGconn *conn, SimplePtrList *relations,
+							 const DatabaseInfo *dat)
+{
+	PGresult   *res;
+	PQExpBufferData sql;
+	int			ntups;
+	int			i;
+	const char *datname;
+
+	initPQExpBuffer(&sql);
+	appendPQExpBufferStr(&sql, "WITH");
+
+	/* Append CTEs for the relation inclusion patterns, if any */
+	if (!opts.allrel)
+	{
+		appendPQExpBufferStr(&sql,
+							 "\ninclude_raw (id, pat, db_regex, nsp_regex, rel_regex, table_only, index_only) AS (");
+		append_rel_pattern_raw_cte(&sql, &opts.include, conn);
+		appendPQExpBufferStr(&sql, "\n),");
+		append_rel_pattern_filtered_cte(&sql, "include_raw", "include_pat", conn);
+	}
+
+	/* Append CTEs for the relation exclusion patterns, if any */
+	if (opts.excludetbl || opts.excludeidx)
+	{
+		appendPQExpBufferStr(&sql,
+							 "\nexclude_raw (id, pat, db_regex, nsp_regex, rel_regex, table_only, index_only) AS (");
+		append_rel_pattern_raw_cte(&sql, &opts.exclude, conn);
+		appendPQExpBufferStr(&sql, "\n),");
+		append_rel_pattern_filtered_cte(&sql, "exclude_raw", "exclude_pat", conn);
+	}
+
+	/* Append the relation CTE. */
+	appendPQExpBufferStr(&sql,
+						 "\nrelation (id, pat, oid, reltoastrelid, relpages, is_table, is_index) AS ("
+						 "\nSELECT DISTINCT ON (c.oid");
+	if (!opts.allrel)
+		appendPQExpBufferStr(&sql, ", ip.id) ip.id, ip.pat,");
+	else
+		appendPQExpBufferStr(&sql, ") NULL::INTEGER AS id, NULL::TEXT AS pat,");
+	appendPQExpBuffer(&sql,
+					  "\nc.oid, c.reltoastrelid, c.relpages,"
+					  "\nc.relam = %u AS is_table,"
+					  "\nc.relam = %u AS is_index"
+					  "\nFROM pg_catalog.pg_class c"
+					  "\nINNER JOIN pg_catalog.pg_namespace n"
+					  "\nON c.relnamespace = n.oid",
+					  HEAP_TABLE_AM_OID, BTREE_AM_OID);
+	if (!opts.allrel)
+		appendPQExpBuffer(&sql,
+						  "\nINNER JOIN include_pat ip"
+						  "\nON (n.nspname ~ ip.nsp_regex OR ip.nsp_regex IS NULL)"
+						  "\nAND (c.relname ~ ip.rel_regex OR ip.rel_regex IS NULL)"
+						  "\nAND (c.relam = %u OR NOT ip.table_only)"
+						  "\nAND (c.relam = %u OR NOT ip.index_only)",
+						  HEAP_TABLE_AM_OID, BTREE_AM_OID);
+	if (opts.excludetbl || opts.excludeidx)
+		appendPQExpBuffer(&sql,
+						  "\nLEFT OUTER JOIN exclude_pat ep"
+						  "\nON (n.nspname ~ ep.nsp_regex OR ep.nsp_regex IS NULL)"
+						  "\nAND (c.relname ~ ep.rel_regex OR ep.rel_regex IS NULL)"
+						  "\nAND (c.relam = %u OR NOT ep.table_only)"
+						  "\nAND (c.relam = %u OR NOT ep.index_only)",
+						  HEAP_TABLE_AM_OID, BTREE_AM_OID);
+
+	if (opts.excludetbl || opts.excludeidx)
+		appendPQExpBufferStr(&sql, "\nWHERE ep.pat IS NULL");
+	else
+		appendPQExpBufferStr(&sql, "\nWHERE true");
+
+	/*
+	 * We need to be careful not to break the --no-toast-expansion and
+	 * --no-index-expansion options.  By default, the indexes, toast tables,
+	 * and toast table indexes associated with primary tables are included,
+	 * using their own CTEs below.  We implement the --exclude-* options by
+	 * not creating those CTEs, but that's no use if we've already selected
+	 * the toast and indexes here.  On the other hand, we want inclusion
+	 * patterns that match indexes or toast tables to be honored.  So, if
+	 * inclusion patterns were given, we want to select all tables, toast
+	 * tables, or indexes that match the patterns.  But if no inclusion
+	 * patterns were given, and we're simply matching all relations, then we
+	 * only want to match the primary tables here.
+	 */
+	if (opts.allrel)
+		appendPQExpBuffer(&sql,
+						  "\nAND c.relam = %u"
+						  "\nAND c.relkind IN ('r', 'm', 't')"
+						  "\nAND c.relnamespace != %u",
+						  HEAP_TABLE_AM_OID, PG_TOAST_NAMESPACE);
+	else
+		appendPQExpBuffer(&sql,
+						  "\nAND c.relam IN (%u, %u)"
+						  "\nAND c.relkind IN ('r', 'm', 't', 'i')"
+						  "\nAND ((c.relam = %u AND c.relkind IN ('r', 'm', 't')) OR"
+						  "\n(c.relam = %u AND c.relkind = 'i'))",
+						  HEAP_TABLE_AM_OID, BTREE_AM_OID,
+						  HEAP_TABLE_AM_OID, BTREE_AM_OID);
+
+	appendPQExpBufferStr(&sql,
+						 "\nORDER BY c.oid"
+						 "\n)");
+
+	if (!opts.no_toast_expansion)
+	{
+		/*
+		 * Include a CTE for toast tables associated with primary tables
+		 * selected above, filtering by exclusion patterns (if any) that match
+		 * toast table names.
+		 */
+		appendPQExpBufferStr(&sql,
+							 ",\ntoast (oid, relpages) AS ("
+							 "\nSELECT t.oid, t.relpages"
+							 "\nFROM pg_catalog.pg_class t"
+							 "\nINNER JOIN relation r"
+							 "\nON r.reltoastrelid = t.oid");
+		if (opts.excludetbl)
+			appendPQExpBufferStr(&sql,
+								 "\nLEFT OUTER JOIN exclude_pat ep"
+								 "\nON ('pg_toast' ~ ep.nsp_regex OR ep.nsp_regex IS NULL)"
+								 "\nAND (t.relname ~ ep.rel_regex OR ep.rel_regex IS NULL)"
+								 "\nAND ep.table_only"
+								 "\nWHERE ep.id IS NULL");
+		appendPQExpBufferStr(&sql,
+							 "\n)");
+	}
+	if (!opts.no_index_expansion)
+	{
+		/*
+		 * Include a CTE for btree indexes associated with primary tables
+		 * selected above, filtering by exclusion patterns (if any) that match
+		 * btree index names.
+		 */
+		appendPQExpBuffer(&sql,
+						  ",\nindex (oid, relpages) AS ("
+						  "\nSELECT c.oid, c.relpages"
+						  "\nFROM relation r"
+						  "\nINNER JOIN pg_catalog.pg_index i"
+						  "\nON r.oid = i.indrelid"
+						  "\nINNER JOIN pg_catalog.pg_class c"
+						  "\nON i.indexrelid = c.oid");
+		if (opts.excludeidx)
+			appendPQExpBufferStr(&sql,
+								 "\nINNER JOIN pg_catalog.pg_namespace n"
+								 "\nON c.relnamespace = n.oid"
+								 "\nLEFT OUTER JOIN exclude_pat ep"
+								 "\nON (n.nspname ~ ep.nsp_regex OR ep.nsp_regex IS NULL)"
+								 "\nAND (c.relname ~ ep.rel_regex OR ep.rel_regex IS NULL)"
+								 "\nAND ep.index_only"
+								 "\nWHERE ep.id IS NULL");
+		else
+			appendPQExpBufferStr(&sql,
+								 "\nWHERE true");
+		appendPQExpBuffer(&sql,
+						  "\nAND c.relam = %u"
+						  "\nAND c.relkind = 'i'",
+						  BTREE_AM_OID);
+		if (opts.no_toast_expansion)
+			appendPQExpBuffer(&sql,
+							  "\nAND c.relnamespace != %u",
+							  PG_TOAST_NAMESPACE);
+		appendPQExpBufferStr(&sql, "\n)");
+	}
+
+	if (!opts.no_toast_expansion && !opts.no_index_expansion)
+	{
+		/*
+		 * Include a CTE for btree indexes associated with toast tables of
+		 * primary tables selected above, filtering by exclusion patterns (if
+		 * any) that match the toast index names.
+		 */
+		appendPQExpBuffer(&sql,
+						  ",\ntoast_index (oid, relpages) AS ("
+						  "\nSELECT c.oid, c.relpages"
+						  "\nFROM toast t"
+						  "\nINNER JOIN pg_catalog.pg_index i"
+						  "\nON t.oid = i.indrelid"
+						  "\nINNER JOIN pg_catalog.pg_class c"
+						  "\nON i.indexrelid = c.oid");
+		if (opts.excludeidx)
+			appendPQExpBufferStr(&sql,
+								 "\nLEFT OUTER JOIN exclude_pat ep"
+								 "\nON ('pg_toast' ~ ep.nsp_regex OR ep.nsp_regex IS NULL)"
+								 "\nAND (c.relname ~ ep.rel_regex OR ep.rel_regex IS NULL)"
+								 "\nAND ep.index_only"
+								 "\nWHERE ep.id IS NULL");
+		else
+			appendPQExpBufferStr(&sql,
+								 "\nWHERE true");
+		appendPQExpBuffer(&sql,
+						  "\nAND c.relam = %u"
+						  "\nAND c.relkind = 'i'"
+						  "\n)",
+						  BTREE_AM_OID);
+	}
+
+	/*
+	 * Roll-up distinct rows from CTEs.
+	 *
+	 * Relations that match more than one pattern may occur more than once in
+	 * the list, and indexes and toast for primary relations may also have
+	 * matched in their own right, so we rely on UNION to deduplicate the
+	 * list.
+	 */
+	appendPQExpBuffer(&sql,
+					  "\nSELECT id, is_table, is_index, oid"
+					  "\nFROM (");
+	appendPQExpBufferStr(&sql,
+	/* Inclusion patterns that failed to match */
+						 "\nSELECT id, is_table, is_index,"
+						 "\nNULL::OID AS oid,"
+						 "\nNULL::INTEGER AS relpages"
+						 "\nFROM relation"
+						 "\nWHERE id IS NOT NULL"
+						 "\nUNION"
+	/* Primary relations */
+						 "\nSELECT NULL::INTEGER AS id,"
+						 "\nis_table, is_index,"
+						 "\noid, relpages"
+						 "\nFROM relation");
+	if (!opts.no_toast_expansion)
+		appendPQExpBufferStr(&sql,
+							 "\nUNION"
+		/* Toast tables for primary relations */
+							 "\nSELECT NULL::INTEGER AS id, TRUE AS is_table,"
+							 "\nFALSE AS is_index, oid, relpages"
+							 "\nFROM toast");
+	if (!opts.no_index_expansion)
+		appendPQExpBufferStr(&sql,
+							 "\nUNION"
+		/* Indexes for primary relations */
+							 "\nSELECT NULL::INTEGER AS id, FALSE AS is_table,"
+							 "\nTRUE AS is_index, oid, relpages"
+							 "\nFROM index");
+	if (!opts.no_toast_expansion && !opts.no_index_expansion)
+		appendPQExpBufferStr(&sql,
+							 "\nUNION"
+		/* Indexes for toast relations */
+							 "\nSELECT NULL::INTEGER AS id, FALSE AS is_table,"
+							 "\nTRUE AS is_index, oid, relpages"
+							 "\nFROM toast_index");
+	appendPQExpBufferStr(&sql,
+						 "\n) AS combined_records"
+						 "\nORDER BY relpages DESC NULLS FIRST, oid");
+
+	res = executeQuery(conn, sql.data, opts.echo);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("query failed: %s", PQerrorMessage(conn));
+		pg_log_error("query was: %s", sql.data);
+		disconnectDatabase(conn);
+		exit(1);
+	}
+	termPQExpBuffer(&sql);
+
+	/*
+	 * Allocate a single copy of the database name to be shared by all nodes
+	 * in the object list, constructed below.
+	 */
+	datname = pstrdup(PQdb(conn));
+
+	ntups = PQntuples(res);
+	for (i = 0; i < ntups; i++)
+	{
+		int			pattern_id = 0;
+		bool		is_table = false;
+		bool		is_index = false;
+		Oid			oid = InvalidOid;
+
+		if (!PQgetisnull(res, i, 0))
+			pattern_id = atoi(PQgetvalue(res, i, 0));
+		if (!PQgetisnull(res, i, 1))
+			is_table = (PQgetvalue(res, i, 1)[0] == 't');
+		if (!PQgetisnull(res, i, 2))
+			is_index = (PQgetvalue(res, i, 2)[0] == 't');
+		if (!PQgetisnull(res, i, 3))
+			oid = atooid(PQgetvalue(res, i, 3));
+
+		if (pattern_id > 0)
+		{
+			/*
+			 * Current record pertains to an inclusion pattern.  Find the
+			 * pattern in the list and record that it matched.  If we expected
+			 * a large number of command-line inclusion pattern arguments, the
+			 * datastructure here might need to be more efficient, but we
+			 * expect the list to be short.
+			 */
+
+			SimplePtrListCell *cell;
+			bool		found;
+
+			for (found = false, cell = opts.include.head; cell; cell = cell->next)
+			{
+				PatternInfo *info = (PatternInfo *) cell->ptr;
+
+				if (info->pattern_id == pattern_id)
+				{
+					info->matched = true;
+					found = true;
+					break;
+				}
+			}
+			if (!found)
+			{
+				pg_log_error("internal error: received unexpected pattern_id %d",
+							 pattern_id);
+				exit(1);
+			}
+		}
+		else
+		{
+			/* Current record pertains to a relation */
+
+			RelationInfo *rel = (RelationInfo *) palloc0(sizeof(RelationInfo));
+
+			Assert(OidIsValid(oid));
+			Assert(is_table ^ is_index);
+
+			rel->datinfo = dat;
+			rel->reloid = oid;
+			rel->is_table = is_table;
+
+			simple_ptr_list_append(relations, rel);
+		}
+	}
+	PQclear(res);
+}
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..8c6e267ee9
--- /dev/null
+++ b/contrib/pg_amcheck/t/002_nonesuch.pl
@@ -0,0 +1,213 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 60;
+
+# 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
+
+# Failing to connect to the initial database is an error.
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'qqq' ],
+	qr/database "qqq" does not exist/,
+	'checking a non-existent database');
+
+# Failing to resolve a secondary database name is also an error, though since
+# the string is treated as a pattern, the error message looks different.
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'postgres', 'qqq' ],
+	qr/pg_amcheck: no checkable database: "qqq"/,
+	'checking a non-existent database');
+
+# Failing to connect to the initial database is still an error when using
+# --no-strict-names.
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-p', $port, 'qqq' ],
+	qr/database "qqq" does not exist/,
+	'checking a non-existent database with --no-strict-names');
+
+# But failing to resolve secondary database names is not an error when using
+# --no-strict-names.  We should still see the message, but as a non-fatal
+# warning
+$node->command_checks_all(
+	[ 'pg_amcheck', '--no-strict-names', '-p', $port, '-d', 'no_such_database', 'postgres', 'qqq' ],
+	0,
+	[ ],
+	[ qr/no checkable database: "qqq"/ ],
+	'checking a non-existent secondary database with --no-strict-names');
+
+# Check that a substring of an existent database name does not get interpreted
+# as a matching pattern.
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'post' ],
+	qr/database "post" does not exist/,
+	'checking a non-existent primary database (substring of existent database)');
+
+# And again, but testing the secondary database name rather than the primary
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'postgres', 'post' ],
+	qr/pg_amcheck: no checkable database: "post"/,
+	'checking a non-existent secondary database (substring of existent database)');
+
+# Likewise, check that a superstring of an existent database name does not get
+# interpreted as a matching pattern.
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'postresql' ],
+	qr/database "postresql" does not exist/,
+	'checking a non-existent primary database (superstring of existent database)');
+
+# And again, but testing the secondary database name rather than the primary
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, 'postgres', 'postgresql' ],
+	qr/pg_amcheck: no checkable database: "postgresql"/,
+	'checking a non-existent secondary database (superstring of existent database)');
+
+#########################################
+# Test connecting with a non-existent user
+
+# Failing to connect to the initial database due to bad username is an error.
+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');
+
+# Failing to connect to the initial database due to bad username is still an
+# error when using --no-strict-names.
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-p', $port, '-U=no_such_user', 'postgres' ],
+	qr/role "=no_such_user" does not exist/,
+	'checking with a non-existent user, --no-strict-names');
+
+#########################################
+# Test checking databases without amcheck installed
+
+# Attempting to check a database by name where amcheck is not installed should
+# raise a warning.  If all databases are skipped, having no relations to check
+# raises an error.
+$node->command_checks_all(
+	[ 'pg_amcheck', '-p', $port, 'template1' ],
+	1,
+	[],
+	[ qr/pg_amcheck: skipping database "template1": amcheck is not installed/,
+	  qr/pg_amcheck: no relations to check/ ],
+	'checking a database by name without amcheck installed');
+
+# Likewise, but by database pattern rather than by name, such that some
+# databases with amcheck installed are included, and so checking occurs and
+# only a warning is raised.
+$node->command_checks_all(
+	[ 'pg_amcheck', '-p', $port, '-d', '*', 'postgres' ],
+	0,
+	[],
+	[ qr/pg_amcheck: skipping database "template1": amcheck is not installed/ ],
+	'checking a database by dbname implication without amcheck installed');
+
+# And again, but by checking all databases.
+$node->command_checks_all(
+	[ 'pg_amcheck', '-p', $port, '--all', 'postgres' ],
+	0,
+	[],
+	[ qr/pg_amcheck: skipping database "template1": amcheck is not installed/ ],
+	'checking a database by --all implication without amcheck installed');
+
+#########################################
+# Test unreasonable patterns
+
+# Check three-part unreasonable pattern that has zero-length names
+$node->command_checks_all(
+	[ 'pg_amcheck', '-p', $port, 'postgres', '-t', '..' ],
+	1,
+	[ qr/^$/ ],
+	[ qr/pg_amcheck: no checkable database: "\.\."/ ],
+	'checking table pattern ".."');
+
+# Again, but with non-trivial schema and relation parts
+$node->command_checks_all(
+	[ 'pg_amcheck', '-p', $port, 'postgres', '-t', '.foo.bar' ],
+	1,
+	[ qr/^$/ ],
+	[ qr/pg_amcheck: no checkable database: "\.foo\.bar"/ ],
+	'checking table pattern ".foo.bar"');
+
+# Check two-part unreasonable pattern that has zero-length names
+$node->command_checks_all(
+	[ 'pg_amcheck', '-p', $port, 'postgres', '-t', '.' ],
+	1,
+	[ qr/^$/ ],
+	[ qr/pg_amcheck: no tables to check for "\."/ ],
+	'checking table pattern "."');
+
+#########################################
+# Test checking non-existent schemas, tables, and indexes
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-s', 'no_such_schema' ],
+	qr/pg_amcheck: no relations to check in schemas for "no_such_schema"/,
+	'checking a non-existent schema');
+
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-v', '-p', $port, '-s', 'no_such_schema' ],
+	qr/pg_amcheck: no relations to check/,
+	'checking a non-existent schema with --no-strict-names -v');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-t', 'no_such_table' ],
+	qr/pg_amcheck: no tables to check for "no_such_table"/,
+	'checking a non-existent table');
+
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-v', '-p', $port, '-t', 'no_such_table' ],
+	qr/pg_amcheck: no relations to check/,
+	'checking a non-existent table with --no-strict-names -v');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-i', 'no_such_index' ],
+	qr/pg_amcheck: no btree indexes to check for "no_such_index"/,
+	'checking a non-existent index');
+
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-v', '-p', $port, '-i', 'no_such_index' ],
+	qr/pg_amcheck: no relations to check/,
+	'checking a non-existent index with --no-strict-names -v');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-s', 'no*such*schema*' ],
+	qr/pg_amcheck: no relations to check in schemas for "no\*such\*schema\*"/,
+	'no matching schemas');
+
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-v', '-p', $port, '-s', 'no*such*schema*' ],
+	qr/pg_amcheck: no relations to check/,
+	'no matching schemas with --no-strict-names -v');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-t', 'no*such*table*' ],
+	qr/pg_amcheck: no tables to check for "no\*such\*table\*"/,
+	'no matching tables');
+
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-v', '-p', $port, '-t', 'no*such*table*' ],
+	qr/pg_amcheck: no relations to check/,
+	'no matching tables with --no-strict-names -v');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', $port, '-i', 'no*such*index*' ],
+	qr/pg_amcheck: no btree indexes to check for "no\*such\*index\*"/,
+	'no matching indexes');
+
+command_fails_like(
+	[ 'pg_amcheck', '--no-strict-names', '-v', '-p', $port, '-i', 'no*such*index*' ],
+	qr/pg_amcheck: no relations to check/,
+	'no matching indexes with --no-strict-names -v');
diff --git a/contrib/pg_amcheck/t/003_check.pl b/contrib/pg_amcheck/t/003_check.pl
new file mode 100644
index 0000000000..502b599fcd
--- /dev/null
+++ b/contrib/pg_amcheck/t/003_check.pl
@@ -0,0 +1,497 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 57;
+
+my ($node, $port, %corrupt_page, %remove_relation);
+
+# 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";
+}
+
+# Adds the relation file for the given (dbname, relname) to the list
+# to be corrupted by means of overwriting junk in the first page.
+#
+# Assumes the test node is running.
+sub plan_to_corrupt_first_page($$)
+{
+	my ($dbname, $relname) = @_;
+	my $relpath = relation_filepath($dbname, $relname);
+	$corrupt_page{$relpath} = 1;
+}
+
+# Adds the relation file for the given (dbname, relname) to the list
+# to be corrupted by means of removing the file..
+#
+# Assumes the test node is running
+sub plan_to_remove_relation_file($$)
+{
+	my ($dbname, $relname) = @_;
+	my $relpath = relation_filepath($dbname, $relname);
+	$remove_relation{$relpath} = 1;
+}
+
+# For the given (dbname, relname), if a corresponding toast table
+# exists, adds that toast table's relation file to the list to be
+# corrupted by means of removing the file.
+#
+# Assumes the test node is running.
+sub plan_to_remove_toast_file($$)
+{
+	my ($dbname, $relname) = @_;
+	my $toastname = relation_toast($dbname, $relname);
+	plan_to_remove_relation_file($dbname, $toastname) if ($toastname);
+}
+
+# Corrupts the first page of the given file path
+sub corrupt_first_page($)
+{
+	my ($relpath) = @_;
+
+	my $fh;
+	open($fh, '+<', $relpath)
+	  or BAIL_OUT("open failed: $!");
+	binmode $fh;
+
+	# Corrupt some line pointers.  The values are chosen to hit the
+	# various line-pointer-corruption checks in verify_heapam.c
+	# on both little-endian and big-endian architectures.
+	seek($fh, 32, 0)
+	  or BAIL_OUT("seek failed: $!");
+	syswrite(
+		$fh,
+		pack("L*",
+			0xAAA15550, 0xAAA0D550, 0x00010000,
+			0x00008000, 0x0000800F, 0x001e8000,
+			0xFFFFFFFF)
+	) or BAIL_OUT("syswrite failed: $!");
+	close($fh)
+	  or BAIL_OUT("close failed: $!");
+}
+
+# Stops the node, performs all the corruptions previously planned, and
+# starts the node again.
+#
+sub perform_all_corruptions()
+{
+	$node->stop();
+	for my $relpath (keys %corrupt_page)
+	{
+		corrupt_first_page($relpath);
+	}
+	for my $relpath (keys %remove_relation)
+	{
+		unlink($relpath);
+	}
+	$node->start;
+}
+
+# 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"
+plan_to_remove_relation_file('db1', 's1.t1_btree');
+plan_to_corrupt_first_page('db1', 's1.t2_btree');
+
+# Corrupt tables in schema "s2"
+plan_to_remove_relation_file('db1', 's2.t1');
+plan_to_corrupt_first_page('db1', 's2.t2');
+
+# Corrupt tables, partitions, matviews, and btrees in schema "s3"
+plan_to_remove_relation_file('db1', 's3.t1');
+plan_to_corrupt_first_page('db1', 's3.t2');
+
+plan_to_remove_relation_file('db1', 's3.t1_mv');
+plan_to_remove_relation_file('db1', 's3.p1_1');
+
+plan_to_corrupt_first_page('db1', 's3.t2_mv');
+plan_to_corrupt_first_page('db1', 's3.p2_1');
+
+plan_to_remove_relation_file('db1', 's3.t1_btree');
+plan_to_corrupt_first_page('db1', 's3.t2_btree');
+
+# Corrupt toast table, partitions, and materialized views in schema "s4"
+plan_to_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
+plan_to_remove_relation_file('db1', 's5.seq1');
+plan_to_remove_relation_file('db1', 's5.t1_hash');
+plan_to_remove_relation_file('db1', 's5.t1_gist');
+plan_to_remove_relation_file('db1', 's5.t1_gin');
+plan_to_remove_relation_file('db1', 's5.t1_brin');
+plan_to_remove_relation_file('db1', 's5.t1_spgist');
+
+plan_to_corrupt_first_page('db1', 's5.seq2');
+plan_to_corrupt_first_page('db1', 's5.t2_hash');
+plan_to_corrupt_first_page('db1', 's5.t2_gist');
+plan_to_corrupt_first_page('db1', 's5.t2_gin');
+plan_to_corrupt_first_page('db1', 's5.t2_brin');
+plan_to_corrupt_first_page('db1', 's5.t2_spgist');
+
+
+# Database 'db2' corruptions
+#
+plan_to_remove_relation_file('db2', 's1.t1');
+plan_to_remove_relation_file('db2', 's1.t1_btree');
+
+
+# Leave 'db3' uncorrupted
+#
+
+# Perform the corruptions we planned above using only a single database restart.
+#
+perform_all_corruptions();
+
+
+# Standard first arguments to TestLib functions
+my @cmd = ('pg_amcheck', '--quiet', '-p', $port);
+
+# Regular expressions to match various expected output
+my $no_output_re = qr/^$/;
+my $line_pointer_corruption_re = qr/line pointer/;
+my $missing_file_re = qr/could not open file ".*": No such file or directory/;
+my $index_missing_relation_fork_re = qr/index ".*" lacks a main relation fork/;
+
+# Checking databases with amcheck installed and corrupt relations, pg_amcheck
+# command itself should return exit status = 2, because tables and indexes are
+# corrupt, not exit status = 1, which would mean the pg_amcheck command itself
+# failed.  Corruption messages should go to stdout, and nothing to stderr.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1' ],
+	2,
+	[ $index_missing_relation_fork_re,
+	  $line_pointer_corruption_re,
+	  $missing_file_re,
+	],
+	[ $no_output_re ],
+	'pg_amcheck all schemas, tables and indexes in database db1');
+
+$node->command_checks_all(
+	[ @cmd, 'db1', 'db2', 'db3' ],
+	2,
+	[ $index_missing_relation_fork_re,
+	  $line_pointer_corruption_re,
+	  $missing_file_re,
+	],
+	[ $no_output_re ],
+	'pg_amcheck all schemas, tables and indexes in databases db1, 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.
+#
+# If we don't limit the check to databases with amcheck installed, we expect
+# complaint on stderr, but otherwise stderr should be quiet.
+#
+$node->command_checks_all(
+	[ @cmd, '--all', '-s', 's1', '-i', 't1_btree' ],
+	2,
+	[ $index_missing_relation_fork_re ],
+	[ qr/pg_amcheck: skipping database "postgres": amcheck is not installed/ ],
+	'pg_amcheck index s1.t1_btree reports missing main relation fork');
+
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's1', '-i', 't2_btree' ],
+	2,
+	[ qr/.+/ ],			# Any non-empty error message is acceptable
+	[ $no_output_re ],
+	'pg_amcheck index s1.s2 reports index corruption');
+
+# Checking db1.s1 with indexes excluded should show no corruptions because we
+# did not corrupt any tables in db1.s1.  Verify that both stdout and stderr
+# are quiet.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1', '-t', 's1.*', '--no-index-expansion' ],
+	0,
+	[ $no_output_re ],
+	[ $no_output_re ],
+	'pg_amcheck of db1.s1 excluding indexes');
+
+# Checking db2.s1 should show table corruptions if indexes are excluded
+#
+$node->command_checks_all(
+	[ @cmd, 'db2', '-t', 's1.*', '--no-index-expansion' ],
+	2,
+	[ $missing_file_re ],
+	[ $no_output_re ],
+	'pg_amcheck of db2.s1 excluding indexes');
+
+# In schema db1.s3, the tables and indexes are both corrupt.  We should see
+# corruption messages on stdout, and nothing on stderr.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's3' ],
+	2,
+	[ $index_missing_relation_fork_re,
+	  $line_pointer_corruption_re,
+	  $missing_file_re,
+	],
+	[ $no_output_re ],
+	'pg_amcheck schema s3 reports table and index errors');
+
+# In schema db1.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_checks_all(
+	[ @cmd, 'db1', '-s', 's4' ],
+	2,
+	[ $missing_file_re ],
+	[ $no_output_re ],
+	'pg_amcheck in schema s4 reports toast corruption');
+
+$node->command_checks_all(
+	[ @cmd, '--no-toast-expansion', '--exclude-toast-pointers', 'db1', '-s', 's4' ],
+	0,
+	[ $no_output_re ],
+	[ $no_output_re ],
+	'pg_amcheck in schema s4 excluding toast reports no corruption');
+
+# Check that no corruption is reported in schema db1.s5
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's5' ],
+	0,
+	[ $no_output_re ],
+	[ $no_output_re ],
+	'pg_amcheck over schema s5 reports no corruption');
+
+# In schema db1.s1, only indexes are corrupt.  Verify that when we exclude
+# the indexes, no corruption is reported about the schema.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's1', '-I', 't1_btree', '-I', 't2_btree' ],
+	0,
+	[ $no_output_re ],
+	[ $no_output_re ],
+	'pg_amcheck over schema s1 with corrupt indexes excluded reports no corruption');
+
+# In schema db1.s1, only indexes are corrupt.  Verify that when we provide only
+# table inclusions, and disable index expansion, no corruption is reported
+# about the schema.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1', '-t', 's1.*', '--no-index-expansion' ],
+	0,
+	[ $no_output_re ],
+	[ $no_output_re ],
+	'pg_amcheck over schema s1 with all indexes excluded reports no corruption');
+
+# In schema db1.s2, only tables are corrupt.  Verify that when we exclude those
+# tables that no corruption is reported.
+#
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's2', '-T', 't1', '-T', 't2' ],
+	0,
+	[ $no_output_re ],
+	[ $no_output_re ],
+	'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_checks_all(
+	[ @cmd, 'db1', '-s', 's1', '-i', 't1_btree', '--parent-check' ],
+	2,
+	[ $index_missing_relation_fork_re ],
+	[ $no_output_re ],
+	'pg_amcheck smoke test --parent-check');
+
+$node->command_checks_all(
+	[ @cmd, 'db1', '-s', 's1', '-i', 't1_btree', '--heapallindexed', '--rootdescend' ],
+	2,
+	[ $index_missing_relation_fork_re ],
+	[ $no_output_re ],
+	'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..d5537a5b37
--- /dev/null
+++ b/contrib/pg_amcheck/t/004_verify_heapam.pl
@@ -0,0 +1,487 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+
+use Test::More tests => 20;
+
+# 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_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 == 15)	# 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', '--no-index-expansion', '-p', $port, 'postgres'],
+	2,
+	[ @expected ],
+	[ ],
+	'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..eba8ea9cae
--- /dev/null
+++ b/contrib/pg_amcheck/t/005_opclass_damage.pl
@@ -0,0 +1,54 @@
+# 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 => 5;
+
+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_checks_all(
+	[ 'pg_amcheck', '-p', $node->port, 'postgres' ],
+	2,
+	[ 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..9bee92c30a
--- /dev/null
+++ b/doc/src/sgml/pgamcheck.sgml
@@ -0,0 +1,668 @@
+<!-- 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>Options</title>
+
+  <para>
+   pg_amcheck accepts the following command-line arguments:
+
+   <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-dbname</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-dbname</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-dbname</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-dbname</option>, the database will
+       be excluded.
+      </para>
+      <para>
+       Examples:
+       <simplelist>
+        <member><literal>--exclude-dbname=america</literal></member>
+        <member><literal>--exclude-dbname="*pacific*"</literal></member>
+       </simplelist>
+      </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>--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, but note that
+       unless <option>--exclude-toast-pointers</option> is given, toast
+       pointers found in the main table will be followed into the toast table
+       without regard for the location in the toast table.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <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 relation.
+      </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>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <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 <xref linkend="amcheck"/>'s
+       <option>heapallindexed</option> option.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <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>-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>-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>-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>
+       The default is to use a single connection.
+      </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>--no-index-expansion</option></term>
+     <listitem>
+      <para>
+       When including a table relation in the list of relations to check, do
+       not automatically include btree indexes associated with table. 
+      </para>
+      <para>
+       By default, all tables to be checked will also have checks performed on
+       their associated btree indexes, if any.  If this option is given, only
+       those indexes which match a <option>--relation</option> or
+       <option>--index</option> pattern will be checked.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><option>--no-strict-names</option></term>
+     <listitem>
+      <para>
+       When calculating the list of databases to check, and the objects within
+       those databases to be checked, do not raise an error for database,
+       schema, relation, table, nor index inclusion patterns which match no
+       corresponding objects.
+      </para>
+      <para>
+       Exclusion patterns are not required to match any objects, but by
+       default unmatched inclusion patterns raise an error, including when
+       they fail to match as a result of an exclusion pattern having
+       prohibited them matching an existent object, and when they fail to
+       match a database because it is unconnectable (datallowconn is false).
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term><option>--no-toast-expansion</option></term>
+     <listitem>
+      <para>
+       When including a table relation in the list of relations to check, do
+       not automatically include toast tables associated with table. 
+      </para>
+      <para>
+       By default, all tables to be checked will also have checks performed on
+       their associated toast tables, if any.  If this option is given, only
+       those toast tables which match a <option>--relation</option> or
+       <option>--table</option> pattern will be checked.
+      </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>--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>-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>--progress</option></term>
+     <listitem>
+      <para>
+       Show progress information about how many relations have been checked.
+      </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>-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>
+       Examples:
+       <simplelist>
+        <member><literal>--relation=accounts_table</literal></member>
+        <member><literal>--relation=accounting_department.accounts_table</literal></member>
+        <member><literal>--relation=corporate_database.accounting_department.*_table</literal></member>
+       </simplelist>
+      </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>--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>--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>
+
+    <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>
+      <para>
+       Note that both tables and indexes are included using this option, which
+       might not be what you want if you are also using
+       <option>--no-index-expansion</option>.  To specify all tables in a schema
+       without also specifying all indexes, <option>--table</option> can be
+       used with a pattern that specifies the schema.  For example, to check
+       all tables in schema <literal>corp</literal>, the option
+       <literal>--table="corp.*"</literal> may be used.
+      </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>--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.  This can be specified as
+       <literal>"none"</literal>, but since this is the default, it need not be
+       mentioned.
+      </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, but note
+       that unless <option>--exclude-toast-pointers</option> is given, toast
+       pointers found in the main table will be followed into the toast table
+       without regard for the location in the toast table.
+      </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>-U</option></term>
+     <term><option>--username=USERNAME</option></term>
+     <listitem>
+      <para>
+       User name to connect as.
+      </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>
+
+    <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>-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>
+
+   </variablelist>
+  </para>
+ </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 49614106dc..f680544e07 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 b1dec43f9d..a0dfe164cd 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -101,6 +101,7 @@ AlterUserMappingStmt
 AlteredTableInfo
 AlternativeSubPlan
 AlternativeSubPlanState
+AmcheckOptions
 AnalyzeAttrComputeStatsFunc
 AnalyzeAttrFetchFunc
 AnalyzeForeignTable_function
@@ -499,6 +500,7 @@ DSA
 DWORD
 DataDumperPtr
 DataPageDeleteStack
+DatabaseInfo
 DateADT
 Datum
 DatumTupleFields
@@ -2084,6 +2086,7 @@ RelToCluster
 RelabelType
 Relation
 RelationData
+RelationInfo
 RelationPtr
 RelationSyncEntry
 RelcacheCallbackFunction
-- 
2.21.1 (Apple Git-122.3)

=

v42-0003-Extending-PostgresNode-to-test-corruption.patch (application/octet-stream, 16.7K) download
From cd0e6068ce7e3d523050467fdf9bc0f21c02c7d8 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 v42 3/3] 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    | 145 ++++++++++
 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, 506 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..45ad223531
--- /dev/null
+++ b/contrib/pg_amcheck/t/006_relfile_damage.pl
@@ -0,0 +1,145 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 22;
+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_checks_all(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--heapallindexed' ],
+	2,
+	[ qr/heap tuple .* from table "t1" lacks matching index tuple within index "t1_idx"/ ],
+	[ ],
+	'pg_amcheck reverted index with --heapallindexed');
+
+$node->command_checks_all(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--heapallindexed', '--rootdescend' ],
+	2,
+	[ 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_checks_all(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', $toastrel ],
+	0,
+	[ qr/^$/ ],
+	[ ],
+	'pg_amcheck reverted toast table');
+
+$node->command_checks_all(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*', '--exclude-toast-pointers' ],
+	0,
+	[ qr/^$/ ],
+	[ ],
+	'pg_amcheck with reverted toast using --exclude-toast-pointers');
+
+$node->command_checks_all(
+	[ 'pg_amcheck', '--quiet', '-p', $port, '-r', 'postgres.t.*' ],
+	2,
+	[ 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 (161+ messages)  latest in thread

Message-ID: <BA592F2D-F928-46FF-9516-2B827F067F57@enterprisedb.com>
Permalink:  ../BA592F2D-F928-46FF-9516-2B827F067F57@enterprisedb.com/
Also on:    postgresql.org/message-id/BA592F2D-F928-46FF-9516-2B827F067F57@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: pg_amcheck contrib application
  In-Reply-To: <BA592F2D-F928-46FF-9516-2B827F067F57@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