From 73a38766c8bc32fc49d3acea32b66b47ff9ab030 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Wed, 11 Jan 2023 02:57:21 +0100
Subject: [PATCH v12 3/6] Use specialized attribute iterators in the
 specialized source files

This is committed separately to make clear what substantial changes were
made to the pre-existing code.

Even though not all nbt*_spec functions have been updated; these functions
can now directly call (and inline, and optimize for) the specialized functions
they call, instead of having to determine the right specialization based on
the (potentially locally unavailable) index relation, making the specialization
of those functions still worth specializing/duplicating.
---
 src/backend/access/nbtree/nbtsearch_spec.c    | 18 +++---
 src/backend/access/nbtree/nbtsort_spec.c      | 24 +++----
 src/backend/access/nbtree/nbtutils_spec.c     | 62 ++++++++++++-------
 .../utils/sort/tuplesortvariants_spec.c       | 53 +++++++++-------
 4 files changed, 92 insertions(+), 65 deletions(-)

diff --git a/src/backend/access/nbtree/nbtsearch_spec.c b/src/backend/access/nbtree/nbtsearch_spec.c
index 5f1ead2400..c03aed3fd7 100644
--- a/src/backend/access/nbtree/nbtsearch_spec.c
+++ b/src/backend/access/nbtree/nbtsearch_spec.c
@@ -668,6 +668,7 @@ _bt_compare(Relation rel,
 	int			ncmpkey;
 	int			ntupatts;
 	int32		result;
+	nbts_attiterdeclare(itup);
 
 	Assert(_bt_check_natts(rel, key->heapkeyspace, page, offnum));
 	Assert(key->keysz <= IndexRelationGetNumberOfKeyAttributes(rel));
@@ -700,23 +701,26 @@ _bt_compare(Relation rel,
 	Assert(!BTreeTupleIsPosting(itup) || key->allequalimage);
 
 	scankey = key->scankeys + ((*comparecol) - 1);
-	for (int i = *comparecol; i <= ncmpkey; i++)
+	nbts_attiterinit(itup, *comparecol, itupdesc);
+
+	nbts_foreachattr(*comparecol, ncmpkey)
 	{
 		Datum		datum;
-		bool		isNull;
 
-		datum = index_getattr(itup, scankey->sk_attno, itupdesc, &isNull);
+		datum = nbts_attiter_nextattdatum(itup, itupdesc);
 
-		if (scankey->sk_flags & SK_ISNULL)	/* key is NULL */
+		/* key is NULL */
+		if (scankey->sk_flags & SK_ISNULL)
 		{
-			if (isNull)
+			if (nbts_attiter_curattisnull(itup))
 				result = 0;		/* NULL "=" NULL */
 			else if (scankey->sk_flags & SK_BT_NULLS_FIRST)
 				result = -1;	/* NULL "<" NOT_NULL */
 			else
 				result = 1;		/* NULL ">" NOT_NULL */
 		}
-		else if (isNull)		/* key is NOT_NULL and item is NULL */
+		/* key is NOT_NULL and item is NULL */
+		else if (nbts_attiter_curattisnull(itup))
 		{
 			if (scankey->sk_flags & SK_BT_NULLS_FIRST)
 				result = 1;		/* NOT_NULL ">" NULL */
@@ -745,7 +749,7 @@ _bt_compare(Relation rel,
 		/* if the keys are unequal, return the difference */
 		if (result != 0)
 		{
-			*comparecol = i;
+			*comparecol = nbts_attiter_attnum;
 			return result;
 		}
 
diff --git a/src/backend/access/nbtree/nbtsort_spec.c b/src/backend/access/nbtree/nbtsort_spec.c
index 368d6f244c..6f33cc4cc2 100644
--- a/src/backend/access/nbtree/nbtsort_spec.c
+++ b/src/backend/access/nbtree/nbtsort_spec.c
@@ -34,8 +34,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 				itup2 = NULL;
 	bool		load1;
 	TupleDesc	tupdes = RelationGetDescr(wstate->index);
-	int			i,
-				keysz = IndexRelationGetNumberOfKeyAttributes(wstate->index);
+	int			keysz = IndexRelationGetNumberOfKeyAttributes(wstate->index);
 	SortSupport sortKeys;
 	int64		tuples_done = 0;
 	bool		deduplicate;
@@ -57,7 +56,7 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 		/* Prepare SortSupport data for each column */
 		sortKeys = (SortSupport) palloc0(keysz * sizeof(SortSupportData));
 
-		for (i = 0; i < keysz; i++)
+		for (int i = 0; i < keysz; i++)
 		{
 			SortSupport sortKey = sortKeys + i;
 			ScanKey		scanKey = wstate->inskey->scankeys + i;
@@ -90,21 +89,24 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 			else if (itup != NULL)
 			{
 				int32		compare = 0;
+				nbts_attiterdeclare(itup);
+				nbts_attiterdeclare(itup2);
 
-				for (i = 1; i <= keysz; i++)
+				nbts_attiterinit(itup, 1, tupdes);
+				nbts_attiterinit(itup2, 1, tupdes);
+
+				nbts_foreachattr(1, keysz)
 				{
 					SortSupport entry;
 					Datum		attrDatum1,
 								attrDatum2;
-					bool		isNull1,
-								isNull2;
 
-					entry = sortKeys + i - 1;
-					attrDatum1 = index_getattr(itup, i, tupdes, &isNull1);
-					attrDatum2 = index_getattr(itup2, i, tupdes, &isNull2);
+					entry = sortKeys + nbts_attiter_attnum - 1;
+					attrDatum1 = nbts_attiter_nextattdatum(itup, tupdes);
+					attrDatum2 = nbts_attiter_nextattdatum(itup2, tupdes);
 
-					compare = ApplySortComparator(attrDatum1, isNull1,
-												  attrDatum2, isNull2,
+					compare = ApplySortComparator(attrDatum1, nbts_attiter_curattisnull(itup),
+												  attrDatum2, nbts_attiter_curattisnull(itup2),
 												  entry);
 					if (compare > 0)
 					{
diff --git a/src/backend/access/nbtree/nbtutils_spec.c b/src/backend/access/nbtree/nbtutils_spec.c
index 0288da22d6..07ca18f404 100644
--- a/src/backend/access/nbtree/nbtutils_spec.c
+++ b/src/backend/access/nbtree/nbtutils_spec.c
@@ -64,7 +64,7 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 	int			indnkeyatts;
 	int16	   *indoption;
 	int			tupnatts;
-	int			i;
+	nbts_attiterdeclare(itup);
 
 	itupdesc = RelationGetDescr(rel);
 	indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
@@ -95,7 +95,10 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 	key->scantid = key->heapkeyspace && itup ?
 				   BTreeTupleGetHeapTID(itup) : NULL;
 	skey = key->scankeys;
-	for (i = 0; i < indnkeyatts; i++)
+
+	nbts_attiterinit(itup, 1, itupdesc);
+
+	nbts_foreachattr(1, indnkeyatts)
 	{
 		FmgrInfo   *procinfo;
 		Datum		arg;
@@ -106,27 +109,30 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 		 * We can use the cached (default) support procs since no cross-type
 		 * comparison can be needed.
 		 */
-		procinfo = index_getprocinfo(rel, i + 1, BTORDER_PROC);
+		procinfo = index_getprocinfo(rel, nbts_attiter_attnum, BTORDER_PROC);
 
 		/*
 		 * Key arguments built from truncated attributes (or when caller
 		 * provides no tuple) are defensively represented as NULL values. They
 		 * should never be used.
 		 */
-		if (i < tupnatts)
-			arg = index_getattr(itup, i + 1, itupdesc, &null);
+		if (nbts_attiter_attnum <= tupnatts)
+		{
+			arg = nbts_attiter_nextattdatum(itup, itupdesc);
+			null = nbts_attiter_curattisnull(itup);
+		}
 		else
 		{
 			arg = (Datum) 0;
 			null = true;
 		}
-		flags = (null ? SK_ISNULL : 0) | (indoption[i] << SK_BT_INDOPTION_SHIFT);
-		ScanKeyEntryInitializeWithInfo(&skey[i],
+		flags = (null ? SK_ISNULL : 0) | (indoption[nbts_attiter_attnum - 1] << SK_BT_INDOPTION_SHIFT);
+		ScanKeyEntryInitializeWithInfo(&skey[nbts_attiter_attnum - 1],
 									   flags,
-									   (AttrNumber) (i + 1),
+									   (AttrNumber) nbts_attiter_attnum,
 									   InvalidStrategy,
 									   InvalidOid,
-									   rel->rd_indcollation[i],
+									   rel->rd_indcollation[nbts_attiter_attnum - 1],
 									   procinfo,
 									   arg);
 		/* Record if any key attribute is NULL (or truncated) */
@@ -675,6 +681,8 @@ _bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
 	TupleDesc	itupdesc = RelationGetDescr(rel);
 	int			keepnatts;
 	ScanKey		scankey;
+	nbts_attiterdeclare(lastleft);
+	nbts_attiterdeclare(firstright);
 
 	/*
 	 * _bt_compare() treats truncated key attributes as having the value minus
@@ -686,20 +694,22 @@ _bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
 
 	scankey = itup_key->scankeys;
 	keepnatts = 1;
-	for (int attnum = 1; attnum <= nkeyatts; attnum++, scankey++)
+
+	nbts_attiterinit(lastleft, 1, itupdesc);
+	nbts_attiterinit(firstright, 1, itupdesc);
+
+	nbts_foreachattr(1, nkeyatts)
 	{
 		Datum		datum1,
 					datum2;
-		bool		isNull1,
-					isNull2;
 
-		datum1 = index_getattr(lastleft, attnum, itupdesc, &isNull1);
-		datum2 = index_getattr(firstright, attnum, itupdesc, &isNull2);
+		datum1 = nbts_attiter_nextattdatum(lastleft, itupdesc);
+		datum2 = nbts_attiter_nextattdatum(firstright, itupdesc);
 
-		if (isNull1 != isNull2)
+		if (nbts_attiter_curattisnull(lastleft) != nbts_attiter_curattisnull(firstright))
 			break;
 
-		if (!isNull1 &&
+		if (!nbts_attiter_curattisnull(lastleft) &&
 			DatumGetInt32(FunctionCall2Coll(&scankey->sk_func,
 											scankey->sk_collation,
 											datum1,
@@ -707,6 +717,7 @@ _bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
 			break;
 
 		keepnatts++;
+		scankey++;
 	}
 
 	/*
@@ -747,24 +758,27 @@ _bt_keep_natts_fast(Relation rel, IndexTuple lastleft, IndexTuple firstright)
 	TupleDesc	itupdesc = RelationGetDescr(rel);
 	int			keysz = IndexRelationGetNumberOfKeyAttributes(rel);
 	int			keepnatts;
+	nbts_attiterdeclare(lastleft);
+	nbts_attiterdeclare(firstright);
 
 	keepnatts = 1;
-	for (int attnum = 1; attnum <= keysz; attnum++)
+	nbts_attiterinit(lastleft, 1, itupdesc);
+	nbts_attiterinit(firstright, 1, itupdesc);
+
+	nbts_foreachattr(1, keysz)
 	{
 		Datum		datum1,
 					datum2;
-		bool		isNull1,
-					isNull2;
 		Form_pg_attribute att;
 
-		datum1 = index_getattr(lastleft, attnum, itupdesc, &isNull1);
-		datum2 = index_getattr(firstright, attnum, itupdesc, &isNull2);
-		att = TupleDescAttr(itupdesc, attnum - 1);
+		datum1 = nbts_attiter_nextattdatum(lastleft, itupdesc);
+		datum2 = nbts_attiter_nextattdatum(firstright, itupdesc);
+		att = TupleDescAttr(itupdesc, nbts_attiter_attnum - 1);
 
-		if (isNull1 != isNull2)
+		if (nbts_attiter_curattisnull(lastleft) != nbts_attiter_curattisnull(firstright))
 			break;
 
-		if (!isNull1 &&
+		if (!nbts_attiter_curattisnull(lastleft) &&
 			!datum_image_eq(datum1, datum2, att->attbyval, att->attlen))
 			break;
 
diff --git a/src/backend/utils/sort/tuplesortvariants_spec.c b/src/backend/utils/sort/tuplesortvariants_spec.c
index 705da09329..cf262eee2d 100644
--- a/src/backend/utils/sort/tuplesortvariants_spec.c
+++ b/src/backend/utils/sort/tuplesortvariants_spec.c
@@ -66,47 +66,54 @@ comparetup_index_btree_tiebreak(const SortTuple *a, const SortTuple *b,
 	bool		equal_hasnull = false;
 	int			nkey;
 	int32		compare;
-	Datum		datum1,
-				datum2;
-	bool		isnull1,
-				isnull2;
+	nbts_attiterdeclare(tuple1);
+	nbts_attiterdeclare(tuple2);
 
 	tuple1 = (IndexTuple) a->tuple;
 	tuple2 = (IndexTuple) b->tuple;
 	keysz = base->nKeys;
 	tupDes = RelationGetDescr(arg->index.indexRel);
 
-	if (sortKey->abbrev_converter)
+	if (!sortKey->abbrev_converter)
 	{
-		datum1 = index_getattr(tuple1, 1, tupDes, &isnull1);
-		datum2 = index_getattr(tuple2, 1, tupDes, &isnull2);
-
-		compare = ApplySortAbbrevFullComparator(datum1, isnull1,
-												datum2, isnull2,
-												sortKey);
-		if (compare != 0)
-			return compare;
+		nkey = 2;
+		sortKey++;
+	}
+	else
+	{
+		nkey = 1;
 	}
 
 	/* they are equal, so we only need to examine one null flag */
 	if (a->isnull1)
 		equal_hasnull = true;
 
-	sortKey++;
-	for (nkey = 2; nkey <= keysz; nkey++, sortKey++)
+	nbts_attiterinit(tuple1, nkey, tupDes);
+	nbts_attiterinit(tuple2, nkey, tupDes);
+
+	nbts_foreachattr(nkey, keysz)
 	{
-		datum1 = index_getattr(tuple1, nkey, tupDes, &isnull1);
-		datum2 = index_getattr(tuple2, nkey, tupDes, &isnull2);
+		Datum	datum1,
+				datum2;
+		datum1 = nbts_attiter_nextattdatum(tuple1, tupDes);
+		datum2 = nbts_attiter_nextattdatum(tuple2, tupDes);
+
+		if (nbts_attiter_attnum == 1)
+			compare = ApplySortAbbrevFullComparator(datum1, nbts_attiter_curattisnull(tuple1),
+													datum2, nbts_attiter_curattisnull(tuple2),
+													sortKey);
+		else
+			compare = ApplySortComparator(datum1, nbts_attiter_curattisnull(tuple1),
+										  datum2, nbts_attiter_curattisnull(tuple2),
+										  sortKey);
 
-		compare = ApplySortComparator(datum1, isnull1,
-									  datum2, isnull2,
-									  sortKey);
 		if (compare != 0)
-			return compare;		/* done when we find unequal attributes */
+			return compare;
 
-		/* they are equal, so we only need to examine one null flag */
-		if (isnull1)
+		if (nbts_attiter_curattisnull(tuple1))
 			equal_hasnull = true;
+
+		sortKey++;
 	}
 
 	/*
-- 
2.40.1

