Re: Composite Datums containing toasted fields are a bad idea(?)

I wrote:
> I'm actually planning to set this patch on the shelf for a bit and go
> investigate the other alternative, ie, not generating composite Datums
> containing toast pointers in the first place.

Here's a draft patch along those lines.  It turned out to be best to
leave heap_form_tuple() alone and instead put the dirty work into
HeapTupleGetDatum().  That has some consequences worth discussing:

* The patch changes HeapTupleGetDatum from a simple inline macro into
a function call.  This means that third-party extensions will not get
protection against creation of toast-pointer-containing composite Datums
until they recompile.  I'm not sure that this is a big deal, though.
After looking through the existing core and contrib code, it seems that
nearly all users of HeapTupleGetDatum are building their tuples from
locally-sourced data that could not possibly be toasted anyway.  (This is
true a-priori for users of BuildTupleFromCStrings, for example, and seems
to be true of all uses of HeapTupleGetDatum in SQL-callable functions.)
So it's entirely possible that there would be no live bug anywhere even
without recompiles.

* If we were sufficiently worried about the previous point, we could
get some protection against unfixed extension code by not removing
toast_flatten_tuple_attribute() in the back branches, only in HEAD.
I'm doubtful that it's worth the cycles though.

* Because HeapTupleGetDatum might allocate a new tuple, the wrong thing
might happen if the caller changes CurrentMemoryContext between
heap_form_tuple and HeapTupleGetDatum.  I could only find two places
that did that, though, both in plpgsql.  I thought about having
HeapTupleGetDatum try to identify the context the passed tuple had been
allocated in, but the problem with that is the passed tuple isn't
necessarily heap-allocated at all --- in fact, one of the two problematic
places in plpgsql passes a pointer to a stack-local variable, so we'd
actually crash if we tried to apply GetMemoryChunkContext() to it.
Of course we can (and I did) change plpgsql, but the question is whether
any third-party code has copied either coding pattern.  On balance it
seems best to just use palloc; that's unlikely to cause anything worse
than a memory leak.

* I was quite pleased with the size of the patch: under 100 net new lines,
half of that being a new regression test case.  And it's worth emphasizing
that this is a *complete* fix, modulo the question of third-party code
recompiles.  The patch I showed a few days ago added several times that
much just to fix arrays of composites, and I wasn't too confident that it
fixed every case even for arrays.

* The cases where an "extra" detoast would be incurred are pretty narrow:
basically, whole-row-Var references, ROW() constructs, and the outputs of
functions returning tuples.  As discussed earlier, whether this would cost
anything or save anything would depend on the number of subsequent uses of
the composite Datum's previously-toasted fields.  But I'm thinking that
the amount of application code that would actually be impacted in either
direction is probably pretty small.  Moreover, we aren't adding any
noticeable overhead in cases where no detoasting occurs, unlike the
situation with the previous patch.  (In fact, we're saving some overhead
by getting rid of syscache lookups in toast_flatten_tuple_attribute.)

So, despite Noah's misgivings, I'm thinking this is the way to proceed.

Comments?

            regards, tom lane

diff --git a/src/backend/access/common/heaptuple.c b/src/backend/access/common/heaptuple.c
index aea9d40..c64ede9 100644
*** a/src/backend/access/common/heaptuple.c
--- b/src/backend/access/common/heaptuple.c
*************** heap_copytuple_with_tuple(HeapTuple src,
*** 617,622 ****
--- 617,657 ----
      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,
*************** heap_form_tuple(TupleDesc tupleDescripto
*** 635,641 ****
                  data_len;
      int            hoff;
      bool        hasnull = false;
-     Form_pg_attribute *att = tupleDescriptor->attrs;
      int            numberOfAttributes = tupleDescriptor->natts;
      int            i;

--- 670,675 ----
*************** heap_form_tuple(TupleDesc tupleDescripto
*** 646,673 ****
                          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.
       */
      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);
          }
      }

--- 680,693 ----
                          numberOfAttributes, MaxTupleAttributeNumber)));

      /*
!      * Check for nulls
       */
      for (i = 0; i < numberOfAttributes; i++)
      {
          if (isnull[i])
          {
!             hasnull = true;
!             break;
          }
      }

