Create a 'type cache' that keeps track of the data needed for any particular

datatype by array_eq and array_cmp; use this to solve problems with memory
leaks in array indexing support.  The parser's equality_oper and ordering_oper
routines also use the cache.  Change the operator search algorithms to look
for appropriate btree or hash index opclasses, instead of assuming operators
named '<' or '=' have the right semantics.  (ORDER BY ASC/DESC now also look
at opclasses, instead of assuming '<' and '>' are the right things.)  Add
several more index opclasses so that there is no regression in functionality
for base datatypes.  initdb forced due to catalog additions.

3 files changed, 196 insertions, 103 deletions

diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index d3060b76c7a..ef6f8a80c5c 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/gram.y,v 2.428 2003/08/04 02:40:01 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/gram.y,v 2.429 2003/08/17 19:58:05 tgl Exp $
  *
  * HISTORY
  *	  AUTHOR			DATE			MAJOR EVENT
@@ -115,7 +115,7 @@ static void doNegateFloat(Value *v);
 
 	TypeName			*typnam;
 	DefElem				*defelt;
-	SortGroupBy			*sortgroupby;
+	SortBy				*sortby;
 	JoinExpr			*jexpr;
 	IndexElem			*ielem;
 	Alias				*alias;
@@ -189,7 +189,7 @@ static void doNegateFloat(Value *v);
 				database_name access_method_clause access_method attr_name
 				index_name name function_name file_name
 
-%type <list>	func_name handler_name qual_Op qual_all_Op OptUseOp
+%type <list>	func_name handler_name qual_Op qual_all_Op
 				opt_class opt_validator
 
 %type <range>	qualified_name OptConstrFromTable
@@ -278,7 +278,7 @@ static void doNegateFloat(Value *v);
 %type <value>	NumericOnly FloatOnly IntegerOnly
 %type <columnref>	columnref
 %type <alias>	alias_clause
-%type <sortgroupby>		sortby
+%type <sortby>	sortby
 %type <ielem>	index_elem
 %type <node>	table_ref
 %type <jexpr>	joined_table
@@ -4577,21 +4577,34 @@ sortby_list:
 			| sortby_list ',' sortby				{ $$ = lappend($1, $3); }
 		;
 
-sortby:		a_expr OptUseOp
+sortby:		a_expr USING qual_all_Op
 				{
-					$$ = makeNode(SortGroupBy);
+					$$ = makeNode(SortBy);
 					$$->node = $1;
-					$$->useOp = $2;
+					$$->sortby_kind = SORTBY_USING;
+					$$->useOp = $3;
+				}
+			| a_expr ASC
+				{
+					$$ = makeNode(SortBy);
+					$$->node = $1;
+					$$->sortby_kind = SORTBY_ASC;
+					$$->useOp = NIL;
+				}
+			| a_expr DESC
+				{
+					$$ = makeNode(SortBy);
+					$$->node = $1;
+					$$->sortby_kind = SORTBY_DESC;
+					$$->useOp = NIL;
+				}
+			| a_expr
+				{
+					$$ = makeNode(SortBy);
+					$$->node = $1;
+					$$->sortby_kind = SORTBY_ASC;	/* default */
+					$$->useOp = NIL;
 				}
-		;
-
-OptUseOp:	USING qual_all_Op						{ $$ = $2; }
-			| ASC
-							{ $$ = makeList1(makeString("<")); }
-			| DESC
-							{ $$ = makeList1(makeString(">")); }
-			| /*EMPTY*/
-							{ $$ = makeList1(makeString("<"));	/*default*/ }
 		;
 
 
