PGPRO-2554: Copy an additional information into the current memory

In case of concurrent updates reading raw additional information from
index pages may lead to undefined behaviour.

Author: za-arthur

src/rum.h

@@ -19,6 +19,7 @@
 #include "access/sdir.h"
 #include "lib/rbtree.h"
 #include "storage/bufmgr.h"
+#include "utils/datum.h"
 
 #include "rumsort.h"
 
@@ -529,7 +530,7 @@ extern void rumEntryFillRoot(RumBtree btree, Buffer root, Buffer lbuf, Buffer rb
 				 Page page, Page lpage, Page rpage);
 extern IndexTuple rumPageGetLinkItup(RumBtree btree, Buffer buf, Page page);
 extern void rumReadTuple(RumState * rumstate, OffsetNumber attnum,
-			 IndexTuple itup, RumItem * items);
+			 IndexTuple itup, RumItem * items, bool copyAddInfo);
 extern void rumReadTuplePointers(RumState * rumstate, OffsetNumber attnum,
 					 IndexTuple itup, ItemPointerData *ipd);
 extern void updateItemIndexes(Page page, OffsetNumber attnum, RumState * rumstate);
@@ -941,10 +942,14 @@ rumDataPageLeafReadItemPointer(char *ptr, ItemPointer iptr, bool *addInfoIsNull)
  * Reads next item pointer and additional information from leaf data page.
  * Replaces current item pointer with the next one. Zero item pointer should be
  * passed in order to read the first item pointer.
+ *
+ * It is necessary to pass copyAddInfo=true if additional information is used
+ * when the data page is unlocked. If the additional information is used without
+ * locking one can get unexpected behaviour.
  */
 static inline Pointer
 rumDataPageLeafRead(Pointer ptr, OffsetNumber attnum, RumItem * item,
-					RumState * rumstate)
+					bool copyAddInfo, RumState * rumstate)
 {
 	Form_pg_attribute attr;
 
@@ -1009,8 +1014,13 @@ rumDataPageLeafRead(Pointer ptr, OffsetNumber attnum, RumItem * item,
 		}
 		else
 		{
-			ptr = (Pointer) att_align_pointer(ptr, attr->attalign, attr->attlen, ptr);
-			item->addInfo = fetch_att(ptr, attr->attbyval, attr->attlen);
+			Datum		addInfo;
+
+			ptr = (Pointer) att_align_pointer(ptr, attr->attalign, attr->attlen,
+											  ptr);
+			addInfo = fetch_att(ptr, attr->attbyval, attr->attlen);
+			item->addInfo = copyAddInfo ?
+				datumCopy(addInfo, attr->attbyval, attr->attlen) : addInfo;
 		}
 
 		ptr = (Pointer) att_addlength_pointer(ptr, attr->attlen, ptr);

src/rumdatapage.c

@@ -589,7 +589,7 @@ findInLeafPage(RumBtree btree, Page page, OffsetNumber *offset,
 		*iptrOut = item.iptr;
 
 		ptr = rumDataPageLeafRead(ptr, btree->entryAttnum, &item,
-								  btree->rumstate);
+								  false, btree->rumstate);
 
 		cmp = compareRumItem(btree->rumstate, btree->entryAttnum,
 							 &btree->items[btree->curitem], &item);
@@ -899,7 +899,8 @@ dataPlaceToPage(RumBtree btree, Page page, OffsetNumber off)
 			if (copyItemEmpty == true && off <= maxoff)
 			{
 				copyPtr = rumDataPageLeafRead(copyPtr, btree->entryAttnum,
-											  &copyItem, btree->rumstate);
+											  &copyItem, false,
+											  btree->rumstate);
 				copyItemEmpty = false;
 			}
 
@@ -1091,7 +1092,7 @@ dataSplitPageLeaf(RumBtree btree, Buffer lbuf, Buffer rbuf,
 
 		prevIptr = item.iptr;
 		copyPtr = rumDataPageLeafRead(copyPtr, btree->entryAttnum, &item,
-									  btree->rumstate);
+									  false, btree->rumstate);
 
 		prevTotalsize = totalsize;
 		totalsize = rumCheckPlaceToDataPageLeaf(btree->entryAttnum,
@@ -1169,7 +1170,7 @@ dataSplitPageLeaf(RumBtree btree, Buffer lbuf, Buffer rbuf,
 		}
 
 		copyPtr = rumDataPageLeafRead(copyPtr, btree->entryAttnum, &item,
-									  btree->rumstate);
+									  false, btree->rumstate);
 
 		curItem = item;
 		ptr = rumPlaceToDataPageLeaf(ptr, btree->entryAttnum, &item,
@@ -1351,7 +1352,7 @@ updateItemIndexes(Page page, OffsetNumber attnum, RumState * rumstate)
 			}
 			j++;
 		}
-		ptr = rumDataPageLeafRead(ptr, attnum, &item, rumstate);
+		ptr = rumDataPageLeafRead(ptr, attnum, &item, false, rumstate);
 	}
 	/* Fill rest of page indexes with InvalidOffsetNumber if any */
 	for (; j < RumDataLeafIndexCount; j++)

src/rumentrypage.c

