v11-0003-Adding-contrib-module-pg_amcheck.patch

From 97126faecea883e072150f0bbecdd323d8a0a646 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Mon, 20 Jul 2020 13:05:26 -0700
Subject: [PATCH v11 3/3] Adding contrib module pg_amcheck

Adding new contrib module pg_amcheck, which is a command line
interface for running amcheck's verifications against tables and
indexes.
---
 contrib/Makefile                          |   1 +
 contrib/pg_amcheck/.gitignore             |   3 +
 contrib/pg_amcheck/Makefile               |  28 +
 contrib/pg_amcheck/pg_amcheck.c           | 900 ++++++++++++++++++++++
 contrib/pg_amcheck/t/001_basic.pl         |   9 +
 contrib/pg_amcheck/t/002_nonesuch.pl      |  55 ++
 contrib/pg_amcheck/t/003_check.pl         |  85 ++
 contrib/pg_amcheck/t/004_verify_heapam.pl | 434 +++++++++++
 doc/src/sgml/contrib.sgml                 |   1 +
 doc/src/sgml/filelist.sgml                |   1 +
 doc/src/sgml/pg_amcheck.sgml              | 136 ++++
 11 files changed, 1653 insertions(+)
 create mode 100644 contrib/pg_amcheck/.gitignore
 create mode 100644 contrib/pg_amcheck/Makefile
 create mode 100644 contrib/pg_amcheck/pg_amcheck.c
 create mode 100644 contrib/pg_amcheck/t/001_basic.pl
 create mode 100644 contrib/pg_amcheck/t/002_nonesuch.pl
 create mode 100644 contrib/pg_amcheck/t/003_check.pl
 create mode 100644 contrib/pg_amcheck/t/004_verify_heapam.pl
 create mode 100644 doc/src/sgml/pg_amcheck.sgml

diff --git a/contrib/Makefile b/contrib/Makefile
index 1846d415b6..c21c27cbeb 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -29,6 +29,7 @@ SUBDIRS = \
 		oid2name	\
 		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..07ad380105
