sort-support-2.patch

diff --git a/doc/src/sgml/xindex.sgml b/doc/src/sgml/xindex.sgml
index 28324361a950f31737c0cbb6909726d02ce1af5f..51f847c9f9777707756a823e1f76ef0275bceed7 100644
*** a/doc/src/sgml/xindex.sgml
--- b/doc/src/sgml/xindex.sgml
*************** DEFAULT FOR TYPE int4 USING btree FAMILY
*** 758,764 ****
    OPERATOR 3 = ,
    OPERATOR 4 >= ,
    OPERATOR 5 > ,
!   FUNCTION 1 btint4cmp(int4, int4) ;
  
  CREATE OPERATOR CLASS int2_ops
  DEFAULT FOR TYPE int2 USING btree FAMILY integer_ops AS
--- 758,765 ----
    OPERATOR 3 = ,
    OPERATOR 4 >= ,
    OPERATOR 5 > ,
!   FUNCTION 1 btint4cmp(int4, int4) ,
!   FUNCTION 2 btint4sortsupport(internal) ;
  
  CREATE OPERATOR CLASS int2_ops
  DEFAULT FOR TYPE int2 USING btree FAMILY integer_ops AS
diff --git a/src/backend/access/nbtree/nbtcompare.c b/src/backend/access/nbtree/nbtcompare.c
index 23f2b61fe902cff7eebcf8526cf8116d90be75e8..62d03f96de07cbbe14dedefb1596220af398caea 100644
*** a/src/backend/access/nbtree/nbtcompare.c
--- b/src/backend/access/nbtree/nbtcompare.c
***************
*** 49,54 ****
--- 49,55 ----
  #include "postgres.h"
  
  #include "utils/builtins.h"
+ #include "utils/sortsupport.h"
  
  
  Datum
*************** btint4cmp(PG_FUNCTION_ARGS)
*** 83,88 ****
--- 84,112 ----
  		PG_RETURN_INT32(-1);
  }
  
+ static int
+ btint4fastcmp(Datum x, Datum y, SortSupportInfo sinfo)
+ {
+ 	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)
+ {
+ 	SortSupportInfo	sinfo = (SortSupportInfo) PG_GETARG_POINTER(0);
+ 
+ 	sinfo->comparator = btint4fastcmp;
+ 	PG_RETURN_VOID();
+ }
+ 
  Datum
  btint8cmp(PG_FUNCTION_ARGS)
  {
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index cb341b8db679382c3ab8de31ff7a0bb57bc8dc50..cb6a38bcfd29fb17fc62040eef79d34aee9fe802 100644
*** a/src/backend/utils/cache/lsyscache.c
--- b/src/backend/utils/cache/lsyscache.c
*************** get_compare_function_for_ordering_op(Oid
*** 287,292 ****
--- 287,348 ----
  }
  
  /*
+  * 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).
+  *
+  * *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 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_sort_function_for_ordering_op(Oid opno, Oid *sortfunc,
+ 								  bool *issupport, bool *reverse)
+ {
+ 	Oid			opfamily;
+ 	Oid			opcintype;
+ 	int16		strategy;
+ 
+ 	/* Find the operator in pg_amop */
+ 	if (get_ordering_op_properties(opno,
+ 								   &opfamily, &opcintype, &strategy))
+ 	{
+ 		/* Found a suitable opfamily, get matching support function */
+ 		*sortfunc = get_opfamily_proc(opfamily,
+ 									  opcintype,
+ 									  opcintype,
+ 									  BTSORTSUPPORT_PROC);
+ 		if (OidIsValid(*sortfunc))
+ 			*issupport = true;
+ 		else
+ 		{
+ 			/* datatype 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 */
+ 	*sortfunc = InvalidOid;
+ 	*issupport = false;
+ 	*reverse = false;
+ 	return false;
+ }
+ 
+ /*
   * get_equality_op_for_ordering_op
   *		Get the OID of the datatype-specific btree equality operator
   *		associated with an ordering operator (a "<" or ">" operator).
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index 3505236e5d3b9c753fed11822e7d4b147b0db0eb..179d607a2e959bf77f79fba85146551595f82ef5 100644
*** a/src/backend/utils/sort/tuplesort.c
--- b/src/backend/utils/sort/tuplesort.c
***************
*** 112,117 ****
--- 112,118 ----
  #include "utils/memutils.h"
  #include "utils/pg_rusage.h"
  #include "utils/rel.h"
+ #include "utils/sortsupport.h"
  #include "utils/tuplesort.h"
  
  
*************** typedef struct
*** 164,169 ****
--- 165,189 ----
  	int			tupindex;		/* see notes above */
  } SortTuple;
  
+ /* 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 */
+ } SortShimInfo;
+ 
+ /*
+  * Information we need for each sort key (column to be sorted).
+  */
+ typedef struct SortKeyData
+ {
+ 	SortSupportInfoData sinfo;	/* sort comparison function */
+ 	AttrNumber	attno;			/* table or index column number to sort */
+ 	bool		reverse;		/* descending-order sort? */
+ 	bool		nulls_first;	/* sort nulls first? */
+ } SortKeyData;
+ 
+ typedef SortKeyData *SortKey;
  
  /*
   * Possible states of a Tuplesort object.  These denote the states that
*************** struct Tuplesortstate
*** 339,345 ****
  	 * tuplesort_begin_heap and used only by the MinimalTuple routines.
  	 */
  	TupleDesc	tupDesc;
