Teach contrib/amcheck to check the unique constraint violation

Add the 'checkunique' argument to bt_index_check() and bt_index_parent_check().
When the flag is specified the procedures will check the unique constraint
violation for unique indexes.  Only one heap entry for all equal keys in
the index should be visible (including posting list entries).  Report an error
otherwise.

pg_amcheck called with the --checkunique option will do the same check for all
the indexes it checks.

Author: Anastasia Lubennikova <lubennikovaav@gmail.com>
Author: Pavel Borisov <pashkin.elfe@gmail.com>
Author: Maxim Orlov <orlovmg@gmail.com>
Reviewed-by: Mark Dilger <mark.dilger@enterprisedb.com>
Reviewed-by: Zhihong Yu <zyu@yugabyte.com>
Reviewed-by: Peter Geoghegan <pg@bowt.ie>
Reviewed-by: Aleksander Alekseev <aleksander@timescale.com>
Discussion: https://postgr.es/m/CALT9ZEHRn5xAM5boga0qnrCmPV52bScEK2QnQ1HmUZDD301JEg%40mail.gmail.com

3 files changed, 161 insertions, 2 deletions

diff --git a/src/bin/pg_amcheck/pg_amcheck.c b/src/bin/pg_amcheck/pg_amcheck.c
index 8ac7051ff4d..57c7c1917c4 100644
--- a/src/bin/pg_amcheck/pg_amcheck.c
+++ b/src/bin/pg_amcheck/pg_amcheck.c
@@ -102,6 +102,7 @@ typedef struct AmcheckOptions
 	bool		parent_check;
 	bool		rootdescend;
 	bool		heapallindexed;
+	bool		checkunique;
 
 	/* heap and btree hybrid option */
 	bool		no_btree_expansion;
@@ -132,6 +133,7 @@ static AmcheckOptions opts = {
 	.parent_check = false,
 	.rootdescend = false,
 	.heapallindexed = false,
+	.checkunique = false,
 	.no_btree_expansion = false
 };
 
@@ -148,6 +150,7 @@ typedef struct DatabaseInfo
 {
 	char	   *datname;
 	char	   *amcheck_schema; /* escaped, quoted literal */
+	bool		is_checkunique;
 } DatabaseInfo;
 
 typedef struct RelationInfo
@@ -267,6 +270,7 @@ main(int argc, char *argv[])
 		{"heapallindexed", no_argument, NULL, 11},
 		{"parent-check", no_argument, NULL, 12},
 		{"install-missing", optional_argument, NULL, 13},
+		{"checkunique", no_argument, NULL, 14},
 
 		{NULL, 0, NULL, 0}
 	};
@@ -434,6 +438,9 @@ main(int argc, char *argv[])
 				if (optarg)
 					opts.install_schema = pg_strdup(optarg);
 				break;
+			case 14:
+				opts.checkunique = true;
+				break;
 			default:
 				/* getopt_long already emitted a complaint */
 				pg_log_error_hint("Try \"%s --help\" for more information.", progname);
@@ -589,6 +596,38 @@ main(int argc, char *argv[])
 						PQdb(conn), PQgetvalue(result, 0, 1), amcheck_schema);
 		dat->amcheck_schema = PQescapeIdentifier(conn, amcheck_schema,
 												 strlen(amcheck_schema));
+
+		/*
+		 * Check the version of amcheck extension. Skip requested unique
+		 * constraint check with warning if it is not yet supported by amcheck.
+		 */
+		if (opts.checkunique == true)
+		{
+			/*
+			 * Now amcheck has only major and minor versions in the string but
+			 * we also support revision just in case. Now it is expected to be
+			 * zero.
+			 */
+			int			vmaj = 0,
+						vmin = 0,
+						vrev = 0;
+			const char *amcheck_version = PQgetvalue(result, 0, 1);
+
+			sscanf(amcheck_version, "%d.%d.%d", &vmaj, &vmin, &vrev);
+
+			/*
+			 * checkunique option is supported in amcheck since version 1.4
+			 */
+			if ((vmaj == 1 && vmin < 4) || vmaj == 0)
+			{
+				pg_log_warning("--checkunique option is not supported by amcheck "
+							   "version \"%s\"", amcheck_version);
+				dat->is_checkunique = false;
+			}
+			else
+				dat->is_checkunique = true;
+		}
+
 		PQclear(result);
 
 		compile_relation_list_one_db(conn, &relations, dat, &pagestotal);