@@ -21,7 +21,7 @@
  */
 void
 rumReadTuple(RumState * rumstate, OffsetNumber attnum,
-			 IndexTuple itup, RumItem * items)
+			 IndexTuple itup, RumItem * items, bool copyAddInfo)
 {
 	Pointer		ptr = RumGetPosting(itup);
 	RumItem		item;
@@ -31,7 +31,7 @@ rumReadTuple(RumState * rumstate, OffsetNumber attnum,
 	ItemPointerSetMin(&item.iptr);
 	for (i = 0; i < nipd; i++)
 	{
-		ptr = rumDataPageLeafRead(ptr, attnum, &item, rumstate);
+		ptr = rumDataPageLeafRead(ptr, attnum, &item, copyAddInfo, rumstate);
 		items[i] = item;
 	}
 }

src/rumget.c

@@ -258,7 +258,8 @@ scanPostingTree(Relation index, RumScanEntry scanEntry,
 			ptr = RumDataPageGetData(page);
 			for (i = FirstOffsetNumber; i <= maxoff; i++)
 			{
-				ptr = rumDataPageLeafRead(ptr, attnum, &item.item, rumstate);
+				ptr = rumDataPageLeafRead(ptr, attnum, &item.item, false,
+										  rumstate);
 				SCAN_ITEM_PUT_KEY(scanEntry, item, idatum, icategory);
 				rum_tuplesort_putrumitem(scanEntry->matchSortstate, &item);
 			}
@@ -468,7 +469,7 @@ collectMatchBitmap(RumBtreeData * btree, RumBtreeStack * stack,
 			for (i = 0; i < RumGetNPosting(itup); i++)
 			{
 				ptr = rumDataPageLeafRead(ptr, scanEntry->attnum, &item.item,
-										  rumstate);
+										  false, rumstate);
 				SCAN_ITEM_PUT_KEY(scanEntry, item, idatum, icategory);
 				rum_tuplesort_putrumitem(scanEntry->matchSortstate, &item);
 			}
@@ -674,7 +675,8 @@ startScanEntry(RumState * rumstate, RumScanEntry entry, Snapshot snapshot)
 
 			for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
 			{
-				ptr = rumDataPageLeafRead(ptr, entry->attnum, &item, rumstate);
+				ptr = rumDataPageLeafRead(ptr, entry->attnum, &item, true,
+										  rumstate);
 				entry->list[i - FirstOffsetNumber] = item;
 			}
 
@@ -689,7 +691,7 @@ startScanEntry(RumState * rumstate, RumScanEntry entry, Snapshot snapshot)
 			entry->predictNumberResult = entry->nlist;
 			entry->list = (RumItem *) palloc(sizeof(RumItem) * entry->nlist);
 
-			rumReadTuple(rumstate, entry->attnum, itup, entry->list);
+			rumReadTuple(rumstate, entry->attnum, itup, entry->list, true);
 			entry->isFinished = setListPositionScanEntry(rumstate, entry);
 			if (!entry->isFinished)
 				entry->curItem = entry->list[entry->offset];
@@ -935,7 +937,8 @@ entryGetNextItem(RumState * rumstate, RumScanEntry entry, Snapshot snapshot)
 
 			for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
 			{
-				ptr = rumDataPageLeafRead(ptr, entry->attnum, &item, rumstate);
+				ptr = rumDataPageLeafRead(ptr, entry->attnum, &item, true,
+										  rumstate);
 				entry->list[i - FirstOffsetNumber] = item;
 
 				if (searchBorder)
@@ -1091,7 +1094,8 @@ entryGetNextItemList(RumState * rumstate, RumScanEntry entry, Snapshot snapshot)
 
 		for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
 		{
-			ptr = rumDataPageLeafRead(ptr, entry->attnum, &item, rumstate);
+			ptr = rumDataPageLeafRead(ptr, entry->attnum, &item, true,
+									  rumstate);
 			entry->list[i - FirstOffsetNumber] = item;
 		}
 
@@ -1104,7 +1108,7 @@ entryGetNextItemList(RumState * rumstate, RumScanEntry entry, Snapshot snapshot)
 		entry->predictNumberResult = entry->nlist;
 		entry->list = (RumItem *) palloc(sizeof(RumItem) * entry->nlist);
 
-		rumReadTuple(rumstate, entry->attnum, itup, entry->list);
+		rumReadTuple(rumstate, entry->attnum, itup, entry->list, true);
 		entry->isFinished = setListPositionScanEntry(rumstate, entry);
 	}
 
@@ -1659,7 +1663,8 @@ scanPage(RumState * rumstate, RumScanEntry entry, RumItem *item, bool equalOk)
 	bound = -1;
 	for (i = first; i <= maxoff; i++)
 	{
-		ptr = rumDataPageLeafRead(ptr, entry->attnum, &iter_item, rumstate);
+		ptr = rumDataPageLeafRead(ptr, entry->attnum, &iter_item, true,
+								  rumstate);
 		entry->list[i - first] = iter_item;
 
 		if (bound != -1)

src/ruminsert.c

@@ -272,7 +272,7 @@ addItemPointersToLeafTuple(RumState * rumstate,
 	newNPosting = oldNPosting + nitem;
 	newItems = (RumItem *) palloc(sizeof(RumItem) * newNPosting);
 
-	rumReadTuple(rumstate, attnum, old, oldItems);
+	rumReadTuple(rumstate, attnum, old, oldItems, false);
 
 	newNPosting = rumMergeRumItems(rumstate, attnum, newItems,
 								   items, nitem, oldItems, oldNPosting);

src/rumvacuum.c

@@ -63,7 +63,7 @@ rumVacuumPostingList(RumVacuumState * gvs, OffsetNumber attnum, Pointer src,
 	for (i = 0; i < nitem; i++)
 	{
 		prev = ptr;
-		ptr = rumDataPageLeafRead(ptr, attnum, &item, &gvs->rumstate);
+		ptr = rumDataPageLeafRead(ptr, attnum, &item, false, &gvs->rumstate);
 		if (gvs->callback(&item.iptr, gvs->callback_state))
 		{
 			gvs->result->tuples_removed += 1;