Improve index AMs' opclass validation procedures.

The amvalidate functions added in commit 65c5fcd3 were on the
crude side.  Improve them in a few ways:

* Perform signature checking for operators and support functions.

* Apply more thorough checks for missing operators and functions,
where possible.

* Instead of reporting problems as ERRORs, report most problems as INFO
messages and make the amvalidate function return FALSE.  This allows
more than one problem to be discovered per run.

* Report object names rather than OIDs, and work a bit harder on making
the messages understandable.

Also, remove a few more opr_sanity regression test queries that are
now superseded by the amvalidate checks.

src/backend/access/gist/gistvalidate.c

@@ -13,12 +13,16 @@
  */
 #include "postgres.h"
 
+#include "access/amvalidate.h"
 #include "access/gist_private.h"
 #include "access/htup_details.h"
 #include "catalog/pg_amop.h"
 #include "catalog/pg_amproc.h"
 #include "catalog/pg_opclass.h"
-#include "utils/catcache.h"
+#include "catalog/pg_opfamily.h"
+#include "catalog/pg_type.h"
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
 
@@ -28,15 +32,22 @@
 bool
 gistvalidate(Oid opclassoid)
 {
+	bool		result = true;
 	HeapTuple	classtup;
 	Form_pg_opclass classform;
 	Oid			opfamilyoid;
 	Oid			opcintype;
-	int			numclassops;
-	int32		classfuncbits;
+	Oid			opckeytype;
+	char	   *opclassname;
+	HeapTuple	familytup;
+	Form_pg_opfamily familyform;
+	char	   *opfamilyname;
 	CatCList   *proclist,
 			   *oprlist;
+	List	   *grouplist;
+	OpFamilyOpFuncGroup *opclassgroup;
 	int			i;
+	ListCell   *lc;
 
 	/* Fetch opclass information */
 	classtup = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclassoid));
@@ -46,88 +57,218 @@ gistvalidate(Oid opclassoid)
 
 	opfamilyoid = classform->opcfamily;
 	opcintype = classform->opcintype;
+	opckeytype = classform->opckeytype;
+	if (!OidIsValid(opckeytype))
+		opckeytype = opcintype;
+	opclassname = NameStr(classform->opcname);
 
-	ReleaseSysCache(classtup);
+	/* Fetch opfamily information */
+	familytup = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(opfamilyoid));
+	if (!HeapTupleIsValid(familytup))
+		elog(ERROR, "cache lookup failed for operator family %u", opfamilyoid);
+	familyform = (Form_pg_opfamily) GETSTRUCT(familytup);
+
+	opfamilyname = NameStr(familyform->opfname);
 
 	/* Fetch all operators and support functions of the opfamily */
 	oprlist = SearchSysCacheList1(AMOPSTRATEGY, ObjectIdGetDatum(opfamilyoid));
 	proclist = SearchSysCacheList1(AMPROCNUM, ObjectIdGetDatum(opfamilyoid));
 
-	/* We'll track the ops and functions belonging to the named opclass */
-	numclassops = 0;
-	classfuncbits = 0;
-
-	/* Check support functions */
+	/* Check individual support functions */
 	for (i = 0; i < proclist->n_members; i++)
 	{
 		HeapTuple	proctup = &proclist->members[i]->tuple;
 		Form_pg_amproc procform = (Form_pg_amproc) GETSTRUCT(proctup);
+		bool		ok;
+
+		/*
+		 * All GiST support functions should be registered with matching
+		 * left/right types
+		 */
+		if (procform->amproclefttype != procform->amprocrighttype)
+		{
+			ereport(INFO,
+					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					 errmsg("gist opfamily %s contains support procedure %s with cross-type registration",
+							opfamilyname,
+							format_procedure(procform->amproc))));
+			result = false;
+		}
+
+		/*
+		 * We can't check signatures except within the specific opclass, since
+		 * we need to know the associated opckeytype in many cases.
+		 */
+		if (procform->amproclefttype != opcintype)
+			continue;
 
-		/* Check that only allowed procedure numbers exist */
-		if (procform->amprocnum < 1 ||
-			procform->amprocnum > GISTNProcs)
-			ereport(ERROR,
+		/* Check procedure numbers and function signatures */
+		switch (procform->amprocnum)
+		{
+			case GIST_CONSISTENT_PROC:
+				ok = check_amproc_signature(procform->amproc, BOOLOID, false,
+											5, 5, INTERNALOID, opcintype,
+											INT2OID, OIDOID, INTERNALOID);
+				break;
+			case GIST_UNION_PROC:
+				ok = check_amproc_signature(procform->amproc, opckeytype, false,
+											2, 2, INTERNALOID, INTERNALOID);
+				break;
+			case GIST_COMPRESS_PROC:
+			case GIST_DECOMPRESS_PROC:
+			case GIST_FETCH_PROC:
+				ok = check_amproc_signature(procform->amproc, INTERNALOID, true,
+											1, 1, INTERNALOID);
+				break;
+			case GIST_PENALTY_PROC:
+				ok = check_amproc_signature(procform->amproc, INTERNALOID, true,
+											3, 3, INTERNALOID,
+											INTERNALOID, INTERNALOID);
+				break;
+			case GIST_PICKSPLIT_PROC:
+				ok = check_amproc_signature(procform->amproc, INTERNALOID, true,
+											2, 2, INTERNALOID, INTERNALOID);
+				break;
+			case GIST_EQUAL_PROC:
+				ok = check_amproc_signature(procform->amproc, INTERNALOID, false,
+											3, 3, opckeytype, opckeytype,
+											INTERNALOID);
+				break;
+			case GIST_DISTANCE_PROC:
+				ok = check_amproc_signature(procform->amproc, FLOAT8OID, false,
+											5, 5, INTERNALOID, opcintype,
+											INT2OID, OIDOID, INTERNALOID);
+				break;
+			default:
+				ereport(INFO,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("gist opfamily %s contains function %s with invalid support number %d",
+								opfamilyname,
+								format_procedure(procform->amproc),
+								procform->amprocnum)));
+				result = false;
+				continue;		/* don't want additional message */
+		}
+
+		if (!ok)
+		{
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("gist opfamily %u contains invalid support number %d for procedure %u",
-							opfamilyoid,
-							procform->amprocnum, procform->amproc)));
-
-		/* Remember functions that are specifically for the named opclass */
-		if (procform->amproclefttype == opcintype &&
-			procform->amprocrighttype == opcintype)
-			classfuncbits |= (1 << procform->amprocnum);
+					 errmsg("gist opfamily %s contains function %s with wrong signature for support number %d",
+							opfamilyname,
+							format_procedure(procform->amproc),
+							procform->amprocnum)));
+			result = false;
+		}
 	}
 
