Use the sortsupport infrastructure in more cases.

This removes some fmgr overhead from cases such as btree index builds.

Peter Geoghegan, reviewed by Andreas Karlsson and me.

src/backend/access/nbtree/nbtsort.c

@@ -686,6 +686,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 	int			i,
 				keysz = RelationGetNumberOfAttributes(wstate->index);
 	ScanKey		indexScanKey = NULL;
+	SortSupport sortKeys;
 
 	if (merge)
 	{
@@ -701,6 +702,31 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 										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 ? */
@@ -713,43 +739,20 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 			{
 				for (i = 1; i <= keysz; i++)
 				{
-					ScanKey		entry;
+					SortSupport	entry;
 					Datum		attrDatum1,
 								attrDatum2;
 					bool		isNull1,
 								isNull2;
 					int32		compare;
 
-					entry = indexScanKey + i - 1;
+					entry = sortKeys + 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;
-					}
+
+					compare = ApplySortComparator(attrDatum1, isNull1,
+												  attrDatum2, isNull2,
+												  entry);
 					if (compare > 0)
 					{
 						load1 = false;
@@ -783,7 +786,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 												true, &should_free2);
 			}
 		}
-		_bt_freeskey(indexScanKey);
+		pfree(sortKeys);
 	}
 	else
 	{

src/backend/utils/sort/sortsupport.c

@@ -21,6 +21,7 @@
 #include "access/nbtree.h"
 #include "fmgr.h"
 #include "utils/lsyscache.h"
+#include "utils/rel.h"
 #include "utils/sortsupport.h"
 
 
@@ -86,28 +87,14 @@ PrepareSortSupportComparisonShim(Oid cmpFunc, SortSupport 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.
+ * Look up and call sortsupport function to setup SortSupport comparator;
+ * or if no such function exists or it declines to set up the appropriate
+ * state, prepare a suitable shim.
  */
-void
-PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
+static void
+FinishSortSupportFunction(Oid opfamily, Oid opcintype, 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,
@@ -136,3 +123,57 @@ PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
 		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);
+}

src/include/utils/sortsupport.h

@@ -48,6 +48,7 @@
 #define SORTSUPPORT_H
 
 #include "access/attnum.h"
+#include "utils/relcache.h"
 
 typedef struct SortSupportData *SortSupport;
 
@@ -152,5 +153,7 @@ ApplySortComparator(Datum datum1, bool isNull1,
 /* 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 */