Create a "sort support" interface API for faster sorting.

This patch creates an API whereby a btree index opclass can optionally
provide non-SQL-callable support functions for sorting.  In the initial
patch, we only use this to provide a directly-callable comparator function,
which can be invoked with a bit less overhead than the traditional
SQL-callable comparator.  While that should be of value in itself, the real
reason for doing this is to provide a datatype-extensible framework for
more aggressive optimizations, as in Peter Geoghegan's recent work.

Robert Haas and Tom Lane

doc/src/sgml/ref/alter_opfamily.sgml

@@ -140,11 +140,12 @@ ALTER OPERATOR FAMILY <replaceable>name</replaceable> USING <replaceable class="
      <para>
       In an <literal>ADD FUNCTION</> clause, the operand data type(s) the
       function is intended to support, if different from
-      the input data type(s) of the function.  For B-tree and hash indexes
+      the input data type(s) of the function.  For B-tree comparison functions
-      it is not necessary to specify <replaceable
+      and hash functions it is not necessary to specify <replaceable
       class="parameter">op_type</replaceable> since the function's input
-      data type(s) are always the correct ones to use.  For GIN and GiST
+      data type(s) are always the correct ones to use.  For B-tree sort
-      indexes it is necessary to specify the input data type the function
+      support functions and all functions in GiST and GIN operator classes,
+      it is necessary to specify the operand data type(s) the function
       is to be used with.
      </para>
 

doc/src/sgml/ref/create_opclass.sgml

@@ -169,13 +169,14 @@ CREATE OPERATOR CLASS <replaceable class="parameter">name</replaceable> [ DEFAUL
      <para>
       In a <literal>FUNCTION</> clause, the operand data type(s) the
       function is intended to support, if different from
-      the input data type(s) of the function (for B-tree and hash indexes)
+      the input data type(s) of the function (for B-tree comparison functions
-      or the class's data type (for GIN and GiST indexes).  These defaults
+      and hash functions)
-      are always correct, so there is no point in specifying <replaceable
+      or the class's data type (for B-tree sort support functions and all
-      class="parameter">op_type</replaceable> in a <literal>FUNCTION</> clause
+      functions in GiST and GIN operator classes).  These defaults
-      in <command>CREATE OPERATOR CLASS</>, but the option is provided
+      are correct, and so <replaceable
-      for consistency with the comparable syntax in
+      class="parameter">op_type</replaceable> need not be specified in
-      <command>ALTER OPERATOR FAMILY</>.
+      <literal>FUNCTION</> clauses, except for the case of a B-tree sort
+      support function that is meant to support cross-data-type comparisons.
      </para>
     </listitem>
    </varlistentry>

doc/src/sgml/xindex.sgml

@@ -311,7 +311,8 @@
   </para>
 
   <para>
-   B-trees require a single support function, shown in <xref
+   B-trees require a single support function, and allow a second one to be
+   supplied at the operator class author's option, as shown in <xref
    linkend="xindex-btree-support-table">.
   </para>
 
@@ -333,12 +334,19 @@
        </entry>
        <entry>1</entry>
       </row>
+      <row>
+       <entry>
+        Return the addresses of C-callable sort support function(s),
+        as documented in <filename>utils/sortsupport.h</> (optional)
+       </entry>
+       <entry>2</entry>
+      </row>
      </tbody>
     </tgroup>
    </table>
 
   <para>
-   Hash indexes likewise require one support function, shown in <xref
+   Hash indexes require one support function, shown in <xref
    linkend="xindex-hash-support-table">.
   </para>
 
@@ -363,6 +371,7 @@
   <para>
    GiST indexes require seven support functions, with an optional eighth, as
    shown in <xref linkend="xindex-gist-support-table">.
+   (For more information see <xref linkend="GiST">.)
   </para>
 
    <table tocentry="1" id="xindex-gist-support-table">
@@ -418,9 +427,7 @@
       </row>
       <row>
        <entry><function>distance</></entry>
-       <entry>
+       <entry>determine distance from key to query value (optional)</entry>
-        (optional method) determine distance from key to query value
-       </entry>
        <entry>8</entry>
       </row>
      </tbody>
@@ -430,6 +437,7 @@
   <para>
    GIN indexes require four support functions, with an optional fifth, as
    shown in <xref linkend="xindex-gin-support-table">.
+   (For more information see <xref linkend="GIN">.)
   </para>
 
    <table tocentry="1" id="xindex-gin-support-table">
@@ -470,10 +478,10 @@
       <row>
        <entry><function>comparePartial</></entry>
        <entry>
-        (optional method) compare partial key from
+        compare partial key from
         query and key from index, and return an integer less than zero, zero,
         or greater than zero, indicating whether GIN should ignore this index
-        entry, treat the entry as a match, or stop the index scan
+        entry, treat the entry as a match, or stop the index scan (optional)
        </entry>
        <entry>5</entry>
       </row>
@@ -486,7 +494,8 @@
    type the particular index method expects; for example in the case
    of the comparison function for B-trees, a signed integer.  The number
    and types of the arguments to each support function are likewise
-   dependent on the index method.  For B-tree and hash the support functions
+   dependent on the index method.  For B-tree and hash the comparison and
+   hashing support functions
    take the same input data types as do the operators included in the operator
    class, but this is not the case for most GIN and GiST support functions.
   </para>
@@ -748,7 +757,8 @@ DEFAULT FOR TYPE int8 USING btree FAMILY integer_ops AS
   OPERATOR 3 = ,
   OPERATOR 4 >= ,
   OPERATOR 5 > ,
-  FUNCTION 1 btint8cmp(int8, int8) ;
+  FUNCTION 1 btint8cmp(int8, int8) ,
+  FUNCTION 2 btint8sortsupport(internal) ;
 
 CREATE OPERATOR CLASS int4_ops
 DEFAULT FOR TYPE int4 USING btree FAMILY integer_ops AS
@@ -758,7 +768,8 @@ DEFAULT FOR TYPE int4 USING btree FAMILY integer_ops AS
   OPERATOR 3 = ,
   OPERATOR 4 >= ,
   OPERATOR 5 > ,
-  FUNCTION 1 btint4cmp(int4, int4) ;
+  FUNCTION 1 btint4cmp(int4, int4) ,
+  FUNCTION 2 btint4sortsupport(internal) ;
 
 CREATE OPERATOR CLASS int2_ops
 DEFAULT FOR TYPE int2 USING btree FAMILY integer_ops AS
@@ -768,7 +779,8 @@ DEFAULT FOR TYPE int2 USING btree FAMILY integer_ops AS
   OPERATOR 3 = ,
   OPERATOR 4 >= ,
   OPERATOR 5 > ,
-  FUNCTION 1 btint2cmp(int2, int2) ;
+  FUNCTION 1 btint2cmp(int2, int2) ,
+  FUNCTION 2 btint2sortsupport(internal) ;
 
 ALTER OPERATOR FAMILY integer_ops USING btree ADD
   -- cross-type comparisons int8 vs int2

src/backend/access/nbtree/nbtcompare.c

@@ -49,6 +49,7 @@
 #include "postgres.h"
 
 #include "utils/builtins.h"
+#include "utils/sortsupport.h"
 
 
 Datum
@@ -69,6 +70,24 @@ btint2cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32((int32) a - (int32) b);
 }
 
+static int
+btint2fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	int16		a = DatumGetInt16(x);
+	int16		b = DatumGetInt16(y);
+
+	return (int) a - (int) b;
+}
+
+Datum
+btint2sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btint2fastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 btint4cmp(PG_FUNCTION_ARGS)
 {
@@ -83,6 +102,29 @@ btint4cmp(PG_FUNCTION_ARGS)
 		PG_RETURN_INT32(-1);
 }
 
+static int
+btint4fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	int32		a = DatumGetInt32(x);
+	int32		b = DatumGetInt32(y);
+
+	if (a > b)
+		return 1;
+	else if (a == b)
+		return 0;
+	else
+		return -1;
+}
+
+Datum
+btint4sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btint4fastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 btint8cmp(PG_FUNCTION_ARGS)
 {
@@ -97,6 +139,29 @@ btint8cmp(PG_FUNCTION_ARGS)
 		PG_RETURN_INT32(-1);
 }
 
+static int
+btint8fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	int64		a = DatumGetInt64(x);
+	int64		b = DatumGetInt64(y);
+
+	if (a > b)
+		return 1;
+	else if (a == b)
+		return 0;
+	else
+		return -1;
+}
+
+Datum
+btint8sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btint8fastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 btint48cmp(PG_FUNCTION_ARGS)
 {
@@ -195,6 +260,29 @@ btoidcmp(PG_FUNCTION_ARGS)
 		PG_RETURN_INT32(-1);
 }
 
+static int
+btoidfastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	Oid			a = DatumGetObjectId(x);
+	Oid			b = DatumGetObjectId(y);
+
+	if (a > b)
+		return 1;
+	else if (a == b)
+		return 0;
+	else
+		return -1;
+}
+
+Datum
+btoidsortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btoidfastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 btoidvectorcmp(PG_FUNCTION_ARGS)
 {
@@ -237,3 +325,21 @@ btnamecmp(PG_FUNCTION_ARGS)
 
 	PG_RETURN_INT32(strncmp(NameStr(*a), NameStr(*b), NAMEDATALEN));
 }
+
+static int
+btnamefastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	Name		a = DatumGetName(x);
+	Name		b = DatumGetName(y);
+
+	return strncmp(NameStr(*a), NameStr(*b), NAMEDATALEN);
+}
+
+Datum
+btnamesortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btnamefastcmp;
+	PG_RETURN_VOID();
+}

src/backend/commands/analyze.c

@@ -47,8 +47,8 @@
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/pg_rusage.h"
+#include "utils/sortsupport.h"
 #include "utils/syscache.h"
-#include "utils/tuplesort.h"
 #include "utils/timestamp.h"
 #include "utils/tqual.h"
 
@@ -1774,8 +1774,7 @@ typedef struct
 
 typedef struct
 {
-	FmgrInfo   *cmpFn;
+	SortSupport ssup;
-	int			cmpFlags;
 	int		   *tupnoLink;
 } CompareScalarsContext;
 
