0008-Specialize-some-sort-search-routines-in-nbtree-code.patch

From 2a5676c8b79449bb563d3aa7a489310926e14f25 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
Date: Sun, 14 Mar 2021 11:02:40 +1300
Subject: [PATCH 8/9] Specialize some sort/search routines in nbtree code.

---
 src/backend/access/nbtree/nbtpage.c  | 29 ++++--------
 src/backend/access/nbtree/nbtutils.c | 68 ++++++++++++++++------------
 2 files changed, 48 insertions(+), 49 deletions(-)

diff --git a/src/backend/access/nbtree/nbtpage.c b/src/backend/access/nbtree/nbtpage.c
index fc744cf9fd..78e04a3e9a 100644
--- a/src/backend/access/nbtree/nbtpage.c
+++ b/src/backend/access/nbtree/nbtpage.c
@@ -1468,24 +1468,16 @@ _bt_delitems_update(BTVacuumPosting *updatable, int nupdatable,
 }
 
 /*
- * Comparator used by _bt_delitems_delete_check() to restore deltids array
- * back to its original leaf-page-wise sort order
+ * Sort used by _bt_delitems_delete_check() to restore deltids array back to
+ * its original leaf-page-wise sort order.  We can safely use a branchless
+ * int-based comparator expression because id is an int16.
  */
-static int
-_bt_delitems_cmp(const void *a, const void *b)
-{
-	TM_IndexDelete *indexdelete1 = (TM_IndexDelete *) a;
-	TM_IndexDelete *indexdelete2 = (TM_IndexDelete *) b;
-
-	if (indexdelete1->id > indexdelete2->id)
-		return 1;
-	if (indexdelete1->id < indexdelete2->id)
-		return -1;
-
-	Assert(false);
-
-	return 0;
-}
+#define ST_SORT qsort_delitems
+#define ST_ELEMENT_TYPE TM_IndexDelete
+#define ST_COMPARE(a, b) ((int) (a)->id - (int) (b)->id)
+#define ST_SCOPE static
+#define ST_DEFINE
+#include "lib/sort_template.h"
 
 /*
  * Try to delete item(s) from a btree leaf page during single-page cleanup.
@@ -1555,8 +1547,7 @@ _bt_delitems_delete_check(Relation rel, Buffer buf, Relation heapRel,
 	 * no entries at all (with bottom-up deletion caller), in which case there
 	 * is nothing left to do.
 	 */
-	qsort(delstate->deltids, delstate->ndeltids, sizeof(TM_IndexDelete),
-		  _bt_delitems_cmp);
+	qsort_delitems(delstate->deltids, delstate->ndeltids);
 	if (delstate->ndeltids == 0)
 	{
 		Assert(delstate->bottomup);
diff --git a/src/backend/access/nbtree/nbtutils.c b/src/backend/access/nbtree/nbtutils.c
index d524310723..3012d9f702 100644
--- a/src/backend/access/nbtree/nbtutils.c
+++ b/src/backend/access/nbtree/nbtutils.c
@@ -22,7 +22,6 @@
 #include "access/relscan.h"
 #include "catalog/catalog.h"
 #include "commands/progress.h"
-#include "lib/qunique.h"
 #include "miscadmin.h"
 #include "utils/array.h"
 #include "utils/datum.h"
@@ -35,7 +34,6 @@ typedef struct BTSortArrayContext
 {
 	FmgrInfo	flinfo;
 	Oid			collation;
-	bool		reverse;
 } BTSortArrayContext;
 
 static Datum _bt_find_extreme_element(IndexScanDesc scan, ScanKey skey,
@@ -44,7 +42,6 @@ static Datum _bt_find_extreme_element(IndexScanDesc scan, ScanKey skey,
 static int	_bt_sort_array_elements(IndexScanDesc scan, ScanKey skey,
 									bool reverse,
 									Datum *elems, int nelems);
-static int	_bt_compare_array_elements(const void *a, const void *b, void *arg);
 static bool _bt_compare_scankey_args(IndexScanDesc scan, ScanKey op,
 									 ScanKey leftarg, ScanKey rightarg,
 									 bool *result);
@@ -429,6 +426,33 @@ _bt_find_extreme_element(IndexScanDesc scan, ScanKey skey,
 	return result;
 }
 
+/* Define a specialized sort function for _bt_sort_array_elements. */
+#define ST_SORT qsort_bt_array_elements
+#define ST_UNIQUE unique_bt_array_elements
+#define ST_ELEMENT_TYPE Datum
+#define ST_COMPARE(a, b, cxt) \
+	DatumGetInt32(FunctionCall2Coll(&cxt->flinfo, cxt->collation, *a, *b))
+#define ST_COMPARE_ARG_TYPE BTSortArrayContext
+#define ST_SCOPE static
+#define ST_DEFINE
+#include "lib/sort_template.h"
+
+/*
+ * Define a reverse sort function for _bt_sort_array_elements.  Note use of
+ * int64 comparator expression, so that we can safely invert even INT_MIN with
+ * simple substraction from zero.
+ */
+#define ST_SORT qsort_bt_array_elements_reverse
+#define ST_UNIQUE unique_bt_array_elements_reverse
+#define ST_ELEMENT_TYPE Datum
+#define ST_COMPARE(a, b, cxt) \
+	(0 - (DatumGetInt32(FunctionCall2Coll(&cxt->flinfo, cxt->collation, *a, *b))))
+#define ST_COMPARE_TYPE int64
+#define ST_COMPARE_ARG_TYPE BTSortArrayContext
+#define ST_SCOPE static
+#define ST_DEFINE
+#include "lib/sort_template.h"
+
 /*
  * _bt_sort_array_elements() -- sort and de-dup array elements
  *
@@ -477,35 +501,19 @@ _bt_sort_array_elements(IndexScanDesc scan, ScanKey skey,
 			 BTORDER_PROC, elemtype, elemtype,
 			 rel->rd_opfamily[skey->sk_attno - 1]);
 
-	/* Sort the array elements */
+	/* Sort the array elements and remove duplicates */
 	fmgr_info(cmp_proc, &cxt.flinfo);
 	cxt.collation = skey->sk_collation;
-	cxt.reverse = reverse;
-	qsort_arg((void *) elems, nelems, sizeof(Datum),
-			  _bt_compare_array_elements, (void *) &cxt);
-
-	/* Now scan the sorted elements and remove duplicates */
-	return qunique_arg(elems, nelems, sizeof(Datum),
-					   _bt_compare_array_elements, &cxt);
-}
-
-/*
- * qsort_arg comparator for sorting array elements
- */
-static int
-_bt_compare_array_elements(const void *a, const void *b, void *arg)
-{
-	Datum		da = *((const Datum *) a);
-	Datum		db = *((const Datum *) b);
-	BTSortArrayContext *cxt = (BTSortArrayContext *) arg;
-	int32		compare;
-
-	compare = DatumGetInt32(FunctionCall2Coll(&cxt->flinfo,
-											  cxt->collation,
-											  da, db));
-	if (cxt->reverse)
-		INVERT_COMPARE_RESULT(compare);
-	return compare;
+	if (reverse)
+	{
+		qsort_bt_array_elements_reverse(elems, nelems, &cxt);
+		return unique_bt_array_elements_reverse(elems, nelems, &cxt);
+	}
+	else
+	{
+		qsort_bt_array_elements(elems, nelems, &cxt);
+		return unique_bt_array_elements(elems, nelems, &cxt);
+	}
 }
 
 /*
-- 
2.30.1