--- /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..74554b9e8d
--- /dev/null
+++ b/contrib/pg_amcheck/Makefile
@@ -0,0 +1,28 @@
+# 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
+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..45cd50c217
--- /dev/null
+++ b/contrib/pg_amcheck/pg_amcheck.c
@@ -0,0 +1,900 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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_am.h"
+#include "catalog/pg_class.h"
+#include "common/logging.h"
+#include "common/username.h"
+#include "fe_utils/connect.h"
+#include "fe_utils/print.h"
+#include "fe_utils/simple_list.h"
+#include "fe_utils/string_utils.h"
+#include "pg_getopt.h"
+
+const char *usage_text[] = {
+	"pg_amcheck is the PostgreSQL command line database corruption checker.",
+	"",
+	"Usage:",
+	"  pg_amcheck [OPTION]... [DBNAME [USERNAME]]",
+	"",
+	"General options:",
+	"  -V, --version            output version information, then exit",
+	"  -?, --help               show this help, then exit",
+	"  -s, --schema=PATTERN     check all relations in the specified schema(s)",
+	"  -N, --exclude-schema=PATTERN do NOT check relations in the specified "
+	"schema(s)",
+	"  -t, --table=PATTERN      check the specified table(s) only",
+	"  -T, --exclude-table=PATTERN    do NOT check the specified table(s)",
+	"  -i, --check-indexes      check associated btree indexes, if any",
+	"  -I, --exclude-indexes    do NOT check associated btree indexes",
+	"  --strict-names           require table and/or schema include patterns "
+	"to match at least one entity each",
+	"  -b, --startblock         check relations beginning at the given "
+	"starting block number",
+	"  -e, --endblock           check relations only up to the given ending "
+	"block number",
+	"  -f, --skip-all-frozen    do not check blocks marked as all frozen",
+	"  -v, --skip-all-visible   do not check blocks marked as all visible",
+	"",
+	"Connection options:",
+	"  -d, --dbname=DBNAME      database name to connect to",
+	"  -h, --host=HOSTNAME      database server host or socket directory",
+	"  -p, --port=PORT          database server port",
+	"  -U, --username=USERNAME  database user name",
+	"  -w, --no-password        never prompt for password",
+	"  -W, --password           force password prompt (should happen "
+	"automatically)",
+	"",
+	NULL						/* sentinel */
+};
+
+typedef struct
+{
+	char	   *dbname;
+	char	   *host;
+	char	   *port;
+	char	   *username;
+}			ConnectOptions;
+
+typedef enum trivalue
+{
+	TRI_DEFAULT,
+	TRI_NO,
+	TRI_YES
+} trivalue;
+
+typedef struct
+{
+	PGconn	   *db;				/* connection to backend */
+	bool		notty;			/* stdin or stdout is not a tty (as determined
+								 * on startup) */
+	trivalue	getPassword;	/* prompt for a username and password */
+	const char *progname;		/* in case you renamed pg_amcheck */
+	bool		strict_names;	/* The specified names/patterns should to
+								 * match at least one entity */
+	bool		on_error_stop;	/* The checking of each table should stop
+								 * after the first corrupt page is found. */
+	bool		skip_frozen;	/* Do not check pages marked all frozen */
+	bool		skip_visible;	/* Do not check pages marked all visible */
+	bool		check_indexes;	/* Check btree indexes for tables */
+	char	   *startblock;		/* Block number where checking begins */
+	char	   *endblock;		/* Block number where checking ends */
+}			AmCheckSettings;
+
+static AmCheckSettings settings;
+
+/*
+ * Object inclusion/exclusion lists
+ *
+ * The string lists record the patterns given by command-line switches,
+ * which we then convert to lists of OIDs of matching objects.
+ */
+static SimpleStringList schema_include_patterns = {NULL, NULL};
+static SimpleOidList schema_include_oids = {NULL, NULL};
+static SimpleStringList schema_exclude_patterns = {NULL, NULL};
+static SimpleOidList schema_exclude_oids = {NULL, NULL};
+
+static SimpleStringList table_include_patterns = {NULL, NULL};
+static SimpleOidList table_include_oids = {NULL, NULL};
+static SimpleStringList table_exclude_patterns = {NULL, NULL};
+static SimpleOidList table_exclude_oids = {NULL, NULL};
+
+/*
+ * List of tables to be checked, compiled from above lists.
+ */
+static SimpleOidList checklist = {NULL, NULL};
+
+
+static void check_tables(SimpleOidList *checklist);
+static void check_table(Oid tbloid);
+static void check_indexes(Oid tbloid);
+static void check_index(Oid tbloid, Oid idxoid);
+
+static void parse_cli_options(int argc, char *argv[],
+							  ConnectOptions * connOpts);
+static void usage(void);
+static void showVersion(void);
+
+static void NoticeProcessor(void *arg, const char *message);
+
+static void expand_schema_name_patterns(SimpleStringList *patterns,
+										SimpleOidList *oids,
+										bool strict_names);
+static void expand_table_name_patterns(SimpleStringList *patterns,
+									   SimpleOidList *oids,
+									   bool strict_names);
+static void get_table_check_list(SimpleOidList *include_nsp,
+								 SimpleOidList *exclude_nsp,
+								 SimpleOidList *include_tbl,
+								 SimpleOidList *exclude_tbl,
+								 SimpleOidList *checklist);
+
+static void die_on_query_failure(const char *query);
+static void ExecuteSqlStatement(const char *query);
+static PGresult *ExecuteSqlQuery(const char *query, ExecStatusType status);
+static PGresult *ExecuteSqlQueryForSingleRow(const char *query);
+
+#define fatal(...) do { pg_log_error(__VA_ARGS__); exit(1); } while(0)
+
+#define NOPAGER		0
+#define EXIT_BADCONN 2
+
+int
+main(int argc, char **argv)
+{
+	ConnectOptions connOpts;
+	bool		have_password = false;
+	char		password[100];
+	bool		new_pass;
+
+	pg_logging_init(argv[0]);
+	set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_amcheck"));
+
+	if (argc > 1)
+	{
+		if ((strcmp(argv[1], "-?") == 0) ||
+			(argc == 2 && (strcmp(argv[1], "--help") == 0)))
+		{
+			usage();
+			exit(EXIT_SUCCESS);
+		}
+		if (strcmp(argv[1], "--version") == 0 || strcmp(argv[1], "-V") == 0)
+		{
+			showVersion();
+			exit(EXIT_SUCCESS);
+		}
+	}
+
+	memset(&settings, 0, sizeof(settings));
+	settings.progname = get_progname(argv[0]);
+
+	settings.db = NULL;
+	setDecimalLocale();
+
+	settings.notty = (!isatty(fileno(stdin)) || !isatty(fileno(stdout)));
+
+	settings.getPassword = TRI_DEFAULT;
+
+	/* Default behaviors */
+	settings.on_error_stop = false;
+	settings.skip_frozen = false;
+	settings.skip_visible = false;
+	settings.check_indexes = true;
+
+	parse_cli_options(argc, argv, &connOpts);
+
+	if (settings.getPassword == TRI_YES)
+	{
+		/*
+		 * We can't be sure yet of the username that will be used, so don't
+		 * offer a potentially wrong one.  Typical uses of this option are
+		 * noninteractive anyway.
+		 */
+		simple_prompt("Password: ", password, sizeof(password), false);
+		have_password = true;
+	}
+
+	/* loop until we have a password if requested by backend */
+	do
+	{
+#define ARRAY_SIZE	8
+		const char **keywords = pg_malloc(ARRAY_SIZE * sizeof(*keywords));
+		const char **values = pg_malloc(ARRAY_SIZE * sizeof(*values));
+
+		keywords[0] = "host";
+		values[0] = connOpts.host;
+		keywords[1] = "port";
+		values[1] = connOpts.port;
+		keywords[2] = "user";
+		values[2] = connOpts.username;
+		keywords[3] = "password";
+		values[3] = have_password ? password : NULL;
+		keywords[4] = "dbname"; /* see do_connect() */
+		if (connOpts.dbname == NULL)
+		{
+			if (getenv("PGDATABASE"))
+				values[4] = getenv("PGDATABASE");
+			else if (getenv("PGUSER"))
+				values[4] = getenv("PGUSER");
+			else
+				values[4] = "postgres";
+		}
+		else
+			values[4] = connOpts.dbname;
+		keywords[5] = "fallback_application_name";
+		values[5] = settings.progname;
+		keywords[6] = "client_encoding";
+		values[6] = (settings.notty ||
+					 getenv("PGCLIENTENCODING")) ? NULL : "auto";
+		keywords[7] = NULL;
+		values[7] = NULL;
+
+		new_pass = false;
+		settings.db = PQconnectdbParams(keywords, values, true);
+		if (settings.db == NULL)
+		{
+			pg_log_error("no connection to server after initial attempt");
+			exit(EXIT_BADCONN);
+		}
+
+		free(keywords);
+		free(values);
+
+		if (PQstatus(settings.db) == CONNECTION_BAD &&
+			PQconnectionNeedsPassword(settings.db) &&
+			!have_password &&
+			settings.getPassword != TRI_NO)
+		{
+			/*
+			 * Before closing the old PGconn, extract the user name that was
+			 * actually connected with.
+			 */
+			const char *realusername = PQuser(settings.db);
+			char	   *password_prompt;
+
+			if (realusername && realusername[0])
+				password_prompt = psprintf(_("Password for user %s: "),
+										   realusername);
+			else
+				password_prompt = pg_strdup(_("Password: "));
+			PQfinish(settings.db);
+
+			simple_prompt(password_prompt, password, sizeof(password), false);
+			free(password_prompt);
+			have_password = true;
+			new_pass = true;
+		}
+	} while (new_pass);
+
+	if (!settings.db)
+	{
+		pg_log_error("no connection to server");
+		exit(EXIT_BADCONN);
+	}
+
+	if (PQstatus(settings.db) == CONNECTION_BAD)
+	{
+		pg_log_error("could not connect to server: %s",
+					 PQerrorMessage(settings.db));
+		PQfinish(settings.db);
+		exit(EXIT_BADCONN);
+	}
+
+	/* Expand schema selection patterns into OID lists */
+	if (schema_include_patterns.head != NULL)
+	{
+		expand_schema_name_patterns(&schema_include_patterns,
+									&schema_include_oids,
+									settings.strict_names);
+		if (schema_include_oids.head == NULL)
+			fatal("no matching schemas were found");
+	}
+	expand_schema_name_patterns(&schema_exclude_patterns,
+								&schema_exclude_oids,
+								false);
+	/* non-matching exclusion patterns aren't an error */
+
+	/* Expand table selection patterns into OID lists */
+	if (table_include_patterns.head != NULL)
+	{
+		expand_table_name_patterns(&table_include_patterns,
+								   &table_include_oids,
+								   settings.strict_names);
+		if (table_include_oids.head == NULL)
+			fatal("no matching tables were found");
+	}
+	expand_table_name_patterns(&table_exclude_patterns,
+							   &table_exclude_oids,
+							   false);
+
+	/*
+	 * Compile list of all tables to be checked based on namespace and table
+	 * includes and excludes.
+	 */
+	get_table_check_list(&schema_include_oids, &schema_exclude_oids,
+						 &table_include_oids, &table_exclude_oids, &checklist);
+
+	PQsetNoticeProcessor(settings.db, NoticeProcessor, NULL);
+
+	check_tables(&checklist);
+
+	return 0;
+}
+
+static void
+check_tables(SimpleOidList *checklist)
+{
+	const SimpleOidListCell *cell;
+
+	for (cell = checklist->head; cell; cell = cell->next)
+	{
+		check_table(cell->val);
+		if (settings.check_indexes)
+			check_indexes(cell->val);
+	}
+}
+
+static void
+check_table(Oid tbloid)
+{
+	PQExpBuffer query;
+	PGresult   *res;
+	int			i;
+	char	   *skip;
+	const char *stop;
+
+	if (settings.db == NULL)
+		fatal("no connection on entry to expand_table_name_patterns");
+
+	if (settings.startblock == NULL)
+		settings.startblock = pg_strdup("NULL");
+	if (settings.endblock == NULL)
+		settings.endblock = pg_strdup("NULL");
+	if (settings.skip_frozen)
+		skip = pg_strdup("'all frozen'");
+	else if (settings.skip_visible)
+		skip = pg_strdup("'all visible'");
+	else
+		skip = pg_strdup("NULL");
+	stop = (settings.on_error_stop) ? "true" : "false";
+
+	query = createPQExpBuffer();
+
+	appendPQExpBuffer(query,
+					  "SELECT c.relname, v.blkno, v.offnum, v.lp_off, "
+					  "v.lp_flags, v.lp_len, v.attnum, v.chunk, v.msg"
+					  "\nFROM verify_heapam(rel := %u, on_error_stop := %s, "
+					  "skip := %s, startblock := %s, endblock := %s) v, "
+					  "pg_class c"
+					  "\nWHERE c.oid = %u",
+					  tbloid, stop, skip, settings.startblock,
+					  settings.endblock, tbloid);
+
+	ExecuteSqlStatement("RESET search_path");
+	res = ExecuteSqlQuery(query->data, PGRES_TUPLES_OK);
+	PQclear(ExecuteSqlQueryForSingleRow(ALWAYS_SECURE_SEARCH_PATH_SQL));
+
+	if (PQntuples(res) > 0)
+	{
+		int			lines = PQntuples(res) * 2;
+		FILE	   *output = PageOutput(lines, NULL);
+
+		for (i = 0; i < PQntuples(res); i++)
+		{
+			fprintf(output,
+					"(relname=%s,blkno=%s,offnum=%s,lp_off=%s,lp_flags=%s,"
+					"lp_len=%s,attnum=%s,chunk=%s)\n%s\n",
+					PQgetvalue(res, i, 0),	/* relname */
+					PQgetvalue(res, i, 1),	/* blkno */
+					PQgetvalue(res, i, 2),	/* offnum */
+					PQgetvalue(res, i, 3),	/* lp_off */
+					PQgetvalue(res, i, 4),	/* lp_flags */
+					PQgetvalue(res, i, 5),	/* lp_len */
+					PQgetvalue(res, i, 6),	/* attnum */
+					PQgetvalue(res, i, 7),	/* chunk */
+					PQgetvalue(res, i, 8)); /* msg */
+		}
+	}
+
+	PQclear(res);
+	resetPQExpBuffer(query);
+	destroyPQExpBuffer(query);
+}
+
+static void
+check_indexes(Oid tbloid)
+{
+	PQExpBuffer query;
+	PGresult   *res;
+	int			i;
+
+	query = createPQExpBuffer();
+	appendPQExpBuffer(query,
+					  "SELECT i.indexrelid"
+					  "\nFROM pg_catalog.pg_index i, pg_catalog.pg_class c"
+					  "\nWHERE i.indexrelid = c.oid"
+					  "\n    AND c.relam = %u"
+					  "\n    AND i.indrelid = %u",
+					  BTREE_AM_OID, tbloid);
+
+	ExecuteSqlStatement("RESET search_path");
+	res = ExecuteSqlQuery(query->data, PGRES_TUPLES_OK);
+	PQclear(ExecuteSqlQueryForSingleRow(ALWAYS_SECURE_SEARCH_PATH_SQL));
+
+	for (i = 0; i < PQntuples(res); i++)
+		check_index(tbloid, atooid(PQgetvalue(res, i, 0)));
+
+	PQclear(res);
+	resetPQExpBuffer(query);
+	destroyPQExpBuffer(query);
+}
+
+static void
+check_index(Oid tbloid, Oid idxoid)
+{
+	PQExpBuffer query;
+	PGresult   *res;
+	int			i;
+
+	query = createPQExpBuffer();
+
+	appendPQExpBuffer(query,
+					  "SELECT ct.relname, ci.relname, blkno, msg"
+					  "\nFROM verify_btreeam(%u,%s),"
+					  "\n     pg_catalog.pg_class ci,"
+					  "\n     pg_catalog.pg_class ct"
+					  "\nWHERE ci.oid = %u"
+					  "\n  AND ct.oid = %u",
+					  idxoid,
+					  settings.on_error_stop ? "true" : "false",
+					  idxoid, tbloid);
+
+	ExecuteSqlStatement("RESET search_path");
+	res = ExecuteSqlQuery(query->data, PGRES_TUPLES_OK);
+	PQclear(ExecuteSqlQueryForSingleRow(ALWAYS_SECURE_SEARCH_PATH_SQL));
+
+	if (PQntuples(res) > 0)
+	{
+		int			lines = PQntuples(res) * 2;
+		FILE	   *output = PageOutput(lines, NULL);
+
+		for (i = 0; i < PQntuples(res); i++)
+		{
+			fprintf(output,
+					"(table=%s,index=%s,blkno=%s)"
+					"\n%s\n",
+					PQgetvalue(res, i, 0),	/* table relname */
+					PQgetvalue(res, i, 1),	/* index relname */
+					PQgetvalue(res, i, 2),	/* index blkno */
+					PQgetvalue(res, i, 3)); /* msg */
+		}
+	}
+
+	PQclear(res);
+	resetPQExpBuffer(query);
+	destroyPQExpBuffer(query);
+}
+
+static void
+parse_cli_options(int argc, char *argv[], ConnectOptions * connOpts)
+{
+	static struct option long_options[] =
+	{
+		{"startblock", required_argument, NULL, 'b'},
+		{"dbname", required_argument, NULL, 'd'},
+		{"endblock", required_argument, NULL, 'e'},
+		{"host", required_argument, NULL, 'h'},
+		{"check-indexes", no_argument, NULL, 'i'},
+		{"exclude-indexes", no_argument, NULL, 'I'},
+		{"skip-all-visible", no_argument, NULL, 'v'},
+		{"skip-all-frozen", no_argument, NULL, 'f'},
+		{"schema", required_argument, NULL, 'n'},
+		{"exclude-schema", required_argument, NULL, 'N'},
+		{"on-error-stop", no_argument, NULL, 'o'},
+		{"port", required_argument, NULL, 'p'},
+		{"strict-names", no_argument, NULL, 's'},
+		{"table", required_argument, NULL, 't'},
+		{"exclude-table", required_argument, NULL, 'T'},
+		{"username", required_argument, NULL, 'U'},
+		{"version", no_argument, NULL, 'V'},
+		{"no-password", no_argument, NULL, 'w'},
+		{"password", no_argument, NULL, 'W'},
+		{"help", optional_argument, NULL, '?'},
+		{NULL, 0, NULL, 0}
+	};
+
+	int			optindex;
+	int			c;
+
+	memset(connOpts, 0, sizeof *connOpts);
+
+	while ((c = getopt_long(argc, argv, "b:d:e:fh:iIn:N:op:st:T:U:vVwW?1",
+							long_options, &optindex)) != -1)
+	{
+		switch (c)
+		{
+			case 'b':
+				settings.startblock = pg_strdup(optarg);
+				break;
+			case 'd':
+				connOpts->dbname = pg_strdup(optarg);
+				break;
+			case 'e':
+				settings.endblock = pg_strdup(optarg);
+				break;
+			case 'f':
+				settings.skip_frozen = true;
+				break;
+			case 'h':
+				connOpts->host = pg_strdup(optarg);
+				break;
+			case 'i':
+				settings.check_indexes = true;
+				break;
+			case 'I':
+				settings.check_indexes = false;
+				break;
+			case 'n':			/* include schema(s) */
+				simple_string_list_append(&schema_include_patterns, optarg);
+				break;
+			case 'N':			/* exclude schema(s) */
+				simple_string_list_append(&schema_exclude_patterns, optarg);
+				break;
+			case 'o':
+				settings.on_error_stop = true;
+				break;
+			case 'p':
+				connOpts->port = pg_strdup(optarg);
+				break;
+			case 's':
+				settings.strict_names = true;
+				break;
+			case 't':			/* include table(s) */
+				simple_string_list_append(&table_include_patterns, optarg);
+				break;
+			case 'T':			/* exclude table(s) */
+				simple_string_list_append(&table_exclude_patterns, optarg);
+				break;
+			case 'U':
+				connOpts->username = pg_strdup(optarg);
+				break;
+			case 'v':
+				settings.skip_visible = true;
+				break;
+			case 'V':
+				showVersion();
+				exit(EXIT_SUCCESS);
+			case 'w':
+				settings.getPassword = TRI_NO;
+				break;
+			case 'W':
+				settings.getPassword = TRI_YES;
+				break;
+			case '?':
+				if (optind <= argc &&
+					strcmp(argv[optind - 1], "-?") == 0)
+				{
+					/* actual help option given */
+					usage();
+					exit(EXIT_SUCCESS);
+				}
+				else
+				{
+					/* getopt error (unknown option or missing argument) */
+					goto unknown_option;
+				}
+				break;
+			case 1:
+				{
+					if (!optarg || strcmp(optarg, "options") == 0)
+						usage();
+					else
+						goto unknown_option;
+
+					exit(EXIT_SUCCESS);
+				}
+				break;
+			default:
+		unknown_option:
+				fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+						settings.progname);
+				exit(EXIT_FAILURE);
+				break;
+		}
+	}
+
+	/*
+	 * if we still have arguments, use it as the database name and username
+	 */
+	while (argc - optind >= 1)
+	{
+		if (!connOpts->dbname)
+			connOpts->dbname = argv[optind];
+		else if (!connOpts->username)
+			connOpts->username = argv[optind];
+		else
+			pg_log_warning("extra command-line argument \"%s\" ignored",
+						   argv[optind]);
+
+		optind++;
+	}
+
+}
+
+/*
+ * usage
+ *
+ * print out command line arguments
+ */
+static void
+usage(void)
+{
+	FILE	   *output;
+	int			lines;
+	int			lineno;
+
+	for (lines = 0; usage_text[lines]; lines++)
+		;
+	output = PageOutput(lines + 2, NULL);
+	for (lineno = 0; usage_text[lineno]; lineno++)
+		fprintf(output, "%s\n", usage_text[lineno]);
+	fprintf(output, "Report bugs to <%s>.\n", PACKAGE_BUGREPORT);
+	fprintf(output, "%s home page: <%s>\n", PACKAGE_NAME, PACKAGE_URL);
+
+	ClosePager(output);
+}
+
+static void
+showVersion(void)
+{
+	puts("pg_amcheck (PostgreSQL) " PG_VERSION);
+}
+
+/*
+ * for backend Notice messages (INFO, WARNING, etc)
+ */
+static void
+NoticeProcessor(void *arg, const char *message)
+{
+	(void) arg;					/* not used */
+	pg_log_info("%s", message);
+}
+
+/*
+ * Find the OIDs of all schemas matching the given list of patterns,
+ * and append them to the given OID list.
+ */
+static void
+expand_schema_name_patterns(SimpleStringList *patterns,
+							SimpleOidList *oids,
+							bool strict_names)
+{
+	PQExpBuffer query;
+	PGresult   *res;
+	SimpleStringListCell *cell;
+	int			i;
+
+	if (settings.db == NULL)
+		fatal("no connection on entry to expand_schema_name_patterns");
+
+	if (patterns->head == NULL)
+		return;					/* nothing to do */
+
+	query = createPQExpBuffer();
+
+	/*
+	 * The loop below runs multiple SELECTs might sometimes result in
+	 * duplicate entries in the OID list, but we don't care.
+	 */
+
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		appendPQExpBufferStr(query,
+							 "SELECT oid FROM pg_catalog.pg_namespace n\n");
+		processSQLNamePattern(settings.db, query, cell->val, false,
+							  false, NULL, "n.nspname", NULL, NULL);
+
+		res = ExecuteSqlQuery(query->data, PGRES_TUPLES_OK);
+		if (strict_names && PQntuples(res) == 0)
+			fatal("no matching schemas were found for pattern \"%s\"",
+				  cell->val);
+
+		for (i = 0; i < PQntuples(res); i++)
+		{
+			simple_oid_list_append(oids, atooid(PQgetvalue(res, i, 0)));
+		}
+
+		PQclear(res);
+		resetPQExpBuffer(query);
+	}
+
+	destroyPQExpBuffer(query);
+}
+
+/*
+ * Find the OIDs of all tables matching the given list of patterns,
+ * and append them to the given OID list. See also expand_dbname_patterns()
+ * in pg_dumpall.c
+ */
+static void
+expand_table_name_patterns(SimpleStringList *patterns, SimpleOidList *oids,
+						   bool strict_names)
+{
+	PQExpBuffer query;
+	PGresult   *res;
+	SimpleStringListCell *cell;
+	int			i;
+
+	if (settings.db == NULL)
+		fatal("no connection on entry to expand_table_name_patterns");
+
+	if (patterns->head == NULL)
+		return;					/* nothing to do */
+
+	query = createPQExpBuffer();
+
+	/*
+	 * this might sometimes result in duplicate entries in the OID list, but
+	 * we don't care.
+	 */
+
+	for (cell = patterns->head; cell; cell = cell->next)
+	{
+		/*
+		 * Query must remain ABSOLUTELY devoid of unqualified names.  This
+		 * would be unnecessary given a pg_table_is_visible() variant taking a
+		 * search_path argument.
+		 */
+		appendPQExpBuffer(query,
+						  "SELECT c.oid"
+						  "\nFROM pg_catalog.pg_class c"
+						  "\n    LEFT JOIN pg_catalog.pg_namespace n"
+						  "\n    ON n.oid OPERATOR(pg_catalog.=) c.relnamespace"
+						  "\nWHERE c.relkind OPERATOR(pg_catalog.=) ANY"
+						  "\n    (array['%c', '%c', '%c'])\n",
+						  RELKIND_RELATION, RELKIND_MATVIEW,
+						  RELKIND_PARTITIONED_TABLE);
+		processSQLNamePattern(settings.db, query, cell->val, true,
+							  false, "n.nspname", "c.relname", NULL, NULL);
+		ExecuteSqlStatement("RESET search_path");
+		res = ExecuteSqlQuery(query->data, PGRES_TUPLES_OK);
+		PQclear(ExecuteSqlQueryForSingleRow(ALWAYS_SECURE_SEARCH_PATH_SQL));
+		if (strict_names && PQntuples(res) == 0)
+			fatal("no matching tables were found for pattern \"%s\"",
+				  cell->val);
+
+		for (i = 0; i < PQntuples(res); i++)
+			simple_oid_list_append(oids, atooid(PQgetvalue(res, i, 0)));
+
+		PQclear(res);
+		resetPQExpBuffer(query);
+	}
+
+	destroyPQExpBuffer(query);
+}
+
+static void
+append_csv_oids(PQExpBuffer query, const SimpleOidList *oids)
+{
+	const SimpleOidListCell *cell;
+	const char *comma;
+
+	for (comma = "", cell = oids->head; cell; comma = ", ", cell = cell->next)
+		appendPQExpBuffer(query, "%s%u", comma, cell->val);
+}
+
+static bool
+append_filter(PQExpBuffer query, const char *lval, const char *operator,
+			  const SimpleOidList *oids)
+{
+	if (!oids->head)
+		return false;
+	appendPQExpBuffer(query, "\nAND %s %s ANY(array[\n", lval, operator);
+	append_csv_oids(query, oids);
+	appendPQExpBuffer(query, "\n])");
+	return true;
+}
+
+static void
+get_table_check_list(SimpleOidList *include_nsp, SimpleOidList *exclude_nsp,
+					 SimpleOidList *include_tbl, SimpleOidList *exclude_tbl,
+					 SimpleOidList *checklist)
+{
+	PQExpBuffer query;
+	PGresult   *res;
+	int			i;
+
+	if (settings.db == NULL)
+		fatal("no connection on entry to expand_table_name_patterns");
+
+	query = createPQExpBuffer();
+
+	appendPQExpBuffer(query,
+					  "SELECT c.oid"
+					  "\nFROM pg_catalog.pg_class c"
+					  "\n    LEFT JOIN pg_catalog.pg_namespace n"
+					  "\n    ON n.oid OPERATOR(pg_catalog.=) c.relnamespace"
+					  "\nWHERE c.relkind OPERATOR(pg_catalog.=) ANY"
+					  "\n    (array['%c', '%c', '%c'])\n",
+					  RELKIND_RELATION, RELKIND_MATVIEW,
+					  RELKIND_PARTITIONED_TABLE);
+	append_filter(query, "n.oid", "OPERATOR(pg_catalog.=)", include_nsp);
+	append_filter(query, "n.oid", "OPERATOR(pg_catalog.!=)", exclude_nsp);
+	append_filter(query, "c.oid", "OPERATOR(pg_catalog.=)", include_tbl);
+	append_filter(query, "c.oid", "OPERATOR(pg_catalog.!=)", exclude_tbl);
+
+	ExecuteSqlStatement("RESET search_path");
+	res = ExecuteSqlQuery(query->data, PGRES_TUPLES_OK);
+	PQclear(ExecuteSqlQueryForSingleRow(ALWAYS_SECURE_SEARCH_PATH_SQL));
+
+	for (i = 0; i < PQntuples(res); i++)
+		simple_oid_list_append(checklist, atooid(PQgetvalue(res, i, 0)));
+
+	PQclear(res);
+	resetPQExpBuffer(query);
+	destroyPQExpBuffer(query);
+}
+
+/* Like fatal(), but with a complaint about a particular query. */
+static void
+die_on_query_failure(const char *query)
+{
+	pg_log_error("query failed: %s",
+				 PQerrorMessage(settings.db));
+	fatal("query was: %s", query);
+}
+
+static void
+ExecuteSqlStatement(const char *query)
+{
+	PGresult   *res;
+
+	res = PQexec(settings.db, query);
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+		die_on_query_failure(query);
+	PQclear(res);
+}
+
+static PGresult *
+ExecuteSqlQuery(const char *query, ExecStatusType status)
+{
+	PGresult   *res;
+
+	res = PQexec(settings.db, query);
+	if (PQresultStatus(res) != status)
+		die_on_query_failure(query);
+	return res;
+}
+
+/*
+ * Execute an SQL query and verify that we got exactly one row back.
+ */
+static PGresult *
+ExecuteSqlQueryForSingleRow(const char *query)
+{
+	PGresult   *res;
+	int			ntups;
+
+	res = ExecuteSqlQuery(query, PGRES_TUPLES_OK);
+
+	/* Expecting a single result only */
+	ntups = PQntuples(res);
+	if (ntups != 1)
+		fatal(ngettext("query returned %d row instead of one: %s",
+					   "query returned %d rows instead of one: %s",
+					   ntups),
+			  ntups, query);
+
+	return res;
+}
diff --git a/contrib/pg_amcheck/t/001_basic.pl b/contrib/pg_amcheck/t/001_basic.pl
new file mode 100644
index 0000000000..dfa0ae9e06
--- /dev/null
+++ b/contrib/pg_amcheck/t/001_basic.pl
@@ -0,0 +1,9 @@
+use strict;
+use warnings;
+
+use TestLib;
+use Test::More tests => 8;
+
+program_help_ok('pg_amcheck');
+program_version_ok('pg_amcheck');
+program_options_handling_ok('pg_amcheck');
diff --git a/contrib/pg_amcheck/t/002_nonesuch.pl b/contrib/pg_amcheck/t/002_nonesuch.pl
new file mode 100644
index 0000000000..c63ba4452e
--- /dev/null
+++ b/contrib/pg_amcheck/t/002_nonesuch.pl
@@ -0,0 +1,55 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 12;
+
+# 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/\Qpg_amcheck: error: could not connect to server: FATAL:  database "qqq" does not exist\E/,
+	'connecting to a non-existent database');
+
+#########################################
+# Test connecting with a non-existent user
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', "$port", '-U=no_such_user' ],
+	qr/\Qpg_amcheck: error: could not connect to server: FATAL:  role "=no_such_user" does not exist\E/,
+	'connecting with a non-existent user');
+
+#########################################
+# Test checking a non-existent schema, table, and patterns with --strict-names
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', "$port", '-n', 'nonexistent' ],
+	qr/\Qpg_amcheck: error: no matching schemas were found\E/,
+	'checking a non-existent schema');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', "$port", '-t', 'nonexistent' ],
+	qr/\Qpg_amcheck: error: no matching tables were found\E/,
+	'checking a non-existent table');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', "$port", '--strict-names', '-n', 'nonexistent*' ],
+	qr/\Qpg_amcheck: error: no matching schemas were found for pattern\E/,
+	'no matching schemas');
+
+command_fails_like(
+	[ 'pg_amcheck', '-p', "$port", '--strict-names', '-t', 'nonexistent*' ],
+	qr/\Qpg_amcheck: error: no matching tables were found for pattern\E/,
+	'no matching tables');
diff --git a/contrib/pg_amcheck/t/003_check.pl b/contrib/pg_amcheck/t/003_check.pl
new file mode 100644
index 0000000000..01531e5c77
--- /dev/null
+++ b/contrib/pg_amcheck/t/003_check.pl
@@ -0,0 +1,85 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+use Test::More tests => 7;
+
+# 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));
+
+# Create schemas and tables for checking pg_amcheck's include
+# and exclude schema and table command line options
+$node->safe_psql('postgres', q(
+CREATE SCHEMA s1;
+CREATE SCHEMA s2;
+CREATE SCHEMA s3;
+CREATE TABLE s1.t1 (a TEXT);
+CREATE TABLE s1.t2 (a TEXT);
+CREATE TABLE s1.t3 (a TEXT);
+CREATE TABLE s2.t1 (a TEXT);
+CREATE TABLE s2.t2 (a TEXT);
+CREATE TABLE s2.t3 (a TEXT);
+CREATE TABLE s3.t1 (a TEXT);
+CREATE TABLE s3.t2 (a TEXT);
+CREATE TABLE s3.t3 (a TEXT);
+CREATE INDEX i1 ON s1.t1(a);
+CREATE INDEX i2 ON s1.t2(a);
+CREATE INDEX i3 ON s1.t3(a);
+CREATE INDEX i1 ON s2.t1(a);
+CREATE INDEX i2 ON s2.t2(a);
+CREATE INDEX i3 ON s2.t3(a);
+CREATE INDEX i1 ON s3.t1(a);
+CREATE INDEX i2 ON s3.t2(a);
+CREATE INDEX i3 ON s3.t3(a);
+INSERT INTO s1.t1 (a) (SELECT gs::TEXT FROM generate_series(1,10000) AS gs);
+));
+
+$node->command_ok(
+	[
+		'pg_amcheck', '-I', '-p', $port, 'postgres'
+	],
+	'pg_amcheck all schemas and tables');
+
+$node->command_ok(
+	[
+		'pg_amcheck', '-i', '-p', $port, 'postgres'
+	],
+	'pg_amcheck all schemas, tables and indexes');
+
+;$node->command_ok(
+	[
+		'pg_amcheck', '-p', $port, 'postgres', '-n', 's1'
+	],
+	'pg_amcheck all objects in schema s1');
+
+$node->command_ok(
+	[
+		'pg_amcheck', '-p', $port, 'postgres', '-N', 's1'
+	],
+	'pg_amcheck all objects not in schema s1');
+
+$node->command_ok(
+	[
+		'pg_amcheck', '-p', $port, 'postgres', '-i', '-n', 's*', '-t', 't1'
+	],
+	'pg_amcheck all tables named t1 and their indexes');
+
+$node->command_ok(
+	[
+		'pg_amcheck', '-I', '-p', $port, 'postgres', '-T', 't1'
+	],
+	'pg_amcheck all tables not named t1');
+
+$node->command_ok(
+	[
+		'pg_amcheck', '-I', '-p', $port, 'postgres', '-N', 's1', '-T', 't1'
+	],
+	'pg_amcheck all tables not named t1 nor in schema s1');
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..58d5ab88cb
--- /dev/null
+++ b/contrib/pg_amcheck/t/004_verify_heapam.pl
@@ -0,0 +1,434 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+
+use Test::More tests => 48;
+
+# This regression test demonstrates that the verify_heapam() function supplied
+# with the amcheck contrib module and depended upon by this 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 verify_heapam
+# 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");
+
+# 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);
+	));
+
+my $rel = $node->safe_psql('postgres', qq(SELECT pg_relation_filepath('public.test')));
+my $relpath = "$pgdata/$rel";
+
+use constant ROWCOUNT => 14;
+$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'));
+
+# 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', '--check-indexes', '-p', $port, 'postgres'],
+				  'pg_amcheck test table and index, prior to corruption');
+
+$node->stop;
+
+# 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;
+
+# 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 $file;
+open($file, '+<', $relpath);
+binmode $file;
+
+for (my $tupidx = 0; $tupidx < ROWCOUNT; $tupidx++)
+{
+	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;
+	}
+
+	if ($tupidx == 0)
+	{
+		# Corruptly set xmin < relfrozenxid
+		$tup->{t_xmin} = 3;
+		$tup->{t_infomask} &= ~HEAP_XMIN_COMMITTED;
+		$tup->{t_infomask} &= ~HEAP_XMIN_INVALID;
+	}
+	elsif ($tupidx == 1)
+	{
+		# Corruptly set xmin < relfrozenxid, further back
+		$tup->{t_xmin} = 4026531839;		# Note circularity of xid comparison
+		$tup->{t_infomask} &= ~HEAP_XMIN_COMMITTED;
+		$tup->{t_infomask} &= ~HEAP_XMIN_INVALID;
+	}
+	elsif ($tupidx == 2)
+	{
+		# Corruptly set xmax < relminmxid;
+		$tup->{t_xmax} = 4026531839;		# Note circularity of xid comparison
+		$tup->{t_infomask} &= ~HEAP_XMAX_INVALID;
+	}
+	elsif ($tupidx == 3)
+	{
+		# Corrupt the tuple t_hoff, but keep it aligned properly
+		$tup->{t_hoff} += 128;
+	}
+	elsif ($tupidx == 4)
+	{
+		# Corrupt the tuple t_hoff, wrong alignment
+		$tup->{t_hoff} += 3;
+	}
+	elsif ($tupidx == 5)
+	{
+		# Corrupt the tuple t_hoff, underflow but correct alignment
+		$tup->{t_hoff} -= 8;
+	}
+	elsif ($tupidx == 6)
+	{
+		# Corrupt the tuple t_hoff, underflow and wrong alignment
+		$tup->{t_hoff} -= 3;
+	}
+	elsif ($tupidx == 7)
+	{
+		# Corrupt the tuple to look like it has lots of attributes, not just 3
+		$tup->{t_infomask2} |= HEAP_NATTS_MASK;
+	}
+	elsif ($tupidx == 8)
+	{
+		# 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;
+	}
+	elsif ($tupidx == 9)
+	{
+		# 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;
+	}
+	elsif ($tupidx == 10)
+	{
+		# Corrupt the bits in column 'b' 1-byte varlena header
+		$tup->{b_header} = 0x80;
+	}
+	elsif ($tupidx == 11)
+	{
+		# Corrupt the bits in column 'c' toast pointer
+		$tup->{c6} = 41;
+		$tup->{c7} = 41;
+	}
+	elsif ($tupidx == 12)
+	{
+		# Set both HEAP_XMAX_LOCK_ONLY and HEAP_KEYS_UPDATED
+		$tup->{t_infomask} |= HEAP_XMAX_LOCK_ONLY;
+		$tup->{t_infomask2} |= HEAP_KEYS_UPDATED;
+	}
+	elsif ($tupidx == 13)
+	{
+		# Set both HEAP_XMAX_COMMITTED and HEAP_XMAX_IS_MULTI
+		$tup->{t_infomask} |= HEAP_XMAX_COMMITTED;
+		$tup->{t_infomask} |= HEAP_XMAX_IS_MULTI;
+	}
+	write_tuple($file, $offset, $tup);
+}
+close($file);
+
+# Run verify_heapam on the corrupted file
+$node->start;
+
+my $result = $node->safe_psql(
+			'postgres',
+			q(SELECT * FROM verify_heapam('test', on_error_stop := false, skip := NULL, startblock := NULL, endblock := NULL)));
+is ($result,
+"0|1|8128|1|58|||tuple xmin 3 precedes relfrozenxid $relfrozenxid
+0|2|8064|1|58|||tuple xmin 4026531839 precedes relfrozenxid $relfrozenxid
+0|3|8000|1|58|||tuple xmax 4026531839 precedes relfrozenxid $relfrozenxid
+0|4|7936|1|58|||tuple's 152 byte header size exceeds the 58 byte length of the entire tuple
+0|4|7936|1|58|||tuple without null values has user data offset 152 rather than the expected offset 24
+0|5|7872|1|58|||tuple's user data offset 27 not maximally aligned to 32
+0|5|7872|1|58|||tuple without null values has user data offset 27 rather than the expected offset 24
+0|6|7808|1|58|||tuple's header size is 16 bytes which is less than the 23 byte minimum valid header size
+0|6|7808|1|58|||tuple without null values has user data offset 16 rather than the expected offset 24
+0|7|7744|1|58|||tuple's header size is 21 bytes which is less than the 23 byte minimum valid header size
+0|7|7744|1|58|||tuple's user data offset 21 not maximally aligned to 24
+0|7|7744|1|58|||tuple without null values has user data offset 21 rather than the expected offset 24
+0|8|7680|1|58|||tuple has 2047 attributes in relation with only 3 attributes
+0|9|7616|1|58|||tuple with null values has user data offset 24 rather than the expected offset 280
+0|10|7552|1|58|||tuple has 67 attributes in relation with only 3 attributes
+0|11|7488|1|58|1||tuple attribute of length 4294967295 ends at offset 416848000, but tuple length is only 58
+0|12|7424|1|58|2|0|final chunk number 0 differs from expected value 6
+0|12|7424|1|58|2|0|toasted value missing from toast table
+0|13|7360|1|58|||tuple xmax marked incompatibly as keys updated and locked only
+0|14|7296|1|58|||tuple xmax 0 precedes relminmxid 1
+0|14|7296|1|58|||tuple xmax marked incompatibly as committed and as a multitransaction ID",
+"Expected verify_heapam output");
+
+# Each table corruption message is returned with a standard header, and we can
+# check for those headers to verify that corruption is being reported.  We can
+# also check for each individual corruption that we would expect to see.
+my @corruption_re = (
+
+	# standard header
+	qr/relname=test,blkno=\d*,offnum=\d*,lp_off=\d*,lp_flags=\d*,lp_len=\d*,attnum=\d*,chunk=\d*/,
+
+	# individual detected corruptions
+	qr/final chunk number \d+ differs from expected value \d+/,
+	qr/toasted value missing from toast table/,
+	qr/tuple attribute of length \d+ ends at offset \d+, but tuple length is only \d+/,
+	qr/tuple has \d+ attributes in relation with only \d+ attributes/,
+	qr/tuple with null values has user data offset \d+ rather than the expected offset \d+/,
+	qr/tuple without null values has user data offset \d+ rather than the expected offset \d+/,
+	qr/tuple xmax \d+ precedes relfrozenxid \d+/,
+	qr/tuple xmax \d+ precedes relminmxid \d+/,
+	qr/tuple xmax marked incompatibly as committed and as a multitransaction ID/,
+	qr/tuple xmax marked incompatibly as keys updated and locked only/,
+	qr/tuple xmin \d+ precedes relfrozenxid \d+/,
+	qr/tuple's \d+ byte header size exceeds the \d+ byte length of the entire tuple/,
+	qr/tuple's header size is \d+ bytes which is less than the \d+ byte minimum valid header size/,
+	qr/tuple's user data offset \d+ not maximally aligned to \d+/,
+);
+
+$node->command_like(
+	['pg_amcheck', '--exclude-indexes', '-p', $port, 'postgres'], $_,
+	"pg_amcheck reports: $_"
+	) for(@corruption_re);
+
+$node->teardown_node;
+$node->clean_node;
diff --git a/doc/src/sgml/contrib.sgml b/doc/src/sgml/contrib.sgml
index 261a559e81..f606e42fb9 100644
--- a/doc/src/sgml/contrib.sgml
+++ b/doc/src/sgml/contrib.sgml
@@ -118,6 +118,7 @@ CREATE EXTENSION <replaceable>module_name</replaceable>;
  &ltree;
  &pageinspect;
  &passwordcheck;
