From 62a8acbc745f3b4a90c9a14b6b61989a9d83bece Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
Date: Sat, 3 Jul 2021 19:02:10 +1200
Subject: [PATCH] WIP: Accelerate tuple sorting for common types.

Several data types have their own fast comparator functions that can be
replaced with common binary or signed comparator functions.  These
functions can then be recognized by tuplesort.c and used to dispatch to
corresponding specialized sort functions, to accelerate sorting.

XXX WIP, experiment grade only, many open questions...
---
 src/backend/access/nbtree/nbtcompare.c |  22 ++--
 src/backend/utils/adt/date.c           |  15 +--
 src/backend/utils/adt/timestamp.c      |  11 ++
 src/backend/utils/adt/varlena.c        |  26 +---
 src/backend/utils/sort/tuplesort.c     | 161 ++++++++++++++++++++++++-
 src/include/utils/sortsupport.h        | 115 ++++++++++++++++++
 6 files changed, 295 insertions(+), 55 deletions(-)

diff --git a/src/backend/access/nbtree/nbtcompare.c b/src/backend/access/nbtree/nbtcompare.c
index 7ac73cb8c2..204cf778fb 100644
--- a/src/backend/access/nbtree/nbtcompare.c
+++ b/src/backend/access/nbtree/nbtcompare.c
@@ -119,26 +119,12 @@ btint4cmp(PG_FUNCTION_ARGS)
 		PG_RETURN_INT32(A_LESS_THAN_B);
 }
 
-static int
-btint4fastcmp(Datum x, Datum y, SortSupport ssup)
-{
-	int32		a = DatumGetInt32(x);
-	int32		b = DatumGetInt32(y);
-
-	if (a > b)
-		return A_GREATER_THAN_B;
-	else if (a == b)
-		return 0;
-	else
-		return A_LESS_THAN_B;
-}
-
 Datum
 btint4sortsupport(PG_FUNCTION_ARGS)
 {
 	SortSupport ssup = (SortSupport) PG_GETARG_POINTER(0);
 
-	ssup->comparator = btint4fastcmp;
+	ssup->comparator = ssup_datum_int32_cmp;
 	PG_RETURN_VOID();
 }
 
@@ -156,6 +142,7 @@ btint8cmp(PG_FUNCTION_ARGS)
 		PG_RETURN_INT32(A_LESS_THAN_B);
 }
 
+#ifndef USE_FLOAT8_BYVAL
 static int
 btint8fastcmp(Datum x, Datum y, SortSupport ssup)
 {
@@ -169,13 +156,18 @@ btint8fastcmp(Datum x, Datum y, SortSupport ssup)
 	else
 		return A_LESS_THAN_B;
 }
+#endif
 
 Datum
 btint8sortsupport(PG_FUNCTION_ARGS)
 {
 	SortSupport ssup = (SortSupport) PG_GETARG_POINTER(0);
 
+#ifdef USE_FLOAT8_BYVAL
+	ssup->comparator = ssup_datum_signed_cmp;
+#else
 	ssup->comparator = btint8fastcmp;
+#endif
 	PG_RETURN_VOID();
 }
 
diff --git a/src/backend/utils/adt/date.c b/src/backend/utils/adt/date.c
index 0bea16cb67..350c662d50 100644
--- a/src/backend/utils/adt/date.c
+++ b/src/backend/utils/adt/date.c
@@ -438,25 +438,12 @@ 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;
+	ssup->comparator = ssup_datum_int32_cmp;
 	PG_RETURN_VOID();
 }
 
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index 79761f809c..c678517db6 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/src/backend/utils/adt/timestamp.c
@@ -37,6 +37,7 @@
 #include "utils/datetime.h"
 #include "utils/float.h"
 #include "utils/numeric.h"
+#include "utils/sortsupport.h"
 
 /*
  * gcc's -ffast-math switch breaks routines that expect exact results from
@@ -2155,6 +2156,7 @@ timestamp_cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(timestamp_cmp_internal(dt1, dt2));
 }
 
+#ifndef USE_FLOAT8_BYVAL
 /* note: this is used for timestamptz also */
 static int
 timestamp_fastcmp(Datum x, Datum y, SortSupport ssup)
@@ -2164,13 +2166,22 @@ timestamp_fastcmp(Datum x, Datum y, SortSupport ssup)
 
 	return timestamp_cmp_internal(a, b);
 }