@@ -845,7 +884,8 @@ prepare_btree_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn)
 	if (opts.parent_check)
 		appendPQExpBuffer(sql,
 						  "SELECT %s.bt_index_parent_check("
-						  "index := c.oid, heapallindexed := %s, rootdescend := %s)"
+						  "index := c.oid, heapallindexed := %s, rootdescend := %s "
+						  "%s)"
 						  "\nFROM pg_catalog.pg_class c, pg_catalog.pg_index i "
 						  "WHERE c.oid = %u "
 						  "AND c.oid = i.indexrelid "
@@ -854,11 +894,13 @@ prepare_btree_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn)
 						  rel->datinfo->amcheck_schema,
 						  (opts.heapallindexed ? "true" : "false"),
 						  (opts.rootdescend ? "true" : "false"),
+						  (rel->datinfo->is_checkunique ? ", checkunique := true" : ""),
 						  rel->reloid);
 	else
 		appendPQExpBuffer(sql,
 						  "SELECT %s.bt_index_check("
-						  "index := c.oid, heapallindexed := %s)"
+						  "index := c.oid, heapallindexed := %s "
+						  "%s)"
 						  "\nFROM pg_catalog.pg_class c, pg_catalog.pg_index i "
 						  "WHERE c.oid = %u "
 						  "AND c.oid = i.indexrelid "
@@ -866,6 +908,7 @@ prepare_btree_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn)
 						  "AND i.indisready AND i.indisvalid AND i.indislive",
 						  rel->datinfo->amcheck_schema,
 						  (opts.heapallindexed ? "true" : "false"),
+						  (rel->datinfo->is_checkunique ? ", checkunique := true" : ""),
 						  rel->reloid);
 }
 
@@ -1163,6 +1206,7 @@ help(const char *progname)
 	printf(_("      --heapallindexed            check that all heap tuples are found within indexes\n"));
 	printf(_("      --parent-check              check index parent/child relationships\n"));
 	printf(_("      --rootdescend               search from root page to refind tuples\n"));
+	printf(_("      --checkunique               check unique constraint if index is unique\n"));
 	printf(_("\nConnection options:\n"));
 	printf(_("  -h, --host=HOSTNAME             database server host or socket directory\n"));
 	printf(_("  -p, --port=PORT                 database server port\n"));
diff --git a/src/bin/pg_amcheck/t/003_check.pl b/src/bin/pg_amcheck/t/003_check.pl
index d577cffa30d..2b7ef198552 100644
--- a/src/bin/pg_amcheck/t/003_check.pl
+++ b/src/bin/pg_amcheck/t/003_check.pl
@@ -257,6 +257,9 @@ for my $dbname (qw(db1 db2 db3))
 
 			CREATE INDEX t1_spgist ON $schema.t1 USING SPGIST (ir);
 			CREATE INDEX t2_spgist ON $schema.t2 USING SPGIST (ir);
+
+			CREATE UNIQUE INDEX t1_btree_unique ON $schema.t1 USING BTREE (i);
+			CREATE UNIQUE INDEX t2_btree_unique ON $schema.t2 USING BTREE (i);
 		));
 	}
 }
@@ -517,4 +520,51 @@ $node->command_checks_all(
 	0, [$no_output_re], [$no_output_re],
 	'pg_amcheck excluding all corrupt schemas');
 
+$node->command_checks_all(
+	[
+		@cmd, '-s', 's1', '-i', 't1_btree', '--parent-check',
+		'--checkunique', 'db1'
+	],
+	2,
+	[$index_missing_relation_fork_re],
+	[$no_output_re],
+	'pg_amcheck smoke test --parent-check --checkunique');
+
+$node->command_checks_all(
+	[
+		@cmd, '-s', 's1', '-i', 't1_btree', '--heapallindexed',
+		'--rootdescend', '--checkunique', 'db1'
+	],
+	2,
+	[$index_missing_relation_fork_re],
+	[$no_output_re],
+	'pg_amcheck smoke test --heapallindexed --rootdescend --checkunique');
+
+$node->command_checks_all(
+	[
+		@cmd, '--checkunique', '-d', 'db1', '-d', 'db2',
+		'-d', 'db3', '-S', 's*'
+	],
+	0,
+	[$no_output_re],
+	[$no_output_re],
+	'pg_amcheck excluding all corrupt schemas with --checkunique option');
+
+#
+# Smoke test for checkunique option for not supported versions.
+#
+$node->safe_psql(
+	'db3', q(
+		DROP EXTENSION amcheck;
+		CREATE EXTENSION amcheck WITH SCHEMA amcheck_schema VERSION '1.3' ;
+));
+
+$node->command_checks_all(
+	[ @cmd, '--checkunique', 'db3' ],
+	0,
+	[$no_output_re],
+	[
+		qr/pg_amcheck: warning: --checkunique option is not supported by amcheck version "1.3"/
+	],
+	'pg_amcheck smoke test --checkunique');
 done_testing();