! 	ScanKey		scanKeys;		/* array of length nKeys */
  
  	/*
  	 * These variables are specific to the CLUSTER case; they are set by
--- 359,365 ----
  	 * tuplesort_begin_heap and used only by the MinimalTuple routines.
  	 */
  	TupleDesc	tupDesc;
! 	SortKey		sortKeys;		/* array of length nKeys */
  
  	/*
  	 * These variables are specific to the CLUSTER case; they are set by
*************** static void tuplesort_heap_insert(Tuples
*** 457,462 ****
--- 477,484 ----
  static void tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex);
  static unsigned int getlen(Tuplesortstate *state, int tapenum, bool eofOK);
  static void markrunend(Tuplesortstate *state, int tapenum);
+ static void PrepareSortSupportComparisonShim(SortSupportInfo sinfo,
+ 											 Oid cmpFunc);
  static int comparetup_heap(const SortTuple *a, const SortTuple *b,
  				Tuplesortstate *state);
  static void copytup_heap(Tuplesortstate *state, SortTuple *stup, void *tup);
*************** tuplesort_begin_heap(TupleDesc tupDesc,
*** 613,653 ****
  	state->reversedirection = reversedirection_heap;
  
  	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
! 	state->scanKeys = (ScanKey) palloc0(nkeys * sizeof(ScanKeyData));
  
  	for (i = 0; i < nkeys; i++)
  	{
  		Oid			sortFunction;
! 		bool		reverse;
! 		int			flags;
  
  		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;
  
! 		/*
! 		 * 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);
--- 635,674 ----
  	state->reversedirection = reversedirection_heap;
  
  	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
! 	state->sortKeys = (SortKey) palloc0(nkeys * sizeof(SortKeyData));
  
  	for (i = 0; i < nkeys; i++)
  	{
+ 		SortKey		sortKey = state->sortKeys + i;
  		Oid			sortFunction;
! 		bool		issupport;
  
  		AssertArg(attNums[i] != 0);
  		AssertArg(sortOperators[i] != 0);
  
! 		if (!get_sort_function_for_ordering_op(sortOperators[i],
! 											   &sortFunction,
! 											   &issupport,
! 											   &sortKey->reverse))
  			elog(ERROR, "operator %u is not a valid ordering operator",
  				 sortOperators[i]);
  
! 		sortKey->sinfo.sinfo_collation = sortCollations[i];
! 		sortKey->sinfo.sinfo_cxt = CurrentMemoryContext;
! 		sortKey->attno = attNums[i];
! 		sortKey->nulls_first = nullsFirstFlags[i];
  
! 		if (issupport)
! 		{
! 			/* The sort support function should provide a comparator */
! 			OidFunctionCall1(sortFunction, PointerGetDatum(&sortKey->sinfo));
! 			Assert(sortKey->sinfo.comparator != NULL);
! 		}
! 		else
! 		{
! 			/* We'll use a shim to call the old-style btree comparator */
! 			PrepareSortSupportComparisonShim(&sortKey->sinfo, sortFunction);
! 		}
  	}
  
  	MemoryContextSwitchTo(oldcontext);
*************** ApplySortFunction(FmgrInfo *sortFunction
*** 2707,2712 ****
--- 2728,2824 ----
  								   datum2, isNull2);
  }
  