-	/* Check operators */
+	/* Check individual operators */
 	for (i = 0; i < oprlist->n_members; i++)
 	{
 		HeapTuple	oprtup = &oprlist->members[i]->tuple;
 		Form_pg_amop oprform = (Form_pg_amop) GETSTRUCT(oprtup);
+		Oid			op_rettype;
 
 		/* TODO: Check that only allowed strategy numbers exist */
 		if (oprform->amopstrategy < 1)
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("gist opfamily %u contains invalid strategy number %d for operator %u",
-							opfamilyoid,
-							oprform->amopstrategy, oprform->amopopr)));
-
-		/* GiST supports ORDER BY operators, but must have distance proc */
-		if (oprform->amoppurpose != AMOP_SEARCH &&
-			oprform->amoplefttype == opcintype &&
-			oprform->amoprighttype == opcintype &&
-			(classfuncbits & (1 << GIST_DISTANCE_PROC)) == 0)
-			ereport(ERROR,
+		{
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("gist opfamily %u contains unsupported ORDER BY specification for operator %u",
-							opfamilyoid, oprform->amopopr)));
+					 errmsg("gist opfamily %s contains operator %s with invalid strategy number %d",
+							opfamilyname,
+							format_operator(oprform->amopopr),
+							oprform->amopstrategy)));
+			result = false;
+		}
 
-		/* Count operators that are specifically for the named opclass */
-		/* XXX we consider only lefttype here */
-		if (oprform->amoplefttype == opcintype)
-			numclassops++;
+		/* GiST supports ORDER BY operators */
+		if (oprform->amoppurpose != AMOP_SEARCH)
+		{
+			/* ... but must have matching distance proc */
+			if (!OidIsValid(get_opfamily_proc(opfamilyoid,
+											  oprform->amoplefttype,
+											  oprform->amoplefttype,
+											  GIST_DISTANCE_PROC)))
+			{
+				ereport(INFO,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("gist opfamily %s contains unsupported ORDER BY specification for operator %s",
+								opfamilyname,
+								format_operator(oprform->amopopr))));
+				result = false;
+			}
+			/* ... and operator result must match the claimed btree opfamily */
+			op_rettype = get_op_rettype(oprform->amopopr);
+			if (!opfamily_can_sort_type(oprform->amopsortfamily, op_rettype))
+			{
+				ereport(INFO,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("gist opfamily %s contains incorrect ORDER BY opfamily specification for operator %s",
+								opfamilyname,
+								format_operator(oprform->amopopr))));
+				result = false;
+			}
+		}
+		else
+		{
+			/* Search operators must always return bool */
+			op_rettype = BOOLOID;
 		}
 
-	/* Check that the named opclass is complete */
-	if (numclassops == 0)
-		ereport(ERROR,
+		/* Check operator signature */
+		if (!check_amop_signature(oprform->amopopr, op_rettype,
+								  oprform->amoplefttype,
+								  oprform->amoprighttype))
+		{
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-				 errmsg("gist opclass %u is missing operator(s)",
-						opclassoid)));
+					 errmsg("gist opfamily %s contains operator %s with wrong signature",
+							opfamilyname,
+							format_operator(oprform->amopopr))));
+			result = false;
+		}
+	}
+
+	/* Now check for inconsistent groups of operators/functions */
+	grouplist = identify_opfamily_groups(oprlist, proclist);
+	opclassgroup = NULL;
+	foreach(lc, grouplist)
+	{
+		OpFamilyOpFuncGroup *thisgroup = (OpFamilyOpFuncGroup *) lfirst(lc);
+
+		/* Remember the group exactly matching the test opclass */
+		if (thisgroup->lefttype == opcintype &&
+			thisgroup->righttype == opcintype)
+			opclassgroup = thisgroup;
+
+		/*
+		 * There is not a lot we can do to check the operator sets, since each
+		 * GiST opclass is more or less a law unto itself, and some contain
+		 * only operators that are binary-compatible with the opclass datatype
+		 * (meaning that empty operator sets can be OK).  That case also means
+		 * that we shouldn't insist on nonempty function sets except for the
+		 * opclass's own group.
+		 */
+	}
+
+	/* Check that the originally-named opclass is complete */
 	for (i = 1; i <= GISTNProcs; i++)
 	{
-		if ((classfuncbits & (1 << i)) != 0)
+		if (opclassgroup && (opclassgroup->functionset & (1 << i)) != 0)
 			continue;			/* got it */
 		if (i == GIST_DISTANCE_PROC || i == GIST_FETCH_PROC)
 			continue;			/* optional methods */
-		ereport(ERROR,
+		ereport(INFO,
 				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-			errmsg("gist opclass %u is missing required support function %d",
-				   opclassoid, i)));
+				 errmsg("gist opclass %s is missing support function %d",
+						opclassname, i)));
+		result = false;
 	}
 
 	ReleaseCatCacheList(proclist);
 	ReleaseCatCacheList(oprlist);
