3 files changed, 92 insertions, 85 deletions

diff --git a/src/backend/access/common/heaptuple.c b/src/backend/access/common/heaptuple.c
index aea9d40d149..c64ede9dac5 100644
--- a/src/backend/access/common/heaptuple.c
+++ b/src/backend/access/common/heaptuple.c
@@ -617,6 +617,41 @@ heap_copytuple_with_tuple(HeapTuple src, HeapTuple dest)
 	memcpy((char *) dest->t_data, (char *) src->t_data, src->t_len);
 }
 
+/* ----------------
+ *		heap_copy_tuple_as_datum
+ *
+ *		copy a tuple as a composite-type Datum
+ * ----------------
+ */
+Datum
+heap_copy_tuple_as_datum(HeapTuple tuple, TupleDesc tupleDesc)
+{
+	HeapTupleHeader td;
+
+	/*
+	 * If the tuple contains any external TOAST pointers, we have to inline
+	 * those fields to meet the conventions for composite-type Datums.
+	 */
+	if (HeapTupleHasExternal(tuple))
+		return toast_flatten_tuple_to_datum(tuple->t_data,
+											tuple->t_len,
+											tupleDesc);
+
+	/*
+	 * Fast path for easy case: just make a palloc'd copy and insert the
+	 * correct composite-Datum header fields (since those may not be set if
+	 * the given tuple came from disk, rather than from heap_form_tuple).
+	 */
+	td = (HeapTupleHeader) palloc(tuple->t_len);
+	memcpy((char *) td, (char *) tuple->t_data, tuple->t_len);
+
+	HeapTupleHeaderSetDatumLength(td, tuple->t_len);
+	HeapTupleHeaderSetTypeId(td, tupleDesc->tdtypeid);
+	HeapTupleHeaderSetTypMod(td, tupleDesc->tdtypmod);
+
+	return PointerGetDatum(td);
+}
+
 /*
  * heap_form_tuple
  *		construct a tuple from the given values[] and isnull[] arrays,
@@ -635,7 +670,6 @@ heap_form_tuple(TupleDesc tupleDescriptor,
 				data_len;
 	int			hoff;
 	bool		hasnull = false;
-	Form_pg_attribute *att = tupleDescriptor->attrs;
 	int			numberOfAttributes = tupleDescriptor->natts;
 	int			i;
 
@@ -646,28 +680,14 @@ heap_form_tuple(TupleDesc tupleDescriptor,
 						numberOfAttributes, MaxTupleAttributeNumber)));
 
 	/*
-	 * Check for nulls and embedded tuples; expand any toasted attributes in
-	 * embedded tuples.  This preserves the invariant that toasting can only
-	 * go one level deep.
-	 *
-	 * We can skip calling toast_flatten_tuple_attribute() if the attribute
-	 * couldn't possibly be of composite type.  All composite datums are
-	 * varlena and have alignment 'd'; furthermore they aren't arrays. Also,
-	 * if an attribute is already toasted, it must have been sent to disk
-	 * already and so cannot contain toasted attributes.
+	 * Check for nulls
 	 */
 	for (i = 0; i < numberOfAttributes; i++)
 	{
 		if (isnull[i])
-			hasnull = true;
-		else if (att[i]->attlen == -1 &&
-				 att[i]->attalign == 'd' &&
-				 att[i]->attndims == 0 &&
-				 !VARATT_IS_EXTENDED(DatumGetPointer(values[i])))
 		{
-			values[i] = toast_flatten_tuple_attribute(values[i],
-													  att[i]->atttypid,
-													  att[i]->atttypmod);
+			hasnull = true;
+			break;
 		}
 	}
 
@@ -697,7 +717,8 @@ heap_form_tuple(TupleDesc tupleDescriptor,
 
 	/*
 	 * And fill in the information.  Note we fill the Datum fields even though
-	 * this tuple may never become a Datum.
+	 * this tuple may never become a Datum.  This lets HeapTupleHeaderGetDatum
+	 * identify the tuple type if needed.
 	 */
 	tuple->t_len = len;
 	ItemPointerSetInvalid(&(tuple->t_self));
@@ -1389,7 +1410,6 @@ heap_form_minimal_tuple(TupleDesc tupleDescriptor,
 				data_len;
 	int			hoff;
 	bool		hasnull = false;
-	Form_pg_attribute *att = tupleDescriptor->attrs;
 	int			numberOfAttributes = tupleDescriptor->natts;
 	int			i;
 
@@ -1400,28 +1420,14 @@ heap_form_minimal_tuple(TupleDesc tupleDescriptor,
 						numberOfAttributes, MaxTupleAttributeNumber)));
 
 	/*
-	 * Check for nulls and embedded tuples; expand any toasted attributes in
-	 * embedded tuples.  This preserves the invariant that toasting can only
-	 * go one level deep.
-	 *
-	 * We can skip calling toast_flatten_tuple_attribute() if the attribute
-	 * couldn't possibly be of composite type.  All composite datums are
-	 * varlena and have alignment 'd'; furthermore they aren't arrays. Also,
-	 * if an attribute is already toasted, it must have been sent to disk
-	 * already and so cannot contain toasted attributes.
+	 * Check for nulls
 	 */
 	for (i = 0; i < numberOfAttributes; i++)
 	{
 		if (isnull[i])
-			hasnull = true;
-		else if (att[i]->attlen == -1 &&
-				 att[i]->attalign == 'd' &&
-				 att[i]->attndims == 0 &&
-				 !VARATT_IS_EXTENDED(values[i]))
 		{
-			values[i] = toast_flatten_tuple_attribute(values[i],
-													  att[i]->atttypid,
-													  att[i]->atttypmod);
+			hasnull = true;
+			break;
 		}
 	}
 
diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index b4c68e9fe27..7da10e9a74a 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -158,6 +158,11 @@ 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.
diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c
index 9a821d3e1cf..dde74d47978 100644
--- a/src/backend/access/heap/tuptoaster.c
+++ b/src/backend/access/heap/tuptoaster.c
@@ -991,6 +991,9 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
  *
  *	"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
@@ -1068,59 +1071,61 @@ toast_flatten_tuple(HeapTuple tup, TupleDesc tupleDesc)
 
 
 /* ----------
- * toast_flatten_tuple_attribute -
+ * 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.
  *
- *	If a Datum is of composite type, "flatten" it to contain no toasted fields.
- *	This must be invoked on any potentially-composite field that is to be
- *	inserted into a tuple.	Doing this preserves the invariant that toasting
- *	goes only one level deep in a tuple.
+ *	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.
  *
- *	Note that flattening does not mean expansion of short-header varlenas,
- *	so in one sense toasting is allowed within composite datums.
+ *	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_attribute(Datum value,
-							  Oid typeId, int32 typeMod)
+toast_flatten_tuple_to_datum(HeapTupleHeader tup,
+							 uint32 tup_len,
+							 TupleDesc tupleDesc)
 {
-	TupleDesc	tupleDesc;
-	HeapTupleHeader olddata;
 	HeapTupleHeader new_data;
 	int32		new_header_len;
 	int32		new_data_len;
 	int32		new_tuple_len;
 	HeapTupleData tmptup;
-	Form_pg_attribute *att;
-	int			numAttrs;
+	Form_pg_attribute *att = tupleDesc->attrs;
+	int			numAttrs = tupleDesc->natts;
 	int			i;
-	bool		need_change = false;
 	bool		has_nulls = false;
 	Datum		toast_values[MaxTupleAttributeNumber];
 	bool		toast_isnull[MaxTupleAttributeNumber];
 	bool		toast_free[MaxTupleAttributeNumber];
 
-	/*
-	 * See if it's a composite type, and get the tupdesc if so.
-	 */
-	tupleDesc = lookup_rowtype_tupdesc_noerror(typeId, typeMod, true);
-	if (tupleDesc == NULL)
-		return value;			/* not a composite type */
-
-	att = tupleDesc->attrs;
-	numAttrs = tupleDesc->natts;
-
-	/*
-	 * Break down the tuple into fields.
-	 */
-	olddata = DatumGetHeapTupleHeader(value);
-	Assert(typeId == HeapTupleHeaderGetTypeId(olddata));
-	Assert(typeMod == HeapTupleHeaderGetTypMod(olddata));
 	/* Build a temporary HeapTuple control structure */
-	tmptup.t_len = HeapTupleHeaderGetDatumLength(olddata);
+	tmptup.t_len = tup_len;
 	ItemPointerSetInvalid(&(tmptup.t_self));
 	tmptup.t_tableOid = InvalidOid;
-	tmptup.t_data = olddata;
+	tmptup.t_data = tup;
 
+	/*
+	 * Break down the tuple into fields.
+	 */
 	Assert(numAttrs <= MaxTupleAttributeNumber);
 	heap_deform_tuple(&tmptup, tupleDesc, toast_values, toast_isnull);
 
@@ -1144,21 +1149,11 @@ toast_flatten_tuple_attribute(Datum value,
 				new_value = heap_tuple_untoast_attr(new_value);
 				toast_values[i] = PointerGetDatum(new_value);
 				toast_free[i] = true;
-				need_change = true;
 			}
 		}
 	}
 
 	/*
-	 * If nothing to untoast, just return the original tuple.
-	 */
-	if (!need_change)
-	{
-		ReleaseTupleDesc(tupleDesc);
-		return value;
-	}
-
-	/*
 	 * Calculate the new size of the tuple.
 	 *
 	 * This should match the reconstruction code in toast_insert_or_update.
@@ -1166,7 +1161,7 @@ toast_flatten_tuple_attribute(Datum value,
 	new_header_len = offsetof(HeapTupleHeaderData, t_bits);
 	if (has_nulls)
 		new_header_len += BITMAPLEN(numAttrs);
-	if (olddata->t_infomask & HEAP_HASOID)
+	if (tup->t_infomask & HEAP_HASOID)
 		new_header_len += sizeof(Oid);
 	new_header_len = MAXALIGN(new_header_len);
 	new_data_len = heap_compute_data_size(tupleDesc,
@@ -1178,14 +1173,16 @@ toast_flatten_tuple_attribute(Datum value,
 	/*
 	 * Copy the existing tuple header, but adjust natts and t_hoff.
 	 */
-	memcpy(new_data, olddata, offsetof(HeapTupleHeaderData, t_bits));
+	memcpy(new_data, tup, offsetof(HeapTupleHeaderData, t_bits));
 	HeapTupleHeaderSetNatts(new_data, numAttrs);
 	new_data->t_hoff = new_header_len;
-	if (olddata->t_infomask & HEAP_HASOID)
-		HeapTupleHeaderSetOid(new_data, HeapTupleHeaderGetOid(olddata));
+	if (tup->t_infomask & HEAP_HASOID)
+		HeapTupleHeaderSetOid(new_data, HeapTupleHeaderGetOid(tup));
 
-	/* Reset the datum length field, too */
+	/* 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,
@@ -1202,7 +1199,6 @@ toast_flatten_tuple_attribute(Datum value,
 	for (i = 0; i < numAttrs; i++)
 		if (toast_free[i])
 			pfree(DatumGetPointer(toast_values[i]));
-	ReleaseTupleDesc(tupleDesc);
 
 	return PointerGetDatum(new_data);
 }