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

27 files changed, 834 insertions, 400 deletions

diff --git a/src/backend/access/nbtree/nbtcompare.c b/src/backend/access/nbtree/nbtcompare.c
index 23f2b61fe90..f86b7551d2c 100644
--- a/src/backend/access/nbtree/nbtcompare.c
+++ b/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();
+}
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 314324618a8..c3d3958da76 100644
--- a/src/backend/commands/analyze.c
+++ b/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;
-	int			cmpFlags;
+	SortSupport ssup;
 	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;
-	int			cmpFlags;
-	FmgrInfo	f_cmpfn;
+	SortSupportData ssup;
 	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);
-	fmgr_info(cmpFn, &f_cmpfn);
+	memset(&ssup, 0, sizeof(ssup));
+	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.cmpFlags = cmpFlags;
+		cxt.ssup = &ssup;
 		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 = ApplySortFunction(cxt->cmpFn, cxt->cmpFlags,
-								DEFAULT_COLLATION_OID,
-								da, false, db, false);
+	compare = ApplySortComparator(da, false, db, false, cxt->ssup);
 	if (compare != 0)
 		return compare;
 
diff --git a/src/backend/commands/opclasscmds.c b/src/backend/commands/opclasscmds.c
index 0ef3584bb2f..b7ab0e3203e 100644
--- a/src/backend/commands/opclasscmds.c
+++ b/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,27 +1152,48 @@ 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
-	 * procs must be 1-arg procs returning int4; otherwise we don't know.
+	 * btree comparison procs must be 2-arg procs returning int4, while btree
+	 * 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 (procform->pronargs != 2)
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("btree procedures must have two arguments")));
-		if (procform->prorettype != INT4OID)
-			ereport(ERROR,
-					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-					 errmsg("btree procedures must return integer")));
+		if (member->number == BTORDER_PROC)
+		{
+			if (procform->pronargs != 2)
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("btree comparison procedures must have two arguments")));
+			if (procform->prorettype != INT4OID)
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("btree comparison procedures must return integer")));
 
-		/*
-		 * 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];
+			/*
+			 * 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)
 	{
@@ -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
-		 * class' opcintype as lefttype and righttype.  In CREATE or ALTER
-		 * OPERATOR FAMILY, opcintype isn't available, so make the user
-		 * 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")));
-	}
+
+	/*
+	 * The default in CREATE OPERATOR CLASS is to use the class' opcintype as
+	 * lefttype and righttype.  In CREATE or ALTER OPERATOR FAMILY, opcintype
+	 * isn't available, so make the user 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);
 }
diff --git a/src/backend/executor/nodeMergeAppend.c b/src/backend/executor/nodeMergeAppend.c
index 43059664b93..6065e2176ac 100644
--- a/src/backend/executor/nodeMergeAppend.c
+++ b/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;
-		bool		reverse;
-		int			flags;
-
-		if (!get_compare_function_for_ordering_op(node->sortOperators[i],
-												  &sortFunction, &reverse))
-			elog(ERROR, "operator %u is not a valid ordering operator",
-				 node->sortOperators[i]);
-
-		/* We use btree's conventions for encoding directionality */
-		flags = 0;
-		if (reverse)
-			flags |= SK_BT_DESC;
-		if (node->nullsFirst[i])
-			flags |= SK_BT_NULLS_FIRST;
+		SortSupport	sortKey = mergestate->ms_sortkeys + i;
 
-		/*
-		 * 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);
+		sortKey->ssup_cxt = CurrentMemoryContext;
+		sortKey->ssup_collation = node->collations[i];
+		sortKey->ssup_nulls_first = node->nullsFirst[i];
+		sortKey->ssup_attno = node->sortColIdx[i];
+
+		PrepareSortSupportFromOrderingOp(node->sortOperators[i], sortKey);
 	}
 
 	/*
@@ -298,45 +276,22 @@ heap_compare_slots(MergeAppendState *node, SlotNumber slot1, SlotNumber slot2)
 
 	for (nkey = 0; nkey < node->ms_nkeys; nkey++)
 	{
-		ScanKey		scankey = node->ms_scankeys + nkey;
-		AttrNumber	attno = scankey->sk_attno;
+		SortSupport	sortKey = node->ms_sortkeys + nkey;
+		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)
-		{
-			if (isNull2)
-				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;
-			}
-		}
+		compare = ApplySortComparator(datum1, isNull1,
+									  datum2, isNull2,
+									  sortKey);
+		if (compare != 0)
+			return compare;
 	}
 	return 0;
 }
diff --git a/src/backend/executor/nodeMergejoin.c b/src/backend/executor/nodeMergejoin.c
index deaa79ed9fb..634931da425 100644
--- a/src/backend/executor/nodeMergejoin.c
+++ b/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
-	 * btree comparison support function, and the collation to use.
+	 * Everything we need to know to compare the left and right values is
+	 * stored here.
 	 */