+	ReleaseSysCache(familytup);
+	ReleaseSysCache(classtup);
 
-	return true;
+	return result;
 }

src/backend/access/index/Makefile

@@ -12,6 +12,6 @@ subdir = src/backend/access/index
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = amapi.o genam.o indexam.o
+OBJS = amapi.o amvalidate.o genam.o indexam.o
 
 include $(top_srcdir)/src/backend/common.mk

src/backend/access/index/amvalidate.c

+/*-------------------------------------------------------------------------
+ *
+ * amvalidate.c
+ *	  Support routines for index access methods' amvalidate functions.
+ *
+ * Copyright (c) 2016, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/index/amvalidate.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/amvalidate.h"
+#include "access/htup_details.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_amop.h"
+#include "catalog/pg_amproc.h"
+#include "catalog/pg_opclass.h"
+#include "catalog/pg_operator.h"
+#include "catalog/pg_proc.h"
+#include "parser/parse_coerce.h"
+#include "utils/syscache.h"
+
+
+/*
+ * identify_opfamily_groups() returns a List of OpFamilyOpFuncGroup structs,
+ * one for each combination of lefttype/righttype present in the family's
+ * operator and support function lists.  If amopstrategy K is present for
+ * this datatype combination, we set bit 1 << K in operatorset, and similarly
+ * for the support functions.  With uint64 fields we can handle operator and
+ * function numbers up to 63, which is plenty for the foreseeable future.
+ *
+ * The given CatCLists are expected to represent a single opfamily fetched
+ * from the AMOPSTRATEGY and AMPROCNUM caches, so that they will be in
+ * order by those caches' second and third cache keys, namely the datatypes.
+ */
+List *
+identify_opfamily_groups(CatCList *oprlist, CatCList *proclist)
+{
+	List	   *result = NIL;
+	OpFamilyOpFuncGroup *thisgroup;
+	Form_pg_amop oprform;
+	Form_pg_amproc procform;
+	int			io,
+				ip;
+
+	/* We need the lists to be ordered; should be true in normal operation */
+	if (!oprlist->ordered || !proclist->ordered)
+		elog(ERROR, "cannot validate operator family without ordered data");
+
+	/*
+	 * Advance through the lists concurrently.  Thanks to the ordering, we
+	 * should see all operators and functions of a given datatype pair
+	 * consecutively.
+	 */
+	thisgroup = NULL;
+	io = ip = 0;
+	if (io < oprlist->n_members)
+	{
+		oprform = (Form_pg_amop) GETSTRUCT(&oprlist->members[io]->tuple);
+		io++;
+	}
+	else
+		oprform = NULL;
+	if (ip < proclist->n_members)
+	{
+		procform = (Form_pg_amproc) GETSTRUCT(&proclist->members[ip]->tuple);
+		ip++;
+	}
+	else
+		procform = NULL;
+
+	while (oprform || procform)
+	{
+		if (oprform && thisgroup &&
+			oprform->amoplefttype == thisgroup->lefttype &&
+			oprform->amoprighttype == thisgroup->righttype)
+		{
+			/* Operator belongs to current group; include it and advance */
+
+			/* Ignore strategy numbers outside supported range */
+			if (oprform->amopstrategy > 0 && oprform->amopstrategy < 64)
+				thisgroup->operatorset |= ((uint64) 1) << oprform->amopstrategy;
+
+			if (io < oprlist->n_members)
+			{
+				oprform = (Form_pg_amop) GETSTRUCT(&oprlist->members[io]->tuple);
+				io++;
+			}
+			else
+				oprform = NULL;
+			continue;
+		}
+
+		if (procform && thisgroup &&
+			procform->amproclefttype == thisgroup->lefttype &&
+			procform->amprocrighttype == thisgroup->righttype)
+		{
+			/* Procedure belongs to current group; include it and advance */
+
+			/* Ignore function numbers outside supported range */
+			if (procform->amprocnum > 0 && procform->amprocnum < 64)
+				thisgroup->functionset |= ((uint64) 1) << procform->amprocnum;
+
+			if (ip < proclist->n_members)
+			{
+				procform = (Form_pg_amproc) GETSTRUCT(&proclist->members[ip]->tuple);
+				ip++;
+			}
+			else
+				procform = NULL;
+			continue;
+		}
+
+		/* Time for a new group */
+		thisgroup = (OpFamilyOpFuncGroup *) palloc(sizeof(OpFamilyOpFuncGroup));
+		if (oprform &&
+			(!procform ||
+			 (oprform->amoplefttype < procform->amproclefttype ||
+			  (oprform->amoplefttype == procform->amproclefttype &&
+			   oprform->amoprighttype < procform->amprocrighttype))))
+		{
+			thisgroup->lefttype = oprform->amoplefttype;
+			thisgroup->righttype = oprform->amoprighttype;
+		}
+		else
+		{
+			thisgroup->lefttype = procform->amproclefttype;
+			thisgroup->righttype = procform->amprocrighttype;
+		}
+		thisgroup->operatorset = thisgroup->functionset = 0;
+		result = lappend(result, thisgroup);
+	}
+
+	return result;
+}
+
+/*
+ * Validate the signature (argument and result types) of an opclass support
+ * function.  Return TRUE if OK, FALSE if not.
+ *
+ * The "..." represents maxargs argument-type OIDs.  If "exact" is TRUE, they
+ * must match the function arg types exactly, else only binary-coercibly.
+ * In any case the function result type must match restype exactly.
+ */
+bool
+check_amproc_signature(Oid funcid, Oid restype, bool exact,
+					   int minargs, int maxargs,...)
+{
+	bool		result = true;
+	HeapTuple	tp;
+	Form_pg_proc procform;
+	va_list		ap;
+	int			i;
+
+	tp = SearchSysCache1(PROCOID, ObjectIdGetDatum(funcid));
+	if (!HeapTupleIsValid(tp))
+		elog(ERROR, "cache lookup failed for function %u", funcid);
+	procform = (Form_pg_proc) GETSTRUCT(tp);
+
+	if (procform->prorettype != restype || procform->proretset ||
+		procform->pronargs < minargs || procform->pronargs > maxargs)
+		result = false;
+
+	va_start(ap, maxargs);
+	for (i = 0; i < maxargs; i++)
+	{
+		Oid			argtype = va_arg(ap, Oid);
+
+		if (i >= procform->pronargs)
+			continue;
+		if (exact ? (argtype != procform->proargtypes.values[i]) :
+			!IsBinaryCoercible(argtype, procform->proargtypes.values[i]))
+			result = false;
+	}
+	va_end(ap);
+
+	ReleaseSysCache(tp);
+	return result;
+}
+
+/*
+ * Validate the signature (argument and result types) of an opclass operator.
+ * Return TRUE if OK, FALSE if not.
+ *
+ * Currently, we can hard-wire this as accepting only binary operators.  Also,
+ * we can insist on exact type matches, since the given lefttype/righttype
+ * come from pg_amop and should always match the operator exactly.
+ */
+bool
+check_amop_signature(Oid opno, Oid restype, Oid lefttype, Oid righttype)
+{
+	bool		result = true;
+	HeapTuple	tp;
+	Form_pg_operator opform;
+
+	tp = SearchSysCache1(OPEROID, ObjectIdGetDatum(opno));
+	if (!HeapTupleIsValid(tp))	/* shouldn't happen */
+		elog(ERROR, "cache lookup failed for operator %u", opno);
+	opform = (Form_pg_operator) GETSTRUCT(tp);
+
+	if (opform->oprresult != restype || opform->oprkind != 'b' ||
+		opform->oprleft != lefttype || opform->oprright != righttype)
+		result = false;
+
+	ReleaseSysCache(tp);
+	return result;
+}
+
+/*
+ * Is the datatype a legitimate input type for the btree opfamily?
+ */
+bool
+opfamily_can_sort_type(Oid opfamilyoid, Oid datatypeoid)
+{
+	bool		result = false;
+	CatCList   *opclist;
+	int			i;
+
+	/*
+	 * We search through all btree opclasses to see if one matches.  This is a
+	 * bit inefficient but there is no better index available.  It also saves
+	 * making an explicit check that the opfamily belongs to btree.
+	 */
+	opclist = SearchSysCacheList1(CLAAMNAMENSP, ObjectIdGetDatum(BTREE_AM_OID));
+
+	for (i = 0; i < opclist->n_members; i++)
+	{
+		HeapTuple	classtup = &opclist->members[i]->tuple;
+		Form_pg_opclass classform = (Form_pg_opclass) GETSTRUCT(classtup);
+
+		if (classform->opcfamily == opfamilyoid &&
+			classform->opcintype == datatypeoid)
+		{
+			result = true;
+			break;
+		}
+	}
+
+	ReleaseCatCacheList(opclist);
+
+	return result;
+}