+ /*
+  * 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, SortSupportInfo sinfo)
+ {
+ 	SortShimInfo *extra = (SortShimInfo *) sinfo->sinfo_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.
+  */
+ static void
+ PrepareSortSupportComparisonShim(SortSupportInfo sinfo, Oid cmpFunc)
+ {
+ 	SortShimInfo   *extra;
+ 
+ 	extra = (SortShimInfo *) MemoryContextAlloc(sinfo->sinfo_cxt,
+ 												sizeof(SortShimInfo));
+ 
+ 	/* Lookup the comparison function */
+ 	fmgr_info_cxt(cmpFunc, &extra->flinfo, sinfo->sinfo_cxt);
+ 
+ 	/* We can initialize the callinfo just once and re-use it */
+ 	InitFunctionCallInfoData(extra->fcinfo, &extra->flinfo, 2,
+ 							 sinfo->sinfo_collation, NULL, NULL);
+ 	extra->fcinfo.argnull[0] = false;
+ 	extra->fcinfo.argnull[1] = false;
+ 
+ 	sinfo->sinfo_extra = extra;
+ 	sinfo->comparator = comparison_shim;
+ }
+ 
+ /*
+  * 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
+ inlineApplyComparator(SortKey sortKey,
+ 					  Datum datum1, bool isNull1,
+ 					  Datum datum2, bool isNull2)
+ {
+ 	int			compare;
+ 
+ 	if (isNull1)
+ 	{
+ 		if (isNull2)
+ 			compare = 0;		/* NULL "=" NULL */
+ 		else if (sortKey->nulls_first)
+ 			compare = -1;		/* NULL "<" NOT_NULL */
+ 		else
+ 			compare = 1;		/* NULL ">" NOT_NULL */
+ 	}
+ 	else if (isNull2)
+ 	{
+ 		if (sortKey->nulls_first)
+ 			compare = 1;		/* NOT_NULL ">" NULL */
+ 		else
+ 			compare = -1;		/* NOT_NULL "<" NULL */
+ 	}
+ 	else
+ 	{
+ 		compare = (*sortKey->sinfo.comparator) (datum1, datum2,
+ 												&sortKey->sinfo);
+ 		if (sortKey->reverse)
+ 			compare = -compare;
+ 	}
+ 
+ 	return compare;
+ }
+ 
  
  /*
   * Routines specialized for HeapTuple (actually MinimalTuple) case
*************** ApplySortFunction(FmgrInfo *sortFunction
*** 2715,2721 ****
  static int
  comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
  {
! 	ScanKey		scanKey = state->scanKeys;
  	HeapTupleData ltup;
  	HeapTupleData rtup;
  	TupleDesc	tupDesc;
--- 2827,2833 ----
  static int
  comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
  {
! 	SortKey		sortKey = state->sortKeys;
  	HeapTupleData ltup;
  	HeapTupleData rtup;
  	TupleDesc	tupDesc;
*************** comparetup_heap(const SortTuple *a, cons
*** 2726,2735 ****
  	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);
  	if (compare != 0)
  		return compare;
  
--- 2838,2846 ----
  	CHECK_FOR_INTERRUPTS();
  
  	/* Compare the leading sort key */
! 	compare = inlineApplyComparator(sortKey,
! 									a->datum1, a->isnull1,
! 									b->datum1, b->isnull1);
  	if (compare != 0)
  		return compare;
  
*************** comparetup_heap(const SortTuple *a, cons
*** 2739,2748 ****
  	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++)
  	{
! 		AttrNumber	attno = scanKey->sk_attno;
  		Datum		datum1,
  					datum2;
  		bool		isnull1,
--- 2850,2859 ----
  	rtup.t_len = ((MinimalTuple) b->tuple)->t_len + MINIMAL_TUPLE_OFFSET;
  	rtup.t_data = (HeapTupleHeader) ((char *) b->tuple - MINIMAL_TUPLE_OFFSET);
  	tupDesc = state->tupDesc;
! 	sortKey++;
! 	for (nkey = 1; nkey < state->nKeys; nkey++, sortKey++)
  	{
! 		AttrNumber	attno = sortKey->attno;
  		Datum		datum1,
  					datum2;
  		bool		isnull1,
*************** comparetup_heap(const SortTuple *a, cons
*** 2751,2760 ****
  		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);
  		if (compare != 0)
  			return compare;
  	}
--- 2862,2870 ----
  		datum1 = heap_getattr(&ltup, attno, tupDesc, &isnull1);
  		datum2 = heap_getattr(&rtup, attno, tupDesc, &isnull2);
  
! 		compare = inlineApplyComparator(sortKey,
! 										datum1, isnull1,
! 										datum2, isnull2);
  		if (compare != 0)
  			return compare;
  	}