@@ -2222,9 +2221,7 @@ compute_scalar_stats(VacAttrStatsP stats,
 	bool		is_varwidth = (!stats->attrtype->typbyval &&
 							   stats->attrtype->typlen < 0);
 	double		corr_xysum;
-	Oid			cmpFn;
+	SortSupportData ssup;
-	int			cmpFlags;
-	FmgrInfo	f_cmpfn;
 	ScalarItem *values;
 	int			values_cnt = 0;
 	int		   *tupnoLink;
@@ -2238,8 +2235,13 @@ compute_scalar_stats(VacAttrStatsP stats,
 	tupnoLink = (int *) palloc(samplerows * sizeof(int));
 	track = (ScalarMCVItem *) palloc(num_mcv * sizeof(ScalarMCVItem));
 
-	SelectSortFunction(mystats->ltopr, false, &cmpFn, &cmpFlags);
+	memset(&ssup, 0, sizeof(ssup));
-	fmgr_info(cmpFn, &f_cmpfn);
+	ssup.ssup_cxt = CurrentMemoryContext;
+	/* We always use the default collation for statistics */
+	ssup.ssup_collation = DEFAULT_COLLATION_OID;
+	ssup.ssup_nulls_first = false;
+
+	PrepareSortSupportFromOrderingOp(mystats->ltopr, &ssup);
 
 	/* Initial scan to find sortable values */
 	for (i = 0; i < samplerows; i++)
@@ -2307,8 +2309,7 @@ compute_scalar_stats(VacAttrStatsP stats,
 		CompareScalarsContext cxt;
 
 		/* Sort the collected values */
-		cxt.cmpFn = &f_cmpfn;
+		cxt.ssup = &ssup;
-		cxt.cmpFlags = cmpFlags;
 		cxt.tupnoLink = tupnoLink;
 		qsort_arg((void *) values, values_cnt, sizeof(ScalarItem),
 				  compare_scalars, (void *) &cxt);
@@ -2712,12 +2713,9 @@ compare_scalars(const void *a, const void *b, void *arg)
 	Datum		db = ((const ScalarItem *) b)->value;
 	int			tb = ((const ScalarItem *) b)->tupno;
 	CompareScalarsContext *cxt = (CompareScalarsContext *) arg;
-	int32		compare;
+	int			compare;
 
-	/* We always use the default collation for statistics */
+	compare = ApplySortComparator(da, false, db, false, cxt->ssup);
-	compare = ApplySortFunction(cxt->cmpFn, cxt->cmpFlags,
-								DEFAULT_COLLATION_OID,
-								da, false, db, false);
 	if (compare != 0)
 		return compare;
 

src/backend/commands/opclasscmds.c

@@ -19,6 +19,7 @@
 
 #include "access/genam.h"
 #include "access/heapam.h"
+#include "access/nbtree.h"
 #include "access/sysattr.h"
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
@@ -1151,28 +1152,49 @@ assignProcTypes(OpFamilyMember *member, Oid amoid, Oid typeoid)
 	procform = (Form_pg_proc) GETSTRUCT(proctup);
 
 	/*
-	 * btree support procs must be 2-arg procs returning int4; hash support
+	 * btree comparison procs must be 2-arg procs returning int4, while btree
-	 * procs must be 1-arg procs returning int4; otherwise we don't know.
+	 * sortsupport procs must take internal and return void.  hash support
+	 * procs must be 1-arg procs returning int4.  Otherwise we don't know.
 	 */
 	if (amoid == BTREE_AM_OID)
+	{
+		if (member->number == BTORDER_PROC)
 		{
 			if (procform->pronargs != 2)
 				ereport(ERROR,
 						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("btree procedures must have two arguments")));
+						 errmsg("btree comparison procedures must have two arguments")));
 			if (procform->prorettype != INT4OID)
 				ereport(ERROR,
 						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("btree procedures must return integer")));
+						 errmsg("btree comparison procedures must return integer")));
 
 			/*
-		 * If lefttype/righttype isn't specified, use the proc's input types
+			 * If lefttype/righttype isn't specified, use the proc's input
+			 * types
 			 */
 			if (!OidIsValid(member->lefttype))
 				member->lefttype = procform->proargtypes.values[0];
 			if (!OidIsValid(member->righttype))
 				member->righttype = procform->proargtypes.values[1];
 		}
+		else if (member->number == BTSORTSUPPORT_PROC)
+		{
+			if (procform->pronargs != 1 ||
+				procform->proargtypes.values[0] != INTERNALOID)
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("btree sort support procedures must accept type \"internal\"")));
+			if (procform->prorettype != VOIDOID)
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("btree sort support procedures must return void")));
+
+			/*
+			 * Can't infer lefttype/righttype from proc, so use default rule
+			 */
+		}
+	}
 	else if (amoid == HASH_AM_OID)
 	{
 		if (procform->pronargs != 1)
@@ -1192,23 +1214,21 @@ assignProcTypes(OpFamilyMember *member, Oid amoid, Oid typeoid)
 		if (!OidIsValid(member->righttype))
 			member->righttype = procform->proargtypes.values[0];
 	}
-	else
+
-	{
 	/*
-		 * The default for GiST and GIN in CREATE OPERATOR CLASS is to use the
+	 * The default in CREATE OPERATOR CLASS is to use the class' opcintype as
-		 * class' opcintype as lefttype and righttype.  In CREATE or ALTER
+	 * lefttype and righttype.  In CREATE or ALTER OPERATOR FAMILY, opcintype
-		 * OPERATOR FAMILY, opcintype isn't available, so make the user
+	 * isn't available, so make the user specify the types.
-		 * specify the types.
 	 */
 	if (!OidIsValid(member->lefttype))
 		member->lefttype = typeoid;
 	if (!OidIsValid(member->righttype))
 		member->righttype = typeoid;
+
 	if (!OidIsValid(member->lefttype) || !OidIsValid(member->righttype))
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 				 errmsg("associated data types must be specified for index support procedure")));
-	}
 
 	ReleaseSysCache(proctup);
 }

src/backend/executor/nodeMergeAppend.c

@@ -38,10 +38,8 @@
 
 #include "postgres.h"
 
-#include "access/nbtree.h"
 #include "executor/execdebug.h"
 #include "executor/nodeMergeAppend.h"
-#include "utils/lsyscache.h"
 
 /*
  * It gets quite confusing having a heap array (indexed by integers) which
@@ -128,38 +126,18 @@ ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
 	 * initialize sort-key information
 	 */
 	mergestate->ms_nkeys = node->numCols;
-	mergestate->ms_scankeys = palloc0(sizeof(ScanKeyData) * node->numCols);
+	mergestate->ms_sortkeys = palloc0(sizeof(SortSupportData) * node->numCols);
 
 	for (i = 0; i < node->numCols; i++)
 	{
-		Oid			sortFunction;
+		SortSupport	sortKey = mergestate->ms_sortkeys + i;
-		bool		reverse;
-		int			flags;
 
-		if (!get_compare_function_for_ordering_op(node->sortOperators[i],
+		sortKey->ssup_cxt = CurrentMemoryContext;
-												  &sortFunction, &reverse))
+		sortKey->ssup_collation = node->collations[i];
-			elog(ERROR, "operator %u is not a valid ordering operator",
+		sortKey->ssup_nulls_first = node->nullsFirst[i];
-				 node->sortOperators[i]);
+		sortKey->ssup_attno = node->sortColIdx[i];
 
-		/* We use btree's conventions for encoding directionality */
+		PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
-		flags = 0;
-		if (reverse)
-			flags |= SK_BT_DESC;
-		if (node->nullsFirst[i])
-			flags |= SK_BT_NULLS_FIRST;
-
-		/*
-		 * We needn't fill in sk_strategy or sk_subtype since these scankeys
-		 * will never be passed to an index.
-		 */
-		ScanKeyEntryInitialize(&mergestate->ms_scankeys[i],
-							   flags,
-							   node->sortColIdx[i],
-							   InvalidStrategy,
-							   InvalidOid,
-							   node->collations[i],
-							   sortFunction,
-							   (Datum) 0);
 	}
 
 	/*
@@ -298,46 +276,23 @@ heap_compare_slots(MergeAppendState *node, SlotNumber slot1, SlotNumber slot2)
 
 	for (nkey = 0; nkey < node->ms_nkeys; nkey++)
 	{
-		ScanKey		scankey = node->ms_scankeys + nkey;
+		SortSupport	sortKey = node->ms_sortkeys + nkey;
-		AttrNumber	attno = scankey->sk_attno;
+		AttrNumber	attno = sortKey->ssup_attno;
 		Datum		datum1,
 					datum2;
 		bool		isNull1,
 					isNull2;
-		int32		compare;
+		int			compare;
 
 		datum1 = slot_getattr(s1, attno, &isNull1);
 		datum2 = slot_getattr(s2, attno, &isNull2);
 
-		if (isNull1)
+		compare = ApplySortComparator(datum1, isNull1,
-		{
+									  datum2, isNull2,
-			if (isNull2)
+									  sortKey);
-				continue;		/* NULL "=" NULL */
-			else if (scankey->sk_flags & SK_BT_NULLS_FIRST)
-				return -1;		/* NULL "<" NOT_NULL */
-			else
-				return 1;		/* NULL ">" NOT_NULL */
-		}
-		else if (isNull2)
-		{
-			if (scankey->sk_flags & SK_BT_NULLS_FIRST)
-				return 1;		/* NOT_NULL ">" NULL */
-			else
-				return -1;		/* NOT_NULL "<" NULL */
-		}
-		else
-		{
-			compare = DatumGetInt32(FunctionCall2Coll(&scankey->sk_func,
-													  scankey->sk_collation,
-													  datum1, datum2));
 		if (compare != 0)
-			{
-				if (scankey->sk_flags & SK_BT_DESC)
-					compare = -compare;
 			return compare;
 	}
-		}
-	}
 	return 0;
 }
 

src/backend/executor/nodeMergejoin.c

@@ -95,8 +95,6 @@
 #include "access/nbtree.h"
 #include "executor/execdebug.h"
 #include "executor/nodeMergejoin.h"
-#include "miscadmin.h"
-#include "utils/acl.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 
@@ -135,13 +133,10 @@ typedef struct MergeJoinClauseData
 	bool		risnull;
 
 	/*
-	 * The comparison strategy in use, and the lookup info to let us call the
+	 * Everything we need to know to compare the left and right values is
-	 * btree comparison support function, and the collation to use.
+	 * stored here.
 	 */
-	bool		reverse;		/* if true, negate the cmpfn's output */
+	SortSupportData ssup;
-	bool		nulls_first;	/* if true, nulls sort low */
-	FmgrInfo	cmpfinfo;
-	Oid			collation;
 }	MergeJoinClauseData;
 
 /* Result type for MJEvalOuterValues and MJEvalInnerValues */