src/backend/access/spgist/spgvalidate.c

@@ -13,30 +13,44 @@
  */
 #include "postgres.h"
 
+#include "access/amvalidate.h"
 #include "access/htup_details.h"
 #include "access/spgist_private.h"
 #include "catalog/pg_amop.h"
 #include "catalog/pg_amproc.h"
 #include "catalog/pg_opclass.h"
-#include "utils/catcache.h"
+#include "catalog/pg_opfamily.h"
+#include "catalog/pg_type.h"
+#include "utils/builtins.h"
 #include "utils/syscache.h"
 
 
 /*
  * Validator for an SP-GiST opclass.
+ *
+ * Some of the checks done here cover the whole opfamily, and therefore are
+ * redundant when checking each opclass in a family.  But they don't run long
+ * enough to be much of a problem, so we accept the duplication rather than
+ * complicate the amvalidate API.
  */
 bool
 spgvalidate(Oid opclassoid)
 {
+	bool		result = true;
 	HeapTuple	classtup;
 	Form_pg_opclass classform;
 	Oid			opfamilyoid;
 	Oid			opcintype;
-	int			numclassops;
-	int32		classfuncbits;
+	char	   *opclassname;
+	HeapTuple	familytup;
+	Form_pg_opfamily familyform;
+	char	   *opfamilyname;
 	CatCList   *proclist,
 			   *oprlist;
+	List	   *grouplist;
+	OpFamilyOpFuncGroup *opclassgroup;
 	int			i;
+	ListCell   *lc;
 
 	/* Fetch opclass information */
 	classtup = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclassoid));