*************** heap_form_tuple(TupleDesc tupleDescripto
*** 697,703 ****

      /*
       * And fill in the information.  Note we fill the Datum fields even though
!      * this tuple may never become a Datum.
       */
      tuple->t_len = len;
      ItemPointerSetInvalid(&(tuple->t_self));
--- 717,724 ----

      /*
       * And fill in the information.  Note we fill the Datum fields even though
!      * this tuple may never become a Datum.  This lets HeapTupleHeaderGetDatum
!      * identify the tuple type if needed.
       */
      tuple->t_len = len;
      ItemPointerSetInvalid(&(tuple->t_self));
*************** heap_form_minimal_tuple(TupleDesc tupleD
*** 1389,1395 ****
                  data_len;
      int            hoff;
      bool        hasnull = false;
-     Form_pg_attribute *att = tupleDescriptor->attrs;
      int            numberOfAttributes = tupleDescriptor->natts;
      int            i;

--- 1410,1415 ----
*************** heap_form_minimal_tuple(TupleDesc tupleD
*** 1400,1427 ****
                          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.
       */
      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);
          }
      }

--- 1420,1433 ----
                          numberOfAttributes, MaxTupleAttributeNumber)));

      /*
!      * Check for nulls
       */
      for (i = 0; i < numberOfAttributes; i++)
      {
          if (isnull[i])
          {
!             hasnull = true;
!             break;
          }
      }

diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index b4c68e9..7da10e9 100644
*** a/src/backend/access/common/indextuple.c
--- b/src/backend/access/common/indextuple.c
*************** index_form_tuple(TupleDesc tupleDescript
*** 158,163 ****
--- 158,168 ----
      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 9a821d3..dde74d4 100644
*** a/src/backend/access/heap/tuptoaster.c
--- b/src/backend/access/heap/tuptoaster.c
*************** toast_insert_or_update(Relation rel, Hea
*** 991,996 ****
--- 991,999 ----
   *
   *    "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, Tuple
*** 1068,1126 ****


  /* ----------
!  * toast_flatten_tuple_attribute -
   *
!  *    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.
   *
!  *    Note that flattening does not mean expansion of short-header varlenas,
!  *    so in one sense toasting is allowed within composite datums.
   * ----------
   */
  Datum
! toast_flatten_tuple_attribute(Datum value,
!                               Oid typeId, int32 typeMod)
  {
-     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;
      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);
      ItemPointerSetInvalid(&(tmptup.t_self));
      tmptup.t_tableOid = InvalidOid;
!     tmptup.t_data = olddata;

      Assert(numAttrs <= MaxTupleAttributeNumber);
      heap_deform_tuple(&tmptup, tupleDesc, toast_values, toast_isnull);

--- 1071,1131 ----


  /* ----------
!  * 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;
!     Form_pg_attribute *att = tupleDesc->attrs;
!     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);

*************** toast_flatten_tuple_attribute(Datum valu
*** 1144,1164 ****
                  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.
--- 1149,1159 ----
*************** toast_flatten_tuple_attribute(Datum valu
*** 1166,1172 ****
      new_header_len = offsetof(HeapTupleHeaderData, t_bits);
      if (has_nulls)
          new_header_len += BITMAPLEN(numAttrs);
!     if (olddata->t_infomask & HEAP_HASOID)
          new_header_len += sizeof(Oid);
      new_header_len = MAXALIGN(new_header_len);
      new_data_len = heap_compute_data_size(tupleDesc,
--- 1161,1167 ----
      new_header_len = offsetof(HeapTupleHeaderData, t_bits);
      if (has_nulls)
          new_header_len += BITMAPLEN(numAttrs);
!     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,
*************** toast_flatten_tuple_attribute(Datum valu
*** 1178,1191 ****
      /*
       * Copy the existing tuple header, but adjust natts and t_hoff.
       */
!     memcpy(new_data, olddata, 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));

!     /* Reset the datum length field, too */
      HeapTupleHeaderSetDatumLength(new_data, new_tuple_len);

      /* Copy over the data, and fill the null bitmap if needed */
      heap_fill_tuple(tupleDesc,
--- 1173,1188 ----
      /*
       * Copy the existing tuple header, but adjust natts and t_hoff.
       */
!     memcpy(new_data, tup, offsetof(HeapTupleHeaderData, t_bits));
      HeapTupleHeaderSetNatts(new_data, numAttrs);
      new_data->t_hoff = new_header_len;
!     if (tup->t_infomask & HEAP_HASOID)
!         HeapTupleHeaderSetOid(new_data, HeapTupleHeaderGetOid(tup));

!     /* 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_flatten_tuple_attribute(Datum valu
*** 1202,1208 ****
      for (i = 0; i < numAttrs; i++)
          if (toast_free[i])
              pfree(DatumGetPointer(toast_values[i]));
-     ReleaseTupleDesc(tupleDesc);

      return PointerGetDatum(new_data);
  }
--- 1199,1204 ----
diff --git a/src/backend/executor/execQual.c b/src/backend/executor/execQual.c
index 0eba025..833c4ed 100644
*** a/src/backend/executor/execQual.c
--- b/src/backend/executor/execQual.c
*************** ExecEvalWholeRowFast(WholeRowVarExprStat
*** 896,903 ****
  {
      Var           *variable = (Var *) wrvstate->xprstate.expr;
      TupleTableSlot *slot;
-     HeapTuple    tuple;
-     TupleDesc    tupleDesc;
      HeapTupleHeader dtuple;

      if (isDone)
--- 896,901 ----
*************** ExecEvalWholeRowFast(WholeRowVarExprStat
*** 927,958 ****
      if (wrvstate->wrv_junkFilter != NULL)
          slot = ExecFilterJunk(wrvstate->wrv_junkFilter, slot);

-     tuple = ExecFetchSlotTuple(slot);
-     tupleDesc = slot->tts_tupleDescriptor;
-
      /*
!      * We have to make a copy of the tuple so we can safely insert the Datum
!      * overhead fields, which are not set in on-disk tuples.
       */
!     dtuple = (HeapTupleHeader) palloc(tuple->t_len);
!     memcpy((char *) dtuple, (char *) tuple->t_data, tuple->t_len);
!
!     HeapTupleHeaderSetDatumLength(dtuple, tuple->t_len);

      /*
!      * If the Var identifies a named composite type, label the tuple with that
!      * type; otherwise use what is in the tupleDesc.
       */
      if (variable->vartype != RECORDOID)
      {
          HeapTupleHeaderSetTypeId(dtuple, variable->vartype);
          HeapTupleHeaderSetTypMod(dtuple, variable->vartypmod);
      }
-     else
-     {
-         HeapTupleHeaderSetTypeId(dtuple, tupleDesc->tdtypeid);
-         HeapTupleHeaderSetTypMod(dtuple, tupleDesc->tdtypmod);
-     }

      return PointerGetDatum(dtuple);
  }
--- 925,944 ----
      if (wrvstate->wrv_junkFilter != NULL)
          slot = ExecFilterJunk(wrvstate->wrv_junkFilter, slot);

      /*
!      * Copy the slot tuple and make sure any toasted fields get detoasted.
       */
!     dtuple = DatumGetHeapTupleHeader(ExecFetchSlotTupleDatum(slot));

      /*
!      * If the Var identifies a named composite type, label the datum with that
!      * type; otherwise we'll use the slot's info.
       */
      if (variable->vartype != RECORDOID)
      {
          HeapTupleHeaderSetTypeId(dtuple, variable->vartype);
          HeapTupleHeaderSetTypMod(dtuple, variable->vartypmod);
      }

      return PointerGetDatum(dtuple);
  }
*************** ExecEvalWholeRowSlow(WholeRowVarExprStat
*** 1029,1041 ****
      }

      /*
!      * We have to make a copy of the tuple so we can safely insert the Datum
!      * overhead fields, which are not set in on-disk tuples.
       */
!     dtuple = (HeapTupleHeader) palloc(tuple->t_len);
!     memcpy((char *) dtuple, (char *) tuple->t_data, tuple->t_len);

!     HeapTupleHeaderSetDatumLength(dtuple, tuple->t_len);
      HeapTupleHeaderSetTypeId(dtuple, variable->vartype);
      HeapTupleHeaderSetTypMod(dtuple, variable->vartypmod);