+#endif
 
 Datum
 timestamp_sortsupport(PG_FUNCTION_ARGS)
 {
 	SortSupport ssup = (SortSupport) PG_GETARG_POINTER(0);
 
+#ifdef USE_FLOAT8_BYVAL
+	/*
+	 * If this build has pass-by-value timestamps, then we can use a standard
+	 * comparator function.
+	 */
+	ssup->comparator = ssup_datum_signed_cmp;
+#else
 	ssup->comparator = timestamp_fastcmp;
+#endif
 	PG_RETURN_VOID();
 }
 
diff --git a/src/backend/utils/adt/varlena.c b/src/backend/utils/adt/varlena.c
index d2a11b1b5d..4709d8129e 100644
--- a/src/backend/utils/adt/varlena.c
+++ b/src/backend/utils/adt/varlena.c
@@ -127,7 +127,6 @@ static int	namefastcmp_c(Datum x, Datum y, SortSupport ssup);
 static int	varlenafastcmp_locale(Datum x, Datum y, SortSupport ssup);
 static int	namefastcmp_locale(Datum x, Datum y, SortSupport ssup);
 static int	varstrfastcmp_locale(char *a1p, int len1, char *a2p, int len2, SortSupport ssup);
-static int	varstrcmp_abbrev(Datum x, Datum y, SortSupport ssup);
 static Datum varstr_abbrev_convert(Datum original, SortSupport ssup);
 static bool varstr_abbrev_abort(int memtupcount, SortSupport ssup);
 static int32 text_length(Datum str);
@@ -2175,7 +2174,7 @@ varstr_sortsupport(SortSupport ssup, Oid typid, Oid collid)
 			initHyperLogLog(&sss->abbr_card, 10);
 			initHyperLogLog(&sss->full_card, 10);
 			ssup->abbrev_full_comparator = ssup->comparator;
-			ssup->comparator = varstrcmp_abbrev;
+			ssup->comparator = ssup_datum_binary_cmp;
 			ssup->abbrev_converter = varstr_abbrev_convert;
 			ssup->abbrev_abort = varstr_abbrev_abort;
 		}
@@ -2461,27 +2460,6 @@ varstrfastcmp_locale(char *a1p, int len1, char *a2p, int len2, SortSupport ssup)
 	return result;
 }
 
-/*
- * Abbreviated key comparison func
- */
-static int
-varstrcmp_abbrev(Datum x, Datum y, SortSupport ssup)
-{
-	/*
-	 * When 0 is returned, the core system will call varstrfastcmp_c()
-	 * (bpcharfastcmp_c() in BpChar case) or varlenafastcmp_locale().  Even a
-	 * strcmp() on two non-truncated strxfrm() blobs cannot indicate *equality*
-	 * authoritatively, for the same reason that there is a strcoll()
-	 * tie-breaker call to strcmp() in varstr_cmp().
-	 */
-	if (x > y)
-		return 1;
-	else if (x == y)
-		return 0;
-	else
-		return -1;
-}
-
 /*
  * Conversion routine for sortsupport.  Converts original to abbreviated key
  * representation.  Our encoding strategy is simple -- pack the first 8 bytes
@@ -2710,7 +2688,7 @@ done:
 	/*
 	 * Byteswap on little-endian machines.
 	 *
-	 * This is needed so that varstrcmp_abbrev() (an unsigned integer 3-way
+	 * This is needed so that ssup_datum_binary_cmp() (an unsigned integer 3-way
 	 * comparator) works correctly on all platforms.  If we didn't do this,
 	 * the comparator would have to call memcmp() with a pair of pointers to
 	 * the first byte of each abbreviated key, which is slower.
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index 22972071ff..455bb652b8 100644
--- a/src/backend/utils/sort/tuplesort.c
+++ b/src/backend/utils/sort/tuplesort.c
@@ -669,14 +669,101 @@ static void free_sort_tuple(Tuplesortstate *state, SortTuple *stup);
 static void tuplesort_free(Tuplesortstate *state);
 static void tuplesort_updatemax(Tuplesortstate *state);
 
+/*
+ * Specialized comparators that we can inline into specialized sorts.  The goal
+ * is to try to sort two tuples without having to follow the pointers to the
+ * comparator or the tuple.
+ *
+ * XXX: For now, these fall back to comparator functions that will compare the
+ * leading datum a second time.
+ *
+ * XXX: For now, there is no specialization for cases where datum1 is
+ * authoritative and we don't even need to fall back to a callback at all (that
+ * would be true for types like int4/int8/timestamp/date, but not true for
+ * abbreviations of text or multi-key sorts.  There could be!  Is it worth it?
+ */
+
+/* Used if first key's comparator is ssup_datum_binary_compare */
+static pg_attribute_always_inline int
+qsort_tuple_binary_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
+{
+	int			compare;
+
+	compare = ApplyBinarySortComparator(a->datum1, a->isnull1,
+										b->datum1, b->isnull1,
+										&state->sortKeys[0]);
+	if (compare != 0)
+		return compare;
+
+	return state->comparetup(a, b, state);
+}
+
+/* Used if first key's comparator is ssup_datum_signed_compare */
+static pg_attribute_always_inline int
+qsort_tuple_signed_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
+{
+	int			compare;
+
+	compare = ApplySignedSortComparator(a->datum1, a->isnull1,
+										b->datum1, b->isnull1,
+										&state->sortKeys[0]);
+	if (compare != 0)
+		return compare;
+
+	return state->comparetup(a, b, state);
+}
+
+/* Used if first key's comparator is ssup_datum_int32_compare */
+static pg_attribute_always_inline int
+qsort_tuple_int32_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
+{
+	int			compare;
+
+	compare = ApplyInt32SortComparator(a->datum1, a->isnull1,
+										b->datum1, b->isnull1,
+										&state->sortKeys[0]);
+	if (compare != 0)
+		return compare;
+
+	return state->comparetup(a, b, state);
+}
+
 /*
  * Special versions of qsort just for SortTuple objects.  qsort_tuple() sorts
  * any variant of SortTuples, using the appropriate comparetup function.
  * qsort_ssup() is specialized for the case where the comparetup function
  * reduces to ApplySortComparator(), that is single-key MinimalTuple sorts
- * and Datum sorts.
+ * and Datum sorts.  qsort_tuple_{binary,signed,int32} are specialized for
+ * common comparison functions on pass-by-value leading datums.
  */
 