@@ -203,8 +198,7 @@ MJExamineQuals(List *mergeclauses,
 		int			op_strategy;
 		Oid			op_lefttype;
 		Oid			op_righttype;
-		RegProcedure cmpproc;
+		Oid			sortfunc;
-		AclResult	aclresult;
 
 		if (!IsA(qual, OpExpr))
 			elog(ERROR, "mergejoin clause is not an OpExpr");
@@ -215,6 +209,17 @@ MJExamineQuals(List *mergeclauses,
 		clause->lexpr = ExecInitExpr((Expr *) linitial(qual->args), parent);
 		clause->rexpr = ExecInitExpr((Expr *) lsecond(qual->args), parent);
 
+		/* Set up sort support data */
+		clause->ssup.ssup_cxt = CurrentMemoryContext;
+		clause->ssup.ssup_collation = collation;
+		if (opstrategy == BTLessStrategyNumber)
+			clause->ssup.ssup_reverse = false;
+		else if (opstrategy == BTGreaterStrategyNumber)
+			clause->ssup.ssup_reverse = true;
+		else	/* planner screwed up */
+			elog(ERROR, "unsupported mergejoin strategy %d", opstrategy);
+		clause->ssup.ssup_nulls_first = nulls_first;
+
 		/* Extract the operator's declared left/right datatypes */
 		get_op_opfamily_properties(qual->opno, opfamily, false,
 								   &op_strategy,
@@ -224,36 +229,30 @@ MJExamineQuals(List *mergeclauses,
 			elog(ERROR, "cannot merge using non-equality operator %u",
 				 qual->opno);
 
-		/* And get the matching support procedure (comparison function) */
+		/* And get the matching support or comparison function */
-		cmpproc = get_opfamily_proc(opfamily,
+		sortfunc = get_opfamily_proc(opfamily,
+									 op_lefttype,
+									 op_righttype,
+									 BTSORTSUPPORT_PROC);
+		if (OidIsValid(sortfunc))
+		{
+			/* The sort support function should provide a comparator */
+			OidFunctionCall1(sortfunc, PointerGetDatum(&clause->ssup));
+			Assert(clause->ssup.comparator != NULL);
+		}
+		else
+		{
+			/* opfamily doesn't provide sort support, get comparison func */
+			sortfunc = get_opfamily_proc(opfamily,
 										 op_lefttype,
 										 op_righttype,
 										 BTORDER_PROC);
-		if (!RegProcedureIsValid(cmpproc))		/* should not happen */
+			if (!OidIsValid(sortfunc))		/* should not happen */
 				elog(ERROR, "missing support function %d(%u,%u) in opfamily %u",
 					 BTORDER_PROC, op_lefttype, op_righttype, opfamily);
-
+			/* We'll use a shim to call the old-style btree comparator */
-		/* Check permission to call cmp function */
+			PrepareSortSupportComparisonShim(sortfunc, &clause->ssup);
-		aclresult = pg_proc_aclcheck(cmpproc, GetUserId(), ACL_EXECUTE);
+		}
-		if (aclresult != ACLCHECK_OK)
-			aclcheck_error(aclresult, ACL_KIND_PROC,
-						   get_func_name(cmpproc));
-
-		/* Set up the fmgr lookup information */
-		fmgr_info(cmpproc, &(clause->cmpfinfo));
-
-		/* Fill the additional comparison-strategy flags */
-		if (opstrategy == BTLessStrategyNumber)
-			clause->reverse = false;
-		else if (opstrategy == BTGreaterStrategyNumber)
-			clause->reverse = true;
-		else	/* planner screwed up */
-			elog(ERROR, "unsupported mergejoin strategy %d", opstrategy);
-
-		clause->nulls_first = nulls_first;
-
-		/* ... and the collation too */
-		clause->collation = collation;
 
 		iClause++;
 	}
@@ -310,7 +309,8 @@ MJEvalOuterValues(MergeJoinState *mergestate)
 		if (clause->lisnull)
 		{
 			/* match is impossible; can we end the join early? */
-			if (i == 0 && !clause->nulls_first && !mergestate->mj_FillOuter)
+			if (i == 0 && !clause->ssup.ssup_nulls_first &&
+				!mergestate->mj_FillOuter)
 				result = MJEVAL_ENDOFJOIN;
 			else if (result == MJEVAL_MATCHABLE)
 				result = MJEVAL_NONMATCHABLE;
@@ -356,7 +356,8 @@ MJEvalInnerValues(MergeJoinState *mergestate, TupleTableSlot *innerslot)
 		if (clause->risnull)
 		{
 			/* match is impossible; can we end the join early? */
-			if (i == 0 && !clause->nulls_first && !mergestate->mj_FillInner)
+			if (i == 0 && !clause->ssup.ssup_nulls_first &&
+				!mergestate->mj_FillInner)
 				result = MJEVAL_ENDOFJOIN;
 			else if (result == MJEVAL_MATCHABLE)
 				result = MJEVAL_NONMATCHABLE;
@@ -373,20 +374,19 @@ MJEvalInnerValues(MergeJoinState *mergestate, TupleTableSlot *innerslot)
  *
  * Compare the mergejoinable values of the current two input tuples
  * and return 0 if they are equal (ie, the mergejoin equalities all
- * succeed), +1 if outer > inner, -1 if outer < inner.
+ * succeed), >0 if outer > inner, <0 if outer < inner.
  *
  * MJEvalOuterValues and MJEvalInnerValues must already have been called
  * for the current outer and inner tuples, respectively.
  */
-static int32
+static int
 MJCompare(MergeJoinState *mergestate)
 {
-	int32		result = 0;
+	int			result = 0;
 	bool		nulleqnull = false;
 	ExprContext *econtext = mergestate->js.ps.ps_ExprContext;
 	int			i;
 	MemoryContext oldContext;
-	FunctionCallInfoData fcinfo;
 
 	/*
 	 * Call the comparison functions in short-lived context, in case they leak
@@ -399,62 +399,28 @@ MJCompare(MergeJoinState *mergestate)
 	for (i = 0; i < mergestate->mj_NumClauses; i++)
 	{
 		MergeJoinClause clause = &mergestate->mj_Clauses[i];
-		Datum		fresult;
 
 		/*
-		 * Deal with null inputs.
+		 * Special case for NULL-vs-NULL, else use standard comparison.
 		 */
-		if (clause->lisnull)
+		if (clause->lisnull && clause->risnull)
-		{
-			if (clause->risnull)
 		{
 			nulleqnull = true;		/* NULL "=" NULL */
 			continue;
 		}
-			if (clause->nulls_first)
-				result = -1;	/* NULL "<" NOT_NULL */
-			else
-				result = 1;		/* NULL ">" NOT_NULL */
-			break;
-		}
-		if (clause->risnull)
-		{
-			if (clause->nulls_first)
-				result = 1;		/* NOT_NULL ">" NULL */
-			else
-				result = -1;	/* NOT_NULL "<" NULL */
-			break;
-		}
 
-		/*
+		result = ApplySortComparator(clause->ldatum, clause->lisnull,
-		 * OK to call the comparison function.
+									 clause->rdatum, clause->risnull,
-		 */
+									 &clause->ssup);
-		InitFunctionCallInfoData(fcinfo, &(clause->cmpfinfo), 2,
-								 clause->collation, NULL, NULL);
-		fcinfo.arg[0] = clause->ldatum;
-		fcinfo.arg[1] = clause->rdatum;
-		fcinfo.argnull[0] = false;
-		fcinfo.argnull[1] = false;
-		fresult = FunctionCallInvoke(&fcinfo);
-		if (fcinfo.isnull)
-		{
-			nulleqnull = true;	/* treat like NULL = NULL */
-			continue;
-		}
-		result = DatumGetInt32(fresult);
-
-		if (clause->reverse)
-			result = -result;
 
 		if (result != 0)
 			break;
 	}
 
 	/*
-	 * If we had any null comparison results or NULL-vs-NULL inputs, we do not
+	 * If we had any NULL-vs-NULL inputs, we do not want to report that the
-	 * want to report that the tuples are equal.  Instead, if result is still
+	 * tuples are equal.  Instead, if result is still 0, change it to +1.
-	 * 0, change it to +1.	This will result in advancing the inner side of
+	 * This will result in advancing the inner side of the join.
-	 * the join.
 	 *
 	 * Likewise, if there was a constant-false joinqual, do not report
 	 * equality.  We have to check this as part of the mergequals, else the
@@ -647,7 +613,7 @@ ExecMergeJoin(MergeJoinState *node)
 	List	   *joinqual;
 	List	   *otherqual;
 	bool		qualResult;
-	int32		compareResult;
+	int			compareResult;
 	PlanState  *innerPlan;
 	TupleTableSlot *innerTupleSlot;
 	PlanState  *outerPlan;

src/backend/utils/adt/date.c

@@ -29,6 +29,7 @@
 #include "utils/date.h"
 #include "utils/datetime.h"
 #include "utils/nabstime.h"
+#include "utils/sortsupport.h"
 
 /*
  * gcc's -ffast-math switch breaks routines that expect exact results from
@@ -320,6 +321,28 @@ date_cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(0);
 }
 
+static int
+date_fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	DateADT		a = DatumGetDateADT(x);
+	DateADT		b = DatumGetDateADT(y);
+
+	if (a < b)
+		return -1;
+	else if (a > b)
+		return 1;
+	return 0;
+}
+
+Datum
+date_sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = date_fastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 date_finite(PG_FUNCTION_ARGS)
 {

src/backend/utils/adt/float.c

@@ -23,6 +23,7 @@
 #include "libpq/pqformat.h"
 #include "utils/array.h"
 #include "utils/builtins.h"
+#include "utils/sortsupport.h"
 
 
 #ifndef M_PI
@@ -936,6 +937,24 @@ btfloat4cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(float4_cmp_internal(arg1, arg2));
 }
 
+static int
+btfloat4fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	float4		arg1 = DatumGetFloat4(x);
+	float4		arg2 = DatumGetFloat4(y);
+
+	return float4_cmp_internal(arg1, arg2);
+}
+
+Datum
+btfloat4sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btfloat4fastcmp;
+	PG_RETURN_VOID();
+}
+
 /*
  *		float8{eq,ne,lt,le,gt,ge}		- float8/float8 comparison operations
  */
@@ -1032,6 +1051,24 @@ btfloat8cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(float8_cmp_internal(arg1, arg2));
 }
 
+static int
+btfloat8fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	float8		arg1 = DatumGetFloat8(x);
+	float8		arg2 = DatumGetFloat8(y);
+
+	return float8_cmp_internal(arg1, arg2);
+}
+
+Datum
+btfloat8sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = btfloat8fastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 btfloat48cmp(PG_FUNCTION_ARGS)
 {

src/backend/utils/adt/timestamp.c

@@ -1830,6 +1830,25 @@ timestamp_cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(timestamp_cmp_internal(dt1, dt2));
 }
 
+/* note: this is used for timestamptz also */
+static int
+timestamp_fastcmp(Datum x, Datum y, SortSupport ssup)
+{
+	Timestamp	a = DatumGetTimestamp(x);
+	Timestamp	b = DatumGetTimestamp(y);
+
+	return timestamp_cmp_internal(a, b);
+}
+
+Datum
+timestamp_sortsupport(PG_FUNCTION_ARGS)
+{
+	SortSupport	ssup = (SortSupport) PG_GETARG_POINTER(0);
+
+	ssup->comparator = timestamp_fastcmp;
+	PG_RETURN_VOID();
+}
+
 Datum
 timestamp_hash(PG_FUNCTION_ARGS)
 {

src/backend/utils/cache/lsyscache.c

@@ -244,19 +244,22 @@ get_ordering_op_properties(Oid opno,
 }
 
 /*
- * get_compare_function_for_ordering_op
+ * get_sort_function_for_ordering_op
- *		Get the OID of the datatype-specific btree comparison function
+ *		Get the OID of the datatype-specific btree sort support function,
+ *		or if there is none, the btree comparison function,
  *		associated with an ordering operator (a "<" or ">" operator).
  *
- * *cmpfunc receives the comparison function OID.
+ * *sortfunc receives the support or comparison function OID.
+ * *issupport is set TRUE if it's a support func, FALSE if a comparison func.
  * *reverse is set FALSE if the operator is "<", TRUE if it's ">"
- * (indicating the comparison result must be negated before use).
+ * (indicating that comparison results 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_compare_function_for_ordering_op(Oid opno, Oid *cmpfunc, bool *reverse)
+get_sort_function_for_ordering_op(Oid opno, Oid *sortfunc,
+								  bool *issupport, bool *reverse)
 {
 	Oid			opfamily;
 	Oid			opcintype;
@@ -267,21 +270,31 @@ get_compare_function_for_ordering_op(Oid opno, Oid *cmpfunc, bool *reverse)
 								   &opfamily, &opcintype, &strategy))
 	{
 		/* Found a suitable opfamily, get matching support function */
-		*cmpfunc = get_opfamily_proc(opfamily,
+		*sortfunc = get_opfamily_proc(opfamily,
+									  opcintype,
+									  opcintype,
+									  BTSORTSUPPORT_PROC);
+		if (OidIsValid(*sortfunc))
+			*issupport = true;
+		else
+		{
+			/* opfamily doesn't provide sort support, get comparison func */
+			*sortfunc = get_opfamily_proc(opfamily,
 										  opcintype,
 										  opcintype,
 										  BTORDER_PROC);
-
+			if (!OidIsValid(*sortfunc))		/* should not happen */
-		if (!OidIsValid(*cmpfunc))		/* should not happen */
 				elog(ERROR, "missing support function %d(%u,%u) in opfamily %u",
 					 BTORDER_PROC, opcintype, opcintype, opfamily);
+			*issupport = false;
+		}
 		*reverse = (strategy == BTGreaterStrategyNumber);
 		return true;
 	}
 
 	/* ensure outputs are set on failure */
-	*cmpfunc = InvalidOid;
+	*sortfunc = InvalidOid;
-
+	*issupport = false;
 	*reverse = false;
 	return false;
 }

src/backend/utils/sort/Makefile

@@ -12,6 +12,6 @@ subdir = src/backend/utils/sort
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = logtape.o tuplesort.o tuplestore.o
+OBJS = logtape.o sortsupport.o tuplesort.o tuplestore.o
 
 include $(top_srcdir)/src/backend/common.mk

src/backend/utils/sort/sortsupport.c

@@ -0,0 +1,160 @@
+/*-------------------------------------------------------------------------
+ *
+ * sortsupport.c
+ *	  Support routines for accelerated sorting.
+ *
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *	  src/backend/utils/sort/sortsupport.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "fmgr.h"
+#include "utils/lsyscache.h"
+#include "utils/sortsupport.h"
+
+
+/* Info needed to use an old-style comparison function as a sort comparator */
+typedef struct
+{
+	FunctionCallInfoData fcinfo;	/* reusable callinfo structure */
+	FmgrInfo	flinfo;				/* lookup data for comparison function */
+} SortShimExtra;
+
+
+/*
+ * sortsupport.h defines inline versions of these functions if allowed by the
+ * compiler; in which case the definitions below are skipped.
+ */
+#ifndef USE_INLINE
+
+/*
+ * Apply a sort comparator function and return a 3-way comparison result.
+ * This takes care of handling reverse-sort and NULLs-ordering properly.
+ */
+int
+ApplySortComparator(Datum datum1, bool isNull1,
+					Datum datum2, bool isNull2,
+					SortSupport ssup)
+{
+	int			compare;
+
+	if (isNull1)
+	{
+		if (isNull2)
+			compare = 0;		/* NULL "=" NULL */
+		else if (ssup->ssup_nulls_first)
+			compare = -1;		/* NULL "<" NOT_NULL */
+		else
+			compare = 1;		/* NULL ">" NOT_NULL */
+	}
+	else if (isNull2)
+	{
+		if (ssup->ssup_nulls_first)
+			compare = 1;		/* NOT_NULL ">" NULL */
+		else
+			compare = -1;		/* NOT_NULL "<" NULL */
+	}
+	else
+	{
+		compare = (*ssup->comparator) (datum1, datum2, ssup);
+		if (ssup->ssup_reverse)
+			compare = -compare;
+	}
+
+	return compare;
+}
+
+#endif   /* ! USE_INLINE */
+
+/*
+ * Shim function for calling an old-style comparator
+ *
+ * This is essentially an inlined version of FunctionCall2Coll(), except
+ * we assume that the FunctionCallInfoData was already mostly set up by
+ * PrepareSortSupportComparisonShim.
+ */
+static int
+comparison_shim(Datum x, Datum y, SortSupport ssup)
+{
+	SortShimExtra *extra = (SortShimExtra *) ssup->ssup_extra;
+	Datum		result;
+
+	extra->fcinfo.arg[0] = x;
+	extra->fcinfo.arg[1] = y;
+
+	/* just for paranoia's sake, we reset isnull each time */
+	extra->fcinfo.isnull = false;
+
+	result = FunctionCallInvoke(&extra->fcinfo);
+
+	/* Check for null result, since caller is clearly not expecting one */
+	if (extra->fcinfo.isnull)
+		elog(ERROR, "function %u returned NULL", extra->flinfo.fn_oid);
+
+	return result;
+}
+
+/*
+ * Set up a shim function to allow use of an old-style btree comparison
+ * function as if it were a sort support comparator.
+ */
+void
+PrepareSortSupportComparisonShim(Oid cmpFunc, SortSupport ssup)
+{
+	SortShimExtra   *extra;
+
+	extra = (SortShimExtra *) MemoryContextAlloc(ssup->ssup_cxt,
+												 sizeof(SortShimExtra));
+
+	/* Lookup the comparison function */
+	fmgr_info_cxt(cmpFunc, &extra->flinfo, ssup->ssup_cxt);
+
+	/* We can initialize the callinfo just once and re-use it */
+	InitFunctionCallInfoData(extra->fcinfo, &extra->flinfo, 2,
+							 ssup->ssup_collation, NULL, NULL);
+	extra->fcinfo.argnull[0] = false;
+	extra->fcinfo.argnull[1] = false;
+
+	ssup->ssup_extra = extra;
+	ssup->comparator = comparison_shim;
+}
+
+/*
+ * Fill in SortSupport given an ordering operator (btree "<" or ">" operator).
+ *
+ * Caller must previously have zeroed the SortSupportData structure and then
+ * filled in ssup_cxt, ssup_collation, and ssup_nulls_first.  This will fill
+ * in ssup_reverse as well as the comparator function pointer.
+ */
+void
+PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
+{
+	Oid			sortFunction;
+	bool		issupport;
+
+	if (!get_sort_function_for_ordering_op(orderingOp,
+										   &sortFunction,
+										   &issupport,
+										   &ssup->ssup_reverse))
+		elog(ERROR, "operator %u is not a valid ordering operator",
+			 orderingOp);
+
+	if (issupport)
+	{
+		/* The sort support function should provide a comparator */
+		OidFunctionCall1(sortFunction, PointerGetDatum(ssup));
+		Assert(ssup->comparator != NULL);
+	}
+	else
+	{
+		/* We'll use a shim to call the old-style btree comparator */
+		PrepareSortSupportComparisonShim(sortFunction, ssup);
+	}
+}

src/backend/utils/sort/tuplesort.c

@@ -112,6 +112,7 @@
 #include "utils/memutils.h"
 #include "utils/pg_rusage.h"
 #include "utils/rel.h"
+#include "utils/sortsupport.h"
 #include "utils/tuplesort.h"
 
 
@@ -339,7 +340,7 @@ struct Tuplesortstate
 	 * tuplesort_begin_heap and used only by the MinimalTuple routines.
 	 */
 	TupleDesc	tupDesc;
-	ScanKey		scanKeys;		/* array of length nKeys */
+	SortSupport	sortKeys;		/* array of length nKeys */
 
 	/*
 	 * These variables are specific to the CLUSTER case; they are set by
@@ -367,9 +368,7 @@ struct Tuplesortstate
 	 * tuplesort_begin_datum and used only by the DatumTuple routines.
 	 */
 	Oid			datumType;
-	FmgrInfo	sortOpFn;		/* cached lookup data for sortOperator */
+	SortSupport	datumKey;
-	int			sortFnFlags;	/* equivalent to sk_flags */
-	Oid			sortCollation;	/* equivalent to sk_collation */
 	/* we need typelen and byval in order to know how to copy the Datums. */
 	int			datumTypeLen;
 	bool		datumTypeByVal;
@@ -613,41 +612,23 @@ tuplesort_begin_heap(TupleDesc tupDesc,
 	state->reversedirection = reversedirection_heap;
 
 	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
-	state->scanKeys = (ScanKey) palloc0(nkeys * sizeof(ScanKeyData));
+
+	/* Prepare SortSupport data for each column */
+	state->sortKeys = (SortSupport) palloc0(nkeys * sizeof(SortSupportData));
 
 	for (i = 0; i < nkeys; i++)
 	{
-		Oid			sortFunction;
+		SortSupport	sortKey = state->sortKeys + i;
-		bool		reverse;
-		int			flags;
 
 		AssertArg(attNums[i] != 0);
 		AssertArg(sortOperators[i] != 0);
 
-		if (!get_compare_function_for_ordering_op(sortOperators[i],
+		sortKey->ssup_cxt = CurrentMemoryContext;
-												  &sortFunction, &reverse))
+		sortKey->ssup_collation = sortCollations[i];
-			elog(ERROR, "operator %u is not a valid ordering operator",
+		sortKey->ssup_nulls_first = nullsFirstFlags[i];
-				 sortOperators[i]);
+		sortKey->ssup_attno = attNums[i];
 
-		/* We use btree's conventions for encoding directionality */
+		PrepareSortSupportFromOrderingOp(sortOperators[i], sortKey);
-		flags = 0;
-		if (reverse)
-			flags |= SK_BT_DESC;
-		if (nullsFirstFlags[i])
-			flags |= SK_BT_NULLS_FIRST;
-
-		/*
-		 * We needn't fill in sk_strategy or sk_subtype since these scankeys
-		 * will never be passed to an index.
-		 */
-		ScanKeyEntryInitialize(&state->scanKeys[i],
-							   flags,
-							   attNums[i],
-							   InvalidStrategy,
-							   InvalidOid,
-							   sortCollations[i],
-							   sortFunction,
-							   (Datum) 0);
 	}
 
 	MemoryContextSwitchTo(oldcontext);
@@ -799,8 +780,6 @@ tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
 	MemoryContext oldcontext;
-	Oid			sortFunction;
-	bool		reverse;
 	int16		typlen;
 	bool		typbyval;
 
@@ -829,18 +808,14 @@ tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 
 	state->datumType = datumType;
 
-	/* lookup the ordering function */
+	/* Prepare SortSupport data */
-	if (!get_compare_function_for_ordering_op(sortOperator,
+	state->datumKey = (SortSupport) palloc0(sizeof(SortSupportData));
-											  &sortFunction, &reverse))
-		elog(ERROR, "operator %u is not a valid ordering operator",
-			 sortOperator);
-	fmgr_info(sortFunction, &state->sortOpFn);
 
-	/* set ordering flags and collation */
+	state->datumKey->ssup_cxt = CurrentMemoryContext;
-	state->sortFnFlags = reverse ? SK_BT_DESC : 0;
+	state->datumKey->ssup_collation = sortCollation;
-	if (nullsFirstFlag)
+	state->datumKey->ssup_nulls_first = nullsFirstFlag;
-		state->sortFnFlags |= SK_BT_NULLS_FIRST;
+
-	state->sortCollation = sortCollation;
+	PrepareSortSupportFromOrderingOp(sortOperator, state->datumKey);
 
 	/* lookup necessary attributes of the datum type */
 	get_typlenbyval(datumType, &typlen, &typbyval);
@@ -2604,29 +2579,6 @@ markrunend(Tuplesortstate *state, int tapenum)
 }
 
 
-/*
- * Set up for an external caller of ApplySortFunction.	This function
- * basically just exists to localize knowledge of the encoding of sk_flags
- * used in this module.
- */
-void
-SelectSortFunction(Oid sortOperator,
-				   bool nulls_first,
-				   Oid *sortFunction,
-				   int *sortFlags)
-{
-	bool		reverse;
-
-	if (!get_compare_function_for_ordering_op(sortOperator,
-											  sortFunction, &reverse))
-		elog(ERROR, "operator %u is not a valid ordering operator",
-			 sortOperator);
-
-	*sortFlags = reverse ? SK_BT_DESC : 0;
-	if (nulls_first)
-		*sortFlags |= SK_BT_NULLS_FIRST;
-}
-
 /*
  * Inline-able copy of FunctionCall2Coll() to save some cycles in sorting.
  */
@@ -2693,20 +2645,6 @@ inlineApplySortFunction(FmgrInfo *sortFunction, int sk_flags, Oid collation,
 	return compare;
 }
 
-/*
- * Non-inline ApplySortFunction() --- this is needed only to conform to
- * C99's brain-dead notions about how to implement inline functions...
- */
-int32
-ApplySortFunction(FmgrInfo *sortFunction, int sortFlags, Oid collation,
-				  Datum datum1, bool isNull1,
-				  Datum datum2, bool isNull2)
-{
-	return inlineApplySortFunction(sortFunction, sortFlags, collation,
-								   datum1, isNull1,
-								   datum2, isNull2);
-}
-
 
 /*
  * Routines specialized for HeapTuple (actually MinimalTuple) case
@@ -2715,7 +2653,7 @@ ApplySortFunction(FmgrInfo *sortFunction, int sortFlags, Oid collation,
 static int
 comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 {
-	ScanKey		scanKey = state->scanKeys;
+	SortSupport	sortKey = state->sortKeys;
 	HeapTupleData ltup;
 	HeapTupleData rtup;
 	TupleDesc	tupDesc;
@@ -2726,10 +2664,9 @@ comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 	CHECK_FOR_INTERRUPTS();
 
 	/* Compare the leading sort key */
-	compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
+	compare = ApplySortComparator(a->datum1, a->isnull1,
-									  scanKey->sk_collation,
+								  b->datum1, b->isnull1,
-									  a->datum1, a->isnull1,
+								  sortKey);
-									  b->datum1, b->isnull1);
 	if (compare != 0)
 		return compare;
 
@@ -2739,10 +2676,10 @@ comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 	rtup.t_len = ((MinimalTuple) b->tuple)->t_len + MINIMAL_TUPLE_OFFSET;
 	rtup.t_data = (HeapTupleHeader) ((char *) b->tuple - MINIMAL_TUPLE_OFFSET);
 	tupDesc = state->tupDesc;
-	scanKey++;
+	sortKey++;
-	for (nkey = 1; nkey < state->nKeys; nkey++, scanKey++)
+	for (nkey = 1; nkey < state->nKeys; nkey++, sortKey++)
 	{
-		AttrNumber	attno = scanKey->sk_attno;
+		AttrNumber	attno = sortKey->ssup_attno;
 		Datum		datum1,
 					datum2;
 		bool		isnull1,
@@ -2751,10 +2688,9 @@ comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 		datum1 = heap_getattr(&ltup, attno, tupDesc, &isnull1);
 		datum2 = heap_getattr(&rtup, attno, tupDesc, &isnull2);
 
-		compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
+		compare = ApplySortComparator(datum1, isnull1,
-										  scanKey->sk_collation,
+									  datum2, isnull2,
-										  datum1, isnull1,
+									  sortKey);
-										  datum2, isnull2);
 		if (compare != 0)
 			return compare;
 	}
