v11-0002-Adding-function-verify_heapam-to-amcheck-module.patch

From a92f063a0deac08503d7f605207fca60542b325b Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Mon, 20 Jul 2020 12:37:48 -0700
Subject: [PATCH v11 2/3] Adding function verify_heapam to amcheck module

Adding new function verify_heapam for checking a heap relation and
associated toast relation, if any, to contrib/amcheck.
---
 contrib/amcheck/Makefile                      |    5 +-
 contrib/amcheck/amcheck--1.2--1.3.sql         |   54 +
 contrib/amcheck/amcheck.h                     |    5 +
 contrib/amcheck/expected/check_heap.out       |   67 +
 .../amcheck/expected/disallowed_reltypes.out  |   48 +
 contrib/amcheck/sql/check_heap.sql            |   19 +
 contrib/amcheck/sql/disallowed_reltypes.sql   |   48 +
 contrib/amcheck/t/001_verify_heapam.pl        |   94 ++
 contrib/amcheck/verify_heapam.c               | 1153 +++++++++++++++++
 doc/src/sgml/amcheck.sgml                     |  106 +-
 src/backend/access/heap/hio.c                 |   11 +
 11 files changed, 1608 insertions(+), 2 deletions(-)
 create mode 100644 contrib/amcheck/amcheck--1.2--1.3.sql
 create mode 100644 contrib/amcheck/amcheck.h
 create mode 100644 contrib/amcheck/expected/check_heap.out
 create mode 100644 contrib/amcheck/expected/disallowed_reltypes.out
 create mode 100644 contrib/amcheck/sql/check_heap.sql
 create mode 100644 contrib/amcheck/sql/disallowed_reltypes.sql
 create mode 100644 contrib/amcheck/t/001_verify_heapam.pl
 create mode 100644 contrib/amcheck/verify_heapam.c

diff --git a/contrib/amcheck/Makefile b/contrib/amcheck/Makefile
index b288c28fa0..27d38b2e86 100644
--- a/contrib/amcheck/Makefile
+++ b/contrib/amcheck/Makefile
@@ -3,13 +3,16 @@
 MODULE_big	= amcheck
 OBJS = \
 	$(WIN32RES) \
+	verify_heapam.o \
 	verify_nbtree.o
 
 EXTENSION = amcheck
 DATA = amcheck--1.2--1.3.sql amcheck--1.1--1.2.sql amcheck--1.0--1.1.sql amcheck--1.0.sql
 PGFILEDESC = "amcheck - function for verifying relation integrity"
 
-REGRESS = check check_btree
+REGRESS = check check_btree check_heap disallowed_reltypes
+
+TAP_TESTS = 1
 
 ifdef USE_PGXS
 PG_CONFIG = pg_config