*************** copytup_heap(Tuplesortstate *state, Sort
*** 2781,2787 ****
  	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->tupDesc,
  								&stup->isnull1);
  }
--- 2891,2897 ----
  	htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
  	htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
  	stup->datum1 = heap_getattr(&htup,
! 								state->sortKeys[0].attno,
  								state->tupDesc,
  								&stup->isnull1);
  }
*************** readtup_heap(Tuplesortstate *state, Sort
*** 2833,2839 ****
  	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->tupDesc,
  								&stup->isnull1);
  }
--- 2943,2949 ----
  	htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
  	htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
  	stup->datum1 = heap_getattr(&htup,
! 								state->sortKeys[0].attno,
  								state->tupDesc,
  								&stup->isnull1);
  }
*************** readtup_heap(Tuplesortstate *state, Sort
*** 2841,2852 ****
  static void
  reversedirection_heap(Tuplesortstate *state)
  {
! 	ScanKey		scanKey = state->scanKeys;
  	int			nkey;
  
! 	for (nkey = 0; nkey < state->nKeys; nkey++, scanKey++)
  	{
! 		scanKey->sk_flags ^= (SK_BT_DESC | SK_BT_NULLS_FIRST);
  	}
  }
  
--- 2951,2963 ----
  static void
  reversedirection_heap(Tuplesortstate *state)
  {
! 	SortKey		sortKey = state->sortKeys;
  	int			nkey;
  
! 	for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
  	{
! 		sortKey->reverse = !sortKey->reverse;
! 		sortKey->nulls_first = !sortKey->nulls_first;
  	}
  }
  
diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h
index 347d9423ba3d092235033d2447fe12f21f43011d..9a751ce100455f81740fa29491d03b050572bee9 100644
*** a/src/include/access/nbtree.h
--- b/src/include/access/nbtree.h
*************** typedef struct xl_btree_newroot
*** 417,429 ****
  
  /*
   *	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.
   */
  
! #define BTORDER_PROC	1
  
  /*
   *	We need to be able to tell the difference between read and write
--- 417,434 ----
  
  /*
   *	When a new operator class is declared, we require that the user
!  *	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 BTSORTSUPPORT_PROC	2
  
  /*
   *	We need to be able to tell the difference between read and write
diff --git a/src/include/catalog/pg_am.h b/src/include/catalog/pg_am.h
index 8b075d30d689a9f7fa022d3f554626b6f5086883..ddacdf274c49ac7e83b941aa983dbf96214f5812 100644
*** a/src/include/catalog/pg_am.h
--- b/src/include/catalog/pg_am.h
*************** typedef FormData_pg_am *Form_pg_am;
*** 117,123 ****
   * ----------------
   */
  
! 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 ));
  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 ));
--- 117,123 ----
   * ----------------
   */
  