-	bool		reverse;		/* if true, negate the cmpfn's output */
-	bool		nulls_first;	/* if true, nulls sort low */
-	FmgrInfo	cmpfinfo;
-	Oid			collation;
+	SortSupportData ssup;
 }	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;
-		AclResult	aclresult;
+		Oid			sortfunc;
 
 		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) */
-		cmpproc = get_opfamily_proc(opfamily,
-									op_lefttype,
-									op_righttype,
-									BTORDER_PROC);
-		if (!RegProcedureIsValid(cmpproc))		/* should not happen */
-			elog(ERROR, "missing support function %d(%u,%u) in opfamily %u",
-				 BTORDER_PROC, op_lefttype, op_righttype, opfamily);
-
-		/* Check permission to call cmp function */
-		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;
+		/* And get the matching support or comparison function */
+		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 (!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 */
+			PrepareSortSupportComparisonShim(sortfunc, &clause->ssup);
+		}
 
 		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.
-		 */
-		if (clause->lisnull)
-		{
-			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;
-		}
 
 		/*
-		 * OK to call the comparison function.
+		 * Special case for NULL-vs-NULL, else use standard comparison.
 		 */
-		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)
+		if (clause->lisnull && clause->risnull)
 		{
-			nulleqnull = true;	/* treat like NULL = NULL */
+			nulleqnull = true;		/* NULL "=" NULL */
 			continue;
 		}
-		result = DatumGetInt32(fresult);
 
-		if (clause->reverse)
-			result = -result;
+		result = ApplySortComparator(clause->ldatum, clause->lisnull,
+									 clause->rdatum, clause->risnull,
+									 &clause->ssup);
 
 		if (result != 0)
 			break;
 	}
 
 	/*
-	 * If we had any null comparison results or NULL-vs-NULL inputs, we do not
-	 * want to report that the tuples are equal.  Instead, if result is still
-	 * 0, change it to +1.	This will result in advancing the inner side of
-	 * the join.
+	 * If we had any NULL-vs-NULL inputs, we do not want to report that the
+	 * tuples are equal.  Instead, if result is still 0, change it to +1.
+	 * This will result in advancing the inner side of 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;
diff --git a/src/backend/utils/adt/date.c b/src/backend/utils/adt/date.c
index 271b57a8302..ff4e3044f07 100644
--- a/src/backend/utils/adt/date.c
+++ b/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)
 {
diff --git a/src/backend/utils/adt/float.c b/src/backend/utils/adt/float.c
index e3c84fd6c55..63b09a4ecac 100644
--- a/src/backend/utils/adt/float.c
+++ b/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)
 {
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index 45e70029e53..450dcd47277 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/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)
 {
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index cb341b8db67..7a4306e93f5 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -244,19 +244,22 @@ get_ordering_op_properties(Oid opno,
 }
 
 /*
- * get_compare_function_for_ordering_op
- *		Get the OID of the datatype-specific btree comparison function
+ * get_sort_function_for_ordering_op
+ *		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,
-									 opcintype,
-									 opcintype,
-									 BTORDER_PROC);
-
-		if (!OidIsValid(*cmpfunc))		/* should not happen */
-			elog(ERROR, "missing support function %d(%u,%u) in opfamily %u",
-				 BTORDER_PROC, opcintype, opcintype, 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 */
+				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;
 }