diff --git a/src/backend/parser/parse_clause.c b/src/backend/parser/parse_clause.c
index ebc3ed23eec..b31e70205dd 100644
--- a/src/backend/parser/parse_clause.c
+++ b/src/backend/parser/parse_clause.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_clause.c,v 1.121 2003/08/07 19:20:22 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_clause.c,v 1.122 2003/08/17 19:58:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1295,7 +1295,7 @@ transformSortClause(ParseState *pstate,
 
 	foreach(olitem, orderlist)
 	{
-		SortGroupBy *sortby = lfirst(olitem);
+		SortBy	   *sortby = lfirst(olitem);
 		TargetEntry *tle;
 
 		tle = findTargetlistEntry(pstate, sortby->node,
@@ -1303,7 +1303,9 @@ transformSortClause(ParseState *pstate,
 
 		sortlist = addTargetToSortList(pstate, tle,
 									   sortlist, targetlist,
-									   sortby->useOp, resolveUnknown);
+									   sortby->sortby_kind,
+									   sortby->useOp,
+									   resolveUnknown);
 	}
 
 	return sortlist;
@@ -1409,7 +1411,7 @@ transformDistinctClause(ParseState *pstate, List *distinctlist,
 			{
 				*sortClause = addTargetToSortList(pstate, tle,
 												  *sortClause, targetlist,
-												  NIL, true);
+												  SORTBY_ASC, NIL, true);
 
 				/*
 				 * Probably, the tle should always have been added at the
@@ -1457,7 +1459,8 @@ addAllTargetsToSortList(ParseState *pstate, List *sortlist,
 		if (!tle->resdom->resjunk)
 			sortlist = addTargetToSortList(pstate, tle,
 										   sortlist, targetlist,
-										   NIL, resolveUnknown);
+										   SORTBY_ASC, NIL,
+										   resolveUnknown);
 	}
 	return sortlist;
 }
@@ -1478,7 +1481,8 @@ addAllTargetsToSortList(ParseState *pstate, List *sortlist,
 List *
 addTargetToSortList(ParseState *pstate, TargetEntry *tle,
 					List *sortlist, List *targetlist,
-					List *opname, bool resolveUnknown)
+					int sortby_kind, List *sortby_opname,
+					bool resolveUnknown)
 {
 	/* avoid making duplicate sortlist entries */
 	if (!targetIsInSortList(tle, sortlist))
@@ -1499,13 +1503,25 @@ addTargetToSortList(ParseState *pstate, TargetEntry *tle,
 
 		sortcl->tleSortGroupRef = assignSortGroupRef(tle, targetlist);
 
-		if (opname)
-			sortcl->sortop = compatible_oper_opid(opname,
-												  restype,
-												  restype,
-												  false);
-		else
-			sortcl->sortop = ordering_oper_opid(restype);
+		switch (sortby_kind)
+		{
+			case SORTBY_ASC:
+				sortcl->sortop = ordering_oper_opid(restype);
+				break;
+			case SORTBY_DESC:
+				sortcl->sortop = reverse_ordering_oper_opid(restype);
+				break;
+			case SORTBY_USING:
+				Assert(sortby_opname != NIL);
+				sortcl->sortop = compatible_oper_opid(sortby_opname,
+													  restype,
+													  restype,
+													  false);
+				break;
+			default:
+				elog(ERROR, "unrecognized sortby_kind: %d", sortby_kind);
+				break;
+		}
 
 		sortlist = lappend(sortlist, sortcl);
 	}
diff --git a/src/backend/parser/parse_oper.c b/src/backend/parser/parse_oper.c
index e5d2ab2d05f..244cd769646 100644
--- a/src/backend/parser/parse_oper.c
+++ b/src/backend/parser/parse_oper.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_oper.c,v 1.73 2003/08/04 02:40:02 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/parser/parse_oper.c,v 1.74 2003/08/17 19:58:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -26,6 +26,7 @@
 #include "utils/fmgroids.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
+#include "utils/typcache.h"
 
 
 static Oid binary_oper_exact(Oid arg1, Oid arg2,
@@ -135,52 +136,49 @@ LookupOperNameTypeNames(List *opername, TypeName *oprleft,
 Operator
 equality_oper(Oid argtype, bool noError)
 {
+	TypeCacheEntry *typentry;
+	Oid			oproid;
 	Operator	optup;
-	Oid			elem_type;
+
+	/*
+	 * Look for an "=" operator for the datatype.  We require it to be
+	 * an exact or binary-compatible match, since most callers are not
+	 * prepared to cope with adding any run-time type coercion steps.
+	 */
+	typentry = lookup_type_cache(argtype, TYPECACHE_EQ_OPR);
+	oproid = typentry->eq_opr;
 
 	/*
 	 * If the datatype is an array, then we can use array_eq ... but only
-	 * if there is a suitable equality operator for the element type. (We
-	 * must run this test first, since compatible_oper will find array_eq,
-	 * but would not notice the lack of an element operator.)
+	 * if there is a suitable equality operator for the element type.
+	 * (This check is not in the raw typcache.c code ... should it be?)
 	 */
-	elem_type = get_element_type(argtype);
-	if (OidIsValid(elem_type))
+	if (oproid == ARRAY_EQ_OP)
 	{
-		optup = equality_oper(elem_type, true);
-		if (optup != NULL)
+		Oid		elem_type = get_element_type(argtype);
+
+		if (OidIsValid(elem_type))
 		{
-			ReleaseSysCache(optup);
-			return SearchSysCache(OPEROID,
-								  ObjectIdGetDatum(ARRAY_EQ_OP),
-								  0, 0, 0);
+			optup = equality_oper(elem_type, true);
+			if (optup != NULL)
+				ReleaseSysCache(optup);
+			else
+				oproid = InvalidOid;	/* element type has no "=" */
 		}
+		else
+			oproid = InvalidOid;		/* bogus array type? */
 	}
-	else
-	{
-		/*
-		 * Look for an "=" operator for the datatype.  We require it to be
-		 * an exact or binary-compatible match, since most callers are not
-		 * prepared to cope with adding any run-time type coercion steps.
-		 */
-		optup = compatible_oper(makeList1(makeString("=")),
-								argtype, argtype, true);
-		if (optup != NULL)
-		{
-			/*
-			 * Only believe that it's equality if it's mergejoinable,
-			 * hashjoinable, or uses eqsel() as oprrest.
-			 */
-			Form_pg_operator pgopform = (Form_pg_operator) GETSTRUCT(optup);
-
-			if (OidIsValid(pgopform->oprlsortop) ||
-				pgopform->oprcanhash ||
-				pgopform->oprrest == F_EQSEL)
-				return optup;
 
-			ReleaseSysCache(optup);
-		}
+	if (OidIsValid(oproid))
+	{
+		optup = SearchSysCache(OPEROID,
+							   ObjectIdGetDatum(oproid),
+							   0, 0, 0);
+		if (optup == NULL)		/* should not fail */
+			elog(ERROR, "cache lookup failed for operator %u", oproid);
+		return optup;
 	}
+
 	if (!noError)
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_FUNCTION),
@@ -197,53 +195,119 @@ equality_oper(Oid argtype, bool noError)
 Operator
 ordering_oper(Oid argtype, bool noError)
 {
+	TypeCacheEntry *typentry;
+	Oid			oproid;
 	Operator	optup;
-	Oid			elem_type;
+
+	/*
+	 * Look for a "<" operator for the datatype.  We require it to be
+	 * an exact or binary-compatible match, since most callers are not
+	 * prepared to cope with adding any run-time type coercion steps.
+	 *
+	 * Note: the search algorithm used by typcache.c ensures that if a "<"
+	 * operator is returned, it will be consistent with the "=" operator
+	 * returned by equality_oper.  This is critical for sorting and grouping
+	 * purposes.
+	 */
+	typentry = lookup_type_cache(argtype, TYPECACHE_LT_OPR);
+	oproid = typentry->lt_opr;
 
 	/*
 	 * If the datatype is an array, then we can use array_lt ... but only
-	 * if there is a suitable ordering operator for the element type. (We
-	 * must run this test first, since the code below would find array_lt
-	 * if there's an element = operator, but would not notice the lack of
-	 * an element < operator.)
+	 * if there is a suitable less-than operator for the element type.
+	 * (This check is not in the raw typcache.c code ... should it be?)
 	 */
-	elem_type = get_element_type(argtype);
-	if (OidIsValid(elem_type))
+	if (oproid == ARRAY_LT_OP)
 	{
-		optup = ordering_oper(elem_type, true);
-		if (optup != NULL)
+		Oid		elem_type = get_element_type(argtype);
+
+		if (OidIsValid(elem_type))
 		{
-			ReleaseSysCache(optup);
-			return SearchSysCache(OPEROID,
-								  ObjectIdGetDatum(ARRAY_LT_OP),
-								  0, 0, 0);
+			optup = ordering_oper(elem_type, true);
+			if (optup != NULL)
+				ReleaseSysCache(optup);
+			else
+				oproid = InvalidOid;	/* element type has no "<" */
 		}
+		else
+			oproid = InvalidOid;		/* bogus array type? */
 	}
-	else
+
+	if (OidIsValid(oproid))
 	{
-		/*
-		 * Find the type's equality operator, and use its lsortop (it
-		 * *must* be mergejoinable).  We use this definition because for
-		 * sorting and grouping purposes, it's important that the equality
-		 * and ordering operators are consistent.
-		 */
-		optup = equality_oper(argtype, noError);
-		if (optup != NULL)
-		{
-			Oid			lsortop;
+		optup = SearchSysCache(OPEROID,
+							   ObjectIdGetDatum(oproid),
+							   0, 0, 0);
+		if (optup == NULL)		/* should not fail */
+			elog(ERROR, "cache lookup failed for operator %u", oproid);
+		return optup;
+	}
 
-			lsortop = ((Form_pg_operator) GETSTRUCT(optup))->oprlsortop;
-			ReleaseSysCache(optup);
-			if (OidIsValid(lsortop))
-			{
-				optup = SearchSysCache(OPEROID,
-									   ObjectIdGetDatum(lsortop),
-									   0, 0, 0);
-				if (optup != NULL)
-					return optup;
-			}
+	if (!noError)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_FUNCTION),
+			errmsg("could not identify an ordering operator for type %s",
+				   format_type_be(argtype)),
+				 errhint("Use an explicit ordering operator or modify the query.")));
+	return NULL;
+}
+
+/*
+ * reverse_ordering_oper - identify DESC sort operator (">") for a datatype
+ *
+ * On failure, return NULL if noError, else report a standard error
+ */
+Operator
+reverse_ordering_oper(Oid argtype, bool noError)
+{
+	TypeCacheEntry *typentry;
+	Oid			oproid;
+	Operator	optup;
+
+	/*
+	 * Look for a ">" operator for the datatype.  We require it to be
+	 * an exact or binary-compatible match, since most callers are not
+	 * prepared to cope with adding any run-time type coercion steps.
+	 *
+	 * Note: the search algorithm used by typcache.c ensures that if a ">"
+	 * operator is returned, it will be consistent with the "=" operator
+	 * returned by equality_oper.  This is critical for sorting and grouping
+	 * purposes.
+	 */
+	typentry = lookup_type_cache(argtype, TYPECACHE_GT_OPR);
+	oproid = typentry->gt_opr;
+
+	/*
+	 * If the datatype is an array, then we can use array_gt ... but only
+	 * if there is a suitable greater-than operator for the element type.
+	 * (This check is not in the raw typcache.c code ... should it be?)
+	 */
+	if (oproid == ARRAY_GT_OP)
+	{
+		Oid		elem_type = get_element_type(argtype);
+
+		if (OidIsValid(elem_type))
+		{
+			optup = reverse_ordering_oper(elem_type, true);
+			if (optup != NULL)
+				ReleaseSysCache(optup);
+			else
+				oproid = InvalidOid;	/* element type has no ">" */
 		}
+		else
+			oproid = InvalidOid;		/* bogus array type? */
 	}
+
+	if (OidIsValid(oproid))
+	{
+		optup = SearchSysCache(OPEROID,
+							   ObjectIdGetDatum(oproid),
+							   0, 0, 0);
+		if (optup == NULL)		/* should not fail */
+			elog(ERROR, "cache lookup failed for operator %u", oproid);
+		return optup;
+	}
+
 	if (!noError)
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_FUNCTION),
@@ -286,16 +350,16 @@ ordering_oper_opid(Oid argtype)
 }
 
 /*
- * ordering_oper_funcid - convenience routine for oprfuncid(ordering_oper())
+ * reverse_ordering_oper_opid - convenience routine for oprid(reverse_ordering_oper())
  */
 Oid
-ordering_oper_funcid(Oid argtype)
+reverse_ordering_oper_opid(Oid argtype)
 {
 	Operator	optup;
 	Oid			result;
 
-	optup = ordering_oper(argtype, false);
-	result = oprfuncid(optup);
+	optup = reverse_ordering_oper(argtype, false);
+	result = oprid(optup);
 	ReleaseSysCache(optup);
 	return result;
 }