--- 1015,1027 ----
      }

      /*
!      * Copy the slot tuple and make sure any toasted fields get detoasted.
       */
!     dtuple = DatumGetHeapTupleHeader(ExecFetchSlotTupleDatum(slot));

!     /*
!      * Reset datum's type ID fields to match the Var.
!      */
      HeapTupleHeaderSetTypeId(dtuple, variable->vartype);
      HeapTupleHeaderSetTypMod(dtuple, variable->vartypmod);

diff --git a/src/backend/executor/execTuples.c b/src/backend/executor/execTuples.c
index af74202..928b5e3 100644
*** a/src/backend/executor/execTuples.c
--- b/src/backend/executor/execTuples.c
***************
*** 89,94 ****
--- 89,95 ----
  #include "postgres.h"

  #include "access/htup_details.h"
+ #include "access/tuptoaster.h"
  #include "funcapi.h"
  #include "catalog/pg_type.h"
  #include "nodes/nodeFuncs.h"
*************** ExecFetchSlotMinimalTuple(TupleTableSlot
*** 700,726 ****
   *        ExecFetchSlotTupleDatum
   *            Fetch the slot's tuple as a composite-type Datum.
   *
!  *        We convert the slot's contents to local physical-tuple form,
!  *        and fill in the Datum header fields.  Note that the result
!  *        always points to storage owned by the slot.
   * --------------------------------
   */
  Datum
  ExecFetchSlotTupleDatum(TupleTableSlot *slot)
  {
      HeapTuple    tup;
-     HeapTupleHeader td;
      TupleDesc    tupdesc;

!     /* Make sure we can scribble on the slot contents ... */
!     tup = ExecMaterializeSlot(slot);
!     /* ... and set up the composite-Datum header fields, in case not done */
!     td = tup->t_data;
      tupdesc = slot->tts_tupleDescriptor;
!     HeapTupleHeaderSetDatumLength(td, tup->t_len);
!     HeapTupleHeaderSetTypeId(td, tupdesc->tdtypeid);
!     HeapTupleHeaderSetTypMod(td, tupdesc->tdtypmod);
!     return PointerGetDatum(td);
  }

  /* --------------------------------
--- 701,721 ----
   *        ExecFetchSlotTupleDatum
   *            Fetch the slot's tuple as a composite-type Datum.
   *
!  *        The result is always freshly palloc'd in the caller's memory context.
   * --------------------------------
   */
  Datum
  ExecFetchSlotTupleDatum(TupleTableSlot *slot)
  {
      HeapTuple    tup;
      TupleDesc    tupdesc;

!     /* Fetch slot's contents in regular-physical-tuple form */
!     tup = ExecFetchSlotTuple(slot);
      tupdesc = slot->tts_tupleDescriptor;
!
!     /* Convert to Datum form */
!     return heap_copy_tuple_as_datum(tup, tupdesc);
  }

  /* --------------------------------
*************** BuildTupleFromCStrings(AttInMetadata *at
*** 1133,1138 ****
--- 1128,1193 ----
  }

  /*
+  * HeapTupleHeaderGetDatum - convert a HeapTupleHeader pointer to a Datum.
+  *
+  * This must *not* get applied to an on-disk tuple; the tuple should be
+  * freshly made by heap_form_tuple or some wrapper routine for it (such as
+  * BuildTupleFromCStrings).  Be sure also that the tupledesc used to build
+  * the tuple has a properly "blessed" rowtype.
+  *
+  * Formerly this was a macro equivalent to PointerGetDatum, relying on the
+  * fact that heap_form_tuple fills in the appropriate tuple header fields
+  * for a composite Datum.  However, we now require that composite Datums not
+  * contain any external TOAST pointers.  We do not want heap_form_tuple itself
+  * to enforce that; more specifically, the rule applies only to actual Datums
+  * and not to HeapTuple structures.  Therefore, HeapTupleHeaderGetDatum is
+  * now a function that detects whether there are externally-toasted fields
+  * and constructs a new tuple with inlined fields if so.  We still need
+  * heap_form_tuple to insert the Datum header fields, because otherwise this
+  * code would have no way to obtain a tupledesc for the tuple.
+  *
+  * Note that if we do build a new tuple, it's palloc'd in the current
+  * memory context.    Beware of code that changes context between the initial
+  * heap_form_tuple/etc call and calling HeapTuple(Header)GetDatum.
+  *
+  * For performance-critical callers, it could be worthwhile to take extra
+  * steps to ensure that there aren't TOAST pointers in the output of
+  * heap_form_tuple to begin with.  It's likely however that the costs of the
+  * typcache lookup and tuple disassembly/reassembly are swamped by TOAST
+  * dereference costs, so that the benefits of such extra effort would be
+  * minimal.
+  *
+  * XXX it would likely be better to create wrapper functions that produce
+  * a composite Datum from the field values in one step.  However, there's
+  * enough code using the existing APIs that we couldn't get rid of this
+  * hack anytime soon.
+  */
+ Datum
+ HeapTupleHeaderGetDatum(HeapTupleHeader tuple)
+ {
+     Datum        result;
+     TupleDesc    tupDesc;
+
+     /* No work if there are no external TOAST pointers in the tuple */
+     if (!HeapTupleHeaderHasExternal(tuple))
+         return PointerGetDatum(tuple);
+
+     /* Use the type data saved by heap_form_tuple to look up the rowtype */
+     tupDesc = lookup_rowtype_tupdesc(HeapTupleHeaderGetTypeId(tuple),
+                                      HeapTupleHeaderGetTypMod(tuple));
+
+     /* And do the flattening */
+     result = toast_flatten_tuple_to_datum(tuple,
+                                         HeapTupleHeaderGetDatumLength(tuple),
+                                           tupDesc);
+
+     ReleaseTupleDesc(tupDesc);
+
+     return result;
+ }
+
+
+ /*
   * Functions for sending tuples to the frontend (or other specified destination)
   * as though it is a SELECT result. These are used by utility commands that
   * need to project directly to the destination and don't need or want full
diff --git a/src/backend/executor/functions.c b/src/backend/executor/functions.c
index b5af19a..f0a89d2 100644
*** a/src/backend/executor/functions.c
--- b/src/backend/executor/functions.c
*************** postquel_get_single_result(TupleTableSlo
*** 954,960 ****
          /* We must return the whole tuple as a Datum. */
          fcinfo->isnull = false;
          value = ExecFetchSlotTupleDatum(slot);
-         value = datumCopy(value, fcache->typbyval, fcache->typlen);
      }
      else
      {
--- 954,959 ----
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 36dcfcf..e0325c4 100644
*** a/src/backend/executor/spi.c
--- b/src/backend/executor/spi.c
*************** SPI_returntuple(HeapTuple tuple, TupleDe
*** 742,753 ****
          oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
      }

!     dtup = (HeapTupleHeader) palloc(tuple->t_len);
!     memcpy((char *) dtup, (char *) tuple->t_data, tuple->t_len);
!
!     HeapTupleHeaderSetDatumLength(dtup, tuple->t_len);
!     HeapTupleHeaderSetTypeId(dtup, tupdesc->tdtypeid);
!     HeapTupleHeaderSetTypMod(dtup, tupdesc->tdtypmod);

      if (oldcxt)
          MemoryContextSwitchTo(oldcxt);
--- 742,748 ----
          oldcxt = MemoryContextSwitchTo(_SPI_current->savedcxt);
      }