diff --git a/src/backend/utils/sort/Makefile b/src/backend/utils/sort/Makefile
index 9e20ecb3267..2ef4965ee6d 100644
--- a/src/backend/utils/sort/Makefile
+++ b/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
diff --git a/src/backend/utils/sort/sortsupport.c b/src/backend/utils/sort/sortsupport.c
new file mode 100644
index 00000000000..4a87f11827f
--- /dev/null
+++ b/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);
+	}
+}
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index 3505236e5d3..28d585fc3b5 100644
--- a/src/backend/utils/sort/tuplesort.c
+++ b/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 */
-	int			sortFnFlags;	/* equivalent to sk_flags */
-	Oid			sortCollation;	/* equivalent to sk_collation */
+	SortSupport	datumKey;
 	/* 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;
-		bool		reverse;
-		int			flags;
+		SortSupport	sortKey = state->sortKeys + i;
 
 		AssertArg(attNums[i] != 0);
 		AssertArg(sortOperators[i] != 0);
 
-		if (!get_compare_function_for_ordering_op(sortOperators[i],
-												  &sortFunction, &reverse))
-			elog(ERROR, "operator %u is not a valid ordering operator",
-				 sortOperators[i]);
-
-		/* We use btree's conventions for encoding directionality */
-		flags = 0;
-		if (reverse)
-			flags |= SK_BT_DESC;
-		if (nullsFirstFlags[i])
-			flags |= SK_BT_NULLS_FIRST;
+		sortKey->ssup_cxt = CurrentMemoryContext;
+		sortKey->ssup_collation = sortCollations[i];
+		sortKey->ssup_nulls_first = nullsFirstFlags[i];
+		sortKey->ssup_attno = attNums[i];
 
-		/*
-		 * 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);
+		PrepareSortSupportFromOrderingOp(sortOperators[i], sortKey);
 	}
 
 	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 */
-	if (!get_compare_function_for_ordering_op(sortOperator,
-											  &sortFunction, &reverse))
-		elog(ERROR, "operator %u is not a valid ordering operator",
-			 sortOperator);
-	fmgr_info(sortFunction, &state->sortOpFn);
+	/* Prepare SortSupport data */
+	state->datumKey = (SortSupport) palloc0(sizeof(SortSupportData));
 
-	/* set ordering flags and collation */
-	state->sortFnFlags = reverse ? SK_BT_DESC : 0;
-	if (nullsFirstFlag)
-		state->sortFnFlags |= SK_BT_NULLS_FIRST;
-	state->sortCollation = sortCollation;
+	state->datumKey->ssup_cxt = CurrentMemoryContext;
+	state->datumKey->ssup_collation = sortCollation;
+	state->datumKey->ssup_nulls_first = nullsFirstFlag;
+
+	PrepareSortSupportFromOrderingOp(sortOperator, state->datumKey);
 
 	/* lookup necessary attributes of the datum type */
 	get_typlenbyval(datumType, &typlen, &typbyval);
@@ -2605,29 +2580,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.
  */
 static inline Datum
@@ -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,
-									  scanKey->sk_collation,
-									  a->datum1, a->isnull1,
-									  b->datum1, b->isnull1);
+	compare = ApplySortComparator(a->datum1, a->isnull1,
+								  b->datum1, b->isnull1,
+								  sortKey);
 	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++;
-	for (nkey = 1; nkey < state->nKeys; nkey++, scanKey++)
+	sortKey++;
+	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,
-										  scanKey->sk_collation,
-										  datum1, isnull1,
-										  datum2, isnull2);
+		compare = ApplySortComparator(datum1, isnull1,
+									  datum2, isnull2,
+									  sortKey);
 		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,