+#define ST_SORT qsort_tuple_binary
+#define ST_ELEMENT_TYPE SortTuple
+#define ST_COMPARE(a, b, state) qsort_tuple_binary_compare(a, b, state)
+#define ST_COMPARE_ARG_TYPE Tuplesortstate
+#define ST_CHECK_FOR_INTERRUPTS
+#define ST_SCOPE static
+#define ST_DEFINE
+#include "lib/sort_template.h"
+
+#define ST_SORT qsort_tuple_signed
+#define ST_ELEMENT_TYPE SortTuple
+#define ST_COMPARE(a, b, state) qsort_tuple_signed_compare(a, b, state)
+#define ST_COMPARE_ARG_TYPE Tuplesortstate
+#define ST_CHECK_FOR_INTERRUPTS
+#define ST_SCOPE static
+#define ST_DEFINE
+#include "lib/sort_template.h"
+
+#define ST_SORT qsort_tuple_int32
+#define ST_ELEMENT_TYPE SortTuple
+#define ST_COMPARE(a, b, state) qsort_tuple_int32_compare(a, b, state)
+#define ST_COMPARE_ARG_TYPE Tuplesortstate
+#define ST_CHECK_FOR_INTERRUPTS
+#define ST_SCOPE static
+#define ST_DEFINE
+#include "lib/sort_template.h"
+
 #define ST_SORT qsort_tuple
 #define ST_ELEMENT_TYPE SortTuple
 #define ST_COMPARE_RUNTIME_POINTER
@@ -698,6 +785,7 @@ static void tuplesort_updatemax(Tuplesortstate *state);
 #define ST_DEFINE
 #include "lib/sort_template.h"
 
+
 /*
  *		tuplesort_begin_xxx
  *
@@ -3558,15 +3646,40 @@ tuplesort_sort_memtuples(Tuplesortstate *state)
 
 	if (state->memtupcount > 1)
 	{
+		/* Do we have a specialization for the leading column's comparator? */
+		if (state->sortKeys &&
+			state->sortKeys[0].comparator == ssup_datum_binary_cmp)
+		{
+			elog(DEBUG1, "qsort_tuple_binary");
+			qsort_tuple_binary(state->memtuples, state->memtupcount, state);
+		}
+		else if (state->sortKeys &&
+				 state->sortKeys[0].comparator == ssup_datum_signed_cmp)
+		{
+			elog(DEBUG1, "qsort_tuple_signed");
+			qsort_tuple_signed(state->memtuples, state->memtupcount, state);
+		}
+		else if (state->sortKeys &&
+				 state->sortKeys[0].comparator == ssup_datum_int32_cmp)
+		{
+			elog(DEBUG1, "qsort_tuple_int32");
+			qsort_tuple_int32(state->memtuples, state->memtupcount, state);
+		}
 		/* Can we use the single-key sort function? */