diff --git a/contrib/amcheck/amcheck--1.2--1.3.sql b/contrib/amcheck/amcheck--1.2--1.3.sql
new file mode 100644
index 0000000000..df418a850b
--- /dev/null
+++ b/contrib/amcheck/amcheck--1.2--1.3.sql
@@ -0,0 +1,54 @@
+/* contrib/amcheck/amcheck--1.2--1.3.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "ALTER EXTENSION amcheck UPDATE TO '1.3'" to load this file. \quit
+
+-- In order to avoid issues with dependencies when updating amcheck to 1.3,
+-- create new, overloaded version of the 1.2 function signature
+
+--
+-- verify_heapam()
+--
+CREATE FUNCTION verify_heapam(rel regclass,
+							  on_error_stop boolean default false,
+							  skip cstring default 'none',
+							  startblock bigint default null,
+							  endblock bigint default null,
+							  blkno OUT bigint,
+							  offnum OUT integer,
+							  lp_off OUT smallint,
+							  lp_flags OUT smallint,
+							  lp_len OUT smallint,
+							  attnum OUT integer,
+							  chunk OUT integer,
+							  msg OUT text
+							  )
+RETURNS SETOF record
+AS 'MODULE_PATHNAME', 'verify_heapam'
+LANGUAGE C;
+
+-- Don't want this to be available to public
+REVOKE ALL ON FUNCTION verify_heapam(regclass, boolean, cstring, bigint, bigint)
+FROM PUBLIC;
+
+--
+-- verify_btreeam()
+--
+CREATE FUNCTION verify_btreeam(rel regclass,
+							   blkno OUT bigint,
+							   msg OUT text)
+RETURNS SETOF record
+AS 'MODULE_PATHNAME', 'verify_btreeam'
+LANGUAGE C;
+
+CREATE FUNCTION verify_btreeam(rel regclass,
+							   on_error_stop boolean,
+							   blkno OUT bigint,
+							   msg OUT text)
+RETURNS SETOF record
+AS 'MODULE_PATHNAME', 'verify_btreeam'
+LANGUAGE C;
+
+-- Don't want this to be available to public
+REVOKE ALL ON FUNCTION verify_btreeam(regclass) FROM PUBLIC;
+REVOKE ALL ON FUNCTION verify_btreeam(regclass, boolean) FROM PUBLIC;
diff --git a/contrib/amcheck/amcheck.h b/contrib/amcheck/amcheck.h
new file mode 100644
index 0000000000..74edfc2f65
--- /dev/null
+++ b/contrib/amcheck/amcheck.h
@@ -0,0 +1,5 @@
+#include "postgres.h"
+
+Datum		verify_heapam(PG_FUNCTION_ARGS);
+Datum		bt_index_check(PG_FUNCTION_ARGS);
+Datum		bt_index_parent_check(PG_FUNCTION_ARGS);
diff --git a/contrib/amcheck/expected/check_heap.out b/contrib/amcheck/expected/check_heap.out
new file mode 100644
index 0000000000..4175bb2d37
--- /dev/null
+++ b/contrib/amcheck/expected/check_heap.out
@@ -0,0 +1,67 @@
+CREATE TABLE heaptest (a integer, b text);
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'rope');
+ERROR:  unrecognized parameter for 'skip': rope
+HINT:  please choose from 'all-visible', 'all-frozen', or 'none'
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'none');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-frozen');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-visible');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 0, endblock := 0);
+ERROR:  starting block 0 is out of bounds for relation with no blocks
+INSERT INTO heaptest (a, b)
+	(SELECT gs, repeat('x', gs)
+		FROM generate_series(1,10000) gs);
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 100000, endblock := 200000);
+ERROR:  block range 100000 .. 200000 is out of bounds for relation with block count 370
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'none');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-frozen');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-visible');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 0, endblock := 0);
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+VACUUM FREEZE heaptest;
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'none');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-frozen');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-visible');
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 0, endblock := 0);
+ blkno | offnum | lp_off | lp_flags | lp_len | attnum | chunk | msg 
+-------+--------+--------+----------+--------+--------+-------+-----
+(0 rows)
+
diff --git a/contrib/amcheck/expected/disallowed_reltypes.out b/contrib/amcheck/expected/disallowed_reltypes.out
new file mode 100644
index 0000000000..892ae89652
--- /dev/null
+++ b/contrib/amcheck/expected/disallowed_reltypes.out
@@ -0,0 +1,48 @@
+--
+-- check that using the module's functions with unsupported relations will fail
+--
+-- partitioned tables (the parent ones) don't have visibility maps
+create table test_partitioned (a int, b text default repeat('x', 5000))
+			 partition by list (a);
+-- these should all fail
+select * from verify_heapam('test_partitioned',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+ERROR:  "test_partitioned" is not a table, materialized view, or TOAST table
+create table test_partition partition of test_partitioned for values in (1);
+create index test_index on test_partition (a);
+-- indexes do not, so these all fail
+select * from verify_heapam('test_index',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+ERROR:  "test_index" is not a table, materialized view, or TOAST table
+create view test_view as select 1;
+-- views do not have vms, so these all fail
+select * from verify_heapam('test_view',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+ERROR:  "test_view" is not a table, materialized view, or TOAST table
+create sequence test_sequence;
+-- sequences do not have vms, so these all fail
+select * from verify_heapam('test_sequence',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+ERROR:  "test_sequence" is not a table, materialized view, or TOAST table
+create foreign data wrapper dummy;
+create server dummy_server foreign data wrapper dummy;
+create foreign table test_foreign_table () server dummy_server;
+-- foreign tables do not have vms, so these all fail
+select * from verify_heapam('test_foreign_table',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+ERROR:  "test_foreign_table" is not a table, materialized view, or TOAST table
diff --git a/contrib/amcheck/sql/check_heap.sql b/contrib/amcheck/sql/check_heap.sql
new file mode 100644
index 0000000000..c75f5ff869
--- /dev/null
+++ b/contrib/amcheck/sql/check_heap.sql
@@ -0,0 +1,19 @@
+CREATE TABLE heaptest (a integer, b text);
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'rope');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'none');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-frozen');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-visible');
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 0, endblock := 0);
+INSERT INTO heaptest (a, b)
+	(SELECT gs, repeat('x', gs)
+		FROM generate_series(1,10000) gs);
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 100000, endblock := 200000);
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'none');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-frozen');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-visible');
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 0, endblock := 0);
+VACUUM FREEZE heaptest;
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'none');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-frozen');
+SELECT * FROM verify_heapam(rel := 'heaptest', skip := 'all-visible');
+SELECT * FROM verify_heapam(rel := 'heaptest', startblock := 0, endblock := 0);
diff --git a/contrib/amcheck/sql/disallowed_reltypes.sql b/contrib/amcheck/sql/disallowed_reltypes.sql
new file mode 100644
index 0000000000..fc90e6ca33
--- /dev/null
+++ b/contrib/amcheck/sql/disallowed_reltypes.sql
@@ -0,0 +1,48 @@
+--
+-- check that using the module's functions with unsupported relations will fail
+--
+
+-- partitioned tables (the parent ones) don't have visibility maps
+create table test_partitioned (a int, b text default repeat('x', 5000))
+			 partition by list (a);
+-- these should all fail
+select * from verify_heapam('test_partitioned',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+
+create table test_partition partition of test_partitioned for values in (1);
+create index test_index on test_partition (a);
+-- indexes do not, so these all fail
+select * from verify_heapam('test_index',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+
+create view test_view as select 1;
+-- views do not have vms, so these all fail
+select * from verify_heapam('test_view',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+
+create sequence test_sequence;
+-- sequences do not have vms, so these all fail
+select * from verify_heapam('test_sequence',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
+
+create foreign data wrapper dummy;
+create server dummy_server foreign data wrapper dummy;
+create foreign table test_foreign_table () server dummy_server;
+-- foreign tables do not have vms, so these all fail
+select * from verify_heapam('test_foreign_table',
+							on_error_stop := false,
+							skip := NULL,
+							startblock := NULL,
+							endblock := NULL);
diff --git a/contrib/amcheck/t/001_verify_heapam.pl b/contrib/amcheck/t/001_verify_heapam.pl
new file mode 100644
index 0000000000..c2d890bcd9
--- /dev/null
+++ b/contrib/amcheck/t/001_verify_heapam.pl
@@ -0,0 +1,94 @@
+use strict;
+use warnings;
+
+use PostgresNode;
+use TestLib;
+
+use Test::More tests => 48;
+
+my ($node, $result);
+
+# Check various options are stable (don't abort) when running verify_heapam on
+# the test table.  For uncorrupted tables, there isn't anything to check except
+# that it runs without crashing.
+sub check_all_options
+{
+	for my $stop (qw(NULL true false))
+	{
+		for my $skip ("'none'", "'all-frozen'", "'all-visible'")
+		{
+			my $check = "SELECT verify_heapam('test', $stop, $skip)";
+			$result = $node->safe_psql('postgres', "$check; SELECT 1");
+			is ($result, 1, "checked: $check");
+		}
+	}
+}
+
+# Stops the server and writes nulls in the first page of the table,
+# assuming page size is large enough for offset 1000..1016 to be
+# in the midst of the first page of data.
+sub corrupt_first_page
+{
+	my $pgdata = $node->data_dir;
+	my $rel = $node->safe_psql('postgres',
+							   qq(SELECT pg_relation_filepath('test')));
+	my $relpath = "$pgdata/$rel";
+	$node->stop;
+
+	my $fh;
+	open($fh, '+<', $relpath);
+	binmode $fh;
+	seek($fh, 1000, 0);
+	syswrite($fh, '\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0', 16);
+	close($fh);
+
+	$node->start;
+}
+
+# Test set-up
+$node = get_new_node('test');
+$node->init;
+$node->append_conf('postgresql.conf', 'autovacuum=off');
+$node->start;
+$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
+
+# Check empty table
+$node->safe_psql('postgres', q(
+	CREATE TABLE test (a integer);
+	ALTER TABLE public.test SET (autovacuum_enabled=false);
+));
+check_all_options();
+
+# Check table with trivial data
+$node->safe_psql('postgres', q(INSERT INTO test VALUES (0)));
+check_all_options();
+
+# Check table with non-trivial data (more than a page worth) but
+# without any all-frozen or all-visible
+$node->safe_psql('postgres', q(
+INSERT INTO test SELECT generate_series(1,10000)));
+check_all_options();
+
+# Check table with all-visible data
+$node->safe_psql('postgres', q(VACUUM test));
+check_all_options();
+
+# Check table with all-frozen data
+$node->safe_psql('postgres', q(VACUUM FREEZE test));
+check_all_options();
+
+# Check table with corruption, no skipping
+corrupt_first_page();
+$result = $node->safe_psql('postgres', q(
+SELECT COUNT(*) > 0 FROM verify_heapam('test', on_error_stop := false, skip := NULL, startblock := NULL, endblock := NULL)));
+is($result, 't', 'corruption detected on first page');
+
+# Check table with corruption, skipping all-visible blocks
+$result = $node->safe_psql('postgres', q(
+SELECT COUNT(*) > 0 FROM verify_heapam('test', on_error_stop := false, skip := 'all-visible', startblock := NULL, endblock := NULL)));
+is($result, 'f', 'skipping all-visible first page');
+
+# Check table with corruption, skipping all-frozen blocks
+$result = $node->safe_psql('postgres', q(
+SELECT COUNT(*) > 0 FROM verify_heapam('test', on_error_stop := false, skip := 'all-frozen', startblock := NULL, endblock := NULL)));
+is($result, 'f', 'skipping all-frozen first page');
diff --git a/contrib/amcheck/verify_heapam.c b/contrib/amcheck/verify_heapam.c
new file mode 100644
index 0000000000..007241d333
--- /dev/null
+++ b/contrib/amcheck/verify_heapam.c
@@ -0,0 +1,1153 @@
+/*-------------------------------------------------------------------------
+ *
+ * verify_heapam.c
+ *	  Functions to check postgresql heap relations for corruption
+ *
+ * Copyright (c) 2016-2020, PostgreSQL Global Development Group
+ *
+ *	  contrib/amcheck/verify_heapam.c
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/detoast.h"
+#include "access/genam.h"
+#include "access/heapam.h"
+#include "access/heaptoast.h"
+#include "access/htup_details.h"
+#include "access/multixact.h"
+#include "access/toast_internals.h"
+#include "access/visibilitymap.h"
+#include "access/xact.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_type.h"
+#include "catalog/storage_xlog.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "storage/bufmgr.h"
+#include "storage/procarray.h"
+#include "storage/smgr.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+#include "utils/syscache.h"
+#include "amcheck.h"
+
+PG_FUNCTION_INFO_V1(verify_heapam);
+
+/* The number of columns in tuples returned by verify_heapam */
+#define HEAPCHECK_RELATION_COLS 8
+
+/*
+ * Struct holding the running context information during
+ * a lifetime of a verify_heapam execution.
+ */
+typedef struct HeapCheckContext
+{
+	/*
+	 * While verifying a table, we check whether any xid we encounter is
+	 * either too old or too new.  We could naively check that by taking the
+	 * XidGenLock each time and reading ShmemVariableCache.  We instead cache
+	 * the values and rely on the fact that we have the table locked
+	 * sufficiently that the oldest xid in the table cannot change
+	 * mid-verification, and although the newest xid in the table may advance,
+	 * it cannot retreat.  As such, whenever we encounter an xid older than
+	 * our cached oldest xid, we know it is invalid, and when we encounter an
+	 * xid newer than our cached newest xid, we recheck the
+	 * ShmemVariableCache.
+	 */
+	TransactionId nextKnownValidXid;
+	TransactionId oldestValidXid;
+
+	/* Values concerning the heap relation being checked */
+	Relation	rel;
+	TransactionId relfrozenxid;
+	TransactionId relminmxid;
+	Relation	toastrel;
+	Relation   *toast_indexes;
+	Relation	valid_toast_index;
+	int			num_toast_indexes;
+
+	/* Values for iterating over pages in the relation */
+	BlockNumber nblocks;
+	BlockNumber blkno;
+	BufferAccessStrategy bstrategy;
+	Buffer		buffer;
+	Page		page;
+
+	/* Values for iterating over tuples within a page */
+	OffsetNumber offnum;
+	ItemId		itemid;
+	uint16		lp_len;
+	HeapTupleHeader tuphdr;
+	int			natts;
+
+	/* Values for iterating over attributes within the tuple */
+	uint32		offset;			/* offset in tuple data */
+	AttrNumber	attnum;
+
+	/* Values for iterating over toast for the attribute */
+	int32		chunkno;
+	int32		attrsize;
+	int32		endchunk;
+	int32		totalchunks;
+
+	/* Whether verify_heapam has yet encountered any corrupt tuples */
+	bool		is_corrupt;
+
+	/* The descriptor and tuplestore for verify_heapam's result tuples */
+	TupleDesc	tupdesc;
+	Tuplestorestate *tupstore;
+}			HeapCheckContext;
+
+/* Internal implementation */
+static void check_relation_relkind_and_relam(Relation rel);
+
+static void confess(HeapCheckContext * ctx, char *msg);
+static TupleDesc verify_heapam_tupdesc(void);
+
+static bool TransactionIdValidInRel(TransactionId xid, HeapCheckContext * ctx);
+static bool tuple_is_visible(HeapTupleHeader tuphdr, HeapCheckContext * ctx);
+static void check_toast_tuple(HeapTuple toasttup, HeapCheckContext * ctx);
+static bool check_tuple_attribute(HeapCheckContext * ctx);
+static void check_tuple(HeapCheckContext * ctx);
+
+typedef enum SkipPages
+{
+	SKIP_ALL_FROZEN_PAGES,
+	SKIP_ALL_VISIBLE_PAGES,
+	SKIP_PAGES_NONE
+}			SkipPages;
+
+/*
+ * verify_heapam
+ *
+ *   Scan and report corruption in heap pages or in associated toast relation.
+ */
+Datum
+verify_heapam(PG_FUNCTION_ARGS)
+{
+	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	MemoryContext oldcontext;
+	bool		randomAccess;
+	HeapCheckContext ctx;
+	FullTransactionId nextFullXid;
+	Buffer		vmbuffer = InvalidBuffer;
+	Oid			relid;
+	bool		fatal = false;
+	bool		on_error_stop;
+	SkipPages	skip_option = SKIP_PAGES_NONE;
+	int64		startblock;
+	int64		endblock;
+
+	/* check to see if caller supports us returning a tuplestore */
+	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("set-valued function called in context that cannot "
+						"accept a set")));
+	if (!(rsinfo->allowedModes & SFRM_Materialize))
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("materialize mode required, but it is not allowed "
+						"in this context")));
+
+	/* check supplied arguments */
+	if (PG_ARGISNULL(0))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("missing required parameter for 'rel'")));
+	relid = PG_GETARG_OID(0);
+	on_error_stop = PG_ARGISNULL(1) ? false : PG_GETARG_BOOL(1);
+	if (!PG_ARGISNULL(2))
+	{
+		const char *skip = PG_GETARG_CSTRING(2);
+
+		if (pg_strcasecmp(skip, "all-visible") == 0)
+			skip_option = SKIP_ALL_VISIBLE_PAGES;
+		else if (pg_strcasecmp(skip, "all-frozen") == 0)
+			skip_option = SKIP_ALL_FROZEN_PAGES;
+		else if (pg_strcasecmp(skip, "none") == 0)
+			skip_option = SKIP_PAGES_NONE;
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("unrecognized parameter for 'skip': %s", skip),
+					 errhint("please choose from 'all-visible', 'all-frozen', or 'none'")));
+	}
+
+	memset(&ctx, 0, sizeof(HeapCheckContext));
+
+	/* The tupdesc and tuplestore must be created in ecxt_per_query_memory */
+	oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
+	randomAccess = (rsinfo->allowedModes & SFRM_Materialize_Random) != 0;
+	ctx.tupdesc = verify_heapam_tupdesc();
+	ctx.tupstore = tuplestore_begin_heap(randomAccess, false, work_mem);
+	rsinfo->returnMode = SFRM_Materialize;
+	rsinfo->setResult = ctx.tupstore;
+	rsinfo->setDesc = ctx.tupdesc;
+
+	MemoryContextSwitchTo(oldcontext);
+
+	/*
+	 * Open the relation.  We use ShareUpdateExclusive to prevent concurrent
+	 * vacuums from changing the relfrozenxid, relminmxid, or advancing the
+	 * global oldestXid to be newer than those.
+	 */
+	ctx.rel = relation_open(relid, ShareUpdateExclusiveLock);
+	check_relation_relkind_and_relam(ctx.rel);
+	ctx.nblocks = RelationGetNumberOfBlocks(ctx.rel);
+
+	/* Early exit if the relation is empty */
+	if (!ctx.nblocks)
+	{
+		/*
+		 * For consistency, we need to enforce that the startblock and
+		 * endblock are within the valid range if the user specified them.
+		 * Yet, for an empty table with no blocks, no specified block can be
+		 * in range.
+		 */
+		if (!PG_ARGISNULL(3))
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("starting block " INT64_FORMAT
+							" is out of bounds for relation with no blocks",
+							PG_GETARG_INT64(3))));
+		if (!PG_ARGISNULL(4))
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("ending block " INT64_FORMAT
+							" is out of bounds for relation with no blocks",
+							PG_GETARG_INT64(4))));
+		relation_close(ctx.rel, ShareUpdateExclusiveLock);
+		PG_RETURN_NULL();
+	}
+
+	ctx.bstrategy = GetAccessStrategy(BAS_BULKREAD);
+	ctx.buffer = InvalidBuffer;
+	ctx.page = NULL;
+
+	/* If we get this far, we know the relation has at least one block */
+	startblock = PG_ARGISNULL(3) ? 0 : PG_GETARG_INT64(3);
+	endblock = PG_ARGISNULL(4) ? ((int64) ctx.nblocks) - 1 : PG_GETARG_INT64(4);
+	if (startblock < 0 || endblock >= ctx.nblocks || startblock > endblock)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("block range " INT64_FORMAT " .. " INT64_FORMAT
+						" is out of bounds for relation with block count %u",
+						startblock, endblock, ctx.nblocks)));
+
+	/*
+	 * Open the toast relation, if any, also protected from concurrent
+	 * vacuums.
+	 */
+	if (ctx.rel->rd_rel->reltoastrelid)
+	{
+		int			offset;
+
+		/* Main relation has associated toast relation */
+		ctx.toastrel = table_open(ctx.rel->rd_rel->reltoastrelid,
+								  ShareUpdateExclusiveLock);
+		offset = toast_open_indexes(ctx.toastrel,
+									ShareUpdateExclusiveLock,
+									&(ctx.toast_indexes),
+									&(ctx.num_toast_indexes));
+		ctx.valid_toast_index = ctx.toast_indexes[offset];
+	}
+	else
+	{
+		/* Main relation has no associated toast relation */
+		ctx.toastrel = NULL;
+		ctx.toast_indexes = NULL;
+		ctx.num_toast_indexes = 0;
+	}
+
+	/*
+	 * Now that we have our relation(s) locked, oldestXid cannot advance
+	 * beyond the oldest valid xid in our table, nor can our relfrozenxid
+	 * advance.
+	 *
+	 * If relfrozenxid is normal, it contains the oldest valid xid we may
+	 * encounter in the table.  If not, the oldest xid for our database is the
+	 * oldest we should encounter.
+	 *
+	 * Bugs in pg_upgrade are reported (see commands/vacuum.c circa line 1572)
+	 * to have sometimes rendered the oldest xid value for a database invalid.
+	 * It seems unwise to report rows as corrupt for failing to be newer than
+	 * a value which itself may be corrupt.  We instead use the oldest xid for
+	 * the entire cluster, which must be at least as old as the oldest xid for
+	 * our database.
+	 *
+	 * If neither the value for the database nor the xids for any row are
+	 * corrupt, then this gives the right answer.  If the rows disagree with
+	 * the value for the database, how can we know which one is wrong?
+	 */
+	ctx.relfrozenxid = ctx.rel->rd_rel->relfrozenxid;
+	ctx.relminmxid = ctx.rel->rd_rel->relminmxid;
+
+	LWLockAcquire(XidGenLock, LW_SHARED);
+	nextFullXid = ShmemVariableCache->nextFullXid;
+	ctx.oldestValidXid = ShmemVariableCache->oldestXid;
+	LWLockRelease(XidGenLock);
+	ctx.nextKnownValidXid = XidFromFullTransactionId(nextFullXid);
+
+	if (TransactionIdIsNormal(ctx.relfrozenxid) &&
+		TransactionIdPrecedes(ctx.relfrozenxid, ctx.oldestValidXid))
+	{
+		confess(&ctx, psprintf("relfrozenxid %u precedes global oldest valid xid %u",
+							   ctx.relfrozenxid, ctx.oldestValidXid));
+		fatal = true;
+	}
+	else if (TransactionIdIsNormal(ctx.relminmxid) &&
+			 TransactionIdPrecedes(ctx.relminmxid, ctx.oldestValidXid))
+	{
+		confess(&ctx, psprintf("relminmxid %u precedes global oldest valid xid %u",
+							   ctx.relminmxid, ctx.oldestValidXid));
+		fatal = true;
+	}
+
+	if (fatal)
+	{
+		if (ctx.toast_indexes)
+			toast_close_indexes(ctx.toast_indexes, ctx.num_toast_indexes,
+								ShareUpdateExclusiveLock);
+		if (ctx.toastrel)
+			table_close(ctx.toastrel, ShareUpdateExclusiveLock);
+		relation_close(ctx.rel, ShareUpdateExclusiveLock);
+		PG_RETURN_NULL();
+	}
+
+	if (TransactionIdIsNormal(ctx.relfrozenxid))
+		ctx.oldestValidXid = ctx.relfrozenxid;
+
+	if (startblock < 0)
+		startblock = 0;
+	if (endblock < 0 || endblock > ctx.nblocks)
+		endblock = ctx.nblocks;
+
+	for (ctx.blkno = startblock; ctx.blkno <= endblock; ctx.blkno++)
+	{
+		int32		mapbits;
+		OffsetNumber maxoff;
+		PageHeader	ph;
+
+		/* Optionally skip over all-frozen or all-visible blocks */
+		if (skip_option != SKIP_PAGES_NONE)
+		{
+			bool		all_frozen,
+						all_visible;
+
+			mapbits = (int32) visibilitymap_get_status(ctx.rel, ctx.blkno,
+													   &vmbuffer);
+			all_frozen = mapbits & VISIBILITYMAP_ALL_VISIBLE;
+			all_visible = mapbits & VISIBILITYMAP_ALL_FROZEN;
+
+			if ((all_frozen && skip_option == SKIP_ALL_FROZEN_PAGES) ||
+				(all_visible && skip_option == SKIP_ALL_VISIBLE_PAGES))
+			{
+				continue;
+			}
+		}
+
+		/* Read and lock the next page. */
+		ctx.buffer = ReadBufferExtended(ctx.rel, MAIN_FORKNUM, ctx.blkno,
+										RBM_NORMAL, ctx.bstrategy);
+		LockBuffer(ctx.buffer, BUFFER_LOCK_SHARE);
+		ctx.page = BufferGetPage(ctx.buffer);
+		ph = (PageHeader) ctx.page;
+
+		/* We rely on this math property for the first iteration */
+		StaticAssertStmt(InvalidOffsetNumber + 1 == FirstOffsetNumber,
+						 "InvalidOffsetNumber increments to FirstOffsetNumber");
+
+		ctx.offnum = InvalidOffsetNumber;
+		ctx.itemid = NULL;
+		ctx.lp_len = 0;
+		ctx.tuphdr = NULL;
+		ctx.natts = 0;
+
+		/* Perform tuple checks */
+		maxoff = PageGetMaxOffsetNumber(ctx.page);
+		for (ctx.offnum = FirstOffsetNumber; ctx.offnum <= maxoff;
+			 ctx.offnum = OffsetNumberNext(ctx.offnum))
+		{
+			ctx.itemid = PageGetItemId(ctx.page, ctx.offnum);
+
+			/* Skip over unused/dead line pointers */
+			if (!ItemIdIsUsed(ctx.itemid) || ItemIdIsDead(ctx.itemid))
+				continue;
+
+			/*
+			 * If this line pointer has been redirected, check that it
+			 * redirects to a valid offset within the line pointer array.
+			 */
+			if (ItemIdIsRedirected(ctx.itemid))
+			{
+				OffsetNumber rdoffnum = ItemIdGetRedirect(ctx.itemid);
+				ItemId		rditem;
+
+				if (rdoffnum < FirstOffsetNumber || rdoffnum > maxoff)
+				{
+					confess(&ctx, psprintf(
+										   "line pointer redirection to item at offset number %u is outside valid bounds %u .. %u",
+										   (unsigned) rdoffnum, (unsigned) FirstOffsetNumber,
+										   (unsigned) maxoff));
+					continue;
+				}
+				rditem = PageGetItemId(ctx.page, rdoffnum);
+				if (!ItemIdIsUsed(rditem))
+					confess(&ctx, psprintf(
+										   "line pointer redirection to unused item at offset %u",
+										   (unsigned) rdoffnum));
+				continue;
+			}
+
+			/* Set up context information about this next tuple */
+			ctx.lp_len = ItemIdGetLength(ctx.itemid);
+			ctx.tuphdr = (HeapTupleHeader) PageGetItem(ctx.page, ctx.itemid);
+			ctx.natts = HeapTupleHeaderGetNatts(ctx.tuphdr);
+
+			/*
+			 * Reset information about individual attributes and related toast
+			 * values, so they show as NULL in the corruption report if we
+			 * record a corruption before beginning to iterate over the
+			 * attributes.
+			 */
+			ctx.attnum = -1;
+			ctx.chunkno = -1;
+
+			/* Ok, ready to check this next tuple */
+			check_tuple(&ctx);
+		}
+
+		/* clean up */
+		ctx.itemid = NULL;
+		ctx.lp_len = 0;
+		UnlockReleaseBuffer(ctx.buffer);
+
+		if (on_error_stop && ctx.is_corrupt)
+			break;
+	}
+
+	if (vmbuffer != InvalidBuffer)
+		ReleaseBuffer(vmbuffer);
+
+	/* Close the associated toast table and indexes, if any. */
+	if (ctx.rel->rd_rel->reltoastrelid)
+	{
+		toast_close_indexes(ctx.toast_indexes, ctx.num_toast_indexes,
+							ShareUpdateExclusiveLock);
+		table_close(ctx.toastrel, ShareUpdateExclusiveLock);
+	}
+
+	/* Close the main relation */
+	relation_close(ctx.rel, ShareUpdateExclusiveLock);
+
+	PG_RETURN_NULL();
+}
+
+/*
+ * check_relation_relkind_and_relam
+ *
+ *   convenience routine to check that relation is of a supported relkind.
+ */
+static void
+check_relation_relkind_and_relam(Relation rel)
+{
+	if (rel->rd_rel->relkind != RELKIND_RELATION &&
+		rel->rd_rel->relkind != RELKIND_MATVIEW &&
+		rel->rd_rel->relkind != RELKIND_TOASTVALUE)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("\"%s\" is not a table, materialized view, or TOAST table",
+						RelationGetRelationName(rel))));
+	if (rel->rd_rel->relam != HEAP_TABLE_AM_OID)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("\"%s\" is not a heap",
+						RelationGetRelationName(rel))));
+}
+
+/*
+ * confess
+ *
+ *   Return a message about corruption, including information
+ *   about where in the relation the corruption was found.
+ *
+ *   The msg argument is pfree'd by this function.
+ */
+static void
+confess(HeapCheckContext * ctx, char *msg)
+{
+	Datum		values[HEAPCHECK_RELATION_COLS];
+	bool		nulls[HEAPCHECK_RELATION_COLS];
+	HeapTuple	tuple;
+	int16		lp_off = ItemIdGetOffset(ctx->itemid);
+	int16		lp_flags = ItemIdGetFlags(ctx->itemid);
+	int16		lp_len = ItemIdGetLength(ctx->itemid);
+
+	MemSet(values, 0, sizeof(values));
+	MemSet(nulls, 0, sizeof(nulls));
+	values[0] = Int64GetDatum(ctx->blkno);
+	values[1] = Int32GetDatum(ctx->offnum);
+	nulls[1] = (ctx->offnum < 0);
+	values[2] = Int16GetDatum(lp_off);
+	nulls[2] = (lp_off < 0);
+	values[3] = Int16GetDatum(lp_flags);
+	nulls[3] = (lp_flags < 0);
+	values[4] = Int16GetDatum(lp_len);
+	nulls[4] = (lp_len < 0);
+	values[5] = Int32GetDatum(ctx->attnum);
+	nulls[5] = (ctx->attnum < 0);
+	values[6] = Int32GetDatum(ctx->chunkno);
+	nulls[6] = (ctx->chunkno < 0);
+	values[7] = CStringGetTextDatum(msg);
+
+	/*
+	 * In principle, there is nothing to prevent a scan over a large, highly
+	 * corrupted table from using workmem worth of memory building up the
+	 * tuplestore.  That's ok, but if we also leak the msg argument memory
+	 * until the end of the query, we could exceed workmem by more than a
+	 * trivial amount.  Therefore, free the msg argument each time we are
+	 * called rather than waiting for our current memory context to be freed.
+	 */
+	pfree(msg);
+
+	tuple = heap_form_tuple(ctx->tupdesc, values, nulls);
+	tuplestore_puttuple(ctx->tupstore, tuple);
+	ctx->is_corrupt = true;
+}
+
+/*
+ * Helper function to construct the TupleDesc needed by verify_heapam.
+ */
+static TupleDesc
+verify_heapam_tupdesc(void)
+{
+	TupleDesc	tupdesc;
+	AttrNumber	a = 0;
+
+	tupdesc = CreateTemplateTupleDesc(HEAPCHECK_RELATION_COLS);
+	TupleDescInitEntry(tupdesc, ++a, "blkno", INT8OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "offnum", INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "lp_off", INT2OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "lp_flags", INT2OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "lp_len", INT2OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "attnum", INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "chunk", INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, ++a, "msg", TEXTOID, -1, 0);
+	Assert(a == HEAPCHECK_RELATION_COLS);
+
+	return BlessTupleDesc(tupdesc);
+}
+
+static inline bool
+XidInValidRange(TransactionId xid, HeapCheckContext * ctx)
+{
+	return (TransactionIdPrecedesOrEquals(ctx->oldestValidXid, xid) &&
+			TransactionIdPrecedes(xid, ctx->nextKnownValidXid));
+}
+
+/*
+ * Given a TransactionId, attempt to interpret it as a valid
+ * FullTransactionId, neither in the future nor overlong in
+ * the past.  Stores the inferred FullTransactionId in *fxid.
+ *
+ * Returns whether the xid is newer than the oldest clog xid.
+ */
+static bool
+TransactionIdValidInRel(TransactionId xid, HeapCheckContext * ctx)
+{
+	/* Quick return for special oids */
+	switch (xid)
+	{
+		case InvalidTransactionId:
+			return false;
+		case BootstrapTransactionId:
+		case FrozenTransactionId:
+			return true;
+	}
+
+	/*
+	 * If this xid is within the last known valid range of xids, then it has
+	 * to be ok.  The oldest valid xid cannot advance, because we have too
+	 * strong a lock on the relation for that, and although the newest valid
+	 * xid may advance, that doesn't invalidate anything from the range we've
+	 * already identified.
+	 */
+	if (XidInValidRange(xid, ctx))
+		return true;
+
+	/* The latest valid xid may have advanced.  Recheck. */
+	ctx->nextKnownValidXid =
+		XidFromFullTransactionId(ReadNextFullTransactionId());
+	if (XidInValidRange(xid, ctx))
+		return true;
+
+	/* No good.  This xid is invalid. */
+	return false;
+}
+
+/*
+ * tuple_is_visible
+ *
+ *	Determine whether tuples are visible for verification.  Similar to
+ *  HeapTupleSatisfiesVacuum, but with critical differences.
+ *
+ *  1) Does not touch hint bits.  It seems imprudent to write hint bits
+ *     to a table during a corruption check.
+ *  2) Only makes a boolean determination of whether verification should
+ *     see the tuple, rather than doing extra work for vacuum-related
+ *     categorization.
+ *
+ *  The caller should already have checked that xmin and xmax are not out of
+ *  bounds for the relation.
+ */
+static bool
+tuple_is_visible(HeapTupleHeader tuphdr, HeapCheckContext * ctx)
+{
+	uint16		infomask = tuphdr->t_infomask;
+
+	if (!HeapTupleHeaderXminCommitted(tuphdr))
+	{
+		TransactionId raw_xmin = HeapTupleHeaderGetRawXmin(tuphdr);
+
+		if (HeapTupleHeaderXminInvalid(tuphdr))
+		{
+			return false;		/* HEAPTUPLE_DEAD */
+		}
+		/* Used by pre-9.0 binary upgrades */
+		else if (infomask & HEAP_MOVED_OFF ||
+				 infomask & HEAP_MOVED_IN)
+		{
+			TransactionId xvac = HeapTupleHeaderGetXvac(tuphdr);
+
+			if (TransactionIdIsCurrentTransactionId(xvac))
+				return true;	/* HEAPTUPLE_DELETE_IN_PROGRESS */
+			if (TransactionIdIsInProgress(xvac))
+				return true;	/* HEAPTUPLE_DELETE_IN_PROGRESS */
+
+			if (!TransactionIdValidInRel(xvac, ctx))
+			{
+				confess(ctx, psprintf("old-style VACUUM FULL transaction ID %u is invalid in this relation",
+									  xvac));
+				return false;
+			}
+			else if (TransactionIdDidCommit(xvac))
+				return false;	/* HEAPTUPLE_DEAD */
+		}
+		else if (TransactionIdIsCurrentTransactionId(raw_xmin))
+			return true;		/* insert or delete in progress */
+		else if (TransactionIdIsInProgress(raw_xmin))
+			return true;		/* HEAPTUPLE_INSERT_IN_PROGRESS */
+		else if (!TransactionIdDidCommit(raw_xmin))
+		{
+			return false;		/* HEAPTUPLE_DEAD */
+		}
+	}
+
+	if (!(infomask & HEAP_XMAX_INVALID) && !HEAP_XMAX_IS_LOCKED_ONLY(infomask))
+	{
+		if (infomask & HEAP_XMAX_IS_MULTI)
+		{
+			TransactionId xmax = HeapTupleGetUpdateXid(tuphdr);
+
+			/* not LOCKED_ONLY, so it has to have an xmax */
+			if (!TransactionIdIsValid(xmax))
+			{
+				confess(ctx, pstrdup(
+									 "heap tuple with XMAX_IS_MULTI is neither LOCKED_ONLY nor has a valid xmax"));
+				return false;
+			}
+			if (TransactionIdIsInProgress(xmax))
+				return true;	/* HEAPTUPLE_DELETE_IN_PROGRESS */
+
+			else if (TransactionIdDidCommit(xmax))
+			{
+				return false;	/* HEAPTUPLE_RECENTLY_DEAD or HEAPTUPLE_DEAD */
+			}
+			/* Ok, the tuple is live */
+		}
+		else if (!(infomask & HEAP_XMAX_COMMITTED))
+		{
+			if (TransactionIdIsInProgress(HeapTupleHeaderGetRawXmax(tuphdr)))
+				return true;	/* HEAPTUPLE_DELETE_IN_PROGRESS */
+			/* Ok, the tuple is live */
+		}
+		else
+			return false;		/* HEAPTUPLE_RECENTLY_DEAD or HEAPTUPLE_DEAD */
+	}
+	return true;
+}
+
+/*
+ * check_toast_tuple
+ *
+ *   Checks the current toast tuple as tracked in ctx for corruption.  Records
+ *   any corruption found in ctx->corruption.
+ */
+static void
+check_toast_tuple(HeapTuple toasttup, HeapCheckContext * ctx)
+{
+	int32		curchunk;
+	Pointer		chunk;
+	bool		isnull;
+	char	   *chunkdata;
+	int32		chunksize;
+	int32		expected_size;
+
+	/*
+	 * Have a chunk, extract the sequence number and the data
+	 */
+	curchunk = DatumGetInt32(fastgetattr(toasttup, 2,
+										 ctx->toastrel->rd_att, &isnull));
+	if (isnull)
+	{
+		confess(ctx,
+				pstrdup("toast chunk sequence number is null"));
+		return;
+	}
+	chunk = DatumGetPointer(fastgetattr(toasttup, 3,
+										ctx->toastrel->rd_att, &isnull));
+	if (isnull)
+	{
+		confess(ctx, pstrdup("toast chunk data is null"));
+		return;
+	}
+	if (!VARATT_IS_EXTENDED(chunk))
+	{
+		chunksize = VARSIZE(chunk) - VARHDRSZ;
+		chunkdata = VARDATA(chunk);
+	}
+	else if (VARATT_IS_SHORT(chunk))
+	{
+		/*
+		 * could happen due to heap_form_tuple doing its thing
+		 */
+		chunksize = VARSIZE_SHORT(chunk) - VARHDRSZ_SHORT;
+		chunkdata = VARDATA_SHORT(chunk);
+	}
+	else
+	{
+		/* should never happen */
+		uint32		header = ((varattrib_4b *) chunk)->va_4byte.va_header;
+
+		confess(ctx, psprintf(
+							  "corrupt extended toast chunk with sequence number %d has invalid varlena header %0x",
+							  curchunk, header));
+		return;
+	}
+
+	/*
+	 * Some checks on the data we've found
+	 */
+	if (curchunk != ctx->chunkno)
+	{
+		confess(ctx, psprintf(
+							  "toast chunk sequence number %u not the expected sequence number %u",
+							  curchunk, ctx->chunkno));
+		return;
+	}
+	if (curchunk > ctx->endchunk)
+	{
+		confess(ctx, psprintf(
+							  "toast chunk sequence number %u exceeds the end chunk sequence number %u",
+							  curchunk, ctx->endchunk));
+		return;
+	}
+
+	expected_size = curchunk < ctx->totalchunks - 1 ? TOAST_MAX_CHUNK_SIZE
+		: ctx->attrsize - ((ctx->totalchunks - 1) * TOAST_MAX_CHUNK_SIZE);
+	if (chunksize != expected_size)
+	{
+		confess(ctx, psprintf("toast chunk size %u differs from expected size %u",
+							  chunksize, expected_size));
+		return;
+	}
+
+	ctx->chunkno++;
+}
+
+/*
+ * check_tuple_attribute
+ *
+ *   Checks the current attribute as tracked in ctx for corruption.  Records
+ *   any corruption found in ctx->corruption.
+ *
+ *   This function follows the logic performed by heap_deform_tuple(), and in
+ *   the case of a toasted value, continues along the logic of
+ *   detoast_external_attr(), checking for any conditions that would result in
+ *   either of those functions Asserting or crashing the backend.  The checks
+ *   performed by Asserts present in those two functions are also performed
+ *   here.  In cases where those two functions are a bit cavalier in their
+ *   assumptions about data being correct, we perform additional checks not
+ *   present in either of those two functions.  Where some condition is checked
+ *   in both of those functions, we perform it here twice, as we parallel the
+ *   logical flow of those two functions.  The presence of duplicate checks
+ *   seems a reasonable price to pay for keeping this code tightly coupled with
+ *   the code it protects.
+ *
+ *   Returns true if the tuple attribute is sane enough for processing to
+ *   continue on to the next attribute, false otherwise.
+ */
+static bool
+check_tuple_attribute(HeapCheckContext * ctx)
+{
+	struct varatt_external toast_pointer;
+	ScanKeyData toastkey;
+	SysScanDesc toastscan;
+	SnapshotData SnapshotToast;
+	HeapTuple	toasttup;
+	bool		found_toasttup;
+	Datum		attdatum;
+	struct varlena *attr;
+	char	   *tp;				/* pointer to the tuple data */
+	uint16		infomask;
+	Form_pg_attribute thisatt;
+
+	infomask = ctx->tuphdr->t_infomask;
+	thisatt = TupleDescAttr(RelationGetDescr(ctx->rel), ctx->attnum);
+
+	tp = (char *) ctx->tuphdr + ctx->tuphdr->t_hoff;
+
+	if (ctx->tuphdr->t_hoff + ctx->offset > ctx->lp_len)
+	{
+		confess(ctx, psprintf(
+							  "tuple attribute should start at offset %u, but tuple length is only %u",
+							  ctx->tuphdr->t_hoff + ctx->offset, ctx->lp_len));
+		return false;
+	}
+
+	/* Skip null values */
+	if (infomask & HEAP_HASNULL && att_isnull(ctx->attnum, ctx->tuphdr->t_bits))
+		return true;
+
+	/* Skip non-varlena values, but update offset first */
+	if (thisatt->attlen != -1)
+	{
+		ctx->offset = att_align_nominal(ctx->offset, thisatt->attalign);
+		ctx->offset = att_addlength_pointer(ctx->offset, thisatt->attlen,
+											tp + ctx->offset);
+		return true;
+	}
+
+	/* Ok, we're looking at a varlena attribute. */
+	ctx->offset = att_align_pointer(ctx->offset, thisatt->attalign, -1,
+									tp + ctx->offset);
+
+	/* Get the (possibly corrupt) varlena datum */
+	attdatum = fetchatt(thisatt, tp + ctx->offset);
+
+	/*
+	 * We have the datum, but we cannot decode it carelessly, as it may still
+	 * be corrupt.
+	 */
+
+	/*
+	 * Check that VARTAG_SIZE won't hit a TrapMacro on a corrupt va_tag before
+	 * risking a call into att_addlength_pointer
+	 */
+	if (VARATT_IS_EXTERNAL(tp + ctx->offset))
+	{
+		uint8		va_tag = VARTAG_EXTERNAL(tp + ctx->offset);
+
+		if (va_tag != VARTAG_ONDISK)
+		{
+			confess(ctx, psprintf(
+								  "%s toast at offset %u is unexpected",
+								  va_tag == VARTAG_INDIRECT ? "indirect" :
+								  va_tag == VARTAG_EXPANDED_RO ? "expanded" :
+								  va_tag == VARTAG_EXPANDED_RW ? "expanded" :
+								  "unexpected",
+								  ctx->tuphdr->t_hoff + ctx->offset));
+			/* We can't know where the next attribute begins */
+			return false;
+		}
+	}
+
+	/* Ok, should be safe now */
+	ctx->offset = att_addlength_pointer(ctx->offset, thisatt->attlen,
+										tp + ctx->offset);
+
+	if (ctx->tuphdr->t_hoff + ctx->offset > ctx->lp_len)
+	{
+		confess(ctx, psprintf(
+							  "tuple attribute of length %u ends at offset %u, but tuple length is only %u",
+							  thisatt->attlen, ctx->tuphdr->t_hoff + ctx->offset,
+							  ctx->lp_len));
+		return false;
+	}
+
+	/*
+	 * heap_deform_tuple would be done with this attribute at this point,
+	 * having stored it in values[], and would continue to the next attribute.
+	 * We go further, because we need to check if the toast datum is corrupt.
+	 */
+
+	attr = (struct varlena *) DatumGetPointer(attdatum);
+
+	/*
+	 * Now we follow the logic of detoast_external_attr(), with the same
+	 * caveats about being paranoid about corruption.
+	 */
+
+	/* Skip values that are not external */
+	if (!VARATT_IS_EXTERNAL(attr))
+		return true;
+
+	/* It is external, and we're looking at a page on disk */
+
+	/* The tuple header better claim to contain toasted values */
+	if (!(infomask & HEAP_HASEXTERNAL))
+	{
+		confess(ctx, pstrdup(
+							 "attribute is external but tuple header flag HEAP_HASEXTERNAL not set"));
+		return true;
+	}
+
+	/* The relation better have a toast table */
+	if (!ctx->rel->rd_rel->reltoastrelid)
+	{
+		confess(ctx, pstrdup(
+							 "attribute is external but relation has no toast relation"));
+		return true;
+	}
+
+	/*
+	 * Must copy attr into toast_pointer for alignment considerations
+	 */
+	VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr);
+
+	ctx->attrsize = toast_pointer.va_extsize;
+	ctx->endchunk = (ctx->attrsize - 1) / TOAST_MAX_CHUNK_SIZE;
+	ctx->totalchunks = ctx->endchunk + 1;
+
+	/*
+	 * Setup a scan key to find chunks in toast table with matching va_valueid
+	 */
+	ScanKeyInit(&toastkey,
+				(AttrNumber) 1,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(toast_pointer.va_valueid));
+
+	/*
+	 * Check if any chunks for this toasted object exist in the toast table,
+	 * accessible via the index.
+	 */
+	init_toast_snapshot(&SnapshotToast);
+	toastscan = systable_beginscan_ordered(ctx->toastrel,
+										   ctx->valid_toast_index,
+										   &SnapshotToast, 1,
+										   &toastkey);
+	ctx->chunkno = 0;
+
+	found_toasttup = false;
+	while ((toasttup =
+			systable_getnext_ordered(toastscan,
+									 ForwardScanDirection)) != NULL)
+	{
+		found_toasttup = true;
+		check_toast_tuple(toasttup, ctx);
+	}
+	if (ctx->chunkno != (ctx->endchunk + 1))
+		confess(ctx, psprintf(
+							  "final chunk number %u differs from expected value %u",
+							  ctx->chunkno, (ctx->endchunk + 1)));
+	if (!found_toasttup)
+		confess(ctx, pstrdup("toasted value missing from toast table"));
+	systable_endscan_ordered(toastscan);
+
+	return true;
+}
+
+/*
+ * check_tuple
+ *
+ *   Checks the current tuple as tracked in ctx for corruption.  Records any
+ *   corruption found in ctx->corruption.
+ */
+static void
+check_tuple(HeapCheckContext * ctx)
+{
+	TransactionId xmin;
+	TransactionId xmax;
+	bool		fatal = false;
+	uint16		infomask = ctx->tuphdr->t_infomask;
+
+	/*
+	 * If the line pointer for this tuple does not reserve enough space for a
+	 * complete tuple header, we dare not read the tuple header.
+	 */
+	if (ctx->lp_len < MAXALIGN(SizeofHeapTupleHeader))
+	{
+		confess(ctx, psprintf(
+							  "tuple's %u byte line pointer length is less than the %u byte minimum tuple header size",
+							  ctx->lp_len, (uint32) MAXALIGN(SizeofHeapTupleHeader)));
+		return;
+	}
+
+	/* Check relminmxid against mxid, if any */
+	xmax = HeapTupleHeaderGetRawXmax(ctx->tuphdr);
+	if (infomask & HEAP_XMAX_IS_MULTI &&
+		MultiXactIdPrecedes(xmax, ctx->relminmxid))
+	{
+		confess(ctx, psprintf(
+							  "tuple xmax %u precedes relminmxid %u",
+							  xmax, ctx->relminmxid));
+		fatal = true;
+	}
+
+	/* Check xmin against relfrozenxid */
+	xmin = HeapTupleHeaderGetXmin(ctx->tuphdr);
+	if (TransactionIdIsNormal(ctx->relfrozenxid) &&
+		TransactionIdIsNormal(xmin))
+	{
+		if (TransactionIdPrecedes(xmin, ctx->relfrozenxid))
+		{
+			confess(ctx, psprintf(
+								  "tuple xmin %u precedes relfrozenxid %u",
+								  xmin, ctx->relfrozenxid));
+			fatal = true;
+		}
+		else if (!TransactionIdValidInRel(xmin, ctx))
+		{
+			confess(ctx, psprintf(
+								  "tuple xmin %u follows last assigned xid %u",
+								  xmin, ctx->nextKnownValidXid));
+			fatal = true;
+		}
+	}
+
+	/* Check xmax against relfrozenxid */
+	if (TransactionIdIsNormal(ctx->relfrozenxid) &&
+		TransactionIdIsNormal(xmax))
+	{
+		if (TransactionIdPrecedes(xmax, ctx->relfrozenxid))
+		{
+			confess(ctx, psprintf(
+								  "tuple xmax %u precedes relfrozenxid %u",
+								  xmax, ctx->relfrozenxid));
+			fatal = true;
+		}
+		else if (!TransactionIdValidInRel(xmax, ctx))
+		{
+			confess(ctx, psprintf(
+								  "tuple xmax %u follows last assigned xid %u",
+								  xmax, ctx->nextKnownValidXid));
+			fatal = true;
+		}
+	}
+
+	/* Check for tuple header corruption */
+	if (ctx->tuphdr->t_hoff < SizeofHeapTupleHeader)
+	{
+		confess(ctx,
+				psprintf("tuple's header size is %u bytes which is less than the %u byte minimum valid header size",
+						 ctx->tuphdr->t_hoff,
+						 (unsigned) SizeofHeapTupleHeader));
+		fatal = true;
+	}
+	if (ctx->tuphdr->t_hoff > ctx->lp_len)
+	{
+		confess(ctx, psprintf(
+							  "tuple's %u byte header size exceeds the %u byte length of the entire tuple",
+							  ctx->tuphdr->t_hoff, ctx->lp_len));
+		fatal = true;
+	}
+	if (ctx->tuphdr->t_hoff != MAXALIGN(ctx->tuphdr->t_hoff))
+	{
+		confess(ctx, psprintf(
+							  "tuple's user data offset %u not maximally aligned to %u",
+							  ctx->tuphdr->t_hoff, (uint32) MAXALIGN(ctx->tuphdr->t_hoff)));
+		fatal = true;
+	}
+	if ((ctx->tuphdr->t_infomask & HEAP_XMAX_LOCK_ONLY) &&
+		(ctx->tuphdr->t_infomask2 & HEAP_KEYS_UPDATED))
+	{
+		confess(ctx,
+				psprintf("tuple xmax marked incompatibly as keys updated and locked only"));
+	}
+	if ((ctx->tuphdr->t_infomask & HEAP_XMAX_COMMITTED) &&
+		(ctx->tuphdr->t_infomask & HEAP_XMAX_IS_MULTI))
+	{
+		confess(ctx,
+				psprintf("tuple xmax marked incompatibly as committed and as a multitransaction ID"));
+	}
+
+	/*
+	 * If the tuple has nulls, check that the implied length of the variable
+	 * length nulls bitmap field t_bits does not overflow the allowed space.
+	 * We don't know if the corruption is in the t_hoff field or the infomask
+	 * bit HEAP_HASNULL.
+	 *
+	 * If the tuple does not have nulls, check that no space has been reserved
+	 * for the null bitmap.
+	 */
+	if ((infomask & HEAP_HASNULL) &&
+		(ctx->tuphdr->t_hoff != MAXALIGN(SizeofHeapTupleHeader + BITMAPLEN(ctx->natts))))
+	{
+		confess(ctx, psprintf(
+							  "tuple with null values has user data offset %u rather than the expected offset %u",
+							  ctx->tuphdr->t_hoff,
+							  (uint32) MAXALIGN(SizeofHeapTupleHeader + BITMAPLEN(ctx->natts))));
+		fatal = true;
+	}
+	else if (!(infomask & HEAP_HASNULL) &&
+			 (ctx->tuphdr->t_hoff != MAXALIGN(SizeofHeapTupleHeader)))
+	{
+		confess(ctx, psprintf(
+							  "tuple without null values has user data offset %u rather than the expected offset %u",
+							  ctx->tuphdr->t_hoff,
+							  (uint32) MAXALIGN(SizeofHeapTupleHeader)));
+		fatal = true;
+	}
+
+	/*
+	 * Cannot process tuple data if tuple header was corrupt, as the offsets
+	 * within the page cannot be trusted, leaving too much risk of reading
+	 * garbage if we continue.
+	 *
+	 * We also cannot process the tuple if the xmin or xmax were invalid
+	 * relative to relfrozenxid or relminmxid, as clog entries for the xids
+	 * may already be gone.
+	 */
+	if (fatal)
+		return;
+
+	/*
+	 * Skip tuples that are invisible, as we cannot assume the TupleDesc we
+	 * are using is appropriate.
+	 */
+	if (!tuple_is_visible(ctx->tuphdr, ctx))
+		return;
+
+	/*
+	 * If we get this far, the tuple is visible to us, so it must not be
+	 * incompatible with our relDesc.  The natts field could be legitimately
+	 * shorter than rel's natts, but it cannot be longer than rel's natts.
+	 */
+	if (RelationGetDescr(ctx->rel)->natts < ctx->natts)
+	{
+		confess(ctx, psprintf(
+							  "tuple has %u attributes in relation with only %u attributes",
+							  ctx->natts,
+							  RelationGetDescr(ctx->rel)->natts));
+		return;
+	}
+
+	/*
+	 * Iterate over the attributes looking for broken toast values. This
+	 * roughly follows the logic of heap_deform_tuple, except that it doesn't
+	 * bother building up isnull[] and values[] arrays, since nobody wants
+	 * them, and it unrolls anything that might trip over an Assert when
+	 * processing corrupt data.
+	 */
+	ctx->offset = 0;
+	for (ctx->attnum = 0; ctx->attnum < ctx->natts; ctx->attnum++)
+	{
+		if (!check_tuple_attribute(ctx))
+			break;
+	}
+}
diff --git a/doc/src/sgml/amcheck.sgml b/doc/src/sgml/amcheck.sgml
index a9df2c1a9d..b8170bbfdf 100644
--- a/doc/src/sgml/amcheck.sgml
+++ b/doc/src/sgml/amcheck.sgml
@@ -69,7 +69,7 @@ AND c.relpersistence != 't'
 -- Function may throw an error when this is omitted:
 AND c.relkind = 'i' AND i.indisready AND i.indisvalid
 ORDER BY c.relpages DESC LIMIT 10;