-								   state->sortCollation,
-								   a->datum1, a->isnull1,
-								   b->datum1, b->isnull1);
+	return ApplySortComparator(a->datum1, a->isnull1,
+							   b->datum1, b->isnull1,
+							   state->datumKey);
 }
 
 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;
 }
 
 /*
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index cbe529e1568..afeae6f9f7f 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/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",
-						  amprocnum, amproc);
+		appendPQExpBuffer(q, "FUNCTION %s", amprocnum);
+
+		if (*amproclefttype && *amprocrighttype)
+			appendPQExpBuffer(q, " (%s, %s)", amproclefttype, amprocrighttype);
+
+		appendPQExpBuffer(q, " %s", amproc);
 
 		needComma = true;
 	}
diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h
index 347d9423ba3..9a751ce1004 100644
--- a/src/include/access/nbtree.h
+++ b/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
- *	two keys a and b, a < b, a = b, or a > b.  This routine must
- *	return < 0, 0, > 0, respectively, in these three cases.  Since we
- *	only have one such proc in amproc, it's number 1.
+ *	supply us with an amproc procedure (BTORDER_PROC) for determining
+ *	whether, for two keys a and b, a < b, a = b, or a > b.  This routine
+ *	must return < 0, 0, > 0, respectively, in these three cases.  (It must
+ *	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 BTORDER_PROC		1
+#define BTSORTSUPPORT_PROC	2
 
 /*
  *	We need to be able to tell the difference between read and write
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 5a73726ba31..95c424b81fa 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201111271
+#define CATALOG_VERSION_NO	201112061
 
 #endif
diff --git a/src/include/catalog/pg_am.h b/src/include/catalog/pg_am.h
index 8b075d30d68..ddacdf274c4 100644
--- a/src/include/catalog/pg_am.h
+++ b/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 ));
diff --git a/src/include/catalog/pg_amproc.h b/src/include/catalog/pg_amproc.h
index e5d43f7c1d3..8571dd08709 100644
--- a/src/include/catalog/pg_amproc.h
+++ b/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 ));
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 28e53b7b7c3..7451a12908a 100644
--- a/src/include/catalog/pg_proc.h
+++ b/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_ ));
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 0a89f189d7c..f5935e2106f 100644
--- a/src/include/nodes/execnodes.h
+++ b/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[] */
diff --git a/src/include/utils/builtins.h b/src/include/utils/builtins.h
index 47a14125c48..17550e0b3fd 100644
--- a/src/include/utils/builtins.h
+++ b/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;
 
diff --git a/src/include/utils/date.h b/src/include/utils/date.h
index 9e10e1afce5..13e0db4e1cb 100644
--- a/src/include/utils/date.h
+++ b/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);
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index d3ad4f14282..311aa25862e 100644
--- a/src/include/utils/lsyscache.h
+++ b/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,
-									 Oid *cmpfunc, bool *reverse);
+extern bool get_sort_function_for_ordering_op(Oid opno, Oid *sortfunc,
+								  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);
diff --git a/src/include/utils/sortsupport.h b/src/include/utils/sortsupport.h
new file mode 100644
index 00000000000..bd6dde34729
--- /dev/null
+++ b/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 */
diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h
index dc91f3eaaa1..7661744e4d4 100644
--- a/src/include/utils/timestamp.h
+++ b/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);
diff --git a/src/include/utils/tuplesort.h b/src/include/utils/tuplesort.h
index 1ebcbfe1724..5c48e0e10f5 100644
--- a/src/include/utils/tuplesort.h
+++ b/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 */
diff --git a/src/test/regress/expected/opr_sanity.out b/src/test/regress/expected/opr_sanity.out
index b915134fff3..a0ffd77e0ed 100644
--- a/src/test/regress/expected/opr_sanity.out
+++ b/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
-     OR proretset
-     OR prorettype != 'int4'::regtype
-     OR pronargs != 2
-     OR proargtypes[0] != amproclefttype
-     OR proargtypes[1] != amprocrighttype);
+    (CASE WHEN amprocnum = 1
+          THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
+               OR proargtypes[0] != amproclefttype
+               OR proargtypes[1] != amprocrighttype
+          WHEN amprocnum = 2
+          THEN prorettype != 'void'::regtype OR proretset OR pronargs != 1
+               OR proargtypes[0] != 'internal'::regtype
+          ELSE true END);
  amprocfamily | amprocnum | oid | proname | opfname 
 --------------+-----------+-----+---------+---------
 (0 rows)
diff --git a/src/test/regress/sql/opr_sanity.sql b/src/test/regress/sql/opr_sanity.sql
index b0d143087e8..6a79ea180c1 100644
--- a/src/test/regress/sql/opr_sanity.sql
+++ b/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
-     OR proretset
-     OR prorettype != 'int4'::regtype
-     OR pronargs != 2
-     OR proargtypes[0] != amproclefttype
-     OR proargtypes[1] != amprocrighttype);
+    (CASE WHEN amprocnum = 1
+          THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
+               OR proargtypes[0] != amproclefttype
+               OR proargtypes[1] != amprocrighttype
+          WHEN amprocnum = 2
+          THEN prorettype != 'void'::regtype OR proretset OR pronargs != 1
+               OR proargtypes[0] != 'internal'::regtype
+          ELSE true END);
 
 -- For hash we can also do a little better: the support routines must be
 -- of the form hash(lefttype) returns int4.  There are several cases where