From bda1f3a0bd35c1762e47aaef8381f67ba241ba24 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Sun, 24 Jan 2021 13:42:24 -0800
Subject: [PATCH v33 8/8] 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            | 1380 ++++++++++++++++++++
 contrib/pg_amcheck/pg_amcheck.h            |  130 ++
 contrib/pg_amcheck/t/001_basic.pl          |    9 +
 contrib/pg_amcheck/t/002_nonesuch.pl       |   59 +
 contrib/pg_amcheck/t/003_check.pl          |  428 ++++++
 contrib/pg_amcheck/t/004_verify_heapam.pl  |  496 +++++++
 contrib/pg_amcheck/t/005_opclass_damage.pl |   52 +
 doc/src/sgml/contrib.sgml                  |    1 +
 doc/src/sgml/filelist.sgml                 |    1 +
 doc/src/sgml/pgamcheck.sgml                | 1004 ++++++++++++++
 src/tools/msvc/Install.pm                  |    4 +-
 src/tools/msvc/Mkvcbuild.pm                |    6 +-
 src/tools/pgindent/typedefs.list           |    3 +
 16 files changed, 3601 insertions(+), 5 deletions(-)
 create mode 100644 contrib/pg_amcheck/.gitignore
 create mode 100644 contrib/pg_amcheck/Makefile
 create mode 100644 contrib/pg_amcheck/pg_amcheck.c
 create mode 100644 contrib/pg_amcheck/pg_amcheck.h
 create mode 100644 contrib/pg_amcheck/t/001_basic.pl
 create mode 100644 contrib/pg_amcheck/t/002_nonesuch.pl
 create mode 100644 contrib/pg_amcheck/t/003_check.pl
 create mode 100644 contrib/pg_amcheck/t/004_verify_heapam.pl
 create mode 100644 contrib/pg_amcheck/t/005_opclass_damage.pl
 create mode 100644 doc/src/sgml/pgamcheck.sgml

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