-		if (state->onlyKey != NULL)
+		else if (state->onlyKey != NULL)
+		{
+			elog(DEBUG1, "qsort_ssup");
 			qsort_ssup(state->memtuples, state->memtupcount,
 					   state->onlyKey);
+		}
 		else
+		{
+			elog(DEBUG1, "qsort_tuple");
 			qsort_tuple(state->memtuples,
 						state->memtupcount,
 						state->comparetup,
 						state);
+		}
 	}
 }
 
@@ -4776,3 +4889,47 @@ free_sort_tuple(Tuplesortstate *state, SortTuple *stup)
 	FREEMEM(state, GetMemoryChunkSpace(stup->tuple));
 	pfree(stup->tuple);
 }
+
+int
+ssup_datum_binary_cmp(Datum x, Datum y, SortSupport ssup)
+{
+	if (x < y)
+		return -1;
+	else if (x > y)
+		return 1;
+	else
+		return 0;
+}
+
+int
+ssup_datum_signed_cmp(Datum x, Datum y, SortSupport ssup)
+{
+#if SIZEOF_DATUM == 8
+	int64		xx = (int64) x;
+	int64		yy = (int64) y;
+#else
+	int32		xx = (int32) x;
+	int32		yy = (int32) y;
+#endif
+
+	if (xx < yy)
+		return -1;
+	else if (xx > yy)
+		return 1;
+	else
+		return 0;
+}
+
+int
+ssup_datum_int32_cmp(Datum x, Datum y, SortSupport ssup)
+{
+	int32		xx = (int32) x;
+	int32		yy = (int32) y;
+
+	if (xx < yy)
+		return -1;
+	else if (xx > yy)
+		return 1;
+	else
+		return 0;
+}
diff --git a/src/include/utils/sortsupport.h b/src/include/utils/sortsupport.h
index 2f12a8b8eb..8b6a0c5e10 100644
--- a/src/include/utils/sortsupport.h
+++ b/src/include/utils/sortsupport.h
@@ -229,6 +229,112 @@ ApplySortComparator(Datum datum1, bool isNull1,
 	return compare;
 }
 
+static inline int
+ApplyBinarySortComparator(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 = datum1 < datum2 ? -1 : datum1 > datum2 ? 1 : 0;
+		if (ssup->ssup_reverse)
+			INVERT_COMPARE_RESULT(compare);
+	}
+
+	return compare;
+}
+
+static inline int
+ApplySignedSortComparator(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
+	{
+#if SIZEOF_DATUM == 8
+		compare = (int64) datum1 < (int64) datum2 ? -1 :
+			(int64) datum1 > (int64) datum2 ? 1 : 0;
+#else
+		compare = (int32) datum1 < (int32) datum2 ? -1 :
+			(int32) datum1 > (int32) datum2 ? 1 : 0;
+#endif
+		if (ssup->ssup_reverse)
+			INVERT_COMPARE_RESULT(compare);
+	}
+
+	return compare;
+}
+
+static inline int
+ApplyInt32SortComparator(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 = (int32) datum1 < (int32) datum2 ? -1 :
+			(int32) datum1 > (int32) datum2 ? 1 : 0;
+		if (ssup->ssup_reverse)
+			INVERT_COMPARE_RESULT(compare);
+	}
+
+	return compare;
+}
+
 /*
  * Apply a sort comparator function and return a 3-way comparison using full,
  * authoritative comparator.  This takes care of handling reverse-sort and
@@ -267,6 +373,15 @@ ApplySortAbbrevFullComparator(Datum datum1, bool isNull1,
 	return compare;
 }
 
+/*
+ * Datum comparison functions that we have specialized sort routines for.
+ * Datatypes that install these as their comparator or abbrevated comparator
+ * are eligible for faster sorting.
+ */
+extern int ssup_datum_binary_cmp(Datum x, Datum y, SortSupport ssup);
+extern int ssup_datum_signed_cmp(Datum x, Datum y, SortSupport ssup);
+extern int ssup_datum_int32_cmp(Datum x, Datum y, SortSupport ssup);
+
 /* Other functions in utils/sort/sortsupport.c */
 extern void PrepareSortSupportComparisonShim(Oid cmpFunc, SortSupport ssup);
 extern void PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup);
-- 
2.30.2

