From 3507ac6a110ad37a540db03342be8d7135fa7612 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm@gmail.com>
Date: Sat, 10 Apr 2021 19:01:50 +0200
Subject: [PATCH v2 1/2] Implement and use index tuple attribute iteration

index_getattr's fast-path only works for fixed-size prefixes with non-null
attrubutes, and null attributes, but for all other cases this is a O(n)
lookup. Because it is also often called in a loop for each attribute, we
can re-use the offset results from earlier attribute lookups, and using
that speed up this attribute lookup.
---
 src/backend/access/common/indextuple.c | 149 +++++++++++++++++++++++++
 src/backend/access/gist/gistutil.c     |  32 ++++--
 src/backend/access/nbtree/nbtsearch.c  |  11 +-
 src/backend/access/nbtree/nbtsort.c    |  15 ++-
 src/backend/access/nbtree/nbtutils.c   |  75 ++++++++-----
 src/backend/utils/sort/tuplesort.c     |  28 ++---
 src/include/access/itup.h              |  89 +++++++++++++++
 7 files changed, 339 insertions(+), 60 deletions(-)

diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index 8df882da7a..f0bdc5669b 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -424,6 +424,155 @@ nocache_index_getattr(IndexTuple tup,
 	return fetchatt(TupleDescAttr(tupleDesc, attnum), tp + off);
 }
 