!     dtup = DatumGetHeapTupleHeader(heap_copy_tuple_as_datum(tuple, tupdesc));

      if (oldcxt)
          MemoryContextSwitchTo(oldcxt);
diff --git a/src/backend/utils/adt/rowtypes.c b/src/backend/utils/adt/rowtypes.c
index 438cbb3..521c3da 100644
*** a/src/backend/utils/adt/rowtypes.c
--- b/src/backend/utils/adt/rowtypes.c
***************
*** 19,24 ****
--- 19,25 ----
  #include "access/htup_details.h"
  #include "access/tuptoaster.h"
  #include "catalog/pg_type.h"
+ #include "funcapi.h"
  #include "libpq/pqformat.h"
  #include "utils/builtins.h"
  #include "utils/lsyscache.h"
diff --git a/src/include/access/htup_details.h b/src/include/access/htup_details.h
index a3eba98..039d4b4 100644
*** a/src/include/access/htup_details.h
--- b/src/include/access/htup_details.h
*************** do { \
*** 476,481 ****
--- 476,484 ----
      (tup)->t_infomask2 = ((tup)->t_infomask2 & ~HEAP_NATTS_MASK) | (natts) \
  )

+ #define HeapTupleHeaderHasExternal(tup) \
+         (((tup)->t_infomask & HEAP_HASEXTERNAL) != 0)
+

  /*
   * BITMAPLEN(NATTS) -
*************** extern Datum heap_getsysattr(HeapTuple t
*** 730,735 ****
--- 733,739 ----
                  bool *isnull);
  extern HeapTuple heap_copytuple(HeapTuple tuple);
  extern void heap_copytuple_with_tuple(HeapTuple src, HeapTuple dest);
+ extern Datum heap_copy_tuple_as_datum(HeapTuple tuple, TupleDesc tupleDesc);
  extern HeapTuple heap_form_tuple(TupleDesc tupleDescriptor,
                  Datum *values, bool *isnull);
  extern HeapTuple heap_modify_tuple(HeapTuple tuple,
diff --git a/src/include/access/tuptoaster.h b/src/include/access/tuptoaster.h
index 296d016..e038e1a 100644
*** a/src/include/access/tuptoaster.h
--- b/src/include/access/tuptoaster.h
*************** extern struct varlena *heap_tuple_untoas
*** 184,199 ****
  extern HeapTuple toast_flatten_tuple(HeapTuple tup, TupleDesc tupleDesc);

  /* ----------
!  * toast_flatten_tuple_attribute -
   *
!  *    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.
   * ----------
   */
