Merge branch 'PGPRO10' into PGPROEE10

Author: vbwagner

src/backend/access/common/indextuple.c

@@ -38,6 +38,25 @@ IndexTuple
 index_form_tuple(TupleDesc tupleDescriptor,
 				 Datum *values,
 				 bool *isnull)
+{
+	return index_form_tuple_extended(tupleDescriptor, values, isnull, tupleDescriptor->natts, NULL);
+}
+
+/*
+ * Workhorse for index_form_tuple().
+ * This function is needed primarily to avoid de-toasting of the
+ * included attributes for indexes with included columns.
+ *
+ * If sz_check_failed == NULL, then in case of tuple being too big we will print
+ * an error. If it's a pointer to a bool, then we won't print an error but we will
+ * make the bool TRUE if the tuple is too big and FALSE if it's all good.
+ */
+IndexTuple
+index_form_tuple_extended(TupleDesc tupleDescriptor,
+				 Datum *values,
+				 bool *isnull,
+				 int nkeyattrs,
+				 bool *sz_check_failed)
 {
 	char	   *tp;				/* tuple pointer */
 	IndexTuple	tuple;			/* return tuple */
@@ -61,6 +80,8 @@ index_form_tuple(TupleDesc tupleDescriptor,
 				 errmsg("number of index columns (%d) exceeds limit (%d)",
 						numberOfAttributes, INDEX_MAX_KEYS)));
 
+	Assert (nkeyattrs <= tupleDescriptor->natts);
+
 #ifdef TOAST_INDEX_HACK
 	for (i = 0; i < numberOfAttributes; i++)
 	{
@@ -73,6 +94,10 @@ index_form_tuple(TupleDesc tupleDescriptor,
 		if (isnull[i] || att->attlen != -1)
 			continue;
 
+		/* Do not de-toast included attibutes. */
+		if (i >= nkeyattrs)
+			continue;
+
 		/*
 		 * If value is stored EXTERNAL, must fetch it so we are not depending
 		 * on outside storage.  This should be improved someday.
@@ -133,6 +158,7 @@ index_form_tuple(TupleDesc tupleDescriptor,
 	tp = (char *) palloc0(size);
 	tuple = (IndexTuple) tp;
 
+
 	heap_fill_tuple(tupleDescriptor,
 #ifdef TOAST_INDEX_HACK
 					untoasted_values,
@@ -162,20 +188,23 @@ index_form_tuple(TupleDesc tupleDescriptor,
 	if (tupmask & HEAP_HASVARWIDTH)
 		infomask |= INDEX_VAR_MASK;
 
-	/* Also assert we got rid of external attributes */
-#ifdef TOAST_INDEX_HACK
-	Assert((tupmask & HEAP_HASEXTERNAL) == 0);
-#endif
-
 	/*
 	 * Here we make sure that the size will fit in the field reserved for it
 	 * in t_info.
 	 */
+	if(sz_check_failed)
+		*sz_check_failed = false;
+
 	if ((size & INDEX_SIZE_MASK) != size)
-		ereport(ERROR,
-				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
-				 errmsg("index row requires %zu bytes, maximum size is %zu",
-						size, (Size) INDEX_SIZE_MASK)));
+	{
+		if(sz_check_failed)
+			*sz_check_failed = true;
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+					errmsg("index row requires %zu bytes, maximum size is %zu",
+							size, (Size) INDEX_SIZE_MASK)));
+	}
 
 	infomask |= size;
 

src/backend/access/nbtree/nbtinsert.c