+/*
+ * Initiate an index attribute iterator to attribute attnum, 
+ * and return the corresponding datum.
+ *
+ * This is nearly the same as index_deform_tuple, except that this
+ * returns the internal state up to attnum, instead of populating the
+ * datum- and isnull-arrays
+ */
+void
+nocache_index_attiterinit(IndexTuple tup, AttrNumber attnum, TupleDesc tupleDesc, IAttrIterState iter)
+{
+	bool		hasnulls = IndexTupleHasNulls(tup);
+	int			curatt;
+	char	   *tp;				/* ptr to tuple data */
+	int			off;			/* offset in tuple data */
+	bits8	   *bp;				/* ptr to null bitmap in tuple */
+	bool		slow = false;	/* can we use/set attcacheoff? */
+	bool		null = false;
+
+	/* Assert to protect callers */
+	Assert(tupleDesc->natts <= INDEX_MAX_KEYS);
+	Assert(attnum <= tupleDesc->natts);
+	Assert(attnum > 0);
+
+	/* XXX "knows" t_bits are just after fixed tuple header! */
+	bp = (bits8 *) ((char *) tup + sizeof(IndexTupleData));
+
+	tp = (char *) tup + IndexInfoFindDataOffset(tup->t_info);
+	off = 0;
+
+	for (curatt = 0; curatt < attnum; curatt++)
+	{
+		Form_pg_attribute thisatt = TupleDescAttr(tupleDesc, curatt);
+
+		if (hasnulls && att_isnull(curatt, bp))
+		{
+			null = true;
+			slow = true;		/* can't use attcacheoff anymore */
+			continue;
+		}
+
+		null = false;
+
+		if (!slow && thisatt->attcacheoff >= 0)
+			off = thisatt->attcacheoff;
+		else if (thisatt->attlen == -1)
+		{
+			/*
+			 * We can only cache the offset for a varlena attribute if the
+			 * offset is already suitably aligned, so that there would be no
+			 * pad bytes in any case: then the offset will be valid for either
+			 * an aligned or unaligned value.
+			 */
+			if (!slow &&
+				off == att_align_nominal(off, thisatt->attalign))
+				thisatt->attcacheoff = off;
+			else
+			{
+				off = att_align_pointer(off, thisatt->attalign, -1,
+										tp + off);
+				slow = true;
+			}
+		}
+		else
+		{
+			/* not varlena, so safe to use att_align_nominal */
+			off = att_align_nominal(off, thisatt->attalign);
+
+			if (!slow)
+				thisatt->attcacheoff = off;
+		}
+
+		off = att_addlength_pointer(off, thisatt->attlen, tp + off);
+
+		if (thisatt->attlen <= 0)
+			slow = true;		/* can't use attcacheoff anymore */
+	}
+
+	iter->isNull = null;
+	iter->offset = off;
+	iter->slow = slow;
+}
+
+Datum
+nocache_index_attiternext(IndexTuple tup, AttrNumber attnum, TupleDesc tupleDesc, IAttrIterState iter)
+{
+	bool		hasnulls = IndexTupleHasNulls(tup);
+	char	   *tp;				/* ptr to tuple data */
+	bits8	   *bp;				/* ptr to null bitmap in tuple */
+	Datum		datum;
+
+	Assert(tupleDesc->natts <= INDEX_MAX_KEYS);
+	Assert(attnum <= tupleDesc->natts);
+	Assert(attnum > 0);
+
+	bp = (bits8 *) ((char *) tup + sizeof(IndexTupleData));
+
+	tp = (char *) tup + IndexInfoFindDataOffset(tup->t_info);
+	Form_pg_attribute thisatt = TupleDescAttr(tupleDesc, attnum - 1);
+
+	if (hasnulls && att_isnull(attnum - 1, bp))
+	{
+		iter->isNull = true;
+		iter->slow = true;		/* can't use attcacheoff anymore */
+		return (Datum) 0;
+	}
+
+	iter->isNull = false;
+
+	if (!iter->slow && thisatt->attcacheoff >= 0)
+		iter->offset = thisatt->attcacheoff;
+	else if (thisatt->attlen == -1)
+	{
+		/*
+		 * We can only cache the offset for a varlena attribute if the
+		 * offset is already suitably aligned, so that there would be no
+		 * pad bytes in any case: then the offset will be valid for either
+		 * an aligned or unaligned value.
+		 */
+		if (!iter->slow &&
+			iter->offset == att_align_nominal(iter->offset, thisatt->attalign))
+			thisatt->attcacheoff = iter->offset;
+		else
+		{
+			iter->offset = att_align_pointer(iter->offset, thisatt->attalign, -1,
+											 tp + iter->offset);
+			iter->slow = true;
+		}
+	}
+	else
+	{
+		/* not varlena, so safe to use att_align_nominal */
+		iter->offset = att_align_nominal(iter->offset, thisatt->attalign);
+
+		if (!iter->slow)
+			thisatt->attcacheoff = iter->offset;
+	}
+
+	datum = fetchatt(thisatt, tp + iter->offset);
+
+	iter->offset = att_addlength_pointer(iter->offset, thisatt->attlen, tp + iter->offset);
+
+	if (thisatt->attlen <= 0)
+		iter->slow = true;		/* can't use attcacheoff anymore */
+
+	return datum;
+}
+
+
 /*
  * Convert an index tuple into Datum/isnull arrays.
  *
diff --git a/src/backend/access/gist/gistutil.c b/src/backend/access/gist/gistutil.c
index 8dcd53c457..4455ec0ff9 100644
--- a/src/backend/access/gist/gistutil.c
+++ b/src/backend/access/gist/gistutil.c
@@ -296,15 +296,18 @@ gistDeCompressAtt(GISTSTATE *giststate, Relation r, IndexTuple tuple, Page p,
 				  OffsetNumber o, GISTENTRY *attdata, bool *isnull)
 {
 	int			i;
+	IAttrIterStateData iter;
+	index_attiterinit(tuple, 1, giststate->leafTupdesc, &iter);
 
 	for (i = 0; i < IndexRelationGetNumberOfKeyAttributes(r); i++)
 	{
 		Datum		datum;
 
-		datum = index_getattr(tuple, i + 1, giststate->leafTupdesc, &isnull[i]);
+		datum = index_attiternext(tuple, i + 1, giststate->leafTupdesc, &iter);
+		isnull[i] = iter.isNull;
 		gistdentryinit(giststate, i, &attdata[i],
 					   datum, r, p, o,
-					   false, isnull[i]);
+					   false, iter.isNull);
 	}
 }
 
@@ -439,6 +442,9 @@ gistchoose(Relation r, Page p, IndexTuple it,	/* it has compressed entry */
 		IndexTuple	itup = (IndexTuple) PageGetItem(p, PageGetItemId(p, i));
 		bool		zero_penalty;
 		int			j;