@@ -2781,7 +2717,7 @@ copytup_heap(Tuplesortstate *state, SortTuple *stup, void *tup)
 	htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
 	htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
 	stup->datum1 = heap_getattr(&htup,
-								state->scanKeys[0].sk_attno,
+								state->sortKeys[0].ssup_attno,
 								state->tupDesc,
 								&stup->isnull1);
 }
@@ -2833,7 +2769,7 @@ readtup_heap(Tuplesortstate *state, SortTuple *stup,
 	htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
 	htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
 	stup->datum1 = heap_getattr(&htup,
-								state->scanKeys[0].sk_attno,
+								state->sortKeys[0].ssup_attno,
 								state->tupDesc,
 								&stup->isnull1);
 }
@@ -2841,12 +2777,13 @@ readtup_heap(Tuplesortstate *state, SortTuple *stup,
 static void
 reversedirection_heap(Tuplesortstate *state)
 {
-	ScanKey		scanKey = state->scanKeys;
+	SortSupport	sortKey = state->sortKeys;
 	int			nkey;
 
-	for (nkey = 0; nkey < state->nKeys; nkey++, scanKey++)
+	for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
 	{
-		scanKey->sk_flags ^= (SK_BT_DESC | SK_BT_NULLS_FIRST);
+		sortKey->ssup_reverse = !sortKey->ssup_reverse;
+		sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
 	}
 }
 
@@ -3297,10 +3234,9 @@ comparetup_datum(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 	/* Allow interrupting long sorts */
 	CHECK_FOR_INTERRUPTS();
 
-	return inlineApplySortFunction(&state->sortOpFn, state->sortFnFlags,
+	return ApplySortComparator(a->datum1, a->isnull1,
-								   state->sortCollation,
+							   b->datum1, b->isnull1,
-								   a->datum1, a->isnull1,
+							   state->datumKey);
-								   b->datum1, b->isnull1);
 }
 
 static void
@@ -3392,7 +3328,8 @@ readtup_datum(Tuplesortstate *state, SortTuple *stup,
 static void
 reversedirection_datum(Tuplesortstate *state)
 {
-	state->sortFnFlags ^= (SK_BT_DESC | SK_BT_NULLS_FIRST);
+	state->datumKey->ssup_reverse = !state->datumKey->ssup_reverse;
+	state->datumKey->ssup_nulls_first = !state->datumKey->ssup_nulls_first;
 }
 
 /*

src/bin/pg_dump/pg_dump.c

@@ -9892,6 +9892,8 @@ dumpOpclass(Archive *fout, OpclassInfo *opcinfo)
 	int			i_sortfamilynsp;
 	int			i_amprocnum;
 	int			i_amproc;
+	int			i_amproclefttype;
+	int			i_amprocrighttype;
 	char	   *opcintype;
 	char	   *opckeytype;
 	char	   *opcdefault;
@@ -9906,6 +9908,8 @@ dumpOpclass(Archive *fout, OpclassInfo *opcinfo)
 	char	   *sortfamilynsp;
 	char	   *amprocnum;
 	char	   *amproc;
+	char	   *amproclefttype;
+	char	   *amprocrighttype;
 	bool		needComma;
 	int			i;
 
@@ -10150,13 +10154,21 @@ dumpOpclass(Archive *fout, OpclassInfo *opcinfo)
 	 *
 	 * Print only those opfamily members that are tied to the opclass by
 	 * pg_depend entries.
+	 *
+	 * We print the amproclefttype/amprocrighttype even though in most cases
+	 * the backend could deduce the right values, because of the corner case
+	 * of a btree sort support function for a cross-type comparison.  That's
+	 * only allowed in 9.2 and later, but for simplicity print them in all
+	 * versions that have the columns.
 	 */
 	resetPQExpBuffer(query);
 
 	if (g_fout->remoteVersion >= 80300)
 	{
 		appendPQExpBuffer(query, "SELECT amprocnum, "
-						  "amproc::pg_catalog.regprocedure "
+						  "amproc::pg_catalog.regprocedure, "
+						  "amproclefttype::pg_catalog.regtype, "
+						  "amprocrighttype::pg_catalog.regtype "
 						"FROM pg_catalog.pg_amproc ap, pg_catalog.pg_depend "
 		   "WHERE refclassid = 'pg_catalog.pg_opclass'::pg_catalog.regclass "
 						  "AND refobjid = '%u'::pg_catalog.oid "
@@ -10168,7 +10180,9 @@ dumpOpclass(Archive *fout, OpclassInfo *opcinfo)
 	else
 	{
 		appendPQExpBuffer(query, "SELECT amprocnum, "
-						  "amproc::pg_catalog.regprocedure "
+						  "amproc::pg_catalog.regprocedure, "
+						  "'' AS amproclefttype, "
+						  "'' AS amprocrighttype "
 						  "FROM pg_catalog.pg_amproc "
 						  "WHERE amopclaid = '%u'::pg_catalog.oid "
 						  "ORDER BY amprocnum",
@@ -10182,17 +10196,25 @@ dumpOpclass(Archive *fout, OpclassInfo *opcinfo)
 
 	i_amprocnum = PQfnumber(res, "amprocnum");
 	i_amproc = PQfnumber(res, "amproc");
+	i_amproclefttype = PQfnumber(res, "amproclefttype");
+	i_amprocrighttype = PQfnumber(res, "amprocrighttype");
 
 	for (i = 0; i < ntups; i++)
 	{
 		amprocnum = PQgetvalue(res, i, i_amprocnum);
 		amproc = PQgetvalue(res, i, i_amproc);
+		amproclefttype = PQgetvalue(res, i, i_amproclefttype);
+		amprocrighttype = PQgetvalue(res, i, i_amprocrighttype);
 
 		if (needComma)
 			appendPQExpBuffer(q, " ,\n    ");
 
-		appendPQExpBuffer(q, "FUNCTION %s %s",
+		appendPQExpBuffer(q, "FUNCTION %s", amprocnum);
-						  amprocnum, amproc);
+
+		if (*amproclefttype && *amprocrighttype)
+			appendPQExpBuffer(q, " (%s, %s)", amproclefttype, amprocrighttype);
+
+		appendPQExpBuffer(q, " %s", amproc);
 
 		needComma = true;
 	}

src/include/access/nbtree.h

@@ -417,13 +417,18 @@ typedef struct xl_btree_newroot
 
 /*
  *	When a new operator class is declared, we require that the user
- *	supply us with an amproc procedure for determining whether, for
+ *	supply us with an amproc procedure (BTORDER_PROC) for determining
- *	two keys a and b, a < b, a = b, or a > b.  This routine must
+ *	whether, for two keys a and b, a < b, a = b, or a > b.  This routine
- *	return < 0, 0, > 0, respectively, in these three cases.  Since we
+ *	must return < 0, 0, > 0, respectively, in these three cases.  (It must
- *	only have one such proc in amproc, it's number 1.
+ *	not return INT_MIN, since we may negate the result before using it.)
+ *
+ *	To facilitate accelerated sorting, an operator class may choose to
+ *	offer a second procedure (BTSORTSUPPORT_PROC).  For full details, see
+ *	src/include/utils/sortsupport.h.
  */
 
 #define BTORDER_PROC		1
+#define BTSORTSUPPORT_PROC	2
 
 /*
  *	We need to be able to tell the difference between read and write

src/include/catalog/catversion.h

@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201111271
+#define CATALOG_VERSION_NO	201112061
 
 #endif

src/include/catalog/pg_am.h

@@ -117,7 +117,7 @@ typedef FormData_pg_am *Form_pg_am;
  * ----------------
  */
 
-DATA(insert OID = 403 (  btree	5 1 t f t t t t t t t f t t 0 btinsert btbeginscan btgettuple btgetbitmap btrescan btendscan btmarkpos btrestrpos btbuild btbuildempty btbulkdelete btvacuumcleanup btcostestimate btoptions ));
+DATA(insert OID = 403 (  btree	5 2 t f t t t t t t t f t t 0 btinsert btbeginscan btgettuple btgetbitmap btrescan btendscan btmarkpos btrestrpos btbuild btbuildempty btbulkdelete btvacuumcleanup btcostestimate btoptions ));
 DESCR("b-tree index access method");
 #define BTREE_AM_OID 403
 DATA(insert OID = 405 (  hash	1 1 f f t f f f f f f f f f 23 hashinsert hashbeginscan hashgettuple hashgetbitmap hashrescan hashendscan hashmarkpos hashrestrpos hashbuild hashbuildempty hashbulkdelete hashvacuumcleanup hashcostestimate hashoptions ));

src/include/catalog/pg_amproc.h

@@ -83,33 +83,43 @@ DATA(insert (	426   1042 1042 1 1078 ));
 DATA(insert (	428   17 17 1 1954 ));
 DATA(insert (	429   18 18 1 358 ));
 DATA(insert (	434   1082 1082 1 1092 ));
+DATA(insert (	434   1082 1082 2 3136 ));
 DATA(insert (	434   1082 1114 1 2344 ));
 DATA(insert (	434   1082 1184 1 2357 ));
 DATA(insert (	434   1114 1114 1 2045 ));
+DATA(insert (	434   1114 1114 2 3137 ));
 DATA(insert (	434   1114 1082 1 2370 ));
 DATA(insert (	434   1114 1184 1 2526 ));
 DATA(insert (	434   1184 1184 1 1314 ));
+DATA(insert (	434   1184 1184 2 3137 ));
 DATA(insert (	434   1184 1082 1 2383 ));
 DATA(insert (	434   1184 1114 1 2533 ));
 DATA(insert (	1970   700 700 1 354 ));
+DATA(insert (	1970   700 700 2 3132 ));
 DATA(insert (	1970   700 701 1 2194 ));
 DATA(insert (	1970   701 701 1 355 ));
+DATA(insert (	1970   701 701 2 3133 ));
 DATA(insert (	1970   701 700 1 2195 ));
 DATA(insert (	1974   869 869 1 926 ));
 DATA(insert (	1976   21 21 1 350 ));
+DATA(insert (	1976   21 21 2 3129 ));
 DATA(insert (	1976   21 23 1 2190 ));
 DATA(insert (	1976   21 20 1 2192 ));
 DATA(insert (	1976   23 23 1 351 ));
+DATA(insert (	1976   23 23 2 3130 ));
 DATA(insert (	1976   23 20 1 2188 ));
 DATA(insert (	1976   23 21 1 2191 ));
 DATA(insert (	1976   20 20 1 842 ));
+DATA(insert (	1976   20 20 2 3131 ));
 DATA(insert (	1976   20 23 1 2189 ));
 DATA(insert (	1976   20 21 1 2193 ));
 DATA(insert (	1982   1186 1186 1 1315 ));
 DATA(insert (	1984   829 829 1 836 ));
 DATA(insert (	1986   19 19 1 359 ));
+DATA(insert (	1986   19 19 2 3135 ));
 DATA(insert (	1988   1700 1700 1 1769 ));
 DATA(insert (	1989   26 26 1 356 ));
+DATA(insert (	1989   26 26 2 3134 ));
 DATA(insert (	1991   30 30 1 404 ));
 DATA(insert (	2994   2249 2249 1 2987 ));
 DATA(insert (	1994   25 25 1 360 ));

src/include/catalog/pg_proc.h

@@ -566,16 +566,28 @@ DESCR("I/O");
 
 DATA(insert OID = 350 (  btint2cmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "21 21" _null_ _null_ _null_ _null_ btint2cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3129 ( btint2sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btint2sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 351 (  btint4cmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "23 23" _null_ _null_ _null_ _null_ btint4cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3130 ( btint4sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btint4sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 842 (  btint8cmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "20 20" _null_ _null_ _null_ _null_ btint8cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3131 ( btint8sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btint8sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 354 (  btfloat4cmp	   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "700 700" _null_ _null_ _null_ _null_ btfloat4cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3132 ( btfloat4sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btfloat4sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 355 (  btfloat8cmp	   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "701 701" _null_ _null_ _null_ _null_ btfloat8cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3133 ( btfloat8sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btfloat8sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 356 (  btoidcmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "26 26" _null_ _null_ _null_ _null_ btoidcmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3134 ( btoidsortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btoidsortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 404 (  btoidvectorcmp    PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "30 30" _null_ _null_ _null_ _null_ btoidvectorcmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
 DATA(insert OID = 357 (  btabstimecmp	   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "702 702" _null_ _null_ _null_ _null_ btabstimecmp _null_ _null_ _null_ ));
@@ -584,6 +596,8 @@ DATA(insert OID = 358 (  btcharcmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23
 DESCR("less-equal-greater");
 DATA(insert OID = 359 (  btnamecmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "19 19" _null_ _null_ _null_ _null_ btnamecmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3135 ( btnamesortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ btnamesortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 360 (  bttextcmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "25 25" _null_ _null_ _null_ _null_ bttextcmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
 DATA(insert OID = 377 (  cash_cmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "790 790" _null_ _null_ _null_ _null_ cash_cmp _null_ _null_ _null_ ));
@@ -1122,6 +1136,8 @@ DATA(insert OID = 1090 (  date_ge		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 16
 DATA(insert OID = 1091 (  date_ne		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 16 "1082 1082" _null_ _null_ _null_ _null_ date_ne _null_ _null_ _null_ ));
 DATA(insert OID = 1092 (  date_cmp		   PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "1082 1082" _null_ _null_ _null_ _null_ date_cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3136 (  date_sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ date_sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 
 /* OIDS 1100 - 1199 */
 
@@ -2769,6 +2785,8 @@ DATA(insert OID = 2044 ( overlaps			PGNSP PGUID 14 1 0 0 0 f f f f f i 4 0 16 "1
 DESCR("intervals overlap?");
 DATA(insert OID = 2045 (  timestamp_cmp		PGNSP PGUID 12 1 0 0 0 f f f t f i 2 0 23 "1114 1114" _null_ _null_ _null_ _null_ timestamp_cmp _null_ _null_ _null_ ));
 DESCR("less-equal-greater");
+DATA(insert OID = 3137 (  timestamp_sortsupport PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 2278 "2281" _null_ _null_ _null_ _null_ timestamp_sortsupport _null_ _null_ _null_ ));
+DESCR("sort support");
 DATA(insert OID = 2046 (  time				PGNSP PGUID 12 1 0 0 0 f f f t f i 1 0 1083 "1266" _null_ _null_ _null_ _null_ timetz_time _null_ _null_ _null_ ));
 DESCR("convert time with time zone to time");
 DATA(insert OID = 2047 (  timetz			PGNSP PGUID 12 1 0 0 0 f f f t f s 1 0 1266 "1083" _null_ _null_ _null_ _null_ time_timetz _null_ _null_ _null_ ));

src/include/nodes/execnodes.h

@@ -20,6 +20,7 @@
 #include "nodes/params.h"
 #include "nodes/plannodes.h"
 #include "utils/reltrigger.h"
+#include "utils/sortsupport.h"
 #include "utils/tuplestore.h"
 
 
@@ -1087,7 +1088,7 @@ typedef struct AppendState
  *
  *		nplans			how many plans are in the array
  *		nkeys			number of sort key columns
- *		scankeys		sort keys in ScanKey representation
+ *		sortkeys		sort keys in SortSupport representation
  *		slots			current output tuple of each subplan
  *		heap			heap of active tuples (represented as array indexes)
  *		heap_size		number of active heap entries
@@ -1101,7 +1102,7 @@ typedef struct MergeAppendState
 	PlanState **mergeplans;		/* array of PlanStates for my inputs */
 	int			ms_nplans;
 	int			ms_nkeys;
-	ScanKey		ms_scankeys;	/* array of length ms_nkeys */
+	SortSupport	ms_sortkeys;	/* array of length ms_nkeys */
 	TupleTableSlot **ms_slots;	/* array of length ms_nplans */
 	int		   *ms_heap;		/* array of length ms_nplans */
 	int			ms_heap_size;	/* current active length of ms_heap[] */

src/include/utils/builtins.h

@@ -279,7 +279,7 @@ extern void pg_lltoa(int64 ll, char *a);
 
 /*
  *		Per-opclass comparison functions for new btrees.  These are
- *		stored in pg_amproc and defined in access/nbtree/nbtcompare.c
+ *		stored in pg_amproc; most are defined in access/nbtree/nbtcompare.c
  */
 extern Datum btboolcmp(PG_FUNCTION_ARGS);
 extern Datum btint2cmp(PG_FUNCTION_ARGS);
@@ -304,6 +304,19 @@ extern Datum btcharcmp(PG_FUNCTION_ARGS);
 extern Datum btnamecmp(PG_FUNCTION_ARGS);
 extern Datum bttextcmp(PG_FUNCTION_ARGS);
 
+/*
+ *		Per-opclass sort support functions for new btrees.  Like the
+ *		functions above, these are stored in pg_amproc; most are defined in
+ *		access/nbtree/nbtcompare.c
+ */
+extern Datum btint2sortsupport(PG_FUNCTION_ARGS);
+extern Datum btint4sortsupport(PG_FUNCTION_ARGS);
+extern Datum btint8sortsupport(PG_FUNCTION_ARGS);
+extern Datum btfloat4sortsupport(PG_FUNCTION_ARGS);
+extern Datum btfloat8sortsupport(PG_FUNCTION_ARGS);
+extern Datum btoidsortsupport(PG_FUNCTION_ARGS);
+extern Datum btnamesortsupport(PG_FUNCTION_ARGS);
+
 /* float.c */
 extern PGDLLIMPORT int extra_float_digits;
 

src/include/utils/date.h

@@ -104,6 +104,7 @@ extern Datum date_le(PG_FUNCTION_ARGS);
 extern Datum date_gt(PG_FUNCTION_ARGS);
 extern Datum date_ge(PG_FUNCTION_ARGS);
 extern Datum date_cmp(PG_FUNCTION_ARGS);
+extern Datum date_sortsupport(PG_FUNCTION_ARGS);
 extern Datum date_finite(PG_FUNCTION_ARGS);
 extern Datum date_larger(PG_FUNCTION_ARGS);
 extern Datum date_smaller(PG_FUNCTION_ARGS);

src/include/utils/lsyscache.h

@@ -48,8 +48,8 @@ extern Oid get_opfamily_member(Oid opfamily, Oid lefttype, Oid righttype,
 					int16 strategy);
 extern bool get_ordering_op_properties(Oid opno,
 						   Oid *opfamily, Oid *opcintype, int16 *strategy);
-extern bool get_compare_function_for_ordering_op(Oid opno,
+extern bool get_sort_function_for_ordering_op(Oid opno, Oid *sortfunc,
-									 Oid *cmpfunc, bool *reverse);
+								  bool *issupport, bool *reverse);
 extern Oid	get_equality_op_for_ordering_op(Oid opno, bool *reverse);
 extern Oid	get_ordering_op_for_equality_op(Oid opno, bool use_lhs_type);
 extern List *get_mergejoin_opfamilies(Oid opno);

src/include/utils/sortsupport.h

@@ -0,0 +1,156 @@
+/*-------------------------------------------------------------------------
+ *
+ * sortsupport.h
+ *	  Framework for accelerated sorting.
+ *
+ * Traditionally, PostgreSQL has implemented sorting by repeatedly invoking
+ * an SQL-callable comparison function "cmp(x, y) returns int" on pairs of
+ * values to be compared, where the comparison function is the BTORDER_PROC
+ * pg_amproc support function of the appropriate btree index opclass.
+ *
+ * This file defines alternative APIs that allow sorting to be performed with
+ * reduced overhead.  To support lower-overhead sorting, a btree opclass may
+ * provide a BTSORTSUPPORT_PROC pg_amproc entry, which must take a single
+ * argument of type internal and return void.  The argument is actually a
+ * pointer to a SortSupportData struct, which is defined below.
+ *
+ * If provided, the BTSORTSUPPORT function will be called during sort setup,
+ * and it must initialize the provided struct with pointers to function(s)
+ * that can be called to perform sorting.  This API is defined to allow
+ * multiple acceleration mechanisms to be supported, but no opclass is
+ * required to provide all of them.  The BTSORTSUPPORT function should
+ * simply not set any function pointers for mechanisms it doesn't support.
+ * (However, all opclasses that provide BTSORTSUPPORT are required to provide
+ * the comparator function.)
+ *
+ * All sort support functions will be passed the address of the
+ * SortSupportData struct when called, so they can use it to store
+ * additional private data as needed.  In particular, for collation-aware
+ * datatypes, the ssup_collation field is set before calling BTSORTSUPPORT
+ * and is available to all support functions.  Additional opclass-dependent
+ * data can be stored using the ssup_extra field.  Any such data
+ * should be allocated in the ssup_cxt memory context.
+ *
+ * Note: since pg_amproc functions are indexed by (lefttype, righttype)
+ * it is possible to associate a BTSORTSUPPORT function with a cross-type
+ * comparison.  This could sensibly be used to provide a fast comparator
+ * function for such cases, but probably not any other acceleration method.
+ *
+ *
+ * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/utils/sortsupport.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SORTSUPPORT_H
+#define SORTSUPPORT_H
+
+#include "access/attnum.h"
+
+typedef struct SortSupportData *SortSupport;
+
+typedef struct SortSupportData
+{
+	/*
+	 * These fields are initialized before calling the BTSORTSUPPORT function
+	 * and should not be changed later.
+	 */
+	MemoryContext ssup_cxt;				/* Context containing sort info */
+	Oid			ssup_collation;			/* Collation to use, or InvalidOid */
+
+	/*
+	 * Additional sorting parameters; but unlike ssup_collation, these can
+	 * be changed after BTSORTSUPPORT is called, so don't use them in
+	 * selecting sort support functions.
+	 */
+	bool		ssup_reverse;			/* descending-order sort? */
+	bool		ssup_nulls_first;		/* sort nulls first? */
+
+	/*
+	 * These fields are workspace for callers, and should not be touched by
+	 * opclass-specific functions.
+	 */
+	AttrNumber	ssup_attno;				/* column number to sort */
+
+	/*
+	 * ssup_extra is zeroed before calling the BTSORTSUPPORT function, and
+	 * is not touched subsequently by callers.
+	 */
+	void	   *ssup_extra;				/* Workspace for opclass functions */
+
+	/*
+	 * Function pointers are zeroed before calling the BTSORTSUPPORT function,
+	 * and must be set by it for any acceleration methods it wants to supply.
+	 * The comparator pointer must be set, others are optional.
+	 */
+
+	/*
+	 * Comparator function has the same API as the traditional btree
+	 * comparison function, ie, return <0, 0, or >0 according as x is less
+	 * than, equal to, or greater than y.  Note that x and y are guaranteed
+	 * not null, and there is no way to return null either.  Do not return
+	 * INT_MIN, as callers are allowed to negate the result before using it.
+	 */
+	int			(*comparator) (Datum x, Datum y, SortSupport ssup);
+
+	/*
+	 * Additional sort-acceleration functions might be added here later.
+	 */
+} SortSupportData;
+
+
+/* ApplySortComparator should be inlined if possible */
+#ifdef USE_INLINE
+
+/*
+ * Apply a sort comparator function and return a 3-way comparison result.
+ * This takes care of handling reverse-sort and NULLs-ordering properly.
+ */
+static inline int
+ApplySortComparator(Datum datum1, bool isNull1,
+					Datum datum2, bool isNull2,
+					SortSupport ssup)
+{
+	int			compare;
+
+	if (isNull1)
+	{
+		if (isNull2)
+			compare = 0;		/* NULL "=" NULL */
+		else if (ssup->ssup_nulls_first)
+			compare = -1;		/* NULL "<" NOT_NULL */
+		else
+			compare = 1;		/* NULL ">" NOT_NULL */
+	}
+	else if (isNull2)
+	{
+		if (ssup->ssup_nulls_first)
+			compare = 1;		/* NOT_NULL ">" NULL */
+		else
+			compare = -1;		/* NOT_NULL "<" NULL */
+	}
+	else
+	{
+		compare = (*ssup->comparator) (datum1, datum2, ssup);
+		if (ssup->ssup_reverse)
+			compare = -compare;
+	}
+
+	return compare;
+}
+
+#else
+
+extern int	ApplySortComparator(Datum datum1, bool isNull1,
+					Datum datum2, bool isNull2,
+					SortSupport ssup);
+
+#endif   /* USE_INLINE */
+
+/* Other functions in utils/sort/sortsupport.c */
+extern void PrepareSortSupportComparisonShim(Oid cmpFunc, SortSupport ssup);
+extern void PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup);
+
+#endif   /* SORTSUPPORT_H */

src/include/utils/timestamp.h

@@ -109,6 +109,7 @@ extern Datum timestamp_ge(PG_FUNCTION_ARGS);
 extern Datum timestamp_gt(PG_FUNCTION_ARGS);
 extern Datum timestamp_finite(PG_FUNCTION_ARGS);
 extern Datum timestamp_cmp(PG_FUNCTION_ARGS);
+extern Datum timestamp_sortsupport(PG_FUNCTION_ARGS);
 extern Datum timestamp_hash(PG_FUNCTION_ARGS);
 extern Datum timestamp_smaller(PG_FUNCTION_ARGS);
 extern Datum timestamp_larger(PG_FUNCTION_ARGS);

src/include/utils/tuplesort.h

@@ -116,19 +116,4 @@ extern void tuplesort_rescan(Tuplesortstate *state);
 extern void tuplesort_markpos(Tuplesortstate *state);
 extern void tuplesort_restorepos(Tuplesortstate *state);
 
-/* Setup for ApplySortFunction */
-extern void SelectSortFunction(Oid sortOperator, bool nulls_first,
-				   Oid *sortFunction,
-				   int *sortFlags);
-
-/*
- * Apply a sort function (by now converted to fmgr lookup form)
- * and return a 3-way comparison result.  This takes care of handling
- * reverse-sort and NULLs-ordering properly.
- */
-extern int32 ApplySortFunction(FmgrInfo *sortFunction, int sortFlags,
-				  Oid collation,
-				  Datum datum1, bool isNull1,
-				  Datum datum2, bool isNull2);
-
 #endif   /* TUPLESORT_H */

src/test/regress/expected/opr_sanity.out

@@ -1186,40 +1186,30 @@ WHERE p1.amprocfamily = p3.oid AND p3.opfmethod = p2.oid AND
 
 -- Detect missing pg_amproc entries: should have as many support functions
 -- as AM expects for each datatype combination supported by the opfamily.
--- GiST/GIN are special cases because each has an optional support function.
+-- btree/GiST/GIN each allow one optional support function, though.
 SELECT p1.amname, p2.opfname, p3.amproclefttype, p3.amprocrighttype
 FROM pg_am AS p1, pg_opfamily AS p2, pg_amproc AS p3
 WHERE p2.opfmethod = p1.oid AND p3.amprocfamily = p2.oid AND
-    p1.amname <> 'gist' AND p1.amname <> 'gin' AND
-    p1.amsupport != (SELECT count(*) FROM pg_amproc AS p4
-                     WHERE p4.amprocfamily = p2.oid AND
-                           p4.amproclefttype = p3.amproclefttype AND
-                           p4.amprocrighttype = p3.amprocrighttype);
- amname | opfname | amproclefttype | amprocrighttype 
---------+---------+----------------+-----------------
-(0 rows)
-
--- Similar check for GiST/GIN, allowing one optional proc
-SELECT p1.amname, p2.opfname, p3.amproclefttype, p3.amprocrighttype
-FROM pg_am AS p1, pg_opfamily AS p2, pg_amproc AS p3
-WHERE p2.opfmethod = p1.oid AND p3.amprocfamily = p2.oid AND
-    (p1.amname = 'gist' OR p1.amname = 'gin') AND
     (SELECT count(*) FROM pg_amproc AS p4
      WHERE p4.amprocfamily = p2.oid AND
            p4.amproclefttype = p3.amproclefttype AND
            p4.amprocrighttype = p3.amprocrighttype)
-      NOT IN (p1.amsupport, p1.amsupport - 1);
+    NOT BETWEEN
+      (CASE WHEN p1.amname IN ('btree', 'gist', 'gin') THEN p1.amsupport - 1
+            ELSE p1.amsupport END)
+      AND p1.amsupport;
  amname | opfname | amproclefttype | amprocrighttype 
 --------+---------+----------------+-----------------
 (0 rows)
 
 -- Also, check if there are any pg_opclass entries that don't seem to have
--- pg_amproc support.  Again, GiST/GIN have to be checked specially.
+-- pg_amproc support.  Again, opclasses with an optional support proc have
+-- to be checked specially.
 SELECT amname, opcname, count(*)
 FROM pg_am am JOIN pg_opclass op ON opcmethod = am.oid
      LEFT JOIN pg_amproc p ON amprocfamily = opcfamily AND
          amproclefttype = amprocrighttype AND amproclefttype = opcintype
-WHERE am.amname <> 'gist' AND am.amname <> 'gin'
+WHERE am.amname <> 'btree' AND am.amname <> 'gist' AND am.amname <> 'gin'
 GROUP BY amname, amsupport, opcname, amprocfamily
 HAVING count(*) != amsupport OR amprocfamily IS NULL;
  amname | opcname | count 
@@ -1230,7 +1220,7 @@ SELECT amname, opcname, count(*)
 FROM pg_am am JOIN pg_opclass op ON opcmethod = am.oid
      LEFT JOIN pg_amproc p ON amprocfamily = opcfamily AND
          amproclefttype = amprocrighttype AND amproclefttype = opcintype
-WHERE am.amname = 'gist' OR am.amname = 'gin'
+WHERE am.amname = 'btree' OR am.amname = 'gist' OR am.amname = 'gin'
 GROUP BY amname, amsupport, opcname, amprocfamily
 HAVING (count(*) != amsupport AND count(*) != amsupport - 1)
     OR amprocfamily IS NULL;
@@ -1261,19 +1251,22 @@ WHERE p1.amprocfamily = p3.oid AND p4.amprocfamily = p6.oid AND
 (0 rows)
 
 -- For btree, though, we can do better since we know the support routines
--- must be of the form cmp(lefttype, righttype) returns int4.
+-- 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
-    (amprocnum != 1
+    (CASE WHEN amprocnum = 1
-     OR proretset
+          THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
-     OR prorettype != 'int4'::regtype
-     OR pronargs != 2
                OR proargtypes[0] != amproclefttype
-     OR proargtypes[1] != amprocrighttype);
+               OR proargtypes[1] != amprocrighttype
+          WHEN amprocnum = 2
+          THEN prorettype != 'void'::regtype OR proretset OR pronargs != 1
+               OR proargtypes[0] != 'internal'::regtype
+          ELSE true END);
  amprocfamily | amprocnum | oid | proname | opfname 
 --------------+-----------+-----+---------+---------
 (0 rows)

src/test/regress/sql/opr_sanity.sql

@@ -926,37 +926,29 @@ WHERE p1.amprocfamily = p3.oid AND p3.opfmethod = p2.oid AND
 
 -- Detect missing pg_amproc entries: should have as many support functions
 -- as AM expects for each datatype combination supported by the opfamily.
--- GiST/GIN are special cases because each has an optional support function.
+-- btree/GiST/GIN each allow one optional support function, though.
 
 SELECT p1.amname, p2.opfname, p3.amproclefttype, p3.amprocrighttype
 FROM pg_am AS p1, pg_opfamily AS p2, pg_amproc AS p3
 WHERE p2.opfmethod = p1.oid AND p3.amprocfamily = p2.oid AND
-    p1.amname <> 'gist' AND p1.amname <> 'gin' AND
-    p1.amsupport != (SELECT count(*) FROM pg_amproc AS p4
-                     WHERE p4.amprocfamily = p2.oid AND
-                           p4.amproclefttype = p3.amproclefttype AND
-                           p4.amprocrighttype = p3.amprocrighttype);
-
--- Similar check for GiST/GIN, allowing one optional proc
-
-SELECT p1.amname, p2.opfname, p3.amproclefttype, p3.amprocrighttype
-FROM pg_am AS p1, pg_opfamily AS p2, pg_amproc AS p3
-WHERE p2.opfmethod = p1.oid AND p3.amprocfamily = p2.oid AND
-    (p1.amname = 'gist' OR p1.amname = 'gin') AND
     (SELECT count(*) FROM pg_amproc AS p4
      WHERE p4.amprocfamily = p2.oid AND
            p4.amproclefttype = p3.amproclefttype AND
            p4.amprocrighttype = p3.amprocrighttype)
-      NOT IN (p1.amsupport, p1.amsupport - 1);
+    NOT BETWEEN
+      (CASE WHEN p1.amname IN ('btree', 'gist', 'gin') THEN p1.amsupport - 1
+            ELSE p1.amsupport END)
+      AND p1.amsupport;
 
 -- Also, check if there are any pg_opclass entries that don't seem to have
--- pg_amproc support.  Again, GiST/GIN have to be checked specially.
+-- pg_amproc support.  Again, opclasses with an optional support proc have
+-- to be checked specially.
 
 SELECT amname, opcname, count(*)
 FROM pg_am am JOIN pg_opclass op ON opcmethod = am.oid
      LEFT JOIN pg_amproc p ON amprocfamily = opcfamily AND
          amproclefttype = amprocrighttype AND amproclefttype = opcintype
-WHERE am.amname <> 'gist' AND am.amname <> 'gin'
+WHERE am.amname <> 'btree' AND am.amname <> 'gist' AND am.amname <> 'gin'
 GROUP BY amname, amsupport, opcname, amprocfamily
 HAVING count(*) != amsupport OR amprocfamily IS NULL;
 
@@ -964,7 +956,7 @@ SELECT amname, opcname, count(*)
 FROM pg_am am JOIN pg_opclass op ON opcmethod = am.oid
      LEFT JOIN pg_amproc p ON amprocfamily = opcfamily AND
          amproclefttype = amprocrighttype AND amproclefttype = opcintype
-WHERE am.amname = 'gist' OR am.amname = 'gin'
+WHERE am.amname = 'btree' OR am.amname = 'gist' OR am.amname = 'gin'
 GROUP BY amname, amsupport, opcname, amprocfamily
 HAVING (count(*) != amsupport AND count(*) != amsupport - 1)
     OR amprocfamily IS NULL;
@@ -990,7 +982,8 @@ WHERE p1.amprocfamily = p3.oid AND p4.amprocfamily = p6.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.
+-- must be of the form cmp(lefttype, righttype) returns int4
+-- or sortsupport(internal) returns void.
 
 SELECT p1.amprocfamily, p1.amprocnum,
 	p2.oid, p2.proname,
@@ -998,12 +991,14 @@ SELECT p1.amprocfamily, p1.amprocnum,
 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
-    (amprocnum != 1
+    (CASE WHEN amprocnum = 1
-     OR proretset
+          THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
-     OR prorettype != 'int4'::regtype
-     OR pronargs != 2
                OR proargtypes[0] != amproclefttype
-     OR proargtypes[1] != amprocrighttype);
+               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