B-Tree index builds, CLUSTER, and sortsupport

diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
new file mode 100644
index e8a89d2..a6c44a7
*** a/src/backend/access/nbtree/nbtsort.c
--- b/src/backend/access/nbtree/nbtsort.c
*************** _bt_load(BTWriteState *wstate, BTSpool *
*** 684,689 ****
--- 684,690 ----
  	int			i,
  				keysz = RelationGetNumberOfAttributes(wstate->index);
  	ScanKey		indexScanKey = NULL;
+ 	SortSupport sortKeys;
  
  	if (merge)
  	{
*************** _bt_load(BTWriteState *wstate, BTSpool *
*** 699,704 ****
--- 700,730 ----
  										true, &should_free2);
  		indexScanKey = _bt_mkscankey_nodata(wstate->index);
  
+ 		/* Prepare SortSupport data for each column */
+ 		sortKeys = (SortSupport) palloc0(keysz * sizeof(SortSupportData));
+ 
+ 		for (i = 0; i < keysz; i++)
+ 		{
+ 			SortSupport sortKey = sortKeys + i;
+ 			ScanKey		scanKey = indexScanKey + i;
+ 			int16		strategy;
+ 
+ 			sortKey->ssup_cxt = CurrentMemoryContext;
+ 			sortKey->ssup_collation = scanKey->sk_collation;
+ 			sortKey->ssup_nulls_first =
+ 				(scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
+ 			sortKey->ssup_attno = scanKey->sk_attno;
+ 
+ 			AssertState(sortKey->ssup_attno != 0);
+ 
+ 			strategy  = (scanKey->sk_flags & SK_BT_DESC) != 0 ?
+ 				BTGreaterStrategyNumber : BTLessStrategyNumber;
+ 
+ 			PrepareSortSupportFromIndexRel(wstate->index, strategy, sortKey);
+ 		}
+ 
+ 		_bt_freeskey(indexScanKey);
+ 
  		for (;;)
  		{
  			load1 = true;		/* load BTSpool next ? */
*************** _bt_load(BTWriteState *wstate, BTSpool *
*** 711,753 ****
  			{
  				for (i = 1; i <= keysz; i++)
  				{
! 					ScanKey		entry;
  					Datum		attrDatum1,
  								attrDatum2;
  					bool		isNull1,
  								isNull2;
  					int32		compare;
  
! 					entry = indexScanKey + i - 1;
  					attrDatum1 = index_getattr(itup, i, tupdes, &isNull1);
  					attrDatum2 = index_getattr(itup2, i, tupdes, &isNull2);
- 					if (isNull1)
- 					{
- 						if (isNull2)
- 							compare = 0;		/* NULL "=" NULL */
- 						else if (entry->sk_flags & SK_BT_NULLS_FIRST)
- 							compare = -1;		/* NULL "<" NOT_NULL */
- 						else
- 							compare = 1;		/* NULL ">" NOT_NULL */
- 					}
- 					else if (isNull2)
- 					{
- 						if (entry->sk_flags & SK_BT_NULLS_FIRST)
- 							compare = 1;		/* NOT_NULL ">" NULL */
- 						else
- 							compare = -1;		/* NOT_NULL "<" NULL */
- 					}
- 					else
- 					{
- 						compare =
- 							DatumGetInt32(FunctionCall2Coll(&entry->sk_func,
- 														 entry->sk_collation,
- 															attrDatum1,
- 															attrDatum2));
  
! 						if (entry->sk_flags & SK_BT_DESC)
! 							compare = -compare;
! 					}
  					if (compare > 0)
  					{
  						load1 = false;
--- 737,756 ----
  			{
  				for (i = 1; i <= keysz; i++)
  				{
! 					SortSupport	entry;
  					Datum		attrDatum1,
  								attrDatum2;
  					bool		isNull1,
  								isNull2;
  					int32		compare;
  
! 					entry = sortKeys + i - 1;
  					attrDatum1 = index_getattr(itup, i, tupdes, &isNull1);
  					attrDatum2 = index_getattr(itup2, i, tupdes, &isNull2);
  
! 					compare = ApplySortComparator(attrDatum1, isNull1,
! 												  attrDatum2, isNull2,
! 												  entry);
  					if (compare > 0)
  					{
  						load1 = false;
*************** _bt_load(BTWriteState *wstate, BTSpool *
*** 781,787 ****
  												true, &should_free2);
  			}
  		}
! 		_bt_freeskey(indexScanKey);
  	}
  	else
  	{
--- 784,790 ----
  												true, &should_free2);
  			}
  		}
! 		pfree(sortKeys);
  	}
  	else
  	{
diff --git a/src/backend/utils/sort/sortsupport.c b/src/backend/utils/sort/sortsupport.c
new file mode 100644
index 2240fd0..8391b1c
*** a/src/backend/utils/sort/sortsupport.c
--- b/src/backend/utils/sort/sortsupport.c
***************
*** 21,26 ****
--- 21,27 ----
  #include "access/nbtree.h"
  #include "fmgr.h"
  #include "utils/lsyscache.h"
+ #include "utils/rel.h"
  #include "utils/sortsupport.h"
  
  
*************** PrepareSortSupportComparisonShim(Oid cmp
*** 86,113 ****
  }
  
  /*
!  * 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			sortSupportFunction;
- 	Oid			opfamily;
- 	Oid			opcintype;
- 	int16		strategy;
- 
- 	Assert(ssup->comparator == NULL);
- 
- 	/* Find the operator in pg_amop */
- 	if (!get_ordering_op_properties(orderingOp, &opfamily, &opcintype,
- 									&strategy))
- 		elog(ERROR, "operator %u is not a valid ordering operator",
- 			 orderingOp);
- 	ssup->ssup_reverse = (strategy == BTGreaterStrategyNumber);
  
  	/* Look for a sort support function */
  	sortSupportFunction = get_opfamily_proc(opfamily, opcintype, opcintype,
--- 87,98 ----
  }
  
  /*
!  * Lookup and call sortsupport function to setup state, or create shim
   */
! static void
! FinishSortSupportFunction(Oid opfamily, Oid opcintype, SortSupport ssup)
  {
  	Oid			sortSupportFunction;
  
  	/* Look for a sort support function */
  	sortSupportFunction = get_opfamily_proc(opfamily, opcintype, opcintype,
*************** PrepareSortSupportFromOrderingOp(Oid ord
*** 136,138 ****
--- 121,177 ----
  		PrepareSortSupportComparisonShim(sortFunction, ssup);
  	}
  }
+ 
+ /*
+  * 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			opfamily;
+ 	Oid			opcintype;
+ 	int16		strategy;
+ 
+ 	Assert(ssup->comparator == NULL);
+ 
+ 	/* Find the operator in pg_amop */
+ 	if (!get_ordering_op_properties(orderingOp, &opfamily, &opcintype,
+ 									&strategy))
+ 		elog(ERROR, "operator %u is not a valid ordering operator",
+ 			 orderingOp);
+ 	ssup->ssup_reverse = (strategy == BTGreaterStrategyNumber);
+ 
+ 	FinishSortSupportFunction(opfamily, opcintype, ssup);
+ }
+ 
+ /*
+  * Fill in SortSupport given an index relation, attribute, and strategy.
+  *
+  * Caller must previously have zeroed the SortSupportData structure and then
+  * filled in ssup_cxt, ssup_attno, ssup_collation, and ssup_nulls_first.  This
+  * will fill in ssup_reverse (based on the supplied strategy), as well as the
+  * comparator function pointer.
+  */
+ void
+ PrepareSortSupportFromIndexRel(Relation indexRel, int16 strategy,
+ 							   SortSupport ssup)
+ {
+ 	Oid			opfamily = indexRel->rd_opfamily[ssup->ssup_attno - 1];
+ 	Oid			opcintype = indexRel->rd_opcintype[ssup->ssup_attno - 1];
+ 
+ 	Assert(ssup->comparator == NULL);
+ 
+ 	/* Find the operator in pg_amop */
+ 	if (indexRel->rd_rel->relam != BTREE_AM_OID)
+ 		elog(ERROR, "unexpected non-btree AM: %u", indexRel->rd_rel->relam);
+ 	if (strategy != BTGreaterStrategyNumber &&
+ 		strategy != BTLessStrategyNumber)
+ 		elog(ERROR, "unexpected sort support strategy: %d", strategy);
+ 	ssup->ssup_reverse = (strategy == BTGreaterStrategyNumber);
+ 
+ 	FinishSortSupportFunction(opfamily, opcintype, ssup);
+ }
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
new file mode 100644
index 8e57505..0486bab
*** a/src/backend/utils/sort/tuplesort.c
--- b/src/backend/utils/sort/tuplesort.c
*************** struct Tuplesortstate
*** 257,269 ****
  										int tapenum, unsigned int len);
  
  	/*
- 	 * Function to reverse the sort direction from its current state. (We
- 	 * could dispense with this if we wanted to enforce that all variants
- 	 * represent the sort key information alike.)
- 	 */
- 	void		(*reversedirection) (Tuplesortstate *state);
- 
- 	/*
  	 * This array holds the tuples now in sort memory.  If we are in state
  	 * INITIAL, the tuples are in no particular order; if we are in state
  	 * SORTEDINMEM, the tuples are in final sorted order; in states BUILDRUNS
--- 257,262 ----
*************** struct Tuplesortstate
*** 340,347 ****
  	bool		markpos_eof;	/* saved "eof_reached" */
  
  	/*
! 	 * These variables are specific to the MinimalTuple case; they are set by
! 	 * tuplesort_begin_heap and used only by the MinimalTuple routines.
  	 */
  	TupleDesc	tupDesc;
  	SortSupport sortKeys;		/* array of length nKeys */
--- 333,341 ----
  	bool		markpos_eof;	/* saved "eof_reached" */
  
  	/*
! 	 * The sortKeys variable is used by every case other than the hash index
! 	 * case; it is set by tuplesort_begin_xxx.  tupDesc is only used by the
! 	 * MinimalTuple and CLUSTER routines, though.
  	 */
  	TupleDesc	tupDesc;
  	SortSupport sortKeys;		/* array of length nKeys */
*************** struct Tuplesortstate
*** 354,361 ****
  
  	/*
  	 * These variables are specific to the CLUSTER case; they are set by
! 	 * tuplesort_begin_cluster.  Note CLUSTER also uses tupDesc and
! 	 * indexScanKey.
  	 */
  	IndexInfo  *indexInfo;		/* info about index being used for reference */
  	EState	   *estate;			/* for evaluating index expressions */
--- 348,354 ----
  
  	/*
  	 * These variables are specific to the CLUSTER case; they are set by
! 	 * tuplesort_begin_cluster.
  	 */
  	IndexInfo  *indexInfo;		/* info about index being used for reference */
  	EState	   *estate;			/* for evaluating index expressions */
*************** struct Tuplesortstate
*** 368,374 ****
  	Relation	indexRel;		/* index being built */
  
  	/* These are specific to the index_btree subcase: */
- 	ScanKey		indexScanKey;
  	bool		enforceUnique;	/* complain if we find duplicate tuples */
  
  	/* These are specific to the index_hash subcase: */
--- 361,366 ----
*************** struct Tuplesortstate
*** 395,401 ****
  #define COPYTUP(state,stup,tup) ((*(state)->copytup) (state, stup, tup))
  #define WRITETUP(state,tape,stup)	((*(state)->writetup) (state, tape, stup))
  #define READTUP(state,stup,tape,len) ((*(state)->readtup) (state, stup, tape, len))
- #define REVERSEDIRECTION(state) ((*(state)->reversedirection) (state))
  #define LACKMEM(state)		((state)->availMem < 0)
  #define USEMEM(state,amt)	((state)->availMem -= (amt))
  #define FREEMEM(state,amt)	((state)->availMem += (amt))
--- 387,392 ----
*************** static void sort_bounded_heap(Tuplesorts
*** 464,469 ****
--- 455,461 ----
  static void tuplesort_heap_insert(Tuplesortstate *state, SortTuple *tuple,
  					  int tupleindex, bool checkIndex);
  static void tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex);
+ static void reversedirection(Tuplesortstate *state);
  static unsigned int getlen(Tuplesortstate *state, int tapenum, bool eofOK);
  static void markrunend(Tuplesortstate *state, int tapenum);
  static int comparetup_heap(const SortTuple *a, const SortTuple *b,
*************** static void writetup_heap(Tuplesortstate
*** 473,479 ****
  			  SortTuple *stup);
  static void readtup_heap(Tuplesortstate *state, SortTuple *stup,
  			 int tapenum, unsigned int len);
- static void reversedirection_heap(Tuplesortstate *state);
  static int comparetup_cluster(const SortTuple *a, const SortTuple *b,
  				   Tuplesortstate *state);
  static void copytup_cluster(Tuplesortstate *state, SortTuple *stup, void *tup);
--- 465,470 ----
*************** static void writetup_index(Tuplesortstat
*** 490,497 ****
  			   SortTuple *stup);
  static void readtup_index(Tuplesortstate *state, SortTuple *stup,
  			  int tapenum, unsigned int len);
- static void reversedirection_index_btree(Tuplesortstate *state);
- static void reversedirection_index_hash(Tuplesortstate *state);
  static int comparetup_datum(const SortTuple *a, const SortTuple *b,
  				 Tuplesortstate *state);
  static void copytup_datum(Tuplesortstate *state, SortTuple *stup, void *tup);
--- 481,486 ----
*************** static void writetup_datum(Tuplesortstat
*** 499,505 ****
  			   SortTuple *stup);
  static void readtup_datum(Tuplesortstate *state, SortTuple *stup,
  			  int tapenum, unsigned int len);
- static void reversedirection_datum(Tuplesortstate *state);
  static void free_sort_tuple(Tuplesortstate *state, SortTuple *stup);
  
  /*
--- 488,493 ----
*************** tuplesort_begin_heap(TupleDesc tupDesc,
*** 629,635 ****
  	state->copytup = copytup_heap;
  	state->writetup = writetup_heap;
  	state->readtup = readtup_heap;
- 	state->reversedirection = reversedirection_heap;
  
  	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
  
--- 617,622 ----
*************** tuplesort_begin_cluster(TupleDesc tupDes
*** 665,671 ****
--- 652,660 ----
  						int workMem, bool randomAccess)
  {
  	Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
+ 	ScanKey			indexScanKey;
  	MemoryContext oldcontext;
+ 	int				i;
  
  	Assert(indexRel->rd_rel->relam == BTREE_AM_OID);
  
*************** tuplesort_begin_cluster(TupleDesc tupDes
*** 691,703 ****
  	state->copytup = copytup_cluster;
  	state->writetup = writetup_cluster;
  	state->readtup = readtup_cluster;
- 	state->reversedirection = reversedirection_index_btree;
  
  	state->indexInfo = BuildIndexInfo(indexRel);
- 	state->indexScanKey = _bt_mkscankey_nodata(indexRel);
  
  	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
  
  	if (state->indexInfo->ii_Expressions != NULL)
  	{
  		TupleTableSlot *slot;
--- 680,692 ----
  	state->copytup = copytup_cluster;
  	state->writetup = writetup_cluster;
  	state->readtup = readtup_cluster;
  
  	state->indexInfo = BuildIndexInfo(indexRel);
  
  	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
  
+ 	indexScanKey = _bt_mkscankey_nodata(indexRel);
+ 
  	if (state->indexInfo->ii_Expressions != NULL)
  	{
  		TupleTableSlot *slot;
*************** tuplesort_begin_cluster(TupleDesc tupDes
*** 715,720 ****
--- 704,735 ----
  		econtext->ecxt_scantuple = slot;
  	}
  
+ 	/* Prepare SortSupport data for each column */
+ 	state->sortKeys = (SortSupport) palloc0(state->nKeys *
+ 											sizeof(SortSupportData));
+ 
+ 	for (i = 0; i < state->nKeys; i++)
+ 	{
+ 		SortSupport sortKey = state->sortKeys + i;
+ 		ScanKey		scanKey = indexScanKey + i;
+ 		int16		strategy;
+ 
+ 		sortKey->ssup_cxt = CurrentMemoryContext;
+ 		sortKey->ssup_collation = scanKey->sk_collation;
+ 		sortKey->ssup_nulls_first =
+ 			(scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
+ 		sortKey->ssup_attno = scanKey->sk_attno;
+ 
+ 		AssertState(sortKey->ssup_attno != 0);
+ 
+ 		strategy = (scanKey->sk_flags & SK_BT_DESC) != 0 ?
+ 			BTGreaterStrategyNumber : BTLessStrategyNumber;
+ 
+ 		PrepareSortSupportFromIndexRel(indexRel, strategy, sortKey);
+ 	}
+ 
+ 	_bt_freeskey(indexScanKey);
+ 
  	MemoryContextSwitchTo(oldcontext);
  
  	return state;
*************** tuplesort_begin_index_btree(Relation hea
*** 727,733 ****
--- 742,750 ----
  							int workMem, bool randomAccess)
  {
  	Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
+ 	ScanKey			indexScanKey;
  	MemoryContext oldcontext;
+ 	int				i;
  
  	oldcontext = MemoryContextSwitchTo(state->sortcontext);
  
*************** tuplesort_begin_index_btree(Relation hea
*** 751,763 ****
  	state->copytup = copytup_index;
  	state->writetup = writetup_index;
  	state->readtup = readtup_index;
- 	state->reversedirection = reversedirection_index_btree;
  
  	state->heapRel = heapRel;
  	state->indexRel = indexRel;
- 	state->indexScanKey = _bt_mkscankey_nodata(indexRel);
  	state->enforceUnique = enforceUnique;
  
  	MemoryContextSwitchTo(oldcontext);
  
  	return state;
--- 768,806 ----
  	state->copytup = copytup_index;
  	state->writetup = writetup_index;
  	state->readtup = readtup_index;
  
  	state->heapRel = heapRel;
  	state->indexRel = indexRel;
  	state->enforceUnique = enforceUnique;
  
+ 	indexScanKey = _bt_mkscankey_nodata(indexRel);
+ 	state->nKeys = RelationGetNumberOfAttributes(indexRel);
+ 	/* Prepare SortSupport data for each column */
+ 	state->sortKeys = (SortSupport) palloc0(state->nKeys *
+ 											sizeof(SortSupportData));
+ 
+ 	for (i = 0; i < state->nKeys; i++)
+ 	{
+ 		SortSupport sortKey = state->sortKeys + i;
+ 		ScanKey		scanKey = indexScanKey + i;
+ 		int16		strategy;
+ 
+ 		sortKey->ssup_cxt = CurrentMemoryContext;
+ 		sortKey->ssup_collation = scanKey->sk_collation;
+ 		sortKey->ssup_nulls_first =
+ 			(scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
+ 		sortKey->ssup_attno = scanKey->sk_attno;
+ 
+ 		AssertState(sortKey->ssup_attno != 0);
+ 
+ 		strategy = (scanKey->sk_flags & SK_BT_DESC) != 0 ?
+ 			BTGreaterStrategyNumber : BTLessStrategyNumber;
+ 
+ 		PrepareSortSupportFromIndexRel(indexRel, strategy, sortKey);
+ 	}
+ 
+ 	_bt_freeskey(indexScanKey);
+ 
  	MemoryContextSwitchTo(oldcontext);
  
  	return state;
*************** tuplesort_begin_index_hash(Relation heap
*** 788,794 ****
  	state->copytup = copytup_index;
  	state->writetup = writetup_index;
  	state->readtup = readtup_index;
- 	state->reversedirection = reversedirection_index_hash;
  
  	state->heapRel = heapRel;
  	state->indexRel = indexRel;
--- 831,836 ----
*************** tuplesort_begin_datum(Oid datumType, Oid
*** 831,837 ****
  	state->copytup = copytup_datum;
  	state->writetup = writetup_datum;
  	state->readtup = readtup_datum;
- 	state->reversedirection = reversedirection_datum;
  
  	state->datumType = datumType;
  
--- 873,878 ----
*************** make_bounded_heap(Tuplesortstate *state)
*** 2599,2605 ****
  	Assert(tupcount >= state->bound);
  
  	/* Reverse sort direction so largest entry will be at root */
! 	REVERSEDIRECTION(state);
  
  	state->memtupcount = 0;		/* make the heap empty */
  	for (i = 0; i < tupcount; i++)
--- 2640,2646 ----
  	Assert(tupcount >= state->bound);
  
  	/* Reverse sort direction so largest entry will be at root */
! 	reversedirection(state);
  
  	state->memtupcount = 0;		/* make the heap empty */
  	for (i = 0; i < tupcount; i++)
*************** sort_bounded_heap(Tuplesortstate *state)
*** 2663,2669 ****
  	 * Reverse sort direction back to the original state.  This is not
  	 * actually necessary but seems like a good idea for tidiness.
  	 */
! 	REVERSEDIRECTION(state);
  
  	state->status = TSS_SORTEDINMEM;
  	state->boundUsed = true;
--- 2704,2710 ----
  	 * Reverse sort direction back to the original state.  This is not
  	 * actually necessary but seems like a good idea for tidiness.
  	 */
! 	reversedirection(state);
  
  	state->status = TSS_SORTEDINMEM;
  	state->boundUsed = true;
*************** tuplesort_heap_siftup(Tuplesortstate *st
*** 2753,2758 ****
--- 2794,2817 ----
  	memtuples[i] = *tuple;
  }
  
+ /*
+  * Function to reverse the sort direction from its current state
+  *
+  * It is not safe to call this when performing hash tuplesorts
+  */
+ static void
+ reversedirection(Tuplesortstate *state)
+ {
+ 	SortSupport sortKey = state->sortKeys;
+ 	int			nkey;
+ 
+ 	for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
+ 	{
+ 		sortKey->ssup_reverse = !sortKey->ssup_reverse;
+ 		sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
+ 	}
+ }
+ 
  
  /*
   * Tape interface routines
*************** markrunend(Tuplesortstate *state, int ta
*** 2781,2853 ****
  
  
  /*
-  * Inline-able copy of FunctionCall2Coll() to save some cycles in sorting.
-  */
- static inline Datum
- myFunctionCall2Coll(FmgrInfo *flinfo, Oid collation, Datum arg1, Datum arg2)
- {
- 	FunctionCallInfoData fcinfo;
- 	Datum		result;
- 
- 	InitFunctionCallInfoData(fcinfo, flinfo, 2, collation, NULL, NULL);
- 
- 	fcinfo.arg[0] = arg1;
- 	fcinfo.arg[1] = arg2;
- 	fcinfo.argnull[0] = false;
- 	fcinfo.argnull[1] = false;
- 
- 	result = FunctionCallInvoke(&fcinfo);
- 
- 	/* Check for null result, since caller is clearly not expecting one */
- 	if (fcinfo.isnull)
- 		elog(ERROR, "function %u returned NULL", fcinfo.flinfo->fn_oid);
- 
- 	return result;
- }
- 
- /*
-  * 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.  We assume that DESC and
-  * NULLS_FIRST options are encoded in sk_flags the same way btree does it.
-  */
- static inline int32
- inlineApplySortFunction(FmgrInfo *sortFunction, int sk_flags, Oid collation,
- 						Datum datum1, bool isNull1,
- 						Datum datum2, bool isNull2)
- {
- 	int32		compare;
- 
- 	if (isNull1)
- 	{
- 		if (isNull2)
- 			compare = 0;		/* NULL "=" NULL */
- 		else if (sk_flags & SK_BT_NULLS_FIRST)
- 			compare = -1;		/* NULL "<" NOT_NULL */
- 		else
- 			compare = 1;		/* NULL ">" NOT_NULL */
- 	}
- 	else if (isNull2)
- 	{
- 		if (sk_flags & SK_BT_NULLS_FIRST)
- 			compare = 1;		/* NOT_NULL ">" NULL */
- 		else
- 			compare = -1;		/* NOT_NULL "<" NULL */
- 	}
- 	else
- 	{
- 		compare = DatumGetInt32(myFunctionCall2Coll(sortFunction, collation,
- 													datum1, datum2));
- 
- 		if (sk_flags & SK_BT_DESC)
- 			compare = -compare;
- 	}
- 
- 	return compare;
- }
- 
- 
- /*
   * Routines specialized for HeapTuple (actually MinimalTuple) case
   */
  
--- 2840,2845 ----
*************** readtup_heap(Tuplesortstate *state, Sort
*** 2972,2991 ****
  								&stup->isnull1);
  }
  
- static void
- reversedirection_heap(Tuplesortstate *state)
- {
- 	SortSupport sortKey = state->sortKeys;
- 	int			nkey;
- 
- 	for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
- 	{
- 		sortKey->ssup_reverse = !sortKey->ssup_reverse;
- 		sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
- 	}
- }
- 
- 
  /*
   * Routines specialized for the CLUSTER case (HeapTuple data, with
   * comparisons per a btree index definition)
--- 2964,2969 ----
*************** static int
*** 2995,3001 ****
  comparetup_cluster(const SortTuple *a, const SortTuple *b,
  				   Tuplesortstate *state)
  {
! 	ScanKey		scanKey = state->indexScanKey;
  	HeapTuple	ltup;
  	HeapTuple	rtup;
  	TupleDesc	tupDesc;
--- 2973,2979 ----
  comparetup_cluster(const SortTuple *a, const SortTuple *b,
  				   Tuplesortstate *state)
  {
! 	SortSupport	sortKey = state->sortKeys;
  	HeapTuple	ltup;
  	HeapTuple	rtup;
  	TupleDesc	tupDesc;
*************** comparetup_cluster(const SortTuple *a, c
*** 3005,3018 ****
  	/* Compare the leading sort key, if it's simple */
  	if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
  	{
! 		compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
! 										  scanKey->sk_collation,
! 										  a->datum1, a->isnull1,
! 										  b->datum1, b->isnull1);
  		if (compare != 0 || state->nKeys == 1)
  			return compare;
  		/* Compare additional columns the hard way */
! 		scanKey++;
  		nkey = 1;
  	}
  	else
--- 2983,2995 ----
  	/* Compare the leading sort key, if it's simple */
  	if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
  	{
! 		compare = ApplySortComparator(a->datum1, a->isnull1,
! 									  b->datum1, b->isnull1,
! 									  sortKey);
  		if (compare != 0 || state->nKeys == 1)
  			return compare;
  		/* Compare additional columns the hard way */
! 		sortKey++;
  		nkey = 1;
  	}
  	else
*************** comparetup_cluster(const SortTuple *a, c
*** 3030,3036 ****
  		/* If not expression index, just compare the proper heap attrs */
  		tupDesc = state->tupDesc;
  
! 		for (; nkey < state->nKeys; nkey++, scanKey++)
  		{
  			AttrNumber	attno = state->indexInfo->ii_KeyAttrNumbers[nkey];
  			Datum		datum1,
--- 3007,3013 ----
  		/* If not expression index, just compare the proper heap attrs */
  		tupDesc = state->tupDesc;
  
! 		for (; nkey < state->nKeys; nkey++, sortKey++)
  		{
  			AttrNumber	attno = state->indexInfo->ii_KeyAttrNumbers[nkey];
  			Datum		datum1,
*************** comparetup_cluster(const SortTuple *a, c
*** 3041,3051 ****
  			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;
  		}
--- 3018,3026 ----
  			datum1 = heap_getattr(ltup, attno, tupDesc, &isnull1);
  			datum2 = heap_getattr(rtup, attno, tupDesc, &isnull2);
  
! 			compare = ApplySortComparator(datum1, isnull1,
! 										  datum2, isnull2,
! 										  sortKey);
  			if (compare != 0)
  				return compare;
  		}
*************** comparetup_cluster(const SortTuple *a, c
*** 3077,3091 ****
  		FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
  					   r_index_values, r_index_isnull);
  
! 		for (; nkey < state->nKeys; nkey++, scanKey++)
  		{
! 			compare = inlineApplySortFunction(&scanKey->sk_func,
! 											  scanKey->sk_flags,
! 											  scanKey->sk_collation,
! 											  l_index_values[nkey],
! 											  l_index_isnull[nkey],
! 											  r_index_values[nkey],
! 											  r_index_isnull[nkey]);
  			if (compare != 0)
  				return compare;
  		}
--- 3052,3064 ----
  		FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
  					   r_index_values, r_index_isnull);
  
! 		for (; nkey < state->nKeys; nkey++, sortKey++)
  		{
! 			compare = ApplySortComparator(l_index_values[nkey],
! 										  l_index_isnull[nkey],
! 										  r_index_values[nkey],
! 										  r_index_isnull[nkey],
! 										  sortKey);
  			if (compare != 0)
  				return compare;
  		}
*************** comparetup_index_btree(const SortTuple *
*** 3181,3187 ****
  	 * whether any null fields are present.  Also see the special treatment
  	 * for equal keys at the end.
  	 */
! 	ScanKey		scanKey = state->indexScanKey;
  	IndexTuple	tuple1;
  	IndexTuple	tuple2;
  	int			keysz;
--- 3154,3160 ----
  	 * whether any null fields are present.  Also see the special treatment
  	 * for equal keys at the end.
  	 */
! 	SortSupport	sortKey = state->sortKeys;
  	IndexTuple	tuple1;
  	IndexTuple	tuple2;
  	int			keysz;
*************** comparetup_index_btree(const SortTuple *
*** 3191,3200 ****
  	int32		compare;
  
  	/* 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;
  
--- 3164,3172 ----
  	int32		compare;
  
  	/* Compare the leading sort key */
! 	compare = ApplySortComparator(a->datum1, a->isnull1,
! 								  b->datum1, b->isnull1,
! 								  sortKey);
  	if (compare != 0)
  		return compare;
  
*************** comparetup_index_btree(const SortTuple *
*** 3207,3214 ****
  	tuple2 = (IndexTuple) b->tuple;
  	keysz = state->nKeys;
  	tupDes = RelationGetDescr(state->indexRel);
! 	scanKey++;
! 	for (nkey = 2; nkey <= keysz; nkey++, scanKey++)
  	{
  		Datum		datum1,
  					datum2;
--- 3179,3186 ----
  	tuple2 = (IndexTuple) b->tuple;
  	keysz = state->nKeys;
  	tupDes = RelationGetDescr(state->indexRel);
! 	sortKey++;
! 	for (nkey = 2; nkey <= keysz; nkey++, sortKey++)
  	{
  		Datum		datum1,
  					datum2;
*************** comparetup_index_btree(const SortTuple *
*** 3218,3227 ****
  		datum1 = index_getattr(tuple1, nkey, tupDes, &isnull1);
  		datum2 = index_getattr(tuple2, nkey, tupDes, &isnull2);
  
! 		compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
! 										  scanKey->sk_collation,
! 										  datum1, isnull1,
! 										  datum2, isnull2);
  		if (compare != 0)
  			return compare;		/* done when we find unequal attributes */
  
--- 3190,3198 ----
  		datum1 = index_getattr(tuple1, nkey, tupDes, &isnull1);
  		datum2 = index_getattr(tuple2, nkey, tupDes, &isnull2);
  
! 		compare = ApplySortComparator(datum1, isnull1,
! 									  datum2, isnull2,
! 									  sortKey);
  		if (compare != 0)
  			return compare;		/* done when we find unequal attributes */
  
*************** readtup_index(Tuplesortstate *state, Sor
*** 3395,3420 ****
  								 &stup->isnull1);
  }
  
- static void
- reversedirection_index_btree(Tuplesortstate *state)
- {
- 	ScanKey		scanKey = state->indexScanKey;
- 	int			nkey;
- 
- 	for (nkey = 0; nkey < state->nKeys; nkey++, scanKey++)
- 	{
- 		scanKey->sk_flags ^= (SK_BT_DESC | SK_BT_NULLS_FIRST);
- 	}
- }
- 
- static void
- reversedirection_index_hash(Tuplesortstate *state)
- {
- 	/* We don't support reversing direction in a hash index sort */
- 	elog(ERROR, "reversedirection_index_hash is not implemented");
- }
- 
- 
  /*
   * Routines specialized for DatumTuple case
   */
--- 3366,3371 ----
*************** readtup_datum(Tuplesortstate *state, Sor
*** 3513,3525 ****
  							 &tuplen, sizeof(tuplen));
  }
  
- static void
- reversedirection_datum(Tuplesortstate *state)
- {
- 	state->onlyKey->ssup_reverse = !state->onlyKey->ssup_reverse;
- 	state->onlyKey->ssup_nulls_first = !state->onlyKey->ssup_nulls_first;
- }
- 
  /*
   * Convenience routine to free a tuple previously loaded into sort memory
   */
--- 3464,3469 ----
diff --git a/src/include/utils/sortsupport.h b/src/include/utils/sortsupport.h
new file mode 100644
index 4417143..faae703
*** a/src/include/utils/sortsupport.h
--- b/src/include/utils/sortsupport.h
***************
*** 48,53 ****
--- 48,54 ----
  #define SORTSUPPORT_H
  
  #include "access/attnum.h"
+ #include "utils/relcache.h"
  
  typedef struct SortSupportData *SortSupport;
  
*************** ApplySortComparator(Datum datum1, bool i
*** 152,156 ****
--- 153,159 ----
  /* Other functions in utils/sort/sortsupport.c */
  extern void PrepareSortSupportComparisonShim(Oid cmpFunc, SortSupport ssup);
  extern void PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup);
+ extern void PrepareSortSupportFromIndexRel(Relation indexRel, int16 strategy,
+ 										   SortSupport ssup);
  
  #endif   /* SORTSUPPORT_H */