Split tuptoaster.c into three separate files.

detoast.c/h contain functions required to detoast a datum, partially
or completely, plus a few other utility functions for examining the
size of toasted datums.

toast_internals.c/h contain functions that are used internally to the
TOAST subsystem but which (mostly) do not need to be accessed from
outside.

heaptoast.c/h contains code that is intrinsically specific to the
heap AM, either because it operates on HeapTuples or is based on the
layout of a heap page.

detoast.c and toast_internals.c are placed in
src/backend/access/common rather than src/backend/access/heap.  At
present, both files still have dependencies on the heap, but that will
be improved in a future commit.

Patch by me, reviewed and tested by Prabhat Sabu, Thomas Munro,
Andres Freund, and Álvaro Herrera.

Discussion: http://postgr.es/m/CA+TgmoZv-=2iWM4jcw5ZhJeL18HF96+W1yJeYrnGMYdkFFnEpQ@mail.gmail.com

1 files changed, 917 insertions, 0 deletions

diff --git a/src/backend/access/heap/heaptoast.c b/src/backend/access/heap/heaptoast.c
new file mode 100644
index 00000000000..5d105e35174
--- /dev/null
+++ b/src/backend/access/heap/heaptoast.c
@@ -0,0 +1,917 @@
+/*-------------------------------------------------------------------------
+ *
+ * heaptoast.c
+ *	  Heap-specific definitions for external and compressed storage
+ *	  of variable size attributes.
+ *
+ * Copyright (c) 2000-2019, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/heap/heaptoast.c
+ *
+ *
+ * INTERFACE ROUTINES
+ *		toast_insert_or_update -
+ *			Try to make a given tuple fit into one page by compressing
+ *			or moving off attributes
+ *
+ *		toast_delete -
+ *			Reclaim toast storage when a tuple is deleted
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/detoast.h"
+#include "access/heapam.h"
+#include "access/heaptoast.h"
+#include "access/toast_internals.h"
+
+
+/* ----------
+ * toast_delete -
+ *
+ *	Cascaded delete toast-entries on DELETE
+ * ----------
+ */
+void
+toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
+{
+	TupleDesc	tupleDesc;
+	int			numAttrs;
+	int			i;
+	Datum		toast_values[MaxHeapAttributeNumber];
+	bool		toast_isnull[MaxHeapAttributeNumber];
+
+	/*
+	 * We should only ever be called for tuples of plain relations or
+	 * materialized views --- recursing on a toast rel is bad news.
+	 */
+	Assert(rel->rd_rel->relkind == RELKIND_RELATION ||
+		   rel->rd_rel->relkind == RELKIND_MATVIEW);
+
+	/*
+	 * Get the tuple descriptor and break down the tuple into fields.
+	 *
+	 * NOTE: it's debatable whether to use heap_deform_tuple() here or just
+	 * heap_getattr() only the varlena columns.  The latter could win if there
+	 * are few varlena columns and many non-varlena ones. However,
+	 * heap_deform_tuple costs only O(N) while the heap_getattr way would cost
+	 * O(N^2) if there are many varlena columns, so it seems better to err on
+	 * the side of linear cost.  (We won't even be here unless there's at
+	 * least one varlena column, by the way.)
+	 */
+	tupleDesc = rel->rd_att;
+	numAttrs = tupleDesc->natts;
+
+	Assert(numAttrs <= MaxHeapAttributeNumber);
+	heap_deform_tuple(oldtup, tupleDesc, toast_values, toast_isnull);
+
+	/*
+	 * Check for external stored attributes and delete them from the secondary
+	 * relation.
+	 */
+	for (i = 0; i < numAttrs; i++)
+	{
+		if (TupleDescAttr(tupleDesc, i)->attlen == -1)
+		{
+			Datum		value = toast_values[i];
+
+			if (toast_isnull[i])
+				continue;
+			else if (VARATT_IS_EXTERNAL_ONDISK(PointerGetDatum(value)))
+				toast_delete_datum(rel, value, is_speculative);
+		}
+	}
+}
+
+
+/* ----------
+ * toast_insert_or_update -
+ *
+ *	Delete no-longer-used toast-entries and create new ones to
+ *	make the new tuple fit on INSERT or UPDATE
+ *
+ * Inputs:
+ *	newtup: the candidate new tuple to be inserted
+ *	oldtup: the old row version for UPDATE, or NULL for INSERT
+ *	options: options to be passed to heap_insert() for toast rows
+ * Result:
+ *	either newtup if no toasting is needed, or a palloc'd modified tuple
+ *	that is what should actually get stored
+ *
+ * NOTE: neither newtup nor oldtup will be modified.  This is a change
+ * from the pre-8.1 API of this routine.
+ * ----------
+ */
+HeapTuple
+toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
+					   int options)
+{
+	HeapTuple	result_tuple;
+	TupleDesc	tupleDesc;
+	int			numAttrs;
+	int			i;
+
+	bool		need_change = false;
+	bool		need_free = false;
+	bool		need_delold = false;
+	bool		has_nulls = false;
+
+	Size		maxDataLen;
+	Size		hoff;
+
+	char		toast_action[MaxHeapAttributeNumber];
+	bool		toast_isnull[MaxHeapAttributeNumber];
+	bool		toast_oldisnull[MaxHeapAttributeNumber];
+	Datum		toast_values[MaxHeapAttributeNumber];
+	Datum		toast_oldvalues[MaxHeapAttributeNumber];
+	struct varlena *toast_oldexternal[MaxHeapAttributeNumber];
+	int32		toast_sizes[MaxHeapAttributeNumber];
+	bool		toast_free[MaxHeapAttributeNumber];
+	bool		toast_delold[MaxHeapAttributeNumber];
+
+	/*
+	 * Ignore the INSERT_SPECULATIVE option. Speculative insertions/super
+	 * deletions just normally insert/delete the toast values. It seems
+	 * easiest to deal with that here, instead on, potentially, multiple
+	 * callers.
+	 */
+	options &= ~HEAP_INSERT_SPECULATIVE;
+
+	/*
+	 * We should only ever be called for tuples of plain relations or
+	 * materialized views --- recursing on a toast rel is bad news.
+	 */
+	Assert(rel->rd_rel->relkind == RELKIND_RELATION ||
+		   rel->rd_rel->relkind == RELKIND_MATVIEW);
+
+	/*
+	 * Get the tuple descriptor and break down the tuple(s) into fields.
+	 */
+	tupleDesc = rel->rd_att;
+	numAttrs = tupleDesc->natts;
+
+	Assert(numAttrs <= MaxHeapAttributeNumber);
+	heap_deform_tuple(newtup, tupleDesc, toast_values, toast_isnull);
+	if (oldtup != NULL)
+		heap_deform_tuple(oldtup, tupleDesc, toast_oldvalues, toast_oldisnull);
+
+	/* ----------
+	 * Then collect information about the values given
+	 *
+	 * NOTE: toast_action[i] can have these values:
+	 *		' '		default handling
+	 *		'p'		already processed --- don't touch it
+	 *		'x'		incompressible, but OK to move off
+	 *
+	 * NOTE: toast_sizes[i] is only made valid for varlena attributes with
+	 *		toast_action[i] different from 'p'.
+	 * ----------
+	 */
+	memset(toast_action, ' ', numAttrs * sizeof(char));
+	memset(toast_oldexternal, 0, numAttrs * sizeof(struct varlena *));
+	memset(toast_free, 0, numAttrs * sizeof(bool));
+	memset(toast_delold, 0, numAttrs * sizeof(bool));
+
+	for (i = 0; i < numAttrs; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(tupleDesc, i);
+		struct varlena *old_value;
+		struct varlena *new_value;
+
+		if (oldtup != NULL)
+		{
+			/*
+			 * For UPDATE get the old and new values of this attribute
+			 */
+			old_value = (struct varlena *) DatumGetPointer(toast_oldvalues[i]);
+			new_value = (struct varlena *) DatumGetPointer(toast_values[i]);
+
+			/*
+			 * If the old value is stored on disk, check if it has changed so
+			 * we have to delete it later.
+			 */
+			if (att->attlen == -1 && !toast_oldisnull[i] &&
+				VARATT_IS_EXTERNAL_ONDISK(old_value))
+			{
+				if (toast_isnull[i] || !VARATT_IS_EXTERNAL_ONDISK(new_value) ||
+					memcmp((char *) old_value, (char *) new_value,
+						   VARSIZE_EXTERNAL(old_value)) != 0)
+				{
+					/*
+					 * The old external stored value isn't needed any more
+					 * after the update
+					 */
+					toast_delold[i] = true;
+					need_delold = true;
+				}
+				else
+				{
+					/*
+					 * This attribute isn't changed by this update so we reuse
+					 * the original reference to the old value in the new
+					 * tuple.
+					 */
+					toast_action[i] = 'p';
+					continue;
+				}
+			}
+		}
+		else
+		{
+			/*
+			 * For INSERT simply get the new value
+			 */
+			new_value = (struct varlena *) DatumGetPointer(toast_values[i]);
+		}
+
+		/*
+		 * Handle NULL attributes
+		 */
+		if (toast_isnull[i])
+		{
+			toast_action[i] = 'p';
+			has_nulls = true;
+			continue;
+		}
+
+		/*
+		 * Now look at varlena attributes
+		 */
+		if (att->attlen == -1)
+		{
+			/*
+			 * If the table's attribute says PLAIN always, force it so.
+			 */
+			if (att->attstorage == 'p')
+				toast_action[i] = 'p';
+
+			/*
+			 * We took care of UPDATE above, so any external value we find
+			 * still in the tuple must be someone else's that we cannot reuse
+			 * (this includes the case of an out-of-line in-memory datum).
+			 * Fetch it back (without decompression, unless we are forcing
+			 * PLAIN storage).  If necessary, we'll push it out as a new
+			 * external value below.
+			 */
+			if (VARATT_IS_EXTERNAL(new_value))
+			{
+				toast_oldexternal[i] = new_value;
+				if (att->attstorage == 'p')
+					new_value = heap_tuple_untoast_attr(new_value);
+				else
+					new_value = heap_tuple_fetch_attr(new_value);
+				toast_values[i] = PointerGetDatum(new_value);
+				toast_free[i] = true;
+				need_change = true;
+				need_free = true;
+			}
+
+			/*
+			 * Remember the size of this attribute
+			 */
+			toast_sizes[i] = VARSIZE_ANY(new_value);
+		}
+		else
+		{
+			/*
+			 * Not a varlena attribute, plain storage always
+			 */
+			toast_action[i] = 'p';
+		}
+	}
+
+	/* ----------
+	 * Compress and/or save external until data fits into target length
+	 *
+	 *	1: Inline compress attributes with attstorage 'x', and store very
+	 *	   large attributes with attstorage 'x' or 'e' external immediately
+	 *	2: Store attributes with attstorage 'x' or 'e' external
+	 *	3: Inline compress attributes with attstorage 'm'
+	 *	4: Store attributes with attstorage 'm' external
+	 * ----------
+	 */
+
+	/* compute header overhead --- this should match heap_form_tuple() */
+	hoff = SizeofHeapTupleHeader;
+	if (has_nulls)
+		hoff += BITMAPLEN(numAttrs);
+	hoff = MAXALIGN(hoff);
+	/* now convert to a limit on the tuple data size */
+	maxDataLen = RelationGetToastTupleTarget(rel, TOAST_TUPLE_TARGET) - hoff;
+
+	/*
+	 * Look for attributes with attstorage 'x' to compress.  Also find large
+	 * attributes with attstorage 'x' or 'e', and store them external.
+	 */
+	while (heap_compute_data_size(tupleDesc,
+								  toast_values, toast_isnull) > maxDataLen)
+	{
+		int			biggest_attno = -1;
+		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
+		Datum		old_value;
+		Datum		new_value;
+
+		/*
+		 * Search for the biggest yet unprocessed internal attribute
+		 */
+		for (i = 0; i < numAttrs; i++)
+		{
+			Form_pg_attribute att = TupleDescAttr(tupleDesc, i);
+
+			if (toast_action[i] != ' ')
+				continue;
+			if (VARATT_IS_EXTERNAL(DatumGetPointer(toast_values[i])))
+				continue;		/* can't happen, toast_action would be 'p' */
+			if (VARATT_IS_COMPRESSED(DatumGetPointer(toast_values[i])))
+				continue;
+			if (att->attstorage != 'x' && att->attstorage != 'e')
+				continue;
+			if (toast_sizes[i] > biggest_size)
+			{
+				biggest_attno = i;
+				biggest_size = toast_sizes[i];
+			}
+		}
+
+		if (biggest_attno < 0)
+			break;
+
+		/*
+		 * Attempt to compress it inline, if it has attstorage 'x'
+		 */
+		i = biggest_attno;
+		if (TupleDescAttr(tupleDesc, i)->attstorage == 'x')
+		{
+			old_value = toast_values[i];
+			new_value = toast_compress_datum(old_value);
+
+			if (DatumGetPointer(new_value) != NULL)
+			{
+				/* successful compression */
+				if (toast_free[i])
+					pfree(DatumGetPointer(old_value));
+				toast_values[i] = new_value;
+				toast_free[i] = true;
+				toast_sizes[i] = VARSIZE(DatumGetPointer(toast_values[i]));
+				need_change = true;
+				need_free = true;
+			}
+			else
+			{
+				/* incompressible, ignore on subsequent compression passes */
+				toast_action[i] = 'x';
+			}
+		}
+		else
+		{
+			/* has attstorage 'e', ignore on subsequent compression passes */
+			toast_action[i] = 'x';
+		}
+
+		/*
+		 * If this value is by itself more than maxDataLen (after compression
+		 * if any), push it out to the toast table immediately, if possible.
+		 * This avoids uselessly compressing other fields in the common case
+		 * where we have one long field and several short ones.
+		 *
+		 * XXX maybe the threshold should be less than maxDataLen?
+		 */
+		if (toast_sizes[i] > maxDataLen &&
+			rel->rd_rel->reltoastrelid != InvalidOid)
+		{
+			old_value = toast_values[i];
+			toast_action[i] = 'p';
+			toast_values[i] = toast_save_datum(rel, toast_values[i],
+											   toast_oldexternal[i], options);
+			if (toast_free[i])
+				pfree(DatumGetPointer(old_value));
+			toast_free[i] = true;
+			need_change = true;
+			need_free = true;
+		}
+	}
+
+	/*
+	 * Second we look for attributes of attstorage 'x' or 'e' that are still
+	 * inline.  But skip this if there's no toast table to push them to.
+	 */
+	while (heap_compute_data_size(tupleDesc,
+								  toast_values, toast_isnull) > maxDataLen &&
+		   rel->rd_rel->reltoastrelid != InvalidOid)
+	{
+		int			biggest_attno = -1;
+		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
+		Datum		old_value;
+
+		/*------
+		 * Search for the biggest yet inlined attribute with
+		 * attstorage equals 'x' or 'e'
+		 *------
+		 */
+		for (i = 0; i < numAttrs; i++)
+		{
+			Form_pg_attribute att = TupleDescAttr(tupleDesc, i);
+
+			if (toast_action[i] == 'p')
+				continue;
+			if (VARATT_IS_EXTERNAL(DatumGetPointer(toast_values[i])))
+				continue;		/* can't happen, toast_action would be 'p' */
+			if (att->attstorage != 'x' && att->attstorage != 'e')
+				continue;
+			if (toast_sizes[i] > biggest_size)
+			{
+				biggest_attno = i;
+				biggest_size = toast_sizes[i];
+			}
+		}
+
+		if (biggest_attno < 0)
+			break;
+
+		/*
+		 * Store this external
+		 */
+		i = biggest_attno;
+		old_value = toast_values[i];
+		toast_action[i] = 'p';
+		toast_values[i] = toast_save_datum(rel, toast_values[i],
+										   toast_oldexternal[i], options);
+		if (toast_free[i])
+			pfree(DatumGetPointer(old_value));
+		toast_free[i] = true;
+
+		need_change = true;
+		need_free = true;
+	}
+
+	/*
+	 * Round 3 - this time we take attributes with storage 'm' into
+	 * compression
+	 */
+	while (heap_compute_data_size(tupleDesc,
+								  toast_values, toast_isnull) > maxDataLen)
+	{
+		int			biggest_attno = -1;
+		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
+		Datum		old_value;
+		Datum		new_value;
+
+		/*
+		 * Search for the biggest yet uncompressed internal attribute
+		 */
+		for (i = 0; i < numAttrs; i++)
+		{
+			if (toast_action[i] != ' ')
+				continue;
+			if (VARATT_IS_EXTERNAL(DatumGetPointer(toast_values[i])))
+				continue;		/* can't happen, toast_action would be 'p' */
+			if (VARATT_IS_COMPRESSED(DatumGetPointer(toast_values[i])))
+				continue;
+			if (TupleDescAttr(tupleDesc, i)->attstorage != 'm')
+				continue;
+			if (toast_sizes[i] > biggest_size)
+			{
+				biggest_attno = i;
+				biggest_size = toast_sizes[i];
+			}
+		}
+
+		if (biggest_attno < 0)
+			break;
+
+		/*
+		 * Attempt to compress it inline
+		 */
+		i = biggest_attno;
+		old_value = toast_values[i];
+		new_value = toast_compress_datum(old_value);
+
+		if (DatumGetPointer(new_value) != NULL)
+		{
+			/* successful compression */
+			if (toast_free[i])
+				pfree(DatumGetPointer(old_value));
+			toast_values[i] = new_value;
+			toast_free[i] = true;
+			toast_sizes[i] = VARSIZE(DatumGetPointer(toast_values[i]));
+			need_change = true;
+			need_free = true;
+		}
+		else
+		{
+			/* incompressible, ignore on subsequent compression passes */
+			toast_action[i] = 'x';
+		}
+	}
+
+	/*
+	 * Finally we store attributes of type 'm' externally.  At this point we
+	 * increase the target tuple size, so that 'm' attributes aren't stored
+	 * externally unless really necessary.
+	 */
+	maxDataLen = TOAST_TUPLE_TARGET_MAIN - hoff;
+
+	while (heap_compute_data_size(tupleDesc,
+								  toast_values, toast_isnull) > maxDataLen &&
+		   rel->rd_rel->reltoastrelid != InvalidOid)
+	{
+		int			biggest_attno = -1;
+		int32		biggest_size = MAXALIGN(TOAST_POINTER_SIZE);
+		Datum		old_value;
+
+		/*--------
+		 * Search for the biggest yet inlined attribute with
+		 * attstorage = 'm'
+		 *--------
+		 */
+		for (i = 0; i < numAttrs; i++)
+		{
+			if (toast_action[i] == 'p')
+				continue;
+			if (VARATT_IS_EXTERNAL(DatumGetPointer(toast_values[i])))
+				continue;		/* can't happen, toast_action would be 'p' */
+			if (TupleDescAttr(tupleDesc, i)->attstorage != 'm')
+				continue;
+			if (toast_sizes[i] > biggest_size)
+			{
+				biggest_attno = i;
+				biggest_size = toast_sizes[i];
+			}
+		}
+
+		if (biggest_attno < 0)
+			break;
+
+		/*
+		 * Store this external
+		 */
+		i = biggest_attno;
+		old_value = toast_values[i];
+		toast_action[i] = 'p';
+		toast_values[i] = toast_save_datum(rel, toast_values[i],
+										   toast_oldexternal[i], options);
+		if (toast_free[i])
+			pfree(DatumGetPointer(old_value));
+		toast_free[i] = true;
+
+		need_change = true;
+		need_free = true;
+	}
+
+	/*
+	 * In the case we toasted any values, we need to build a new heap tuple
+	 * with the changed values.
+	 */
+	if (need_change)
+	{
+		HeapTupleHeader olddata = newtup->t_data;
+		HeapTupleHeader new_data;
+		int32		new_header_len;
+		int32		new_data_len;
+		int32		new_tuple_len;
+
+		/*
+		 * Calculate the new size of the tuple.
+		 *
+		 * Note: we used to assume here that the old tuple's t_hoff must equal
+		 * the new_header_len value, but that was incorrect.  The old tuple
+		 * might have a smaller-than-current natts, if there's been an ALTER
+		 * TABLE ADD COLUMN since it was stored; and that would lead to a
+		 * different conclusion about the size of the null bitmap, or even
+		 * whether there needs to be one at all.
+		 */
+		new_header_len = SizeofHeapTupleHeader;
+		if (has_nulls)
+			new_header_len += BITMAPLEN(numAttrs);
+		new_header_len = MAXALIGN(new_header_len);
+		new_data_len = heap_compute_data_size(tupleDesc,
+											  toast_values, toast_isnull);
+		new_tuple_len = new_header_len + new_data_len;
+
+		/*
+		 * Allocate and zero the space needed, and fill HeapTupleData fields.
+		 */
+		result_tuple = (HeapTuple) palloc0(HEAPTUPLESIZE + new_tuple_len);
+		result_tuple->t_len = new_tuple_len;
+		result_tuple->t_self = newtup->t_self;
+		result_tuple->t_tableOid = newtup->t_tableOid;
+		new_data = (HeapTupleHeader) ((char *) result_tuple + HEAPTUPLESIZE);
+		result_tuple->t_data = new_data;
+
+		/*
+		 * Copy the existing tuple header, but adjust natts and t_hoff.
+		 */
+		memcpy(new_data, olddata, SizeofHeapTupleHeader);
+		HeapTupleHeaderSetNatts(new_data, numAttrs);
+		new_data->t_hoff = new_header_len;
+
+		/* Copy over the data, and fill the null bitmap if needed */
+		heap_fill_tuple(tupleDesc,
+						toast_values,
+						toast_isnull,
+						(char *) new_data + new_header_len,
+						new_data_len,
+						&(new_data->t_infomask),
+						has_nulls ? new_data->t_bits : NULL);
+	}
+	else
+		result_tuple = newtup;
+
+	/*
+	 * Free allocated temp values
+	 */
+	if (need_free)
+		for (i = 0; i < numAttrs; i++)
+			if (toast_free[i])
+				pfree(DatumGetPointer(toast_values[i]));
+
+	/*
+	 * Delete external values from the old tuple
+	 */
+	if (need_delold)
+		for (i = 0; i < numAttrs; i++)
+			if (toast_delold[i])
+				toast_delete_datum(rel, toast_oldvalues[i], false);
+
+	return result_tuple;
+}
+
+
+/* ----------
+ * toast_flatten_tuple -
+ *
+ *	"Flatten" a tuple to contain no out-of-line toasted fields.
+ *	(This does not eliminate compressed or short-header datums.)
+ *
+ *	Note: we expect the caller already checked HeapTupleHasExternal(tup),
+ *	so there is no need for a short-circuit path.
+ * ----------
+ */
+HeapTuple
+toast_flatten_tuple(HeapTuple tup, TupleDesc tupleDesc)
+{
+	HeapTuple	new_tuple;
+	int			numAttrs = tupleDesc->natts;
+	int			i;
+	Datum		toast_values[MaxTupleAttributeNumber];
+	bool		toast_isnull[MaxTupleAttributeNumber];
+	bool		toast_free[MaxTupleAttributeNumber];
+
+	/*
+	 * Break down the tuple into fields.
+	 */
+	Assert(numAttrs <= MaxTupleAttributeNumber);
+	heap_deform_tuple(tup, tupleDesc, toast_values, toast_isnull);
+
+	memset(toast_free, 0, numAttrs * sizeof(bool));
+
+	for (i = 0; i < numAttrs; i++)
+	{
+		/*
+		 * Look at non-null varlena attributes
+		 */
+		if (!toast_isnull[i] && TupleDescAttr(tupleDesc, i)->attlen == -1)
+		{
+			struct varlena *new_value;
+
+			new_value = (struct varlena *) DatumGetPointer(toast_values[i]);
+			if (VARATT_IS_EXTERNAL(new_value))
+			{
+				new_value = heap_tuple_fetch_attr(new_value);
+				toast_values[i] = PointerGetDatum(new_value);
+				toast_free[i] = true;
+			}
+		}
+	}
+
+	/*
+	 * Form the reconfigured tuple.
+	 */
+	new_tuple = heap_form_tuple(tupleDesc, toast_values, toast_isnull);
+
+	/*
+	 * Be sure to copy the tuple's identity fields.  We also make a point of
+	 * copying visibility info, just in case anybody looks at those fields in
+	 * a syscache entry.
+	 */
+	new_tuple->t_self = tup->t_self;
+	new_tuple->t_tableOid = tup->t_tableOid;
+
+	new_tuple->t_data->t_choice = tup->t_data->t_choice;
+	new_tuple->t_data->t_ctid = tup->t_data->t_ctid;
+	new_tuple->t_data->t_infomask &= ~HEAP_XACT_MASK;
+	new_tuple->t_data->t_infomask |=
+		tup->t_data->t_infomask & HEAP_XACT_MASK;
+	new_tuple->t_data->t_infomask2 &= ~HEAP2_XACT_MASK;
+	new_tuple->t_data->t_infomask2 |=
+		tup->t_data->t_infomask2 & HEAP2_XACT_MASK;
+
+	/*
+	 * Free allocated temp values
+	 */
+	for (i = 0; i < numAttrs; i++)
+		if (toast_free[i])
+			pfree(DatumGetPointer(toast_values[i]));
+
+	return new_tuple;
+}
+
+
+/* ----------
+ * toast_flatten_tuple_to_datum -
+ *
+ *	"Flatten" a tuple containing out-of-line toasted fields into a Datum.
+ *	The result is always palloc'd in the current memory context.
+ *
+ *	We have a general rule that Datums of container types (rows, arrays,
+ *	ranges, etc) must not contain any external TOAST pointers.  Without
+ *	this rule, we'd have to look inside each Datum when preparing a tuple
+ *	for storage, which would be expensive and would fail to extend cleanly
+ *	to new sorts of container types.
+ *
+ *	However, we don't want to say that tuples represented as HeapTuples
+ *	can't contain toasted fields, so instead this routine should be called
+ *	when such a HeapTuple is being converted into a Datum.
+ *
+ *	While we're at it, we decompress any compressed fields too.  This is not
+ *	necessary for correctness, but reflects an expectation that compression
+ *	will be more effective if applied to the whole tuple not individual
+ *	fields.  We are not so concerned about that that we want to deconstruct
+ *	and reconstruct tuples just to get rid of compressed fields, however.
+ *	So callers typically won't call this unless they see that the tuple has
+ *	at least one external field.
+ *
+ *	On the other hand, in-line short-header varlena fields are left alone.
+ *	If we "untoasted" them here, they'd just get changed back to short-header
+ *	format anyway within heap_fill_tuple.
+ * ----------
+ */
+Datum
+toast_flatten_tuple_to_datum(HeapTupleHeader tup,
+							 uint32 tup_len,
+							 TupleDesc tupleDesc)
+{
+	HeapTupleHeader new_data;
+	int32		new_header_len;
+	int32		new_data_len;
+	int32		new_tuple_len;
+	HeapTupleData tmptup;
+	int			numAttrs = tupleDesc->natts;
+	int			i;
+	bool		has_nulls = false;
+	Datum		toast_values[MaxTupleAttributeNumber];
+	bool		toast_isnull[MaxTupleAttributeNumber];
+	bool		toast_free[MaxTupleAttributeNumber];
+
+	/* Build a temporary HeapTuple control structure */
+	tmptup.t_len = tup_len;
+	ItemPointerSetInvalid(&(tmptup.t_self));
+	tmptup.t_tableOid = InvalidOid;
+	tmptup.t_data = tup;
+
+	/*
+	 * Break down the tuple into fields.
+	 */
+	Assert(numAttrs <= MaxTupleAttributeNumber);
+	heap_deform_tuple(&tmptup, tupleDesc, toast_values, toast_isnull);
+
+	memset(toast_free, 0, numAttrs * sizeof(bool));
+
+	for (i = 0; i < numAttrs; i++)
+	{
+		/*
+		 * Look at non-null varlena attributes
+		 */
+		if (toast_isnull[i])
+			has_nulls = true;
+		else if (TupleDescAttr(tupleDesc, i)->attlen == -1)
+		{
+			struct varlena *new_value;
+
+			new_value = (struct varlena *) DatumGetPointer(toast_values[i]);
+			if (VARATT_IS_EXTERNAL(new_value) ||
+				VARATT_IS_COMPRESSED(new_value))
+			{
+				new_value = heap_tuple_untoast_attr(new_value);
+				toast_values[i] = PointerGetDatum(new_value);
+				toast_free[i] = true;
+			}
+		}
+	}
+
+	/*
+	 * Calculate the new size of the tuple.
+	 *
+	 * This should match the reconstruction code in toast_insert_or_update.
+	 */
+	new_header_len = SizeofHeapTupleHeader;
+	if (has_nulls)
+		new_header_len += BITMAPLEN(numAttrs);
+	new_header_len = MAXALIGN(new_header_len);
+	new_data_len = heap_compute_data_size(tupleDesc,
+										  toast_values, toast_isnull);
+	new_tuple_len = new_header_len + new_data_len;
+
+	new_data = (HeapTupleHeader) palloc0(new_tuple_len);
+
+	/*
+	 * Copy the existing tuple header, but adjust natts and t_hoff.
+	 */
+	memcpy(new_data, tup, SizeofHeapTupleHeader);
+	HeapTupleHeaderSetNatts(new_data, numAttrs);
+	new_data->t_hoff = new_header_len;
+
+	/* Set the composite-Datum header fields correctly */
+	HeapTupleHeaderSetDatumLength(new_data, new_tuple_len);
+	HeapTupleHeaderSetTypeId(new_data, tupleDesc->tdtypeid);
+	HeapTupleHeaderSetTypMod(new_data, tupleDesc->tdtypmod);
+
+	/* Copy over the data, and fill the null bitmap if needed */
+	heap_fill_tuple(tupleDesc,
+					toast_values,
+					toast_isnull,
+					(char *) new_data + new_header_len,
+					new_data_len,
+					&(new_data->t_infomask),
+					has_nulls ? new_data->t_bits : NULL);
+
+	/*
+	 * Free allocated temp values
+	 */
+	for (i = 0; i < numAttrs; i++)
+		if (toast_free[i])
+			pfree(DatumGetPointer(toast_values[i]));
+
+	return PointerGetDatum(new_data);
+}
+
+
+/* ----------
+ * toast_build_flattened_tuple -
+ *
+ *	Build a tuple containing no out-of-line toasted fields.
+ *	(This does not eliminate compressed or short-header datums.)
+ *
+ *	This is essentially just like heap_form_tuple, except that it will
+ *	expand any external-data pointers beforehand.
+ *
+ *	It's not very clear whether it would be preferable to decompress
+ *	in-line compressed datums while at it.  For now, we don't.
+ * ----------
+ */
+HeapTuple
+toast_build_flattened_tuple(TupleDesc tupleDesc,
+							Datum *values,
+							bool *isnull)
+{
+	HeapTuple	new_tuple;
+	int			numAttrs = tupleDesc->natts;
+	int			num_to_free;
+	int			i;
+	Datum		new_values[MaxTupleAttributeNumber];
+	Pointer		freeable_values[MaxTupleAttributeNumber];
+
+	/*
+	 * We can pass the caller's isnull array directly to heap_form_tuple, but
+	 * we potentially need to modify the values array.
+	 */
+	Assert(numAttrs <= MaxTupleAttributeNumber);
+	memcpy(new_values, values, numAttrs * sizeof(Datum));
+
+	num_to_free = 0;
+	for (i = 0; i < numAttrs; i++)
+	{
+		/*
+		 * Look at non-null varlena attributes
+		 */
+		if (!isnull[i] && TupleDescAttr(tupleDesc, i)->attlen == -1)
+		{
+			struct varlena *new_value;
+
+			new_value = (struct varlena *) DatumGetPointer(new_values[i]);
+			if (VARATT_IS_EXTERNAL(new_value))
+			{
+				new_value = heap_tuple_fetch_attr(new_value);
+				new_values[i] = PointerGetDatum(new_value);
+				freeable_values[num_to_free++] = (Pointer) new_value;
+			}
+		}
+	}
+
+	/*
+	 * Form the reconfigured tuple.
+	 */
+	new_tuple = heap_form_tuple(tupleDesc, new_values, isnull);
+
+	/*
+	 * Free allocated temp values
+	 */
+	for (i = 0; i < num_to_free; i++)
+		pfree(freeable_values[i]);
+
+	return new_tuple;
+}