7 files changed, 308 insertions, 119 deletions

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;
 }
 
 /*