diff --git a/src/backend/access/common/heaptuple.c b/src/backend/access/common/heaptuple.c index aea9d40..48b09b8 100644 *** a/src/backend/access/common/heaptuple.c --- b/src/backend/access/common/heaptuple.c *************** heap_form_tuple(TupleDesc tupleDescripto *** 649,674 **** * 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); ! } } /* --- 649,661 ---- * 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. */ for (i = 0; i < numberOfAttributes; i++) { if (isnull[i]) hasnull = true; ! else ! values[i] = toast_flatten_tuple_attribute(values[i], att[i]); } /* *************** heap_form_minimal_tuple(TupleDesc tupleD *** 1403,1428 **** * 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); ! } } /* --- 1390,1402 ---- * 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. */ for (i = 0; i < numberOfAttributes; i++) { if (isnull[i]) hasnull = true; ! else ! values[i] = toast_flatten_tuple_attribute(values[i], att[i]); } /* diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c index 9a821d3..2d1a922 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,1097 **** /* ---------- ! * 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; --- 1071,1110 ---- /* ---------- ! * toast_flatten_tuple_datum - * ! * "Flatten" a composite Datum to contain no toasted fields. ! * This must be invoked on any composite value that is to be inserted into ! * a tuple, array, range, etc. Doing this preserves the invariant that ! * toasting goes only one level deep in a tuple. * ! * Unlike toast_flatten_tuple, this does expand compressed fields. That's ! * not necessary for correctness, but is a policy decision based on the ! * assumption that compression will be more effective if applied to the ! * whole tuple not individual fields. ! * ! * 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. ! * ! * Note: generally speaking, callers should skip applying this to Datums ! * that are toasted overall, even just to the extent of being in short-header ! * format. That implies that the Datum has previously been stored within ! * some tuple, array, range, etc, and therefore it should already not contain ! * any out-of-line fields. * ---------- */ Datum ! toast_flatten_tuple_datum(Datum value, 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 = tupleDesc->attrs; ! int numAttrs = tupleDesc->natts; int i; bool need_change = false; bool has_nulls = false; *************** toast_flatten_tuple_attribute(Datum valu *** 1100,1120 **** 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)); --- 1113,1128 ---- bool toast_free[MaxTupleAttributeNumber]; /* * Break down the tuple into fields. + * + * Note: if the supplied datum is toasted overall, DatumGetHeapTupleHeader + * will detoast it, and then we'll leak the detoasted copy. This is not + * worth fixing because callers shouldn't call us on toasted datums + * anyway. */ olddata = DatumGetHeapTupleHeader(value); ! Assert(HeapTupleHeaderGetTypeId(olddata) == tupleDesc->tdtypeid); ! Assert(HeapTupleHeaderGetTypMod(olddata) == tupleDesc->tdtypmod); /* Build a temporary HeapTuple control structure */ tmptup.t_len = HeapTupleHeaderGetDatumLength(olddata); ItemPointerSetInvalid(&(tmptup.t_self)); *************** toast_flatten_tuple_attribute(Datum valu *** 1150,1162 **** } /* ! * If nothing to untoast, just return the original tuple. */ if (!need_change) - { - ReleaseTupleDesc(tupleDesc); return value; - } /* * Calculate the new size of the tuple. --- 1158,1167 ---- } /* ! * If nothing to untoast, just return the original datum. */ if (!need_change) return value; /* * Calculate the new size of the tuple. *************** toast_flatten_tuple_attribute(Datum valu *** 1202,1214 **** for (i = 0; i < numAttrs; i++) if (toast_free[i]) pfree(DatumGetPointer(toast_values[i])); - ReleaseTupleDesc(tupleDesc); return PointerGetDatum(new_data); } /* ---------- * toast_compress_datum - * * Create a compressed version of a varlena datum --- 1207,1269 ---- for (i = 0; i < numAttrs; i++) if (toast_free[i]) pfree(DatumGetPointer(toast_values[i])); return PointerGetDatum(new_data); } /* ---------- + * toast_flatten_tuple_attribute - + * + * If a Datum is of composite type, "flatten" it to contain no toasted fields. + * This is a convenience routine for doing toast_flatten_tuple_datum() on + * tuple fields. + * ---------- + */ + Datum + toast_flatten_tuple_attribute(Datum value, Form_pg_attribute att) + { + TupleDesc tupleDesc; + + /* + * Exit quickly if the attribute couldn't possibly be of composite type. + * All composite datums are varlena and have alignment 'd'; furthermore + * they aren't arrays. This heuristic is sufficient to eliminate all + * but a few non-composite types. + */ + if (att->attlen != -1 || + att->attalign != 'd' || + att->attndims != 0) + return value; + + /* + * Also, if the datum is already toasted, it must have already been stored + * within some larger tuple (or array, range, etc) and so it doesn't need + * to be flattened again. (It is not our job here to force it to be + * uncompressed, so don't worry about that.) + */ + if (VARATT_IS_EXTENDED(DatumGetPointer(value))) + return value; + + /* + * See if it's a composite type, and get the tupdesc if so. + */ + tupleDesc = lookup_rowtype_tupdesc_noerror(att->atttypid, + att->atttypmod, + true); + if (tupleDesc == NULL) + return value; /* not a composite type */ + + /* OK, flatten it */ + value = toast_flatten_tuple_datum(value, tupleDesc); + + ReleaseTupleDesc(tupleDesc); + + return value; + } + + + /* ---------- * toast_compress_datum - * * Create a compressed version of a varlena datum diff --git a/src/backend/executor/execQual.c b/src/backend/executor/execQual.c index 0eba025..d84c051 100644 *** a/src/backend/executor/execQual.c --- b/src/backend/executor/execQual.c *************** ExecEvalArrayRef(ArrayRefExprState *asta *** 449,462 **** } if (lIndex == NULL) ! resultArray = array_set(array_source, i, ! upper.indx, ! sourceData, ! eisnull, ! astate->refattrlength, ! astate->refelemlength, ! astate->refelembyval, ! astate->refelemalign); else resultArray = array_set_slice(array_source, i, upper.indx, lower.indx, --- 449,492 ---- } if (lIndex == NULL) ! { ! /* ! * If element type is composite, look up the tupledesc. This is ! * not because array_set_element couldn't do so, but because we ! * need to install the ShutdownTupleDescRef callback to clean up ! * the tupledesc reference later, and we don't have another easy ! * way to remember if that's been done already. ! */ ! if (astate->refiscomposite && !eisnull) ! { ! HeapTupleHeader tuple; ! Oid tupType; ! int32 tupTypmod; ! ! tuple = DatumGetHeapTupleHeader(sourceData); ! ! tupType = HeapTupleHeaderGetTypeId(tuple); ! tupTypmod = HeapTupleHeaderGetTypMod(tuple); ! ! (void) get_cached_rowtype(tupType, tupTypmod, ! &astate->refelemtupdesc, ! econtext); ! ! /* If we detoasted sourceData overall, pass that on */ ! sourceData = PointerGetDatum(tuple); ! } ! ! resultArray = array_set_element(array_source, i, ! upper.indx, ! sourceData, ! eisnull, ! astate->refattrlength, ! astate->refelemlength, ! astate->refelembyval, ! astate->refelemalign, ! astate->refiscomposite, ! &astate->refelemtupdesc); ! } else resultArray = array_set_slice(array_source, i, upper.indx, lower.indx, *************** ExecInitExpr(Expr *node, PlanState *pare *** 4510,4519 **** parent); /* do one-time catalog lookups for type info */ astate->refattrlength = get_typlen(aref->refarraytype); ! get_typlenbyvalalign(aref->refelemtype, ! &astate->refelemlength, ! &astate->refelembyval, ! &astate->refelemalign); state = (ExprState *) astate; } break; --- 4540,4551 ---- parent); /* do one-time catalog lookups for type info */ astate->refattrlength = get_typlen(aref->refarraytype); ! get_typlenbyvalaligncomp(aref->refelemtype, ! &astate->refelemlength, ! &astate->refelembyval, ! &astate->refelemalign, ! &astate->refiscomposite); ! astate->refelemtupdesc = NULL; state = (ExprState *) astate; } break; diff --git a/src/backend/utils/adt/arrayfuncs.c b/src/backend/utils/adt/arrayfuncs.c index 91df184..c52a5de 100644 *** a/src/backend/utils/adt/arrayfuncs.c --- b/src/backend/utils/adt/arrayfuncs.c *************** *** 17,22 **** --- 17,23 ---- #include #include "access/htup_details.h" + #include "access/tuptoaster.h" #include "funcapi.h" #include "libpq/pqformat.h" #include "utils/array.h" *************** typedef struct ArrayIteratorData *** 74,79 **** --- 75,130 ---- int current_item; /* the item # we're at in the array */ } ArrayIteratorData; + /* Support for array element detoasting */ + + /* Initialize state needed by FLATTEN_ARRAY_ELEMENT */ + #define INIT_FLATTEN(tupdesc) \ + do { \ + tupdesc = NULL; \ + } while (0) + + /* Call this on each non-NULL value about to be inserted into an array */ + #define FLATTEN_ARRAY_ELEMENT(val, typlen, typbyval, typiscomposite, tupdesc) \ + do { \ + if ((typlen) == -1) \ + { \ + if (typiscomposite) \ + val = flatten_composite_element(val, &(tupdesc)); \ + else \ + val = PointerGetDatum(PG_DETOAST_DATUM(val)); \ + } \ + } while (0) + + /* Variant that guarantees to make a copy */ + #define FLATTEN_ARRAY_ELEMENT_COPY(val, typlen, typbyval, typiscomposite, tupdesc) \ + do { \ + if ((typlen) == -1) \ + { \ + if (typiscomposite) \ + { \ + Datum _orig_val = (val); \ + val = flatten_composite_element(val, &(tupdesc)); \ + if (val == _orig_val) \ + val = datumCopy(val, typbyval, typlen); \ + } \ + else \ + val = PointerGetDatum(PG_DETOAST_DATUM_COPY(val)); \ + } \ + else \ + val = datumCopy(val, typbyval, typlen); \ + } while (0) + + /* Clean up after some FLATTEN_ARRAY_ELEMENT calls */ + #define END_FLATTEN(tupdesc) \ + do { \ + if (tupdesc) \ + { \ + ReleaseTupleDesc(tupdesc); \ + tupdesc = NULL; \ + } \ + } while (0) + + /* Local functions */ static bool array_isspace(char ch); static int ArrayCount(const char *str, int *dim, char typdelim); static void ReadArrayStr(char *arrayStr, const char *origStr, *************** static void CopyArrayEls(ArrayType *arra *** 92,97 **** --- 143,151 ---- Datum *values, bool *nulls, int nitems, int typlen, bool typbyval, char typalign, bool freedata); + static int array_cmp(FunctionCallInfo fcinfo); + static bool element_type_is_composite(Oid elmtype, int16 typlen, char typalign); + static Datum flatten_composite_element(Datum value, TupleDesc *tupdescptr); static bool array_get_isnull(const bits8 *nullbitmap, int offset); static void array_set_isnull(bits8 *nullbitmap, int offset, bool isNull); static Datum ArrayCast(char *value, bool byval, int len); *************** static void array_insert_slice(ArrayType *** 119,125 **** int ndim, int *dim, int *lb, int *st, int *endp, int typlen, bool typbyval, char typalign); - static int array_cmp(FunctionCallInfo fcinfo); static ArrayType *create_array_envelope(int ndims, int *dimv, int *lbv, int nbytes, Oid elmtype, int dataoffset); static ArrayType *array_fill_internal(ArrayType *dims, ArrayType *lbs, --- 173,178 ---- *************** ReadArrayStr(char *arrayStr, *** 864,870 **** hasnull = true; else { ! /* let's just make sure data is not toasted */ if (typlen == -1) values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i])); totbytes = att_addlength_datum(totbytes, typlen, values[i]); --- 917,930 ---- hasnull = true; else { ! /* ! * Let's just make sure data is not toasted. It seems quite ! * unlikely that the input function could have produced a toasted ! * value at all, and certainly it shouldn't have produced ! * something containing nested toasted values, so we don't bother ! * with FLATTEN_ARRAY_ELEMENT(). But a PG_DETOAST_DATUM() call is ! * reasonably cheap when it's a no-op. ! */ if (typlen == -1) values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i])); totbytes = att_addlength_datum(totbytes, typlen, values[i]); *************** ReadArrayBinary(StringInfo buf, *** 1466,1472 **** hasnull = true; else { ! /* let's just make sure data is not toasted */ if (typlen == -1) values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i])); totbytes = att_addlength_datum(totbytes, typlen, values[i]); --- 1526,1539 ---- hasnull = true; else { ! /* ! * Let's just make sure data is not toasted. It seems quite ! * unlikely that the receive function could have produced a ! * toasted value at all, and certainly it shouldn't have produced ! * something containing nested toasted values, so we don't bother ! * with FLATTEN_ARRAY_ELEMENT(). But a PG_DETOAST_DATUM() call is ! * reasonably cheap when it's a no-op. ! */ if (typlen == -1) values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i])); totbytes = att_addlength_datum(totbytes, typlen, values[i]); *************** array_get_slice(ArrayType *array, *** 2019,2025 **** } /* ! * array_set : * This routine sets the value of an array element (specified by * a subscript array) to a new value specified by "dataValue". * --- 2086,2092 ---- } /* ! * array_set_element : * This routine sets the value of an array element (specified by * a subscript array) to a new value specified by "dataValue". * *************** array_get_slice(ArrayType *array, *** 2035,2045 **** --- 2102,2117 ---- * elmlen: pg_type.typlen for the array's element type * elmbyval: pg_type.typbyval for the array's element type * elmalign: pg_type.typalign for the array's element type + * elmiscomposite: true if element type is composite + * element_tupdesc: pointer to where to cache tupledesc, if composite * * Result: * A new array is returned, just like the old except for the one * modified entry. The original array object is not changed. * + * Caller must arrange for any refcount on *element_tupdesc to be released + * when no longer useful. + * * For one-dimensional arrays only, we allow the array to be extended * by assigning to a position outside the existing subscript range; any * positions between the existing elements and the new one are set to NULLs. *************** array_get_slice(ArrayType *array, *** 2049,2063 **** * rather than returning a NULL as the fetch operations do. */ ArrayType * ! array_set(ArrayType *array, ! int nSubscripts, ! int *indx, ! Datum dataValue, ! bool isNull, ! int arraytyplen, ! int elmlen, ! bool elmbyval, ! char elmalign) { ArrayType *newarray; int i, --- 2121,2137 ---- * rather than returning a NULL as the fetch operations do. */ ArrayType * ! array_set_element(ArrayType *array, ! int nSubscripts, ! int *indx, ! Datum dataValue, ! bool isNull, ! int arraytyplen, ! int elmlen, ! bool elmbyval, ! char elmalign, ! bool elmiscomposite, ! TupleDesc *element_tupdesc) { ArrayType *newarray; int i, *************** array_set(ArrayType *array, *** 2116,2122 **** /* make sure item to be inserted is not toasted */ if (elmlen == -1 && !isNull) ! dataValue = PointerGetDatum(PG_DETOAST_DATUM(dataValue)); /* detoast input array if necessary */ array = DatumGetArrayTypeP(PointerGetDatum(array)); --- 2190,2201 ---- /* make sure item to be inserted is not toasted */ if (elmlen == -1 && !isNull) ! { ! FLATTEN_ARRAY_ELEMENT(dataValue, ! elmlen, elmbyval, ! elmiscomposite, ! *element_tupdesc); ! } /* detoast input array if necessary */ array = DatumGetArrayTypeP(PointerGetDatum(array)); *************** array_set(ArrayType *array, *** 2306,2311 **** --- 2385,2446 ---- } /* + * array_set : + * Backwards-compatibility function that doesn't require caller + * to supply info about composite element types. + * + * Identical to array_set_element except elmiscomposite is computed internally + * and there is no chance to cache element tupledesc across calls. + */ + ArrayType * + array_set(ArrayType *array, + int nSubscripts, + int *indx, + Datum dataValue, + bool isNull, + int arraytyplen, + int elmlen, + bool elmbyval, + char elmalign) + { + ArrayType *newarray; + bool elmiscomposite; + TupleDesc elmtupdesc; + + if (arraytyplen < 0) + { + Oid elmtype; + + /* detoast input array if necessary */ + array = DatumGetArrayTypeP(PointerGetDatum(array)); + /* now we can get the element type safely */ + elmtype = ARR_ELEMTYPE(array); + + elmiscomposite = element_type_is_composite(elmtype, elmlen, elmalign); + } + else + elmiscomposite = false; /* not possible for fixed-size array */ + + INIT_FLATTEN(elmtupdesc); + + newarray = array_set_element(array, + nSubscripts, + indx, + dataValue, + isNull, + arraytyplen, + elmlen, + elmbyval, + elmalign, + elmiscomposite, + &elmtupdesc); + + END_FLATTEN(elmtupdesc); + + return newarray; + } + + /* * array_set_slice : * This routine sets the value of a range of array locations (specified * by upper and lower subscript values) to new values passed as *************** array_map(FunctionCallInfo fcinfo, Oid i *** 2683,2688 **** --- 2818,2825 ---- int typlen; bool typbyval; char typalign; + bool typiscomposite; + TupleDesc typtupdesc; char *s; bits8 *bitmap; int bitmask; *************** array_map(FunctionCallInfo fcinfo, Oid i *** 2741,2746 **** --- 2878,2886 ---- typbyval = ret_extra->typbyval; typalign = ret_extra->typalign; + typiscomposite = element_type_is_composite(retType, typlen, typalign); + INIT_FLATTEN(typtupdesc); + /* Allocate temporary arrays for new values */ values = (Datum *) palloc(nitems * sizeof(Datum)); nulls = (bool *) palloc(nitems * sizeof(bool)); *************** array_map(FunctionCallInfo fcinfo, Oid i *** 2800,2807 **** else { /* Ensure data is not toasted */ ! if (typlen == -1) ! values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i])); /* Update total result size */ nbytes = att_addlength_datum(nbytes, typlen, values[i]); nbytes = att_align_nominal(nbytes, typalign); --- 2940,2948 ---- else { /* Ensure data is not toasted */ ! FLATTEN_ARRAY_ELEMENT(values[i], ! typlen, typbyval, ! typiscomposite, typtupdesc); /* Update total result size */ nbytes = att_addlength_datum(nbytes, typlen, values[i]); nbytes = att_align_nominal(nbytes, typalign); *************** array_map(FunctionCallInfo fcinfo, Oid i *** 2825,2830 **** --- 2966,2973 ---- } } + END_FLATTEN(typtupdesc); + /* Allocate and initialize the result array */ if (hasnulls) { *************** array_map(FunctionCallInfo fcinfo, Oid i *** 2868,2874 **** * A palloc'd 1-D array object is constructed and returned. Note that * elem values will be copied into the object even if pass-by-ref type. * ! * NOTE: it would be cleaner to look up the elmlen/elmbval/elmalign info * from the system catalogs, given the elmtype. However, the caller is * in a better position to cache this info across multiple uses, or even * to hard-wire values if the element type is hard-wired. --- 3011,3017 ---- * A palloc'd 1-D array object is constructed and returned. Note that * elem values will be copied into the object even if pass-by-ref type. * ! * NOTE: it would be cleaner to look up the elmlen/elmbyval/elmalign info * from the system catalogs, given the elmtype. However, the caller is * in a better position to cache this info across multiple uses, or even * to hard-wire values if the element type is hard-wired. *************** construct_array(Datum *elems, int nelems *** 2902,2908 **** * A palloc'd ndims-D array object is constructed and returned. Note that * elem values will be copied into the object even if pass-by-ref type. * ! * NOTE: it would be cleaner to look up the elmlen/elmbval/elmalign info * from the system catalogs, given the elmtype. However, the caller is * in a better position to cache this info across multiple uses, or even * to hard-wire values if the element type is hard-wired. --- 3045,3051 ---- * A palloc'd ndims-D array object is constructed and returned. Note that * elem values will be copied into the object even if pass-by-ref type. * ! * NOTE: it would be cleaner to look up the elmlen/elmbyval/elmalign info * from the system catalogs, given the elmtype. However, the caller is * in a better position to cache this info across multiple uses, or even * to hard-wire values if the element type is hard-wired. *************** construct_md_array(Datum *elems, *** 2921,2926 **** --- 3064,3071 ---- int32 dataoffset; int i; int nelems; + bool elmiscomposite; + TupleDesc elmtupdesc; if (ndims < 0) /* we do allow zero-dimension arrays */ ereport(ERROR, *************** construct_md_array(Datum *elems, *** 2938,2943 **** --- 3083,3091 ---- nelems = ArrayGetNItems(ndims, dims); + elmiscomposite = element_type_is_composite(elmtype, elmlen, elmalign); + INIT_FLATTEN(elmtupdesc); + /* compute required space */ nbytes = 0; hasnulls = false; *************** construct_md_array(Datum *elems, *** 2949,2956 **** continue; } /* make sure data is not toasted */ ! if (elmlen == -1) ! elems[i] = PointerGetDatum(PG_DETOAST_DATUM(elems[i])); nbytes = att_addlength_datum(nbytes, elmlen, elems[i]); nbytes = att_align_nominal(nbytes, elmalign); /* check for overflow of total request */ --- 3097,3106 ---- continue; } /* make sure data is not toasted */ ! FLATTEN_ARRAY_ELEMENT(elems[i], ! elmlen, elmbyval, ! elmiscomposite, elmtupdesc); ! /* update total result size */ nbytes = att_addlength_datum(nbytes, elmlen, elems[i]); nbytes = att_align_nominal(nbytes, elmalign); /* check for overflow of total request */ *************** construct_md_array(Datum *elems, *** 2961,2966 **** --- 3111,3118 ---- (int) MaxAllocSize))); } + END_FLATTEN(elmtupdesc); + /* Allocate and initialize result array */ if (hasnulls) { *************** construct_empty_array(Oid elmtype) *** 3020,3026 **** * If array elements are pass-by-ref data type, the returned Datums will * be pointers into the array object. * ! * NOTE: it would be cleaner to look up the elmlen/elmbval/elmalign info * from the system catalogs, given the elmtype. However, in most current * uses the type is hard-wired into the caller and so we can save a lookup * cycle by hard-wiring the type info as well. --- 3172,3178 ---- * If array elements are pass-by-ref data type, the returned Datums will * be pointers into the array object. * ! * NOTE: it would be cleaner to look up the elmlen/elmbyval/elmalign info * from the system catalogs, given the elmtype. However, in most current * uses the type is hard-wired into the caller and so we can save a lookup * cycle by hard-wiring the type info as well. *************** array_free_iterator(ArrayIterator iterat *** 4070,4075 **** --- 4222,4328 ---- /***************************************************************************/ /* + * Array element detoasting ("flattening") support + * + * It is critical that no external TOAST pointers appear within an array, + * either directly as an array element or within a non-scalar element type, + * so we must detoast any value that is about to be inserted into an array. + * + * Since we allow composite-type datums to contain toasted fields (at the + * first level; nested toasted values aren't allowed), we have to work extra + * hard to get rid of possible toasting when the array element type is + * composite. We assume that other composites such as ranges contain no + * external pointers, though. + * + * In addition to the no-external-pointers requirement, we make a policy + * decision that array elements not be compressed. We expect that compression + * would be better applied to the whole array rather than individual elements. + * + * The current implementation of FLATTEN_ARRAY_ELEMENT() also forces varlena + * elements to not be in short-header format, but this is an implementation + * artifact that should probably be removed. + */ + + /* + * Check if array element type is composite. + * + * Currently, we only have typlen/typbyval/typalign information cached for + * the array element type. To minimize unnecessary syscache lookups, this + * function uses the knowledge that a composite type must have typlen -1 + * (varlena) and alignment 'd'. There are only a few non-composite, non-array + * types for which this is true, so though imperfect this heuristic should + * work well enough. + * + * XXX ideally this would go away in favor of always getting typiscomposite + * info in the same syscache lookup that we get typlen etc with. That will + * require additional API changes though, so it doesn't seem back-patchable. + */ + static bool + element_type_is_composite(Oid elmtype, int16 typlen, char typalign) + { + if (typlen == -1 && typalign == 'd' && type_is_rowtype(elmtype)) + return true; + else + return false; + } + + /* + * Detoast ("flatten") a composite value + * + * This is the out-of-line portion of the FLATTEN_ARRAY_ELEMENT macro. + * We assume we have a non-null, known-composite value to work on. + * The tuple descriptor for the type can be cached at *tupdescptr. + * + * Note: for some logic paths, it's important that this does not + * copy the Datum if re-invoked on a previously flattened Datum. + */ + static Datum + flatten_composite_element(Datum value, TupleDesc *tupdescptr) + { + HeapTupleHeader tupdata; + Oid typeid; + int32 typmod; + TupleDesc tupleDesc; + + /* + * If the value is itself toasted (even just to the extent of having a + * short varlena header), then it must have been incorporated into some + * larger composite structure, so it should not contain any embedded toast + * pointers. We need only detoast it overall and we're done. + */ + if (VARATT_IS_EXTENDED(DatumGetPointer(value))) + return PointerGetDatum(PG_DETOAST_DATUM(value)); + + /* + * When working with an array of RECORD, it's possible that each array + * element is of a different record subtype, so we have to be prepared to + * look up the appropriate tupdesc here. But we cache the tupdesc across + * calls as much as we can. For an array of a named composite type, we + * should only have to do one lookup, but it seems prudent to check the + * type every time anyway. + */ + tupdata = DatumGetHeapTupleHeader(value); + typeid = HeapTupleHeaderGetTypeId(tupdata); + typmod = HeapTupleHeaderGetTypMod(tupdata); + + tupleDesc = *tupdescptr; + if (tupleDesc && + (tupleDesc->tdtypeid != typeid || tupleDesc->tdtypmod != typmod)) + { + ReleaseTupleDesc(tupleDesc); + tupleDesc = NULL; + } + if (tupleDesc == NULL) + { + tupleDesc = lookup_rowtype_tupdesc(typeid, typmod); + *tupdescptr = tupleDesc; + } + + /* tuptoaster.c does the actual work */ + return toast_flatten_tuple_datum(value, tupleDesc); + } + + /* * Check whether a specific array element is NULL * * nullbitmap: pointer to array's null bitmap (NULL if none) *************** accumArrayResult(ArrayBuildState *astate *** 4591,4600 **** astate->dnulls = (bool *) palloc(astate->alen * sizeof(bool)); astate->nelems = 0; astate->element_type = element_type; ! get_typlenbyvalalign(element_type, ! &astate->typlen, ! &astate->typbyval, ! &astate->typalign); } else { --- 4844,4855 ---- astate->dnulls = (bool *) palloc(astate->alen * sizeof(bool)); astate->nelems = 0; astate->element_type = element_type; ! get_typlenbyvalaligncomp(element_type, ! &astate->typlen, ! &astate->typbyval, ! &astate->typalign, ! &astate->typiscomposite); ! INIT_FLATTEN(astate->element_tupdesc); } else { *************** accumArrayResult(ArrayBuildState *astate *** 4621,4630 **** */ if (!disnull && !astate->typbyval) { ! if (astate->typlen == -1) ! dvalue = PointerGetDatum(PG_DETOAST_DATUM_COPY(dvalue)); ! else ! dvalue = datumCopy(dvalue, astate->typbyval, astate->typlen); } astate->dvalues[astate->nelems] = dvalue; --- 4876,4885 ---- */ if (!disnull && !astate->typbyval) { ! FLATTEN_ARRAY_ELEMENT_COPY(dvalue, ! astate->typlen, astate->typbyval, ! astate->typiscomposite, ! astate->element_tupdesc); } astate->dvalues[astate->nelems] = dvalue; *************** makeMdArrayResult(ArrayBuildState *astat *** 4691,4697 **** MemoryContextSwitchTo(oldcontext); ! /* Clean up all the junk */ if (release) MemoryContextDelete(astate->mcontext); --- 4946,4959 ---- MemoryContextSwitchTo(oldcontext); ! /* ! * Clean up all the junk. We must release any tupdesc refcount held for ! * composite-flattening purposes even if !release, since we can't be sure ! * we'll get called again. flatten_composite_element can re-acquire the ! * refcount if needed. ! */ ! END_FLATTEN(astate->element_tupdesc); ! if (release) MemoryContextDelete(astate->mcontext); *************** array_fill_internal(ArrayType *dims, Arr *** 5037,5043 **** /* make sure data is not toasted */ if (elmlen == -1) ! value = PointerGetDatum(PG_DETOAST_DATUM(value)); nbytes = att_addlength_datum(0, elmlen, value); nbytes = att_align_nominal(nbytes, elmalign); --- 5299,5315 ---- /* make sure data is not toasted */ if (elmlen == -1) ! { ! bool elmiscomposite; ! TupleDesc elmtupdesc; ! ! elmiscomposite = element_type_is_composite(elmtype, elmlen, elmalign); ! INIT_FLATTEN(elmtupdesc); ! FLATTEN_ARRAY_ELEMENT(value, ! elmlen, elmbyval, ! elmiscomposite, elmtupdesc); ! END_FLATTEN(elmtupdesc); ! } nbytes = att_addlength_datum(0, elmlen, value); nbytes = att_align_nominal(nbytes, elmalign); *************** array_replace_internal(ArrayType *array, *** 5277,5286 **** */ if (typlen == -1) { if (!search_isnull) ! search = PointerGetDatum(PG_DETOAST_DATUM(search)); if (!replace_isnull) ! replace = PointerGetDatum(PG_DETOAST_DATUM(replace)); } /* Prepare to apply the comparison operator */ --- 5549,5569 ---- */ if (typlen == -1) { + bool typiscomposite; + TupleDesc typtupdesc; + + typiscomposite = element_type_is_composite(element_type, + typlen, typalign); + INIT_FLATTEN(typtupdesc); if (!search_isnull) ! FLATTEN_ARRAY_ELEMENT(search, ! typlen, typbyval, ! typiscomposite, typtupdesc); if (!replace_isnull) ! FLATTEN_ARRAY_ELEMENT(replace, ! typlen, typbyval, ! typiscomposite, typtupdesc); ! END_FLATTEN(typtupdesc); } /* Prepare to apply the comparison operator */ diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c index a4ce716..278709d 100644 *** a/src/backend/utils/cache/lsyscache.c --- b/src/backend/utils/cache/lsyscache.c *************** get_typlenbyvalalign(Oid typid, int16 *t *** 1922,1927 **** --- 1922,1953 ---- } /* + * get_typlenbyvalaligncomp + * + * A four-fer: given the type OID, return typlen, typbyval, typalign, + * and typiscomposite. + */ + void + get_typlenbyvalaligncomp(Oid typid, int16 *typlen, bool *typbyval, + char *typalign, bool *typiscomposite) + { + HeapTuple tp; + Form_pg_type typtup; + + tp = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typid)); + if (!HeapTupleIsValid(tp)) + elog(ERROR, "cache lookup failed for type %u", typid); + typtup = (Form_pg_type) GETSTRUCT(tp); + *typlen = typtup->typlen; + *typbyval = typtup->typbyval; + *typalign = typtup->typalign; + /* this computation must match type_is_rowtype(): */ + *typiscomposite = (typid == RECORDOID || + typtup->typtype == TYPTYPE_COMPOSITE); + ReleaseSysCache(tp); + } + + /* * getTypeIOParam * Given a pg_type row, select the type OID to pass to I/O functions * diff --git a/src/include/access/tuptoaster.h b/src/include/access/tuptoaster.h index 296d016..5cdefad 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,208 ---- extern HeapTuple toast_flatten_tuple(HeapTuple tup, TupleDesc tupleDesc); /* ---------- + * toast_flatten_tuple_datum - + * + * "Flatten" a composite Datum to contain no toasted fields. + * This must be invoked on any composite value that is to be inserted into + * a tuple, array, range, etc. Doing this preserves the invariant that + * toasting goes only one level deep in a tuple. + * ---------- + */ + extern Datum toast_flatten_tuple_datum(Datum value, TupleDesc tupleDesc); + + /* ---------- * toast_flatten_tuple_attribute - * * If a Datum is of composite type, "flatten" it to contain no toasted fields. ! * This is a convenience routine for doing toast_flatten_tuple_datum() on ! * tuple fields. * ---------- */ ! extern Datum toast_flatten_tuple_attribute(Datum value, Form_pg_attribute att); /* ---------- * toast_compress_datum - diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 6c94e8a..7309022 100644 *** a/src/include/nodes/execnodes.h --- b/src/include/nodes/execnodes.h *************** typedef struct ArrayRefExprState *** 624,629 **** --- 624,631 ---- int16 refelemlength; /* typlen of the array element type */ bool refelembyval; /* is the element type pass-by-value? */ char refelemalign; /* typalign of the element type */ + bool refiscomposite; /* is the element type composite? */ + TupleDesc refelemtupdesc; /* may be set if refiscomposite */ } ArrayRefExprState; /* ---------------- diff --git a/src/include/utils/array.h b/src/include/utils/array.h index 9bbfaae..63307c1 100644 *** a/src/include/utils/array.h --- b/src/include/utils/array.h *************** typedef struct ArrayBuildState *** 88,93 **** --- 88,96 ---- int16 typlen; /* needed info about datatype */ bool typbyval; char typalign; + bool typiscomposite; + /* use struct pointer to avoid including tupdesc.h here */ + struct tupleDesc *element_tupdesc; /* may be set if typiscomposite */ } ArrayBuildState; /* *************** extern Datum array_replace(PG_FUNCTION_A *** 218,223 **** --- 221,230 ---- extern Datum array_ref(ArrayType *array, int nSubscripts, int *indx, int arraytyplen, int elmlen, bool elmbyval, char elmalign, bool *isNull); + extern ArrayType *array_set_element(ArrayType *array, int nSubscripts, int *indx, + Datum dataValue, bool isNull, + int arraytyplen, int elmlen, bool elmbyval, char elmalign, + bool elmiscomposite, struct tupleDesc **element_tupdesc); extern ArrayType *array_set(ArrayType *array, int nSubscripts, int *indx, Datum dataValue, bool isNull, int arraytyplen, int elmlen, bool elmbyval, char elmalign); diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h index f46460a..30fb0a0 100644 *** a/src/include/utils/lsyscache.h --- b/src/include/utils/lsyscache.h *************** extern bool get_typbyval(Oid typid); *** 109,114 **** --- 109,116 ---- extern void get_typlenbyval(Oid typid, int16 *typlen, bool *typbyval); extern void get_typlenbyvalalign(Oid typid, int16 *typlen, bool *typbyval, char *typalign); + extern void get_typlenbyvalaligncomp(Oid typid, int16 *typlen, bool *typbyval, + char *typalign, bool *typiscomposite); extern Oid getTypeIOParam(HeapTuple typeTuple); extern void get_type_io_data(Oid typid, IOFuncSelector which_func, diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c index 3749fac..27d6b65 100644 *** a/src/pl/plpgsql/src/pl_exec.c --- b/src/pl/plpgsql/src/pl_exec.c *************** exec_assign_value(PLpgSQL_execstate *est *** 4246,4251 **** --- 4246,4252 ---- ArrayType *oldarrayval; ArrayType *newarrayval; SPITupleTable *save_eval_tuptable; + TupleDesc elemtyptupdesc = NULL; MemoryContext oldcontext; /* *************** exec_assign_value(PLpgSQL_execstate *est *** 4297,4302 **** --- 4298,4304 ---- int16 elemtyplen; bool elemtypbyval; char elemtypalign; + bool elemtypiscomposite; /* If target is domain over array, reduce to base type */ arraytypoid = getBaseTypeAndTypmod(parenttypoid, *************** exec_assign_value(PLpgSQL_execstate *est *** 4312,4321 **** /* Collect needed data about the types */ arraytyplen = get_typlen(arraytypoid); ! get_typlenbyvalalign(elemtypoid, ! &elemtyplen, ! &elemtypbyval, ! &elemtypalign); /* Now safe to update the cached data */ arrayelem->parenttypoid = parenttypoid; --- 4314,4324 ---- /* Collect needed data about the types */ arraytyplen = get_typlen(arraytypoid); ! get_typlenbyvalaligncomp(elemtypoid, ! &elemtyplen, ! &elemtypbyval, ! &elemtypalign, ! &elemtypiscomposite); /* Now safe to update the cached data */ arrayelem->parenttypoid = parenttypoid; *************** exec_assign_value(PLpgSQL_execstate *est *** 4327,4332 **** --- 4330,4336 ---- arrayelem->elemtyplen = elemtyplen; arrayelem->elemtypbyval = elemtypbyval; arrayelem->elemtypalign = elemtypalign; + arrayelem->elemtypiscomposite = elemtypiscomposite; } /* *************** exec_assign_value(PLpgSQL_execstate *est *** 4395,4409 **** /* * Build the modified array value. */ ! newarrayval = array_set(oldarrayval, ! nsubscripts, ! subscriptvals, ! coerced_value, ! *isNull, ! arrayelem->arraytyplen, ! arrayelem->elemtyplen, ! arrayelem->elemtypbyval, ! arrayelem->elemtypalign); MemoryContextSwitchTo(oldcontext); --- 4399,4423 ---- /* * Build the modified array value. */ ! newarrayval = array_set_element(oldarrayval, ! nsubscripts, ! subscriptvals, ! coerced_value, ! *isNull, ! arrayelem->arraytyplen, ! arrayelem->elemtyplen, ! arrayelem->elemtypbyval, ! arrayelem->elemtypalign, ! arrayelem->elemtypiscomposite, ! &elemtyptupdesc); ! ! /* ! * Unfortunately, we have no easy way to cache element type ! * tupledescs across calls, because there's no clear place ! * to release them again. Perhaps this can be improved later. ! */ ! if (elemtyptupdesc) ! ReleaseTupleDesc(elemtyptupdesc); MemoryContextSwitchTo(oldcontext); diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h index b4d1498..fc28e07 100644 *** a/src/pl/plpgsql/src/plpgsql.h --- b/src/pl/plpgsql/src/plpgsql.h *************** typedef struct *** 321,326 **** --- 321,327 ---- int16 elemtyplen; /* typlen of element type */ bool elemtypbyval; /* element type is pass-by-value? */ char elemtypalign; /* typalign of element type */ + bool elemtypiscomposite; /* element type is composite? */ } PLpgSQL_arrayelem;