@@ -88,6 +88,48 @@ static bool _bt_isequal(TupleDesc itupdesc, Page page, OffsetNumber offnum,
 static void _bt_vacuum_one_page(Relation rel, Buffer buffer, Relation heapRel);
 
 
+/*
+ *	tuple_too_long_for_btree() -- returns TRUE if too long, FALSE if not
+ *
+ *		Code for the check taken from _bt_findinsertloc() and _bt_doinsert()
+ */
+bool
+tuple_too_long_for_btree(Relation rel, IndexTuple itup)
+{
+	Buffer		buf;
+	Page		page;
+	Size		itemsz;
+	int			indnkeyatts;
+	ScanKey		itup_scankey;
+	BTStack		stack;
+
+	Assert(IndexRelationGetNumberOfAttributes(rel) != 0);
+	indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
+	Assert(indnkeyatts != 0);
+
+	itup_scankey = _bt_mkscankey(rel, itup);
+
+	stack = _bt_search(rel, indnkeyatts, itup_scankey, false, &buf, BT_WRITE, NULL);
+
+	LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+	LockBuffer(buf, BT_WRITE);
+
+	buf = _bt_moveright(rel, buf, indnkeyatts, itup_scankey, false,
+						true, stack, BT_WRITE, NULL);
+
+	itemsz = IndexTupleDSize(*itup);
+	itemsz = MAXALIGN(itemsz);
+
+	page = BufferGetPage(buf);
+
+	_bt_relbuf(rel, buf);
+
+	if (itemsz > BTMaxItemSize(page))
+		return true;
+
+	return false;
+}
+
 /*
  *	_bt_doinsert() -- Handle insertion of a single index tuple in the tree.
  *

src/backend/access/nbtree/nbtree.c

@@ -22,6 +22,7 @@
 #include "access/relscan.h"
 #include "access/xlog.h"
 #include "access/ptrack.h"
+#include "executor/executor.h"		/* for using CreateExecutorState */
 #include "catalog/index.h"
 #include "commands/vacuum.h"
 #include "nodes/primnodes.h"
@@ -339,9 +340,27 @@ btinsert(Relation rel, Datum *values, bool *isnull,
 {
 	bool		result;
 	IndexTuple	itup;
+	int			indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
+	int			indnatts = IndexRelationGetNumberOfAttributes(rel);
+	TupleDesc	tupleDescIndex = RelationGetDescr(rel);
+	bool		data_too_long_for_making_a_tup = true;
+
+
+	/* Forming an index tuple from the values[] and isnull[] arrays */
+	itup = index_form_tuple_extended(tupleDescIndex, values, isnull, indnkeyatts, &data_too_long_for_making_a_tup);
+
+	/* If we discovered that there's an INCLUDED attribute, AND that the data is too big to fit
+	* in the index, we will take the data for building right from the heap. We do it because this
+	* way we can get actual TOAST links rather than detoasted data. We want TOAST links becacuse we
+	* might want to let INCLUDED attributes stay toasted when being put into the index.
+	*/
+	if((indnatts != indnkeyatts) && (data_too_long_for_making_a_tup || tuple_too_long_for_btree(rel, itup)) )
+	{
+		pfree(itup);
+		bt_get_data_from_heap(rel, values, isnull, ht_ctid, heapRel);
+		itup = index_form_tuple_extended(tupleDescIndex, values, isnull, indnkeyatts, NULL);
+	}
 
-	/* generate an index tuple */
-	itup = index_form_tuple(RelationGetDescr(rel), values, isnull);
 	itup->t_tid = *ht_ctid;
 
 	result = _bt_doinsert(rel, itup, checkUnique, heapRel);
@@ -493,6 +512,51 @@ btgetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 	return ntids;
 }
 
+/*
+* Taking the data for building an index tuple right from the heap.
+* We do it in case if detoasted data turned out to be too long, because
+* this way we can get actual TOAST links rather than detoasted data.
+* We need it only becacuse we might want to let INCLUDED
+* attributes stay toasted when being put into the index.
+*
+* It will form data for btree tuple and put it into values[] and isnull[]
+*/
+void
+bt_get_data_from_heap(Relation rel, Datum *values, bool *isnull,
+		 ItemPointer ht_ctid, Relation heapRel)
+{
+	TupleDesc		tupleDescHeap = RelationGetDescr(heapRel);
+	HeapTupleData	htup;
+	Buffer			buffer;
+
+	htup.t_self = *ht_ctid;
+
+	/* Getting the tuple that we want to form an index tuple with from the heap */
+	if (!heap_fetch(heapRel, SnapshotAny, &htup, &buffer, false, NULL))
+		elog(ERROR, "failed to fetch tuple dy tid while tring to insert a tuple into btree");
+	else
+	{
+		EState			*estate = CreateExecutorState();
+		IndexInfo 		*indexInfo = BuildIndexInfo(rel);
+		TupleTableSlot	*slot;
+		/*
+		* We put the tuple taken from the heap into a slot,
+		* and then FormIndexDatum takes the data from that slot
+		* and fills the values[] and isnull[] arrays with the
+		* appropriate values for the column(s) of the index.
+		* IndexInfo explains the structure of the index tuple.
+		*/
+		slot = MakeSingleTupleTableSlot(tupleDescHeap);
+		ExecStoreTuple(&htup, slot, InvalidBuffer, false);
+
+		FormIndexDatum(indexInfo, slot, estate,
+								values, isnull);
+
+		ExecDropSingleTupleTableSlot(slot);
+		ReleaseBuffer(buffer);
+	}
+}
+
 /*
  *	btbeginscan() -- start a scan on a btree index
  */

src/backend/utils/sort/tuplesort.c

@@ -1411,8 +1411,10 @@ tuplesort_putindextuplevalues(Tuplesortstate *state, Relation rel,
 	SortTuple	stup;
 	Datum		original;
 	IndexTuple	tuple;
+	int			indnkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
+
+	stup.tuple = index_form_tuple_extended(RelationGetDescr(rel), values, isnull, indnkeyatts, NULL);
 
-	stup.tuple = index_form_tuple(RelationGetDescr(rel), values, isnull);
 	tuple = ((IndexTuple) stup.tuple);
 	tuple->t_tid = *self;
 	USEMEM(state, GetMemoryChunkSpace(stup.tuple));

src/include/access/itup.h

@@ -142,6 +142,8 @@ typedef IndexAttributeBitMapData * IndexAttributeBitMap;
 /* routines in indextuple.c */
 extern IndexTuple index_form_tuple(TupleDesc tupleDescriptor,
 				 Datum *values, bool *isnull);
+extern IndexTuple index_form_tuple_extended(TupleDesc tupleDescriptor,
+				 Datum *values, bool *isnull, int nkeyattrs, bool *sz_check_failed);
 extern Datum nocache_index_getattr(IndexTuple tup, int attnum,
 					  TupleDesc tupleDesc);
 extern void index_deform_tuple(IndexTuple tup, TupleDesc tupleDescriptor,

src/include/access/nbtree.h

@@ -542,6 +542,7 @@ extern void _bt_pgrewritetup(Relation rel, Buffer buf, Page page, OffsetNumber o
 extern bool _bt_isbinaryequal(TupleDesc itupdesc, IndexTuple itup, int nindatts, IndexTuple ituptoinsert);
 extern bool _bt_pgaddtup(Page page, Size itemsize, IndexTuple itup,
 						 OffsetNumber itup_off);
+extern bool tuple_too_long_for_btree(Relation rel, IndexTuple itup);
 
 /*
  * prototypes for functions in nbtpage.c