! 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 e5d43f7c1d3ef02322ad2c5b376885a17e5f7430..9d0cb28fac8b772ea384ba72aa6ac70fa4174644 100644
*** a/src/include/catalog/pg_amproc.h
--- b/src/include/catalog/pg_amproc.h
*************** DATA(insert (	1976   21 21 1 350 ));
*** 100,105 ****
--- 100,106 ----
  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 321 ));
  DATA(insert (	1976   23 20 1 2188 ));
  DATA(insert (	1976   23 21 1 2191 ));
  DATA(insert (	1976   20 20 1 842 ));
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 28e53b7b7c367731b5b0102f5afe032b9bbf09a0..3bd52e7add9645875c31f518baaff28824bf9f85 100644
*** a/src/include/catalog/pg_proc.h
--- b/src/include/catalog/pg_proc.h
*************** DATA(insert OID = 350 (  btint2cmp		   P
*** 568,573 ****
--- 568,575 ----
  DESCR("less-equal-greater");
  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 = 321 (  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 = 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_ ));
diff --git a/src/include/utils/builtins.h b/src/include/utils/builtins.h
index 47a14125c48fc384b5d5c8d77abad99cec114e22..9a43e490b1d2d91c8d19a9c68524b54a3831a727 100644
*** a/src/include/utils/builtins.h
--- b/src/include/utils/builtins.h
*************** extern void pg_lltoa(int64 ll, char *a);
*** 279,285 ****
  
  /*
   *		Per-opclass comparison functions for new btrees.  These are
!  *		stored in pg_amproc and defined in access/nbtree/nbtcompare.c
   */
  extern Datum btboolcmp(PG_FUNCTION_ARGS);
  extern Datum btint2cmp(PG_FUNCTION_ARGS);
--- 279,285 ----
  
  /*
   *		Per-opclass comparison functions for new btrees.  These are
!  *		stored in pg_amproc; most are defined in access/nbtree/nbtcompare.c
   */
  extern Datum btboolcmp(PG_FUNCTION_ARGS);
  extern Datum btint2cmp(PG_FUNCTION_ARGS);
*************** extern Datum btcharcmp(PG_FUNCTION_ARGS)
*** 304,309 ****
--- 304,316 ----
  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 btint4sortsupport(PG_FUNCTION_ARGS);
+ 
  /* float.c */
  extern PGDLLIMPORT int extra_float_digits;
  
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index d3ad4f14282641fc7ba35f3188d76a8b097dd22e..f5e8e25a38e159ea7364d66917d724771aef02f4 100644
*** a/src/include/utils/lsyscache.h
--- b/src/include/utils/lsyscache.h
*************** extern bool get_ordering_op_properties(O
*** 50,55 ****
--- 50,57 ----
  						   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
index ...7e196a413653fa0de3f10b246078e01be46453a3 .
*** a/src/include/utils/sortsupport.h
--- b/src/include/utils/sortsupport.h
***************
*** 0 ****
--- 1,87 ----
+ /*-------------------------------------------------------------------------
+  *
+  * 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 SortSupportInfoData 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
+  * SortSupportInfoData struct when called, so they can use it to store
+  * additional private data as needed.  In particular, for collation-aware
+  * datatypes, the sinfo_collation field is set before calling BTSORTSUPPORT
+  * and is available to all support functions.  Additional opclass-dependent
+  * data can be stored using the sinfo_extra field.  Any such data
+  * should be allocated in the sinfo_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
+ 
+ typedef struct SortSupportInfoData *SortSupportInfo;
+ 
+ typedef struct SortSupportInfoData
+ {
+ 	/*
+ 	 * These fields are initialized before calling the BTSORTSUPPORT function
+ 	 * and should not be changed later.
+ 	 */
+ 	Oid			sinfo_collation;		/* Collation to use, or InvalidOid */
+ 	MemoryContext	sinfo_cxt;			/* Context containing sort info */
+ 
+ 	/*
+ 	 * sinfo_extra is zeroed before calling the BTSORTSUPPORT function, and
+ 	 * is not touched subsequently by callers.
+ 	 */
+ 	void	   *sinfo_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, SortSupportInfo sinfo);
+ 
+ 	/*
+ 	 * Additional sort-acceleration functions might be added here later.
+ 	 */
+ } SortSupportInfoData;
+ 
+ #endif   /* SORTSUPPORT_H */
diff --git a/src/test/regress/expected/opr_sanity.out b/src/test/regress/expected/opr_sanity.out
index b915134fff33d3154a24d56361d0427810b90af1..a0ffd77e0ed889001b11fe174d62737e0d281d5e 100644
*** a/src/test/regress/expected/opr_sanity.out
--- b/src/test/regress/expected/opr_sanity.out
*************** WHERE p1.amprocfamily = p3.oid AND p3.op
*** 1186,1225 ****
  
  -- 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.
! 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);
   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.
  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'
  GROUP BY amname, amsupport, opcname, amprocfamily
  HAVING count(*) != amsupport OR amprocfamily IS NULL;
   amname | opcname | count 
--- 1186,1215 ----
  
  -- Detect missing pg_amproc entries: should have as many support functions
  -- as AM expects for each datatype combination supported by the opfamily.
! -- 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
      (SELECT count(*) FROM pg_amproc AS p4
       WHERE p4.amprocfamily = p2.oid AND
             p4.amproclefttype = p3.amproclefttype AND
             p4.amprocrighttype = p3.amprocrighttype)
!     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, 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 <> '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 
*************** SELECT amname, opcname, count(*)
*** 1230,1236 ****
  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'
  GROUP BY amname, amsupport, opcname, amprocfamily
  HAVING (count(*) != amsupport AND count(*) != amsupport - 1)
      OR amprocfamily IS NULL;