diff --git a/src/bin/pg_amcheck/t/005_opclass_damage.pl b/src/bin/pg_amcheck/t/005_opclass_damage.pl
index fd476179f49..a5ef2c0f33d 100644
--- a/src/bin/pg_amcheck/t/005_opclass_damage.pl
+++ b/src/bin/pg_amcheck/t/005_opclass_damage.pl
@@ -22,14 +22,33 @@ $node->safe_psql(
 	CREATE FUNCTION int4_asc_cmp (a int4, b int4) RETURNS int LANGUAGE sql AS $$
 		SELECT CASE WHEN $1 = $2 THEN 0 WHEN $1 > $2 THEN 1 ELSE -1 END; $$;
 
+	CREATE FUNCTION ok_cmp (int4, int4)
+	RETURNS int LANGUAGE sql AS
+	$$
+		SELECT
+			CASE WHEN $1 < $2 THEN -1
+				 WHEN $1 > $2 THEN  1
+				 ELSE 0
+			END;
+	$$;
+
 	CREATE OPERATOR CLASS int4_fickle_ops FOR TYPE int4 USING btree AS
 	    OPERATOR 1 < (int4, int4), OPERATOR 2 <= (int4, int4),
 	    OPERATOR 3 = (int4, int4), OPERATOR 4 >= (int4, int4),
 	    OPERATOR 5 > (int4, int4), FUNCTION 1 int4_asc_cmp(int4, int4);
 
+	CREATE OPERATOR CLASS int4_unique_ops FOR TYPE int4 USING btree AS
+		OPERATOR 1 < (int4, int4), OPERATOR 2 <= (int4, int4),
+		OPERATOR 3 = (int4, int4), OPERATOR 4 >= (int4, int4),
+		OPERATOR 5 > (int4, int4), FUNCTION 1 ok_cmp(int4, int4);
+
 	CREATE TABLE int4tbl (i int4);
 	INSERT INTO int4tbl (SELECT * FROM generate_series(1,1000) gs);
 	CREATE INDEX fickleidx ON int4tbl USING btree (i int4_fickle_ops);
+	CREATE UNIQUE INDEX bttest_unique_idx
+						ON int4tbl
+						USING btree (i int4_unique_ops)
+						WITH (deduplicate_items = off);
 ));
 
 # We have not yet broken the index, so we should get no corruption
@@ -57,4 +76,50 @@ $node->command_checks_all(
 	'pg_amcheck all schemas, tables and indexes reports fickleidx corruption'
 );
 
+#
+# Check unique constraints
+#
+
+# Repair broken opclass for check unique tests.
+$node->safe_psql(
+	'postgres', q(
+	UPDATE pg_catalog.pg_amproc
+		SET amproc = 'int4_asc_cmp'::regproc
+		WHERE amproc = 'int4_desc_cmp'::regproc
+));
+
+# We should get no corruptions
+$node->command_like(
+	[ 'pg_amcheck', '--checkunique', '-p', $node->port, 'postgres' ],
+	qr/^$/,
+	'pg_amcheck all schemas, tables and indexes reports no corruption');
+
+# Break opclass for check unique tests.
+$node->safe_psql(
+	'postgres', q(
+	CREATE FUNCTION bad_cmp (int4, int4)
+	RETURNS int LANGUAGE sql AS
+	$$
+		SELECT
+			CASE WHEN ($1 = 768 AND $2 = 769) OR
+					  ($1 = 769 AND $2 = 768) THEN 0
+				 WHEN $1 < $2 THEN -1
+				 WHEN $1 > $2 THEN  1
+				 ELSE 0
+			END;
+	$$;
+
+	UPDATE pg_catalog.pg_amproc
+		SET amproc = 'bad_cmp'::regproc
+		WHERE amproc = 'ok_cmp'::regproc
+));
+
+# Unique index corruption should now be reported
+$node->command_checks_all(
+	[ 'pg_amcheck', '--checkunique', '-p', $node->port, 'postgres' ],
+	2,
+	[qr/index uniqueness is violated for index "bttest_unique_idx"/],
+	[],
+	'pg_amcheck all schemas, tables and indexes reports bttest_unique_idx corruption'
+);
 done_testing();