@@ -46,84 +60,185 @@ spgvalidate(Oid opclassoid)
 
 	opfamilyoid = classform->opcfamily;
 	opcintype = classform->opcintype;
+	opclassname = NameStr(classform->opcname);
 
-	ReleaseSysCache(classtup);
+	/* Fetch opfamily information */
+	familytup = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(opfamilyoid));
+	if (!HeapTupleIsValid(familytup))
+		elog(ERROR, "cache lookup failed for operator family %u", opfamilyoid);
+	familyform = (Form_pg_opfamily) GETSTRUCT(familytup);
+
+	opfamilyname = NameStr(familyform->opfname);
 
 	/* Fetch all operators and support functions of the opfamily */
 	oprlist = SearchSysCacheList1(AMOPSTRATEGY, ObjectIdGetDatum(opfamilyoid));
 	proclist = SearchSysCacheList1(AMPROCNUM, ObjectIdGetDatum(opfamilyoid));
 
-	/* We'll track the ops and functions belonging to the named opclass */
-	numclassops = 0;
-	classfuncbits = 0;
-
-	/* Check support functions */
+	/* Check individual support functions */
 	for (i = 0; i < proclist->n_members; i++)
 	{
 		HeapTuple	proctup = &proclist->members[i]->tuple;
 		Form_pg_amproc procform = (Form_pg_amproc) GETSTRUCT(proctup);
+		bool		ok;
+
+		/*
+		 * All SP-GiST support functions should be registered with matching
+		 * left/right types
+		 */
+		if (procform->amproclefttype != procform->amprocrighttype)
+		{
+			ereport(INFO,
+					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					 errmsg("spgist opfamily %s contains support procedure %s with cross-type registration",
+							opfamilyname,
+							format_procedure(procform->amproc))));
+			result = false;
+		}
 
-		/* Check that only allowed procedure numbers exist */
-		if (procform->amprocnum < 1 ||
-			procform->amprocnum > SPGISTNProc)
-			ereport(ERROR,
+		/* Check procedure numbers and function signatures */
+		switch (procform->amprocnum)
+		{
+			case SPGIST_CONFIG_PROC:
+			case SPGIST_CHOOSE_PROC:
+			case SPGIST_PICKSPLIT_PROC:
+			case SPGIST_INNER_CONSISTENT_PROC:
+				ok = check_amproc_signature(procform->amproc, VOIDOID, true,
+											2, 2, INTERNALOID, INTERNALOID);
+				break;
+			case SPGIST_LEAF_CONSISTENT_PROC:
+				ok = check_amproc_signature(procform->amproc, BOOLOID, true,
+											2, 2, INTERNALOID, INTERNALOID);
+				break;
+			default:
+				ereport(INFO,
 						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("spgist opfamily %u contains invalid support number %d for procedure %u",
-							opfamilyoid,
-							procform->amprocnum, procform->amproc)));
-
-		/* Remember functions that are specifically for the named opclass */
-		if (procform->amproclefttype == opcintype &&
-			procform->amprocrighttype == opcintype)
-			classfuncbits |= (1 << procform->amprocnum);
+						 errmsg("spgist opfamily %s contains function %s with invalid support number %d",
+								opfamilyname,
+								format_procedure(procform->amproc),
+								procform->amprocnum)));
+				result = false;
+				continue;		/* don't want additional message */
 		}
 
-	/* Check operators */
+		if (!ok)
+		{
+			ereport(INFO,
+					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					 errmsg("spgist opfamily %s contains function %s with wrong signature for support number %d",
+							opfamilyname,
+							format_procedure(procform->amproc),
+							procform->amprocnum)));
+			result = false;
+		}
+	}
+
+	/* Check individual operators */
 	for (i = 0; i < oprlist->n_members; i++)
 	{
 		HeapTuple	oprtup = &oprlist->members[i]->tuple;
 		Form_pg_amop oprform = (Form_pg_amop) GETSTRUCT(oprtup);
 
 		/* TODO: Check that only allowed strategy numbers exist */
-		if (oprform->amopstrategy < 1)
-			ereport(ERROR,
+		if (oprform->amopstrategy < 1 || oprform->amopstrategy > 63)
+		{
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("spgist opfamily %u contains invalid strategy number %d for operator %u",
-							opfamilyoid,
-							oprform->amopstrategy, oprform->amopopr)));
+					 errmsg("spgist opfamily %s contains operator %s with invalid strategy number %d",
+							opfamilyname,
+							format_operator(oprform->amopopr),
+							oprform->amopstrategy)));
+			result = false;
+		}
 
 		/* spgist doesn't support ORDER BY operators */
 		if (oprform->amoppurpose != AMOP_SEARCH ||
 			OidIsValid(oprform->amopsortfamily))
-			ereport(ERROR,
+		{
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("spgist opfamily %u contains invalid ORDER BY specification for operator %u",
-							opfamilyoid, oprform->amopopr)));
+					 errmsg("spgist opfamily %s contains invalid ORDER BY specification for operator %s",
+							opfamilyname,
+							format_operator(oprform->amopopr))));
+			result = false;
+		}
 
-		/* Count operators that are specifically for the named opclass */
-		if (oprform->amoplefttype == opcintype &&
-			oprform->amoprighttype == opcintype)
-			numclassops++;
+		/* Check operator signature --- same for all spgist strategies */
+		if (!check_amop_signature(oprform->amopopr, BOOLOID,
+								  oprform->amoplefttype,
+								  oprform->amoprighttype))
+		{
+			ereport(INFO,
+					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					 errmsg("spgist opfamily %s contains operator %s with wrong signature",
+							opfamilyname,
+							format_operator(oprform->amopopr))));
+			result = false;
+		}
 	}
 