! extern Datum toast_flatten_tuple_attribute(Datum value,
!                               Oid typeId, int32 typeMod);

  /* ----------
   * toast_compress_datum -
--- 184,197 ----
  extern HeapTuple toast_flatten_tuple(HeapTuple tup, TupleDesc tupleDesc);

  /* ----------
!  * toast_flatten_tuple_to_datum -
   *
!  *    "Flatten" a tuple containing out-of-line toasted fields into a Datum.
   * ----------
   */
! extern Datum toast_flatten_tuple_to_datum(HeapTupleHeader tup,
!                              uint32 tup_len,
!                              TupleDesc tupleDesc);

  /* ----------
   * toast_compress_datum -
diff --git a/src/include/fmgr.h b/src/include/fmgr.h
index 22539ee..edb97f6 100644
*** a/src/include/fmgr.h
--- b/src/include/fmgr.h
*************** extern struct varlena *pg_detoast_datum_
*** 313,319 ****
  #define PG_RETURN_TEXT_P(x)    PG_RETURN_POINTER(x)
  #define PG_RETURN_BPCHAR_P(x)  PG_RETURN_POINTER(x)
  #define PG_RETURN_VARCHAR_P(x) PG_RETURN_POINTER(x)
! #define PG_RETURN_HEAPTUPLEHEADER(x)  PG_RETURN_POINTER(x)


  /*-------------------------------------------------------------------------
--- 313,319 ----
  #define PG_RETURN_TEXT_P(x)    PG_RETURN_POINTER(x)
  #define PG_RETURN_BPCHAR_P(x)  PG_RETURN_POINTER(x)
  #define PG_RETURN_VARCHAR_P(x) PG_RETURN_POINTER(x)
! #define PG_RETURN_HEAPTUPLEHEADER(x)  return HeapTupleHeaderGetDatum(x)


  /*-------------------------------------------------------------------------
diff --git a/src/include/funcapi.h b/src/include/funcapi.h
index 3610fc8..a3a12f7 100644
*** a/src/include/funcapi.h
--- b/src/include/funcapi.h
*************** extern TupleDesc build_function_result_t
*** 200,205 ****
--- 200,207 ----
   * HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values) -
   *        build a HeapTuple given user data in C string form. values is an array
   *        of C strings, one for each attribute of the return tuple.
+  * Datum HeapTupleHeaderGetDatum(HeapTupleHeader tuple) - convert a
+  *        HeapTupleHeader to a Datum.
   *
   * Macro declarations:
   * HeapTupleGetDatum(HeapTuple tuple) - convert a HeapTuple to a Datum.
*************** extern TupleDesc build_function_result_t
*** 216,224 ****
   *----------
   */