--- 1220,1226 ----
  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 = '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;
*************** WHERE p1.amprocfamily = p3.oid AND p4.am
*** 1261,1279 ****
  (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.
  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);
   amprocfamily | amprocnum | oid | proname | opfname 
  --------------+-----------+-----+---------+---------
  (0 rows)
--- 1251,1272 ----
  (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
! -- or sortsupport(internal) returns void.
  SELECT p1.amprocfamily, p1.amprocnum,
  	p2.oid, p2.proname,
  	p3.opfname
  FROM pg_amproc AS p1, pg_proc AS p2, pg_opfamily AS p3
  WHERE p3.opfmethod = (SELECT oid FROM pg_am WHERE amname = 'btree')
      AND p1.amprocfamily = p3.oid AND p1.amproc = p2.oid AND
!     (CASE WHEN amprocnum = 1
!           THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
!                OR proargtypes[0] != amproclefttype
!                OR proargtypes[1] != amprocrighttype
!           WHEN amprocnum = 2
!           THEN prorettype != 'void'::regtype OR proretset OR pronargs != 1
!                OR proargtypes[0] != 'internal'::regtype
!           ELSE true END);
   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 b0d143087e8fd68b30461c5d59b5bc5fdd3da086..6a79ea180c1c6c2d83891ca32a05e2249202ea5c 100644
*** a/src/test/regress/sql/opr_sanity.sql
--- b/src/test/regress/sql/opr_sanity.sql
*************** WHERE p1.amprocfamily = p3.oid AND p3.op
*** 926,962 ****
  
  -- 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.
! 
! 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);
  
  -- 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.
  
  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'
  GROUP BY amname, amsupport, opcname, amprocfamily
  HAVING count(*) != amsupport OR amprocfamily IS NULL;
  
--- 926,954 ----
  
  -- Detect missing pg_amproc entries: should have as many support functions
  -- as AM expects for each datatype combination supported by the opfamily.
! -- 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
      (SELECT count(*) FROM pg_amproc AS p4
       WHERE p4.amprocfamily = p2.oid AND
             p4.amproclefttype = p3.amproclefttype AND
             p4.amprocrighttype = p3.amprocrighttype)
!     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, 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 <> 'btree' AND am.amname <> 'gist' AND am.amname <> 'gin'
  GROUP BY amname, amsupport, opcname, amprocfamily
  HAVING count(*) != amsupport OR amprocfamily IS NULL;
  
*************** SELECT amname, opcname, count(*)
*** 964,970 ****
  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'
  GROUP BY amname, amsupport, opcname, amprocfamily
  HAVING (count(*) != amsupport AND count(*) != amsupport - 1)
      OR amprocfamily IS NULL;
--- 956,962 ----
  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 = '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;
*************** WHERE p1.amprocfamily = p3.oid AND p4.am
*** 990,996 ****
      (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.
  
  SELECT p1.amprocfamily, p1.amprocnum,
  	p2.oid, p2.proname,
--- 982,989 ----
      (p2.proretset OR p5.proretset OR p2.pronargs != p5.pronargs);
  
  -- For btree, though, we can do better since we know the support routines
! -- must be of the form cmp(lefttype, righttype) returns int4
! -- or sortsupport(internal) returns void.
  
  SELECT p1.amprocfamily, p1.amprocnum,
  	p2.oid, p2.proname,
*************** SELECT p1.amprocfamily, p1.amprocnum,
*** 998,1009 ****
  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);
  
  -- 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
--- 991,1004 ----
  FROM pg_amproc AS p1, pg_proc AS p2, pg_opfamily AS p3
  WHERE p3.opfmethod = (SELECT oid FROM pg_am WHERE amname = 'btree')
      AND p1.amprocfamily = p3.oid AND p1.amproc = p2.oid AND
!     (CASE WHEN amprocnum = 1
!           THEN prorettype != 'int4'::regtype OR proretset OR pronargs != 2
!                OR proargtypes[0] != amproclefttype
!                OR proargtypes[1] != amprocrighttype
!           WHEN amprocnum = 2
!           THEN prorettype != 'void'::regtype OR proretset OR pronargs != 1
!                OR proargtypes[0] != 'internal'::regtype
!           ELSE true END);
  
  -- For hash we can also do a little better: the support routines must be
  -- of the form hash(lefttype) returns int4.  There are several cases where