- bt_index_check |             relname             | relpages 
+ bt_index_check |             relname             | relpages
 ----------------+---------------------------------+----------
                 | pg_depend_reference_index       |       43
                 | pg_depend_depender_index        |       40
@@ -165,6 +165,110 @@ ORDER BY c.relpages DESC LIMIT 10;
      </para>
     </listitem>
    </varlistentry>
+
+   <varlistentry>
+    <term>
+     <function>
+      verify_heapam(relation regclass,
+                    on_error_stop boolean,
+                    skip_all_frozen boolean,
+                    skip_all_visible boolean,
+                    blkno OUT bigint,
+                    offnum OUT integer,
+                    lp_off OUT smallint,
+                    lp_flags OUT smallint,
+                    lp_len OUT smallint,
+                    attnum OUT integer,
+                    chunk OUT integer,
+                    msg OUT text)
+      returns record
+     </function>
+    </term>
+    <listitem>
+     <para>
+      Checks for "logical" corruption, where the page is valid but inconsistent
+      with the rest of the database cluster. This can happen due to faulty or
+      ill-conceived backup and restore tools, or bad storage, or user error, or
+      bugs in the server itself.  It checks xmin and xmax values against
+      relfrozenxid and relminmxid, and also validates TOAST pointers.
+     </para>
+
+     <para>
+      For each block in the relation where corruption is detected, or for just
+      the first block if on_error_stop is true, for each corruption detected,
+      returns one row containing the following fields:
+     </para>
+     <variablelist>
+      <varlistentry>
+       <term>blkno</term>
+       <listitem>
+        <para>
+         The number of the block containing the corrupt page.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>offnum</term>
+       <listitem>
+        <para>
+         The OffsetNumber of the corrupt tuple.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>lp_off</term>
+       <listitem>
+        <para>
+         The offset into the page of the line pointer for the corrupt tuple.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>lp_flags</term>
+       <listitem>
+        <para>
+         The flags in the line pointer for the corrupt tuple.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>lp_len</term>
+       <listitem>
+        <para>
+         The length of the corrupt tuple as recorded in the line pointer.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>attnum</term>
+       <listitem>
+        <para>
+         The attribute number of the corrupt column in the tuple, if the
+         corruption is specific to a column and not the tuple as a whole.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>chunk</term>
+       <listitem>
+        <para>
+         The chunk number of the corrupt toasted attribute, if the corruption
+         is specific to a toasted value.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry>
+       <term>msg</term>
+       <listitem>
+        <para>
+         A human readable message describing the corruption in the page.
+        </para>
+       </listitem>
+      </varlistentry>
+     </variablelist>
+    </listitem>
+   </varlistentry>
+
   </variablelist>
   <tip>
    <para>
diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index aa3f14c019..00de10b7c9 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -47,6 +47,17 @@ RelationPutHeapTuple(Relation relation,
 	 */
 	Assert(!token || HeapTupleHeaderIsSpeculative(tuple->t_data));
 
+	/*
+	 * Do not allow tuples with invalid combinations of hint bits to be placed
+	 * on a page.  These combinations are detected as corruption by the
+	 * contrib/amcheck logic, so if you decide to disable one or more of these
+	 * assertions, make corresponding changes to contrib/amcheck.
+	 */
+	Assert(!((tuple->t_data->t_infomask & HEAP_XMAX_LOCK_ONLY) &&
+			 (tuple->t_data->t_infomask2 & HEAP_KEYS_UPDATED)));
+	Assert(!((tuple->t_data->t_infomask & HEAP_XMAX_COMMITTED) &&
+			 (tuple->t_data->t_infomask & HEAP_XMAX_IS_MULTI)));
+
 	/* Add the tuple to the page */
 	pageHeader = BufferGetPage(buffer);
 
-- 
2.21.1 (Apple Git-122.3)