Support ORDER BY ... NULLS FIRST/LAST, and add ASC/DESC/NULLS FIRST/NULLS LAST

per-column options for btree indexes.  The planner's support for this is still
pretty rudimentary; it does not yet know how to plan mergejoins with
nondefault ordering options.  The documentation is pretty rudimentary, too.
I'll work on improving that stuff later.

Note incompatible change from prior behavior: ORDER BY ... USING will now be
rejected if the operator is not a less-than or greater-than member of some
btree opclass.  This prevents less-than-sane behavior if an operator that
doesn't actually define a proper sort ordering is selected.

2 files changed, 112 insertions, 6 deletions

diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 82dbca9e116..6379e258126 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/lsyscache.c,v 1.141 2007/01/05 22:19:43 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/lsyscache.c,v 1.142 2007/01/09 02:14:14 tgl Exp $
  *
  * NOTES
  *	  Eventually, the index information should go through here, too.
@@ -16,6 +16,7 @@
 #include "postgres.h"
 
 #include "access/hash.h"
+#include "access/nbtree.h"
 #include "bootstrap/bootstrap.h"
 #include "catalog/pg_amop.h"
 #include "catalog/pg_amproc.h"
@@ -192,7 +193,7 @@ get_op_mergejoin_info(Oid eq_op, Oid left_sortop,
 		if (op_form->amopstrategy != BTEqualStrategyNumber)
 			continue;
 
-		/* See if sort operator is also in this opclass with OK semantics */
+		/* See if sort operator is also in this opfamily with OK semantics */
 		opfamily_id = op_form->amopfamily;
 		op_strategy = get_op_opfamily_strategy(left_sortop, opfamily_id);
 		if (op_strategy == BTLessStrategyNumber ||
@@ -285,6 +286,78 @@ get_op_mergejoin_info(Oid eq_op, Oid *left_sortop,
 #endif
 
 /*
+ * get_op_compare_function
+ *		Get the OID of the datatype-specific btree comparison function
+ *		associated with an ordering operator (a "<" or ">" operator).
+ *
+ * *cmpfunc receives the comparison function OID.
+ * *reverse is set FALSE if the operator is "<", TRUE if it's ">"
+ * (indicating the comparison result must be negated before use).
+ *
+ * Returns TRUE if successful, FALSE if no btree function can be found.
+ * (This indicates that the operator is not a valid ordering operator.)
+ */
+bool
+get_op_compare_function(Oid opno, Oid *cmpfunc, bool *reverse)
+{
+	bool		result = false;
+	CatCList   *catlist;
+	int			i;
+
+	/* ensure outputs are set on failure */
+	*cmpfunc = InvalidOid;
+	*reverse = false;
+
+	/*
+	 * Search pg_amop to see if the target operator is registered as the "<"
+	 * or ">" operator of any btree opfamily.  It's possible that it might be
+	 * registered both ways (if someone were to build a "reverse sort"
+	 * opfamily); assume we can use either interpretation.  (Note: the
+	 * existence of a reverse-sort opfamily would result in uncertainty as
+	 * to whether "ORDER BY USING op" would default to NULLS FIRST or NULLS
+	 * LAST.  Since there is no longer any particularly good reason to build
+	 * reverse-sort opfamilies, we don't bother expending any extra work to
+	 * make this more determinate.  In practice, because of the way the
+	 * syscache search works, we'll use the interpretation associated with
+	 * the opfamily with smallest OID, which is probably determinate enough.)
+	 */
+	catlist = SearchSysCacheList(AMOPOPID, 1,
+								 ObjectIdGetDatum(opno),
+								 0, 0, 0);
+
+	for (i = 0; i < catlist->n_members; i++)
+	{
+		HeapTuple	tuple = &catlist->members[i]->tuple;
+		Form_pg_amop aform = (Form_pg_amop) GETSTRUCT(tuple);
+
+		/* must be btree */
+		if (aform->amopmethod != BTREE_AM_OID)
+			continue;
+
+		if (aform->amopstrategy == BTLessStrategyNumber ||
+			aform->amopstrategy == BTGreaterStrategyNumber)
+		{
+			/* Found a suitable opfamily, get matching support function */
+			*reverse = (aform->amopstrategy == BTGreaterStrategyNumber);
+			*cmpfunc = get_opfamily_proc(aform->amopfamily,
+										 aform->amoplefttype,
+										 aform->amoprighttype,
+										 BTORDER_PROC);
+			if (!OidIsValid(*cmpfunc))				/* should not happen */
+				elog(ERROR, "missing support function %d(%u,%u) in opfamily %u",
+					 BTORDER_PROC, aform->amoplefttype, aform->amoprighttype,
+					 aform->amopfamily);
+			result = true;
+			break;
+		}
+	}
+
+	ReleaseSysCacheList(catlist);
+
+	return result;
+}
+
+/*
  * get_op_hash_function
  *		Get the OID of the datatype-specific hash function associated with
  *		a hashable equality operator.
@@ -298,9 +371,9 @@ get_op_mergejoin_info(Oid eq_op, Oid *left_sortop,
 Oid
 get_op_hash_function(Oid opno)
 {
+	Oid			result = InvalidOid;
 	CatCList   *catlist;
 	int			i;
-	Oid			result = InvalidOid;
 
 	/*
 	 * Search pg_amop to see if the target operator is registered as the "="
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index baa45447a29..c43846cd57a 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.253 2007/01/05 22:19:43 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/utils/cache/relcache.c,v 1.254 2007/01/09 02:14:15 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -925,8 +925,10 @@ RelationInitIndexAccessInfo(Relation relation)
 	HeapTuple	tuple;
 	Form_pg_am	aform;
 	Datum		indclassDatum;
+	Datum		indoptionDatum;
 	bool		isnull;
 	oidvector  *indclass;
+	int2vector  *indoption;
 	MemoryContext indexcxt;
 	MemoryContext oldcontext;
 	int			natts;
@@ -1019,6 +1021,9 @@ RelationInitIndexAccessInfo(Relation relation)
 		relation->rd_supportinfo = NULL;
 	}
 
+	relation->rd_indoption = (int16 *)
+		MemoryContextAllocZero(indexcxt, natts * sizeof(int16));
+
 	/*
 	 * indclass cannot be referenced directly through the C struct, because it
 	 * comes after the variable-width indkey field.  Must extract the
@@ -1042,6 +1047,17 @@ RelationInitIndexAccessInfo(Relation relation)
 						   amstrategies, amsupport, natts);
 
 	/*
+	 * Similarly extract indoption and copy it to the cache entry
+	 */
+	indoptionDatum = fastgetattr(relation->rd_indextuple,
+								 Anum_pg_index_indoption,
+								 GetPgIndexDescriptor(),
+								 &isnull);
+	Assert(!isnull);
+	indoption = (int2vector *) DatumGetPointer(indoptionDatum);
+	memcpy(relation->rd_indoption, indoption->values, natts * sizeof(int16));
+
+	/*
 	 * expressions and predicate cache will be filled later
 	 */
 	relation->rd_indexprs = NIL;
@@ -3237,6 +3253,7 @@ load_relcache_init_file(void)
 			Oid		   *operator;
 			RegProcedure *support;
 			int			nsupport;
+			int16	   *indoption;
 
 			/* Count nailed indexes to ensure we have 'em all */
 			if (rel->rd_isnailed)
@@ -3304,7 +3321,7 @@ load_relcache_init_file(void)
 
 			rel->rd_operator = operator;
 
-			/* finally, read the vector of support procedures */
+			/* next, read the vector of support procedures */
 			if ((nread = fread(&len, 1, sizeof(len), fp)) != sizeof(len))
 				goto read_failed;
 			support = (RegProcedure *) MemoryContextAlloc(indexcxt, len);
@@ -3313,6 +3330,16 @@ load_relcache_init_file(void)
 
 			rel->rd_support = support;
 
+			/* finally, read the vector of indoption values */
+			if ((nread = fread(&len, 1, sizeof(len), fp)) != sizeof(len))
+				goto read_failed;
+
+			indoption = (int16 *) MemoryContextAlloc(indexcxt, len);
+			if ((nread = fread(indoption, 1, len, fp)) != len)
+				goto read_failed;
+
+			rel->rd_indoption = indoption;
+
 			/* set up zeroed fmgr-info vectors */
 			rel->rd_aminfo = (RelationAmInfo *)
 				MemoryContextAllocZero(indexcxt, sizeof(RelationAmInfo));
@@ -3336,6 +3363,7 @@ load_relcache_init_file(void)
 			Assert(rel->rd_operator == NULL);
 			Assert(rel->rd_support == NULL);
 			Assert(rel->rd_supportinfo == NULL);
+			Assert(rel->rd_indoption == NULL);
 		}
 
 		/*
@@ -3525,10 +3553,15 @@ write_relcache_init_file(void)
 					   relform->relnatts * (am->amstrategies * sizeof(Oid)),
 					   fp);
 
-			/* finally, write the vector of support procedures */
+			/* next, write the vector of support procedures */
 			write_item(rel->rd_support,
 				  relform->relnatts * (am->amsupport * sizeof(RegProcedure)),
 					   fp);
+
+			/* finally, write the vector of indoption values */
+			write_item(rel->rd_indoption,
+					   relform->relnatts * sizeof(int16),
+					   fp);
 		}
 
 		/* also make a list of their OIDs, for RelationIdIsInInitFile */