! #define HeapTupleGetDatum(_tuple)        PointerGetDatum((_tuple)->t_data)
  /* obsolete version of above */
! #define TupleGetDatum(_slot, _tuple)    PointerGetDatum((_tuple)->t_data)

  extern TupleDesc RelationNameGetTupleDesc(const char *relname);
  extern TupleDesc TypeGetTupleDesc(Oid typeoid, List *colaliases);
--- 218,226 ----
   *----------
   */

! #define HeapTupleGetDatum(tuple)        HeapTupleHeaderGetDatum((tuple)->t_data)
  /* obsolete version of above */
! #define TupleGetDatum(_slot, _tuple)    HeapTupleGetDatum(_tuple)

  extern TupleDesc RelationNameGetTupleDesc(const char *relname);
  extern TupleDesc TypeGetTupleDesc(Oid typeoid, List *colaliases);
*************** extern TupleDesc TypeGetTupleDesc(Oid ty
*** 227,232 ****
--- 229,235 ----
  extern TupleDesc BlessTupleDesc(TupleDesc tupdesc);
  extern AttInMetadata *TupleDescGetAttInMetadata(TupleDesc tupdesc);
  extern HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values);
+ extern Datum HeapTupleHeaderGetDatum(HeapTupleHeader tuple);
  extern TupleTableSlot *TupleDescGetSlot(TupleDesc tupdesc);


diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 3749fac..5d54399 100644
*** a/src/pl/plpgsql/src/pl_exec.c
--- b/src/pl/plpgsql/src/pl_exec.c
*************** exec_eval_datum(PLpgSQL_execstate *estat
*** 4473,4490 ****
                  tup = make_tuple_from_row(estate, row, row->rowtupdesc);
                  if (tup == NULL)    /* should not happen */
                      elog(ERROR, "row not compatible with its own tupdesc");
-                 MemoryContextSwitchTo(oldcontext);
                  *typeid = row->rowtupdesc->tdtypeid;
                  *typetypmod = row->rowtupdesc->tdtypmod;
                  *value = HeapTupleGetDatum(tup);
                  *isnull = false;
                  break;
              }

          case PLPGSQL_DTYPE_REC:
              {
                  PLpgSQL_rec *rec = (PLpgSQL_rec *) datum;
-                 HeapTupleData worktup;

                  if (!HeapTupleIsValid(rec->tup))
                      ereport(ERROR,
--- 4473,4489 ----
                  tup = make_tuple_from_row(estate, row, row->rowtupdesc);
                  if (tup == NULL)    /* should not happen */
                      elog(ERROR, "row not compatible with its own tupdesc");
                  *typeid = row->rowtupdesc->tdtypeid;
                  *typetypmod = row->rowtupdesc->tdtypmod;
                  *value = HeapTupleGetDatum(tup);
                  *isnull = false;
+                 MemoryContextSwitchTo(oldcontext);
                  break;
              }

          case PLPGSQL_DTYPE_REC:
              {
                  PLpgSQL_rec *rec = (PLpgSQL_rec *) datum;

                  if (!HeapTupleIsValid(rec->tup))
                      ereport(ERROR,
*************** exec_eval_datum(PLpgSQL_execstate *estat
*** 4496,4516 ****
                  /* Make sure we have a valid type/typmod setting */
                  BlessTupleDesc(rec->tupdesc);

-                 /*
-                  * In a trigger, the NEW and OLD parameters are likely to be
-                  * on-disk tuples that don't have the desired Datum fields.
-                  * Copy the tuple body and insert the right values.
-                  */
                  oldcontext = MemoryContextSwitchTo(estate->eval_econtext->ecxt_per_tuple_memory);
-                 heap_copytuple_with_tuple(rec->tup, &worktup);
-                 HeapTupleHeaderSetDatumLength(worktup.t_data, worktup.t_len);
-                 HeapTupleHeaderSetTypeId(worktup.t_data, rec->tupdesc->tdtypeid);
-                 HeapTupleHeaderSetTypMod(worktup.t_data, rec->tupdesc->tdtypmod);
-                 MemoryContextSwitchTo(oldcontext);
                  *typeid = rec->tupdesc->tdtypeid;
                  *typetypmod = rec->tupdesc->tdtypmod;
!                 *value = HeapTupleGetDatum(&worktup);
                  *isnull = false;
                  break;
              }

--- 4495,4506 ----
                  /* Make sure we have a valid type/typmod setting */
                  BlessTupleDesc(rec->tupdesc);

                  oldcontext = MemoryContextSwitchTo(estate->eval_econtext->ecxt_per_tuple_memory);
                  *typeid = rec->tupdesc->tdtypeid;
                  *typetypmod = rec->tupdesc->tdtypmod;
!                 *value = heap_copy_tuple_as_datum(rec->tup, rec->tupdesc);
                  *isnull = false;
+                 MemoryContextSwitchTo(oldcontext);
                  break;
              }

diff --git a/src/test/regress/expected/arrays.out b/src/test/regress/expected/arrays.out
index 6dce5c9..4286691 100644
*** a/src/test/regress/expected/arrays.out
--- b/src/test/regress/expected/arrays.out
*************** select * from t1;
*** 1676,1678 ****
--- 1676,1708 ----
   [5:5]={"(42,43)"}
  (1 row)

+ -- Check that arrays of composites are safely detoasted when needed
+ create temp table src (f1 text);
+ insert into src
+   select string_agg(random()::text,'') from generate_series(1,10000);
+ create type textandtext as (c1 text, c2 text);
+ create temp table dest (f1 textandtext[]);
+ insert into dest select array[row(f1,f1)::textandtext] from src;
+ select length(md5((f1[1]).c2)) from dest;
+  length
+ --------
+      32
+ (1 row)
+
+ delete from src;
+ select length(md5((f1[1]).c2)) from dest;
+  length
+ --------
+      32
+ (1 row)
+
+ truncate table src;
+ drop table src;
+ select length(md5((f1[1]).c2)) from dest;
+  length
+ --------
+      32
+ (1 row)
+
+ drop table dest;
+ drop type textandtext;
diff --git a/src/test/regress/regress.c b/src/test/regress/regress.c
index c25bf6e..4b69067 100644
*** a/src/test/regress/regress.c
--- b/src/test/regress/regress.c
*************** make_tuple_indirect(PG_FUNCTION_ARGS)
*** 810,814 ****

      MemoryContextSwitchTo(old_context);

!     PG_RETURN_HEAPTUPLEHEADER(newtup->t_data);
  }
--- 810,823 ----

      MemoryContextSwitchTo(old_context);

!     /*
!      * We intentionally don't use PG_RETURN_HEAPTUPLEHEADER here, because that
!      * would cause the indirect toast pointers to be flattened out of the
!      * tuple immediately, rendering subsequent testing irrelevant.  So just
!      * return the HeapTupleHeader pointer as-is.  This violates the general
!      * rule that composite Datums shouldn't contain toast pointers, but so
!      * long as the regression test scripts don't insert the result of this
!      * function into a container type (record, array, etc) it should be OK.
!      */
!     PG_RETURN_POINTER(newtup->t_data);
  }
diff --git a/src/test/regress/sql/arrays.sql b/src/test/regress/sql/arrays.sql
index 92af172..d9f7cbf 100644
*** a/src/test/regress/sql/arrays.sql
--- b/src/test/regress/sql/arrays.sql
*************** insert into t1 (f1[5].q1) values(42);
*** 459,461 ****
--- 459,478 ----
  select * from t1;
  update t1 set f1[5].q2 = 43;
  select * from t1;
+
+ -- Check that arrays of composites are safely detoasted when needed
+
+ create temp table src (f1 text);
+ insert into src
+   select string_agg(random()::text,'') from generate_series(1,10000);
+ create type textandtext as (c1 text, c2 text);
+ create temp table dest (f1 textandtext[]);
+ insert into dest select array[row(f1,f1)::textandtext] from src;
+ select length(md5((f1[1]).c2)) from dest;
+ delete from src;
+ select length(md5((f1[1]).c2)) from dest;
+ truncate table src;
+ drop table src;
+ select length(md5((f1[1]).c2)) from dest;
+ drop table dest;
+ drop type textandtext;