-	/* Check that the named opclass is complete */
-	if (numclassops == 0)
-		ereport(ERROR,
+	/* Now check for inconsistent groups of operators/functions */
+	grouplist = identify_opfamily_groups(oprlist, proclist);
+	opclassgroup = NULL;
+	foreach(lc, grouplist)
+	{
+		OpFamilyOpFuncGroup *thisgroup = (OpFamilyOpFuncGroup *) lfirst(lc);
+
+		/* Remember the group exactly matching the test opclass */
+		if (thisgroup->lefttype == opcintype &&
+			thisgroup->righttype == opcintype)
+			opclassgroup = thisgroup;
+
+		/*
+		 * Complain if there are any datatype pairs with functions but no
+		 * operators.  This is about the best we can do for now to detect
+		 * missing operators.
+		 */
+		if (thisgroup->operatorset == 0)
+		{
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-				 errmsg("spgist opclass %u is missing operator(s)",
-						opclassoid)));
+					 errmsg("spgist opfamily %s is missing operator(s) for types %s and %s",
+							opfamilyname,
+							format_type_be(thisgroup->lefttype),
+							format_type_be(thisgroup->righttype))));
+			result = false;
+		}
+
+		/*
+		 * Complain if we're missing functions for any datatype, remembering
+		 * that SP-GiST doesn't use cross-type support functions.
+		 */
+		if (thisgroup->lefttype != thisgroup->righttype)
+			continue;
+
 		for (i = 1; i <= SPGISTNProc; i++)
 		{
-		if ((classfuncbits & (1 << i)) != 0)
+			if ((thisgroup->functionset & (1 << i)) != 0)
 				continue;		/* got it */
-		ereport(ERROR,
+			ereport(INFO,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-		  errmsg("spgist opclass %u is missing required support function %d",
-				 opclassoid, i)));
+					 errmsg("spgist opfamily %s is missing support function %d for type %s",
+							opfamilyname, i,
+							format_type_be(thisgroup->lefttype))));
+			result = false;
+		}
+	}
+
+	/* Check that the originally-named opclass is supported */
+	/* (if group is there, we already checked it adequately above) */
+	if (!opclassgroup)
+	{
+		ereport(INFO,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("spgist opclass %s is missing operator(s)",
+						opclassname)));
+		result = false;
 	}
 
 	ReleaseCatCacheList(proclist);
 	ReleaseCatCacheList(oprlist);
+	ReleaseSysCache(familytup);
+	ReleaseSysCache(classtup);
 
-	return true;
+	return result;
 }

src/backend/utils/cache/lsyscache.c

@@ -1102,6 +1102,29 @@ get_opname(Oid opno)
 		return NULL;
 }
 
+/*
+ * get_op_rettype
+ *		Given operator oid, return the operator's result type.
+ */
+Oid
+get_op_rettype(Oid opno)
+{
+	HeapTuple	tp;
+
+	tp = SearchSysCache1(OPEROID, ObjectIdGetDatum(opno));
+	if (HeapTupleIsValid(tp))
+	{
+		Form_pg_operator optup = (Form_pg_operator) GETSTRUCT(tp);
+		Oid			result;
+
+		result = optup->oprresult;
+		ReleaseSysCache(tp);
+		return result;
+	}
+	else
+		return InvalidOid;
+}
+
 /*
  * op_input_types
  *

src/include/access/amvalidate.h

+/*-------------------------------------------------------------------------
+ *
+ * amvalidate.h
+ *	  Support routines for index access methods' amvalidate functions.
+ *
+ * Copyright (c) 2016, PostgreSQL Global Development Group
+ *
+ * src/include/access/amvalidate.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef AMVALIDATE_H
+#define AMVALIDATE_H
+
+#include "utils/catcache.h"
+
+
+/* Struct returned (in a list) by identify_opfamily_groups() */
+typedef struct OpFamilyOpFuncGroup
+{
+	Oid			lefttype;		/* amoplefttype/amproclefttype */
+	Oid			righttype;		/* amoprighttype/amprocrighttype */
+	uint64		operatorset;	/* bitmask of operators with these types */
+	uint64		functionset;	/* bitmask of support funcs with these types */
+} OpFamilyOpFuncGroup;
+
+
+/* Functions in access/index/amvalidate.c */
+extern List *identify_opfamily_groups(CatCList *oprlist, CatCList *proclist);
+extern bool check_amproc_signature(Oid funcid, Oid restype, bool exact,
+					   int minargs, int maxargs,...);
+extern bool check_amop_signature(Oid opno, Oid restype,
+					 Oid lefttype, Oid righttype);
+extern bool opfamily_can_sort_type(Oid opfamilyoid, Oid datatypeoid);
+
+#endif   /* AMVALIDATE_H */

src/include/utils/lsyscache.h

@@ -75,6 +75,7 @@ extern Oid	get_opclass_family(Oid opclass);
 extern Oid	get_opclass_input_type(Oid opclass);
 extern RegProcedure get_opcode(Oid opno);
 extern char *get_opname(Oid opno);
+extern Oid	get_op_rettype(Oid opno);
 extern void op_input_types(Oid opno, Oid *lefttype, Oid *righttype);
 extern bool op_mergejoinable(Oid opno, Oid inputtype);
 extern bool op_hashjoinable(Oid opno, Oid inputtype);

src/test/regress/sql/opr_sanity.sql