+		IAttrIterStateData iter;
+
+		index_attiterinit(itup, 1, giststate->leafTupdesc, &iter);
 
 		zero_penalty = true;
 
@@ -447,14 +453,13 @@ gistchoose(Relation r, Page p, IndexTuple it,	/* it has compressed entry */
 		{
 			Datum		datum;
 			float		usize;
-			bool		IsNull;
 
 			/* Compute penalty for this column. */
-			datum = index_getattr(itup, j + 1, giststate->leafTupdesc,
-								  &IsNull);
+			datum = index_attiternext(itup, j + 1, giststate->leafTupdesc,
+								  &iter);
 			gistdentryinit(giststate, j, &entry, datum, r, p, i,
-						   false, IsNull);
-			usize = gistpenalty(giststate, j, &entry, IsNull,
+						   false, iter.isNull);
+			usize = gistpenalty(giststate, j, &entry, iter.isNull,
 								&identry[j], isnull[j]);
 			if (usize > 0)
 				zero_penalty = false;
@@ -668,13 +673,17 @@ gistFetchTuple(GISTSTATE *giststate, Relation r, IndexTuple tuple)
 	MemoryContext oldcxt = MemoryContextSwitchTo(giststate->tempCxt);
 	Datum		fetchatt[INDEX_MAX_KEYS];
 	bool		isnull[INDEX_MAX_KEYS];
+	IAttrIterStateData iter;
 	int			i;
 
+	index_attiterinit(tuple, 1, giststate->leafTupdesc, &iter);
+
 	for (i = 0; i < IndexRelationGetNumberOfKeyAttributes(r); i++)
 	{
 		Datum		datum;
 
-		datum = index_getattr(tuple, i + 1, giststate->leafTupdesc, &isnull[i]);
+		datum = index_attiternext(tuple, i + 1, giststate->leafTupdesc, &iter);
+		isnull[i] = iter.isNull;
 
 		if (giststate->fetchFn[i].fn_oid != InvalidOid)
 		{
@@ -707,12 +716,13 @@ gistFetchTuple(GISTSTATE *giststate, Relation r, IndexTuple tuple)
 	}
 
 	/*
-	 * Get each included attribute.
+	 * Get each INCLUDEd attribute.
 	 */
 	for (; i < r->rd_att->natts; i++)
 	{
-		fetchatt[i] = index_getattr(tuple, i + 1, giststate->leafTupdesc,
-									&isnull[i]);
+		fetchatt[i] = index_attiternext(tuple, i + 1, giststate->leafTupdesc,
+										&iter);
+		isnull[i] = iter.isNull;
 	}
 	MemoryContextSwitchTo(oldcxt);
 
diff --git a/src/backend/access/nbtree/nbtsearch.c b/src/backend/access/nbtree/nbtsearch.c
index d1177d8772..e002c11e8b 100644
--- a/src/backend/access/nbtree/nbtsearch.c
+++ b/src/backend/access/nbtree/nbtsearch.c
@@ -654,6 +654,7 @@ _bt_compare(Relation rel,
 	int			ncmpkey;
 	int			ntupatts;
 	int32		result;
+	IAttrIterStateData iter;
 
 	Assert(_bt_check_natts(rel, key->heapkeyspace, page, offnum));
 	Assert(key->keysz <= IndexRelationGetNumberOfKeyAttributes(rel));
@@ -685,23 +686,25 @@ _bt_compare(Relation rel,
 	Assert(key->heapkeyspace || ncmpkey == key->keysz);
 	Assert(!BTreeTupleIsPosting(itup) || key->allequalimage);
 	scankey = key->scankeys;
+
+	index_attiterinit(itup, 1, itupdesc, &iter);
+
 	for (int i = 1; i <= ncmpkey; i++)
 	{
 		Datum		datum;
-		bool		isNull;
 
-		datum = index_getattr(itup, scankey->sk_attno, itupdesc, &isNull);
+		datum = index_attiternext(itup, scankey->sk_attno, itupdesc, &iter);
 
 		if (scankey->sk_flags & SK_ISNULL)	/* key is NULL */
 		{
-			if (isNull)
+			if (iter.isNull)
 				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 */
+		else if (iter.isNull)		/* key is NOT_NULL and item is NULL */
 		{
 			if (scankey->sk_flags & SK_BT_NULLS_FIRST)
 				result = 1;		/* NOT_NULL ">" NULL */
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index 5fa6ea8ad9..3806f275f5 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -1242,21 +1242,24 @@ _bt_load(BTWriteState *wstate, BTSpool *btspool, BTSpool *btspool2)
 			else if (itup != NULL)
 			{
 				int32		compare = 0;
+				IAttrIterStateData iter1;
+				IAttrIterStateData iter2;
+
+				index_attiterinit(itup, 1, tupdes, &iter1);
+				index_attiterinit(itup2, 1, tupdes, &iter2);
 
 				for (i = 1; i <= keysz; i++)
 				{
 					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);
+					attrDatum1 = index_attiternext(itup, i, tupdes, &iter1);
+					attrDatum2 = index_attiternext(itup2, i, tupdes, &iter2);
 
-					compare = ApplySortComparator(attrDatum1, isNull1,
-												  attrDatum2, isNull2,
+					compare = ApplySortComparator(attrDatum1, iter1.isNull,
+												  attrDatum2, iter2.isNull,
 												  entry);
 					if (compare > 0)
 					{
diff --git a/src/backend/access/nbtree/nbtutils.c b/src/backend/access/nbtree/nbtutils.c
index d524310723..87b0075fdc 100644
--- a/src/backend/access/nbtree/nbtutils.c
+++ b/src/backend/access/nbtree/nbtutils.c
@@ -96,6 +96,7 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 	int16	   *indoption;
 	int			tupnatts;
 	int			i;
+	IAttrIterStateData iter;
 
 	itupdesc = RelationGetDescr(rel);
 	indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
@@ -126,11 +127,13 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 	key->scantid = key->heapkeyspace && itup ?
 		BTreeTupleGetHeapTID(itup) : NULL;
 	skey = key->scankeys;
+
+	index_attiterinit(itup, 1, itupdesc, &iter);
+
 	for (i = 0; i < indnkeyatts; i++)
 	{
 		FmgrInfo   *procinfo;
 		Datum		arg;
-		bool		null;
 		int			flags;
 
 		/*
@@ -145,13 +148,13 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 		 * should never be used.
 		 */
 		if (i < tupnatts)
-			arg = index_getattr(itup, i + 1, itupdesc, &null);
+			arg = index_attiternext(itup, i + 1, itupdesc, &iter);
 		else
 		{
 			arg = (Datum) 0;
-			null = true;
+			iter.isNull = true;
 		}
-		flags = (null ? SK_ISNULL : 0) | (indoption[i] << SK_BT_INDOPTION_SHIFT);
+		flags = (iter.isNull ? SK_ISNULL : 0) | (indoption[i] << SK_BT_INDOPTION_SHIFT);
 		ScanKeyEntryInitializeWithInfo(&skey[i],
 									   flags,
 									   (AttrNumber) (i + 1),
@@ -161,7 +164,7 @@ _bt_mkscankey(Relation rel, IndexTuple itup)
 									   procinfo,
 									   arg);
 		/* Record if any key attribute is NULL (or truncated) */
-		if (null)
+		if (iter.isNull)
 			key->anynullkeys = true;
 	}
 
@@ -1360,6 +1363,9 @@ _bt_checkkeys(IndexScanDesc scan, IndexTuple tuple, int tupnatts,
 	int			keysz;
 	int			ikey;
 	ScanKey		key;
+	IAttrIterStateData iter;
+	Datum		datum;
+	int			curattnum;
 
 	Assert(BTreeTupleGetNAtts(tuple, scan->indexRelation) == tupnatts);
 
@@ -1369,10 +1375,12 @@ _bt_checkkeys(IndexScanDesc scan, IndexTuple tuple, int tupnatts,
 	so = (BTScanOpaque) scan->opaque;
 	keysz = so->numberOfKeys;
 
+	index_attiterinit(tuple, 1, tupdesc, &iter);
+	curattnum = 0;
+	datum = (Datum) 0;
+
 	for (key = so->keyData, ikey = 0; ikey < keysz; key++, ikey++)
 	{
-		Datum		datum;
-		bool		isNull;
 		Datum		test;
 
 		if (key->sk_attno > tupnatts)
@@ -1397,23 +1405,31 @@ _bt_checkkeys(IndexScanDesc scan, IndexTuple tuple, int tupnatts,
 			return false;
 		}
 
-		datum = index_getattr(tuple,
-							  key->sk_attno,
-							  tupdesc,
-							  &isNull);
+		/*
+		 * Attributes are received in sorded order, so we iterate until we
+		 * have the correct attribute. We will not see preceding attribute
+		 * again.
+		 * 
+		 * Note that we can see the same attribute many times; in which
+		 * case we will skip the index_attiternext call.
+		 */
+		for(; key->sk_attno > curattnum; curattnum++)
+		{
+			datum = index_attiternext(tuple, curattnum + 1, tupdesc, &iter);
+		}
 
 		if (key->sk_flags & SK_ISNULL)
 		{
 			/* Handle IS NULL/NOT NULL tests */
 			if (key->sk_flags & SK_SEARCHNULL)
 			{
-				if (isNull)
+				if (iter.isNull)
 					continue;	/* tuple satisfies this qual */
 			}
 			else
 			{
 				Assert(key->sk_flags & SK_SEARCHNOTNULL);
-				if (!isNull)
+				if (!iter.isNull)
 					continue;	/* tuple satisfies this qual */
 			}
 
@@ -1435,7 +1451,7 @@ _bt_checkkeys(IndexScanDesc scan, IndexTuple tuple, int tupnatts,
 			return false;
 		}
 
-		if (isNull)
+		if (iter.isNull)
 		{
 			if (key->sk_flags & SK_BT_NULLS_FIRST)
 			{
@@ -2349,6 +2365,8 @@ _bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
 	TupleDesc	itupdesc = RelationGetDescr(rel);
 	int			keepnatts;
 	ScanKey		scankey;
+	IAttrIterStateData iter1;
+	IAttrIterStateData iter2;
 
 	/*
 	 * _bt_compare() treats truncated key attributes as having the value minus
@@ -2360,20 +2378,22 @@ _bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
 
 	scankey = itup_key->scankeys;
 	keepnatts = 1;
+
+	index_attiterinit(lastleft, 1, itupdesc, &iter1);
+	index_attiterinit(firstright, 1, itupdesc, &iter2);
+
 	for (int attnum = 1; attnum <= nkeyatts; attnum++, scankey++)
 	{
 		Datum		datum1,
 					datum2;
-		bool		isNull1,
-					isNull2;
 
-		datum1 = index_getattr(lastleft, attnum, itupdesc, &isNull1);
-		datum2 = index_getattr(firstright, attnum, itupdesc, &isNull2);
+		datum1 = index_attiternext(lastleft, attnum, itupdesc, &iter1);
+		datum2 = index_attiternext(firstright, attnum, itupdesc, &iter2);
 
-		if (isNull1 != isNull2)
+		if (iter1.isNull != iter2.isNull)
 			break;
 
-		if (!isNull1 &&
+		if (!iter1.isNull &&
 			DatumGetInt32(FunctionCall2Coll(&scankey->sk_func,
 											scankey->sk_collation,
 											datum1,
@@ -2421,24 +2441,27 @@ _bt_keep_natts_fast(Relation rel, IndexTuple lastleft, IndexTuple firstright)
 	TupleDesc	itupdesc = RelationGetDescr(rel);
 	int			keysz = IndexRelationGetNumberOfKeyAttributes(rel);
 	int			keepnatts;
+	IAttrIterStateData iter1;
+	IAttrIterStateData iter2;
+	
+	index_attiterinit(lastleft, 1, itupdesc, &iter1);
+	index_attiterinit(firstright, 1, itupdesc, &iter2);
 
 	keepnatts = 1;
 	for (int attnum = 1; attnum <= keysz; attnum++)
 	{
 		Datum		datum1,
 					datum2;
-		bool		isNull1,
-					isNull2;
 		Form_pg_attribute att;
 
-		datum1 = index_getattr(lastleft, attnum, itupdesc, &isNull1);
-		datum2 = index_getattr(firstright, attnum, itupdesc, &isNull2);
+		datum1 = index_attiternext(lastleft, attnum, itupdesc, &iter1);
+		datum2 = index_attiternext(firstright, attnum, itupdesc, &iter2);
 		att = TupleDescAttr(itupdesc, attnum - 1);
 
-		if (isNull1 != isNull2)
+		if (iter1.isNull != iter2.isNull)
 			break;
 
-		if (!isNull1 &&
+		if (!iter1.isNull &&
 			!datum_image_eq(datum1, datum2, att->attbyval, att->attlen))
 			break;
 
diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index 22972071ff..18fa22cb26 100644
--- a/src/backend/utils/sort/tuplesort.c
+++ b/src/backend/utils/sort/tuplesort.c
@@ -4215,9 +4215,8 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 	int32		compare;
 	Datum		datum1,
 				datum2;
-	bool		isnull1,
-				isnull2;
-
+	IAttrIterStateData iter1;
+	IAttrIterStateData iter2;
 
 	/* Compare the leading sort key */
 	compare = ApplySortComparator(a->datum1, a->isnull1,
@@ -4231,14 +4230,17 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 	tuple2 = (IndexTuple) b->tuple;
 	keysz = state->nKeys;
 	tupDes = RelationGetDescr(state->indexRel);
+	
+	index_attiterinit(tuple1, 1, tupDes, &iter1);
+	index_attiterinit(tuple2, 1, tupDes, &iter2);
+
+	datum1 = index_attiternext(tuple1, 1, tupDes, &iter1);
+	datum2 = index_attiternext(tuple2, 1, tupDes, &iter2);
 
 	if (sortKey->abbrev_converter)
 	{
-		datum1 = index_getattr(tuple1, 1, tupDes, &isnull1);
-		datum2 = index_getattr(tuple2, 1, tupDes, &isnull2);
-
-		compare = ApplySortAbbrevFullComparator(datum1, isnull1,
-												datum2, isnull2,
+		compare = ApplySortAbbrevFullComparator(datum1, iter1.isNull,
+												datum2, iter2.isNull,
 												sortKey);
 		if (compare != 0)
 			return compare;
@@ -4251,17 +4253,17 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 	sortKey++;
 	for (nkey = 2; nkey <= keysz; nkey++, sortKey++)
 	{
-		datum1 = index_getattr(tuple1, nkey, tupDes, &isnull1);
-		datum2 = index_getattr(tuple2, nkey, tupDes, &isnull2);
+		datum1 = index_attiternext(tuple1, nkey, tupDes, &iter1);
+		datum2 = index_attiternext(tuple2, nkey, tupDes, &iter2);
 
-		compare = ApplySortComparator(datum1, isnull1,
-									  datum2, isnull2,
+		compare = ApplySortComparator(datum1, iter1.isNull,
+									  datum2, iter2.isNull,
 									  sortKey);
 		if (compare != 0)
 			return compare;		/* done when we find unequal attributes */
 
 		/* they are equal, so we only need to examine one null flag */
-		if (isnull1)
+		if (iter1.isNull)
 			equal_hasnull = true;
 	}
 
diff --git a/src/include/access/itup.h b/src/include/access/itup.h
index 1917375cde..224aa61ad8 100644
--- a/src/include/access/itup.h
+++ b/src/include/access/itup.h
@@ -59,6 +59,15 @@ typedef struct IndexAttributeBitMapData
 
 typedef IndexAttributeBitMapData * IndexAttributeBitMap;
 
+typedef struct IAttrIterStateData
+{
+	int			offset;
+	bool		slow;
+	bool		isNull;
+} IAttrIterStateData;
+
+typedef IAttrIterStateData * IAttrIterState;
+
 /*
  * t_info manipulation macros
  */
@@ -126,6 +135,84 @@ typedef IndexAttributeBitMapData * IndexAttributeBitMap;
 	) \
 )
 
+/* ----------------
+ *		index_attiterinit
+ *
+ *		This gets called many times, so we macro the cacheable and NULL
+ *		lookups, and call nocache_index_attiterinit() for the rest.
+ *
+ *		tup - the tuple being iterated on
+ *		attnum - the attribute number that we start the iteration with 
+ *				 in the first index_attiternext call
+ *		tupdesc - the tuple description
+ *
+ * ----------------
+ */
+#define index_attiterinit(tup, attnum, tupleDesc, iter) \
+do { \
+	if ((attnum) == 1) \
+	{ \
+		*(iter) = ((IAttrIterStateData) { \
+			0 /* offset */, \
+			false /* slow */, \
+			false /* isNull */ \
+		}); \
+	} \
+	else if (!IndexTupleHasNulls(tup) && \
+			 TupleDescAttr((tupleDesc), (attnum)-1)->attcacheoff >= 0) \
+	{ \
+		*(iter) = ((IAttrIterStateData) { \
+			TupleDescAttr((tupleDesc), (attnum)-1)->attcacheoff, /* offset */ \
+			TupleDescAttr((tupleDesc), (attnum)-1)->attlen >= 0, /* slow */ \
+			false /* isNull */ \
+		}); \
+	} \
+	else \
+		nocache_index_attiterinit((tup), (attnum) - 1, (tupleDesc), (iter)); \
+} while (false);
+
+/* ----------------
+ *		index_attiternext
+ *
+ *		This gets called many times, so we macro the cacheable and NULL
+ *		lookups, and call nocache_index_attiternext() for the rest.
+ *
+ * ----------------
+ */
+#define index_attiternext(itup, attnum, tupleDesc, iter) \
+( \
+	AssertMacro(PointerIsValid(iter) && (attnum) > 0), \
+	(!IndexTupleHasNulls(itup)) ? \
+	( \
+		!(iter)->slow && TupleDescAttr((tupleDesc), (attnum) - 1)->attcacheoff >= 0 ? \
+		( \
+			(iter)->offset = att_addlength_pointer(TupleDescAttr((tupleDesc), \
+				(attnum) - 1)->attcacheoff, TupleDescAttr((tupleDesc), \
+				(attnum) - 1)->attlen, (char *) (itup) + \
+				IndexInfoFindDataOffset((itup)->t_info) + \
+				TupleDescAttr((tupleDesc), (attnum) - 1)->attcacheoff), \
+			(iter)->isNull = false,\
+			(iter)->slow = TupleDescAttr((tupleDesc), (attnum) - 1)->attlen < 0, \
+			(Datum) fetchatt(TupleDescAttr((tupleDesc), (attnum) - 1), \
+				(char *) (itup) + IndexInfoFindDataOffset((itup)->t_info) + \
+				TupleDescAttr((tupleDesc), (attnum) - 1)->attcacheoff) \
+		) \
+		: \
+		nocache_index_attiternext((itup), (attnum), (tupleDesc), (iter)) \
+	) \
+	: \
+	( \
+		att_isnull((attnum) - 1, (char *) (itup) + sizeof(IndexTupleData)) ? \
+		( \
+			(iter)->isNull = true, \
+			(iter)->slow = true, \
+			(Datum) 0 \
+		) \
+		: \
+		nocache_index_attiternext((itup), (attnum), (tupleDesc), (iter)) \
+	) \
+)
+
 /*
  * MaxIndexTuplesPerPage is an upper bound on the number of tuples that can
  * fit on one index page.  An index tuple must have either data or a null
@@ -160,5 +247,7 @@ extern void index_deform_tuple_internal(TupleDesc tupleDescriptor,
 extern IndexTuple CopyIndexTuple(IndexTuple source);
 extern IndexTuple index_truncate_tuple(TupleDesc sourceDescriptor,
 									   IndexTuple source, int leavenatts);
+extern void nocache_index_attiterinit(IndexTuple tup, AttrNumber attnum, TupleDesc tupleDesc, IAttrIterState iter);
+extern Datum nocache_index_attiternext(IndexTuple tup, AttrNumber attnum, TupleDesc tupleDesc, IAttrIterState iter);
 
 #endif							/* ITUP_H */
-- 
2.20.1