+ &pg_amcheck;
  &pgbuffercache;
  &pgcrypto;
  &pgfreespacemap;
diff --git a/doc/src/sgml/filelist.sgml b/doc/src/sgml/filelist.sgml
index 64b5da0070..10e1ca9663 100644
--- a/doc/src/sgml/filelist.sgml
+++ b/doc/src/sgml/filelist.sgml
@@ -131,6 +131,7 @@
 <!ENTITY oid2name        SYSTEM "oid2name.sgml">
 <!ENTITY pageinspect     SYSTEM "pageinspect.sgml">
 <!ENTITY passwordcheck   SYSTEM "passwordcheck.sgml">
+<!ENTITY pg_amcheck      SYSTEM "pg_amcheck.sgml">
 <!ENTITY pgbuffercache   SYSTEM "pgbuffercache.sgml">
 <!ENTITY pgcrypto        SYSTEM "pgcrypto.sgml">
 <!ENTITY pgfreespacemap  SYSTEM "pgfreespacemap.sgml">
diff --git a/doc/src/sgml/pg_amcheck.sgml b/doc/src/sgml/pg_amcheck.sgml
new file mode 100644
index 0000000000..a0b9c9d19b
--- /dev/null
+++ b/doc/src/sgml/pg_amcheck.sgml
@@ -0,0 +1,136 @@
+<!-- doc/src/sgml/pg_amcheck.sgml -->
+
+<sect1 id="pg_amcheck" xreflabel="pg_amcheck">
+ <title>pg_amcheck</title>
+
+ <indexterm zone="pg_amcheck">
+  <primary>pg_amcheck</primary>
+ </indexterm>
+
+ <para>
+  The <filename>pg_amcheck</filename> module provides a command line interface
+  to the <xref linkend="amcheck"/> corruption checking functionality.
+ </para>
+
+ <para>
+  <application>pg_amcheck</application> is a regular
+  <productname>PostgreSQL</productname> client application. You can perform
+  corruption checks from any remote host that has access to the database
+  connecting as a user with sufficient privileges to check tables and indexes.
+  Currently, this requires superuser privileges.
+ </para>
+
+ <sect2>
+  <title>Options</title>
+
+  <para>
+   To specify which database server <application>pg_amcheck</application> should
+   contact, use the command line options <option>-h</option> or
+   <option>--host</option> and <option>-p</option> or
+   <option>port</option>. The default host is the local host
+   or whatever your <envar>PGHOST</envar> environment variable specifies.
+   Similarly, the default port is indicated by the <envar>PGPORT</envar>
+   environment variable or, failing that, by the compiled-in default.
+  </para>
+ 
+  <para>
+   Like any other <productname>PostgreSQL</productname> client application,
+   <application>pg_amcheck</application> will by default connect with the
+   database user name that is equal to the current operating system user name.
+   To override this, either specify the <option>-U</option> option or set the
+   environment variable <envar>PGUSER</envar>. Remember that
+   <application>pg_amcheck</application> connections are subject to the normal
+   client authentication mechanisms (which are described in <xref
+   linkend="client-authentication"/>).
+  </para>
+ 
+  <para>
+   To restrict checking of tables and indexes to specific schemas, specify the
+   <option>-s</option> or <option>--schema</option> option with a pattern.
+   To exclude checking of tables and indexes within specific schemas, specify
+   the <option>-N</option> or <option>--exclude-schema</option> option with
+   a pattern.
+  </para>
+ 
+  <para>
+   To specify which tables are checked, specify the
+   <option>-t</option> or <option>--table</option> option with a pattern.
+   To exclude checking of tables, specify the
+   <option>-T</option> or <option>--exclude-table</option> option with a
+   pattern.
+  </para>
+ 
+  <para>
+   To check indexes associated with checked tables, specify the
+   <option>-i</option> or <option>--check-indexes</option> option.  Only
+   indexes on tables which are being checked will themselves be checked.  To
+   check all indexes in a database, all tables on which the indexes exist must
+   also be checked.  This restriction may be relaxed in the future.
+  </para>
+ 
+  <para>
+   To restrict the range of blocks within a table that are checked, specify the
+   <option>-b</option> or <option>--startblock</option> and/or
+   <option>-e</option> or <option>--endblock</option> options with numeric
+   values for the starting and ending block numbers.  Although these options
+   make the most sense when applied to a single table, if specified along with
+   options that select multiple tables, each table check will be restricted to
+   the specified blocks.  If <option>--startblock</option> is omitted, checking
+   begins with the first block. If <option>--endblock</option> is omitted,
+   checking continues to the end of the relation.
+  </para>
+ 
+  <para>
+   Some users may wish to periodically check tables without incurring the cost
+   of rechecking older table blocks, presumably because those blocks have
+   already been checked in the past.  There is at present no perfect way to do
+   this.  Although the <option>--startblock</option> and <option>--endblock</option>
+   options can be used to restrict blocks, the user is not expected to have
+   perfect knowledge of which blocks have already been checked, and in any
+   event, some blocks that were previously checked may have been subject to
+   modification since the last check.  As an approximation to the desired
+   functionality, one can specify the
+   <option>-f</option> or <option>--skip-all-frozen</option> option, or
+   alternatively the
+   <option>-v</option> or <option>--skip-all-visible</option> option to skip
+   blocks marked all frozen or all visible, respectively.
+  </para>
+ </sect2>
+
+ <sect2>
+  <title>Example Usage</title>
+
+  <para>
+   Checking an entire database which contains one corrupt table, "corrupted",
+   along with the output:
+  </para>
+
+<screen>
+% pg_amcheck -i test                                             
+(relname=corrupted,blkno=0,offnum=16,lp_off=7680,lp_flags=1,lp_len=31,attnum=,chunk=)
+tuple xmin = 3289393 is in the future
+(relname=corrupted,blkno=0,offnum=17,lp_off=7648,lp_flags=1,lp_len=31,attnum=,chunk=)
+tuple xmax = 0 precedes relation relminmxid = 1
+(relname=corrupted,blkno=0,offnum=17,lp_off=7648,lp_flags=1,lp_len=31,attnum=,chunk=)
+tuple xmin = 12593 is in the future
+</screen>
+
+  <para>
+   .... many pages of output removed for brevity ....
+  </para>
+
+<screen>
+(relname=corrupted,blkno=107,offnum=22,lp_off=7312,lp_flags=1,lp_len=34,attnum=,chunk=)
+tuple xmin = 305 precedes relation relfrozenxid = 487
+(relname=corrupted,blkno=107,offnum=22,lp_off=7312,lp_flags=1,lp_len=34,attnum=,chunk=)
+t_hoff > lp_len (54 > 34)
+(relname=corrupted,blkno=107,offnum=22,lp_off=7312,lp_flags=1,lp_len=34,attnum=,chunk=)
+t_hoff not max-aligned (54)
+</screen>
+
+  <para>
+   Each detected corruption is reported on two lines, the first line shows the
+   location and the second line shows a message describing the problem.
+  </para>
+ </sect2>
+</sect1>
-- 
2.21.1 (Apple Git-122.3)