@@ -1037,6 +1037,7 @@ WHERE p1.oid != p2.oid AND
     p1.opcdefault AND p2.opcdefault;
 
 -- Ask access methods to validate opclasses
+-- (this replaces a lot of SQL-level checks that used to be done in this file)
 
 SELECT oid, opcname FROM pg_opclass WHERE NOT amvalidate(oid);
 
@@ -1073,47 +1074,12 @@ FROM pg_amop as p1
 WHERE NOT ((p1.amoppurpose = 's' AND p1.amopsortfamily = 0) OR
            (p1.amoppurpose = 'o' AND p1.amopsortfamily <> 0));
 
--- amoplefttype/amoprighttype must match the operator
-
-SELECT p1.oid, p2.oid
-FROM pg_amop AS p1, pg_operator AS p2
-WHERE p1.amopopr = p2.oid AND NOT
-    (p1.amoplefttype = p2.oprleft AND p1.amoprighttype = p2.oprright);
 -- amopmethod must match owning opfamily's opfmethod
 
 SELECT p1.oid, p2.oid
 FROM pg_amop AS p1, pg_opfamily AS p2
 WHERE p1.amopfamily = p2.oid AND p1.amopmethod != p2.opfmethod;
 
-
--- amopsortfamily, if present, must reference a btree family
-
-SELECT p1.amopfamily, p1.amopstrategy
-FROM pg_amop AS p1
-WHERE p1.amopsortfamily <> 0 AND NOT EXISTS
-    (SELECT 1 from pg_opfamily op WHERE op.oid = p1.amopsortfamily
-     AND op.opfmethod = (SELECT oid FROM pg_am WHERE amname = 'btree'));
-
--- Check that amopopr points at a reasonable-looking operator, ie a binary
--- operator.  If it's a search operator it had better yield boolean,
--- otherwise an input type of its sort opfamily.
-
-SELECT p1.amopfamily, p1.amopopr, p2.oid, p2.oprname
-FROM pg_amop AS p1, pg_operator AS p2
-WHERE p1.amopopr = p2.oid AND
-    p2.oprkind != 'b';
-
-SELECT p1.amopfamily, p1.amopopr, p2.oid, p2.oprname
-FROM pg_amop AS p1, pg_operator AS p2
-WHERE p1.amopopr = p2.oid AND p1.amoppurpose = 's' AND
-    p2.oprresult != 'bool'::regtype;
-
-SELECT p1.amopfamily, p1.amopopr, p2.oid, p2.oprname
-FROM pg_amop AS p1, pg_operator AS p2
-WHERE p1.amopopr = p2.oid AND p1.amoppurpose = 'o' AND NOT EXISTS
-    (SELECT 1 FROM pg_opclass op
 -- Make a list of all the distinct operator names being used in particular
 -- strategy slots.  This is a bit hokey, since the list might need to change
 -- in future releases, but it's an effective way of spotting mistakes such as
@@ -1171,65 +1137,6 @@ WHERE p1.amopopr = p2.oid AND p2.oprcode = p3.oid AND
     p1.amoplefttype != p1.amoprighttype AND
     p3.provolatile = 'v';
 
--- Multiple-datatype btree opfamilies should provide closed sets of equality
--- operators; that is if you provide int2 = int4 and int4 = int8 then you
--- should also provide int2 = int8 (and commutators of all these).  This is
--- important because the planner tries to deduce additional qual clauses from
--- transitivity of mergejoinable operators.  If there are clauses
--- int2var = int4var and int4var = int8var, the planner will want to deduce
--- int2var = int8var ... so there should be a way to represent that.  While
--- a missing cross-type operator is now only an efficiency loss rather than
--- an error condition, it still seems reasonable to insist that all built-in
--- opfamilies be complete.
-
--- check commutative closure
-SELECT p1.amoplefttype, p1.amoprighttype
-FROM pg_amop AS p1
-WHERE p1.amopmethod = (SELECT oid FROM pg_am WHERE amname = 'btree') AND
-    p1.amopstrategy = 3 AND
-    p1.amoplefttype != p1.amoprighttype AND
-    NOT EXISTS(SELECT 1 FROM pg_amop p2 WHERE
-                 p2.amopfamily = p1.amopfamily AND
-                 p2.amoplefttype = p1.amoprighttype AND
-                 p2.amoprighttype = p1.amoplefttype AND
-                 p2.amopstrategy = 3);
-
--- check transitive closure
-SELECT p1.amoplefttype, p1.amoprighttype, p2.amoprighttype
-FROM pg_amop AS p1, pg_amop AS p2
-WHERE p1.amopfamily = p2.amopfamily AND
-    p1.amoprighttype = p2.amoplefttype AND
-    p1.amopmethod = (SELECT oid FROM pg_am WHERE amname = 'btree') AND
-    p2.amopmethod = (SELECT oid FROM pg_am WHERE amname = 'btree') AND
-    p1.amopstrategy = 3 AND p2.amopstrategy = 3 AND
-    p1.amoplefttype != p1.amoprighttype AND
-    p2.amoplefttype != p2.amoprighttype AND
-    NOT EXISTS(SELECT 1 FROM pg_amop p3 WHERE
-                 p3.amopfamily = p1.amopfamily AND
-                 p3.amoplefttype = p1.amoplefttype AND
-                 p3.amoprighttype = p2.amoprighttype AND
-                 p3.amopstrategy = 3);
-
--- We also expect that built-in multiple-datatype hash opfamilies provide
--- complete sets of cross-type operators.  Again, this isn't required, but
--- it is reasonable to expect it for built-in opfamilies.
-
--- if same family has x=x and y=y, it should have x=y
-SELECT p1.amoplefttype, p2.amoplefttype
-FROM pg_amop AS p1, pg_amop AS p2
-WHERE p1.amopfamily = p2.amopfamily AND
-    p1.amoplefttype = p1.amoprighttype AND
-    p2.amoplefttype = p2.amoprighttype AND
-    p1.amopmethod = (SELECT oid FROM pg_am WHERE amname = 'hash') AND
-    p2.amopmethod = (SELECT oid FROM pg_am WHERE amname = 'hash') AND
-    p1.amopstrategy = 1 AND p2.amopstrategy = 1 AND
-    p1.amoplefttype != p2.amoplefttype AND
-    NOT EXISTS(SELECT 1 FROM pg_amop p3 WHERE
-                 p3.amopfamily = p1.amopfamily AND
-                 p3.amoplefttype = p1.amoplefttype AND
-                 p3.amoprighttype = p2.amoplefttype AND
-                 p3.amopstrategy = 1);
-
 
 -- **************** pg_amproc ****************
 
@@ -1240,82 +1147,6 @@ FROM pg_amproc as p1
 WHERE p1.amprocfamily = 0 OR p1.amproclefttype = 0 OR p1.amprocrighttype = 0
     OR p1.amprocnum < 1 OR p1.amproc = 0;
 
--- Unfortunately, we can't check the amproc link very well because the
--- signature of the function may be different for different support routines
--- or different base data types.
--- We can check that all the referenced instances of the same support
--- routine number take the same number of parameters, but that's about it
--- for a general check...
-
-SELECT p1.amprocfamily, p1.amprocnum,
-	p2.oid, p2.proname,
-	p3.opfname,
-	p4.amprocfamily, p4.amprocnum,
-	p5.oid, p5.proname,
-	p6.opfname
-FROM pg_amproc AS p1, pg_proc AS p2, pg_opfamily AS p3,
-     pg_amproc AS p4, pg_proc AS p5, pg_opfamily AS p6
-WHERE p1.amprocfamily = p3.oid AND p4.amprocfamily = p6.oid AND
-    p3.opfmethod = p6.opfmethod AND p1.amprocnum = p4.amprocnum AND
-    p1.amproc = p2.oid AND p4.amproc = p5.oid AND
-    (p2.proretset OR p5.proretset OR p2.pronargs != p5.pronargs);
-
--- For btree, though, we can do better since we know the support routines
--- must be of the form cmp(lefttype, righttype) returns int4
--- or sortsupport(internal) returns void.
-
-SELECT p1.amprocfamily, p1.amprocnum,
-	p2.oid, p2.proname,
-	p3.opfname
-FROM pg_amproc AS p1, pg_proc AS p2, pg_opfamily AS p3
-WHERE p3.opfmethod = (SELECT oid FROM pg_am WHERE amname = 'btree')
-    AND p1.amprocfamily = p3.oid AND p1.amproc = p2.oid AND
-    (CASE WHEN amprocnum = 1
-          THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
-               OR proargtypes[0] != amproclefttype
-               OR proargtypes[1] != amprocrighttype
-          WHEN amprocnum = 2
-          THEN prorettype != 'void'::regtype OR proretset OR pronargs != 1
-               OR proargtypes[0] != 'internal'::regtype
-          ELSE true END);
-
--- For hash we can also do a little better: the support routines must be
--- of the form hash(lefttype) returns int4.  There are several cases where
--- we cheat and use a hash function that is physically compatible with the
--- datatype even though there's no cast, so this check does find a small
--- number of entries.
-
-SELECT p1.amprocfamily, p1.amprocnum, p2.proname, p3.opfname
-FROM pg_amproc AS p1, pg_proc AS p2, pg_opfamily AS p3
-WHERE p3.opfmethod = (SELECT oid FROM pg_am WHERE amname = 'hash')
-    AND p1.amprocfamily = p3.oid AND p1.amproc = p2.oid AND
-    (amprocnum != 1
-     OR proretset
-     OR prorettype != 'int4'::regtype
-     OR pronargs != 1
-     OR NOT physically_coercible(amproclefttype, proargtypes[0])
-     OR amproclefttype != amprocrighttype)
-ORDER BY 1;
-
--- We can also check SP-GiST carefully, since the support routine signatures
--- are independent of the datatype being indexed.
-
-SELECT p1.amprocfamily, p1.amprocnum,
-	p2.oid, p2.proname,
-	p3.opfname
-FROM pg_amproc AS p1, pg_proc AS p2, pg_opfamily AS p3
-WHERE p3.opfmethod = (SELECT oid FROM pg_am WHERE amname = 'spgist')
-    AND p1.amprocfamily = p3.oid AND p1.amproc = p2.oid AND
-    (CASE WHEN amprocnum = 1 OR amprocnum = 2 OR amprocnum = 3 OR amprocnum = 4
-          THEN prorettype != 'void'::regtype OR proretset OR pronargs != 2
-               OR proargtypes[0] != 'internal'::regtype
-               OR proargtypes[1] != 'internal'::regtype
-          WHEN amprocnum = 5
-          THEN prorettype != 'bool'::regtype OR proretset OR pronargs != 2
-               OR proargtypes[0] != 'internal'::regtype
-               OR proargtypes[1] != 'internal'::regtype
-          ELSE true END);
-
 -- Support routines that are primary members of opfamilies must be immutable
 -- (else it suggests that the index ordering isn't fixed).  But cross-type
 -- members need only be stable, since they are just shorthands