OSDN Git Service

Make the world very nearly safe for composite-type columns in tables.
authorTom Lane <tgl@sss.pgh.pa.us>
Sat, 5 Jun 2004 01:55:05 +0000 (01:55 +0000)
committerTom Lane <tgl@sss.pgh.pa.us>
Sat, 5 Jun 2004 01:55:05 +0000 (01:55 +0000)
1. Solve the problem of not having TOAST references hiding inside composite
values by establishing the rule that toasting only goes one level deep:
a tuple can contain toasted fields, but a composite-type datum that is
to be inserted into a tuple cannot.  Enforcing this in heap_formtuple
is relatively cheap and it avoids a large increase in the cost of running
the tuptoaster during final storage of a row.
2. Fix some interesting problems in expansion of inherited queries that
reference whole-row variables.  We never really did this correctly before,
but it's now relatively painless to solve by expanding the parent's
whole-row Var into a RowExpr() selecting the proper columns from the
child.
If you dike out the preventive check in CheckAttributeType(),
composite-type columns now seem to actually work.  However, we surely
cannot ship them like this --- without I/O for composite types, you
can't get pg_dump to dump tables containing them.  So a little more
work still to do.

12 files changed:
src/backend/access/common/heaptuple.c
src/backend/access/heap/tuptoaster.c
src/backend/optimizer/path/allpaths.c
src/backend/optimizer/path/costsize.c
src/backend/optimizer/path/pathkeys.c
src/backend/optimizer/prep/prepunion.c
src/backend/optimizer/util/relnode.c
src/backend/optimizer/util/tlist.c
src/backend/utils/cache/typcache.c
src/include/access/tuptoaster.h
src/include/nodes/relation.h
src/include/utils/typcache.h

index 266cf3b..88a9631 100644 (file)
@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/access/common/heaptuple.c,v 1.91 2004/06/04 20:35:21 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/access/common/heaptuple.c,v 1.92 2004/06/05 01:55:04 tgl Exp $
  *
  * NOTES
  *       The old interface functions have been converted to macros
@@ -21,6 +21,7 @@
 #include "postgres.h"
 
 #include "access/heapam.h"
+#include "access/tuptoaster.h"
 #include "catalog/pg_type.h"
 
 
@@ -567,8 +568,9 @@ heap_formtuple(TupleDesc tupleDescriptor,
        unsigned long len;
        int                     hoff;
        bool            hasnull = false;
-       int                     i;
+       Form_pg_attribute *att = tupleDescriptor->attrs;
        int                     numberOfAttributes = tupleDescriptor->natts;
+       int                     i;
 
        if (numberOfAttributes > MaxTupleAttributeNumber)
                ereport(ERROR,
@@ -577,17 +579,34 @@ heap_formtuple(TupleDesc tupleDescriptor,
                                                numberOfAttributes, MaxTupleAttributeNumber)));
 
        /*
-        * Determine total space needed
+        * 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 (nulls[i] != ' ')
-               {
                        hasnull = true;
-                       break;
+               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);
                }
        }
 
+       /*
+        * Determine total space needed
+        */
        len = offsetof(HeapTupleHeaderData, t_bits);
 
        if (hasnull)
@@ -744,7 +763,11 @@ heap_deformtuple(HeapTuple tuple,
        bool            slow = false;   /* can we use/set attcacheoff? */
 
        natts = tup->t_natts;
-       /* This min() operation is pure paranoia */
+       /*
+        * In inheritance situations, it is possible that the given tuple actually
+        * has more fields than the caller is expecting.  Don't run off the end
+        * of the caller's arrays.
+        */
        natts = Min(natts, tdesc_natts);
 
        tp = (char *) tup + tup->t_hoff;
index f206a3f..c36cc42 100644 (file)
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/access/heap/tuptoaster.c,v 1.42 2004/06/04 20:35:21 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/access/heap/tuptoaster.c,v 1.43 2004/06/05 01:55:04 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -35,6 +35,7 @@
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/pg_lzcompress.h"
+#include "utils/typcache.h"
 
 
 #undef TOAST_DEBUG
@@ -458,10 +459,10 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup)
                         * still in the tuple must be someone else's we cannot reuse.
                         * Expand it to plain (and, probably, toast it again below).
                         */
-                       if (VARATT_IS_EXTERNAL(DatumGetPointer(toast_values[i])))
+                       if (VARATT_IS_EXTERNAL(new_value))
                        {
-                               toast_values[i] = PointerGetDatum(heap_tuple_untoast_attr(
-                                               (varattrib *) DatumGetPointer(toast_values[i])));
+                               new_value = heap_tuple_untoast_attr(new_value);
+                               toast_values[i] = PointerGetDatum(new_value);
                                toast_free[i] = true;
                                need_change = true;
                                need_free = true;
@@ -470,7 +471,7 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup)
                        /*
                         * Remember the size of this attribute
                         */
-                       toast_sizes[i] = VARATT_SIZE(DatumGetPointer(toast_values[i]));
+                       toast_sizes[i] = VARATT_SIZE(new_value);
                }
                else
                {
@@ -786,6 +787,128 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup)
 
 
 /* ----------
+ * 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.
+ * ----------
+ */
+Datum
+toast_flatten_tuple_attribute(Datum value,
+                                                         Oid typeId, int32 typeMod)
+{
+       TupleDesc       tupleDesc;
+       HeapTupleHeader olddata;
+       HeapTupleHeader new_data;
+       int32           new_len;
+       HeapTupleData tmptup;
+       Form_pg_attribute *att;
+       int                     numAttrs;
+       int                     i;
+       bool            need_change = false;
+       bool            has_nulls = false;
+       Datum           toast_values[MaxTupleAttributeNumber];
+       char            toast_nulls[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_deformtuple(&tmptup, tupleDesc, toast_values, toast_nulls);
+
+       memset(toast_free, 0, numAttrs * sizeof(bool));
+
+       for (i = 0; i < numAttrs; i++)
+       {
+               /*
+                * Look at non-null varlena attributes
+                */
+               if (toast_nulls[i] == 'n')
+                       has_nulls = true;
+               else if (att[i]->attlen == -1)
+               {
+                       varattrib  *new_value;
+
+                       new_value = (varattrib *) DatumGetPointer(toast_values[i]);
+                       if (VARATT_IS_EXTENDED(new_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)
+               return value;
+
+       /*
+        * Calculate the new size of the tuple.  Header size should not
+        * change, but data size might.
+        */
+       new_len = offsetof(HeapTupleHeaderData, t_bits);
+       if (has_nulls)
+               new_len += BITMAPLEN(numAttrs);
+       if (olddata->t_infomask & HEAP_HASOID)
+               new_len += sizeof(Oid);
+       new_len = MAXALIGN(new_len);
+       Assert(new_len == olddata->t_hoff);
+       new_len += ComputeDataSize(tupleDesc, toast_values, toast_nulls);
+
+       new_data = (HeapTupleHeader) palloc0(new_len);
+
+       /*
+        * Put the tuple header and the changed values into place
+        */
+       memcpy(new_data, olddata, olddata->t_hoff);
+
+       HeapTupleHeaderSetDatumLength(new_data, new_len);
+
+       DataFill((char *) new_data + olddata->t_hoff,
+                        tupleDesc,
+                        toast_values,
+                        toast_nulls,
+                        &(new_data->t_infomask),
+                        has_nulls ? new_data->t_bits : NULL);
+
+       /*
+        * Free allocated temp values
+        */
+       for (i = 0; i < numAttrs; i++)
+               if (toast_free[i])
+                       pfree(DatumGetPointer(toast_values[i]));
+
+       return PointerGetDatum(new_data);
+}
+
+
+/* ----------
  * toast_compress_datum -
  *
  *     Create a compressed version of a varlena datum
index dc428ab..24a5114 100644 (file)
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/optimizer/path/allpaths.c,v 1.117 2004/06/01 03:02:51 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/optimizer/path/allpaths.c,v 1.118 2004/06/05 01:55:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -302,11 +302,15 @@ set_inherited_rel_pathlist(Query *root, RelOptInfo *rel,
                {
                        Var                *parentvar = (Var *) lfirst(parentvars);
                        Var                *childvar = (Var *) lfirst(childvars);
-                       int                     parentndx = parentvar->varattno - rel->min_attr;
-                       int                     childndx = childvar->varattno - childrel->min_attr;
 
-                       if (childrel->attr_widths[childndx] > rel->attr_widths[parentndx])
-                               rel->attr_widths[parentndx] = childrel->attr_widths[childndx];
+                       if (IsA(parentvar, Var) && IsA(childvar, Var))
+                       {
+                               int                     pndx = parentvar->varattno - rel->min_attr;
+                               int                     cndx = childvar->varattno - childrel->min_attr;
+
+                               if (childrel->attr_widths[cndx] > rel->attr_widths[pndx])
+                                       rel->attr_widths[pndx] = childrel->attr_widths[cndx];
+                       }
                }
        }
 
index 46a323f..53bd8bd 100644 (file)
@@ -49,7 +49,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/optimizer/path/costsize.c,v 1.129 2004/06/01 03:02:52 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/optimizer/path/costsize.c,v 1.130 2004/06/05 01:55:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1704,11 +1704,18 @@ set_rel_width(Query *root, RelOptInfo *rel)
        foreach(tllist, rel->reltargetlist)
        {
                Var                *var = (Var *) lfirst(tllist);
-               int                     ndx = var->varattno - rel->min_attr;
+               int                     ndx;
                Oid                     relid;
                int32           item_width;
 
-               Assert(IsA(var, Var));
+               /* For now, punt on whole-row child Vars */
+               if (!IsA(var, Var))
+               {
+                       tuple_width += 32;      /* arbitrary */
+                       continue;
+               }
+
+               ndx = var->varattno - rel->min_attr;
 
                /*
                 * The width probably hasn't been cached yet, but may as well
index 363fe54..71f96c1 100644 (file)
@@ -11,7 +11,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/optimizer/path/pathkeys.c,v 1.59 2004/06/01 03:02:52 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/optimizer/path/pathkeys.c,v 1.60 2004/06/05 01:55:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -725,7 +725,8 @@ find_indexkey_var(Query *root, RelOptInfo *rel, AttrNumber varattno)
        {
                Var                *var = (Var *) lfirst(temp);
 
-               if (var->varattno == varattno)
+               if (IsA(var, Var) &&
+                       var->varattno == varattno)
                        return var;
        }
 
index af2bb7e..abc5af7 100644 (file)
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/optimizer/prep/prepunion.c,v 1.112 2004/05/30 23:40:29 neilc Exp $
+ *       $PostgreSQL: pgsql/src/backend/optimizer/prep/prepunion.c,v 1.113 2004/06/05 01:55:04 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
 
 
+#include "access/heapam.h"
 #include "catalog/pg_type.h"
 #include "nodes/makefuncs.h"
 #include "optimizer/clauses.h"
@@ -39,8 +40,10 @@ typedef struct
 {
        Index           old_rt_index;
        Index           new_rt_index;
-       Oid                     old_relid;
-       Oid                     new_relid;
+       TupleDesc       old_tupdesc;
+       TupleDesc       new_tupdesc;
+       char       *old_rel_name;
+       char       *new_rel_name;
 } adjust_inherited_attrs_context;
 
 static Plan *recurse_set_operations(Node *setOp, Query *parse,
@@ -65,7 +68,8 @@ static bool tlist_same_datatypes(List *tlist, List *colTypes, bool junkOK);
 static Node *adjust_inherited_attrs_mutator(Node *node,
                                                           adjust_inherited_attrs_context *context);
 static Relids adjust_relid_set(Relids relids, Index oldrelid, Index newrelid);
-static List *adjust_inherited_tlist(List *tlist, Oid old_relid, Oid new_relid);
+static List *adjust_inherited_tlist(List *tlist,
+                                                                       adjust_inherited_attrs_context *context);
 
 
 /*
@@ -787,17 +791,17 @@ expand_inherited_rtentry(Query *parse, Index rti, bool dup_parent)
  * We also adjust varattno to match the new table by column name, rather
  * than column number. This hack makes it possible for child tables to have
  * different column positions for the "same" attribute as a parent, which
- * helps ALTER TABLE ADD COLUMN.  Unfortunately this isn't nearly enough to
- * make it work transparently; there are other places where things fall down
- * if children and parents don't have the same column numbers for inherited
- * attributes. It'd be better to rip this code out and fix ALTER TABLE...
+ * is necessary for ALTER TABLE ADD COLUMN.
  */
 Node *
 adjust_inherited_attrs(Node *node,
                                           Index old_rt_index, Oid old_relid,
                                           Index new_rt_index, Oid new_relid)
 {
+       Node       *result;
        adjust_inherited_attrs_context context;
+       Relation        oldrelation;
+       Relation        newrelation;
 
        /* Handle simple case simply... */
        if (old_rt_index == new_rt_index)
@@ -806,10 +810,19 @@ adjust_inherited_attrs(Node *node,
                return copyObject(node);
        }
 
+       /*
+        * We assume that by now the planner has acquired at least AccessShareLock
+        * on both rels, and so we need no additional lock now.
+        */
+       oldrelation = heap_open(old_relid, NoLock);
+       newrelation = heap_open(new_relid, NoLock);
+
        context.old_rt_index = old_rt_index;
        context.new_rt_index = new_rt_index;
-       context.old_relid = old_relid;
-       context.new_relid = new_relid;
+       context.old_tupdesc = RelationGetDescr(oldrelation);
+       context.new_tupdesc = RelationGetDescr(newrelation);
+       context.old_rel_name = RelationGetRelationName(oldrelation);
+       context.new_rel_name = RelationGetRelationName(newrelation);
 
        /*
         * Must be prepared to start with a Query or a bare expression tree.
@@ -829,13 +842,109 @@ adjust_inherited_attrs(Node *node,
                        if (newnode->commandType == CMD_UPDATE)
                                newnode->targetList =
                                        adjust_inherited_tlist(newnode->targetList,
-                                                                                  old_relid,
-                                                                                  new_relid);
+                                                                                  &context);
                }
-               return (Node *) newnode;
+               result = (Node *) newnode;
        }
        else
-               return adjust_inherited_attrs_mutator(node, &context);
+               result = adjust_inherited_attrs_mutator(node, &context);
+
+       heap_close(oldrelation, NoLock);
+       heap_close(newrelation, NoLock);
+
+       return result;
+}
+
+/*
+ * Translate parent's attribute number into child's.
+ *
+ * For paranoia's sake, we match type as well as attribute name.
+ */
+static AttrNumber
+translate_inherited_attnum(AttrNumber old_attno,
+                                                  adjust_inherited_attrs_context *context)
+{
+       Form_pg_attribute att;
+       char       *attname;
+       Oid                     atttypid;
+       int32           atttypmod;
+       int                     newnatts;
+       int                     i;
+
+       if (old_attno <= 0 || old_attno > context->old_tupdesc->natts)
+               elog(ERROR, "attribute %d of relation \"%s\" does not exist",
+                        (int) old_attno, context->old_rel_name);
+       att = context->old_tupdesc->attrs[old_attno - 1];
+       if (att->attisdropped)
+               elog(ERROR, "attribute %d of relation \"%s\" does not exist",
+                        (int) old_attno, context->old_rel_name);
+       attname = NameStr(att->attname);
+       atttypid = att->atttypid;
+       atttypmod = att->atttypmod;
+
+       newnatts = context->new_tupdesc->natts;
+       for (i = 0; i < newnatts; i++)
+       {
+               att = context->new_tupdesc->attrs[i];
+               if (att->attisdropped)
+                       continue;
+               if (strcmp(attname, NameStr(att->attname)) == 0)
+               {
+                       /* Found it, check type */
+                       if (atttypid != att->atttypid || atttypmod != att->atttypmod)
+                               elog(ERROR, "attribute \"%s\" of relation \"%s\" does not match parent's type",
+                                        attname, context->new_rel_name);
+                       return (AttrNumber) (i + 1);
+               }
+       }
+
+       elog(ERROR, "attribute \"%s\" of relation \"%s\" does not exist",
+                attname, context->new_rel_name);
+       return 0;                                       /* keep compiler quiet */
+}
+
+/*
+ * Translate a whole-row Var to be correct for a child table.
+ *
+ * In general the child will not have a suitable field layout to be used
+ * directly, so we translate the simple whole-row Var into a ROW() construct.
+ */
+static Node *
+generate_whole_row(Var *var,
+                                  adjust_inherited_attrs_context *context)
+{
+       RowExpr         *rowexpr;
+       List            *fields = NIL;
+       int                     oldnatts = context->old_tupdesc->natts;
+       int                     i;
+
+       for (i = 0; i < oldnatts; i++)
+       {
+               Form_pg_attribute att = context->old_tupdesc->attrs[i];
+               Var             *newvar;
+
+               if (att->attisdropped)
+               {
+                       /*
+                        * can't use atttypid here, but it doesn't really matter
+                        * what type the Const claims to be.
+                        */
+                       newvar = (Var *) makeNullConst(INT4OID);
+               }
+               else
+                       newvar = makeVar(context->new_rt_index,
+                                                        translate_inherited_attnum(i + 1, context),
+                                                        att->atttypid,
+                                                        att->atttypmod,
+                                                        0);
+               fields = lappend(fields, newvar);
+       }
+       rowexpr = makeNode(RowExpr);
+       rowexpr->args = fields;
+       rowexpr->row_typeid = var->vartype;     /* report parent's rowtype */
+       rowexpr->row_format = COERCE_IMPLICIT_CAST;
+
+       return (Node *) rowexpr;
 }
 
 static Node *
@@ -855,17 +964,16 @@ adjust_inherited_attrs_mutator(Node *node,
                        var->varnoold = context->new_rt_index;
                        if (var->varattno > 0)
                        {
-                               char       *attname;
-
-                               attname = get_relid_attribute_name(context->old_relid,
-                                                                                                  var->varattno);
-                               var->varattno = get_attnum(context->new_relid, attname);
-                               if (var->varattno == InvalidAttrNumber)
-                                       elog(ERROR, "attribute \"%s\" of relation \"%s\" does not exist",
-                                                attname, get_rel_name(context->new_relid));
+                               var->varattno = translate_inherited_attnum(var->varattno,
+                                                                                                                  context);
                                var->varoattno = var->varattno;
-                               pfree(attname);
                        }
+                       else if (var->varattno == 0)
+                       {
+                               /* expand whole-row reference into a ROW() construct */
+                               return generate_whole_row(var, context);
+                       }
+                       /* system attributes don't need any translation */
                }
                return (Node *) var;
        }
@@ -1022,7 +1130,8 @@ adjust_relid_set(Relids relids, Index oldrelid, Index newrelid)
  * Note that this is not needed for INSERT because INSERT isn't inheritable.
  */
 static List *
-adjust_inherited_tlist(List *tlist, Oid old_relid, Oid new_relid)
+adjust_inherited_tlist(List *tlist,
+                                          adjust_inherited_attrs_context *context)
 {
        bool            changed_it = false;
        ListCell   *tl;
@@ -1035,26 +1144,19 @@ adjust_inherited_tlist(List *tlist, Oid old_relid, Oid new_relid)
        {
                TargetEntry *tle = (TargetEntry *) lfirst(tl);
                Resdom     *resdom = tle->resdom;
-               char       *attname;
 
                if (resdom->resjunk)
                        continue;                       /* ignore junk items */
 
-               attname = get_relid_attribute_name(old_relid, resdom->resno);
-               attrno = get_attnum(new_relid, attname);
-               if (attrno == InvalidAttrNumber)
-                       elog(ERROR, "attribute \"%s\" of relation \"%s\" does not exist",
-                                attname, get_rel_name(new_relid));
+               attrno = translate_inherited_attnum(resdom->resno, context);
+
                if (resdom->resno != attrno)
                {
                        resdom = (Resdom *) copyObject((Node *) resdom);
                        resdom->resno = attrno;
-                       resdom->resname = attname;
                        tle->resdom = resdom;
                        changed_it = true;
                }
-               else
-                       pfree(attname);
        }
 
        /*
index 36e6881..58d4f58 100644 (file)
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/optimizer/util/relnode.c,v 1.59 2004/06/01 03:03:02 tgl Exp $
+ *       $PostgreSQL: pgsql/src/backend/optimizer/util/relnode.c,v 1.60 2004/06/05 01:55:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -381,6 +381,9 @@ build_joinrel_tlist(Query *root, RelOptInfo *joinrel)
                        Var                *var = (Var *) lfirst(vars);
                        int                     ndx = var->varattno - baserel->min_attr;
 
+                       /* We can't run into any child RowExprs here */
+                       Assert(IsA(var, Var));
+
                        if (bms_nonempty_difference(baserel->attr_needed[ndx], relids))
                        {
                                joinrel->reltargetlist = lappend(joinrel->reltargetlist, var);
index dcee4f8..4878ebb 100644 (file)
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/optimizer/util/tlist.c,v 1.64 2004/05/30 23:40:31 neilc Exp $
+ *       $PostgreSQL: pgsql/src/backend/optimizer/util/tlist.c,v 1.65 2004/06/05 01:55:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -17,6 +17,7 @@
 #include "nodes/makefuncs.h"
 #include "optimizer/tlist.h"
 #include "optimizer/var.h"
+#include "parser/parse_expr.h"
 
 
 /*****************************************************************************
@@ -83,13 +84,28 @@ tlist_member(Node *node, List *targetlist)
  * create_tl_element
  *       Creates a target list entry node and its associated (resdom var) pair
  *       with its resdom number equal to 'resdomno'.
+ *
+ * Note: the argument is almost always a Var, but occasionally not.
  */
 TargetEntry *
 create_tl_element(Var *var, int resdomno)
 {
+       Oid             vartype;
+       int32   vartypmod;
+
+       if (IsA(var, Var))
+       {
+               vartype = var->vartype;
+               vartypmod = var->vartypmod;
+       }
+       else
+       {
+               vartype = exprType((Node *) var);
+               vartypmod = exprTypmod((Node *) var);
+       }
        return makeTargetEntry(makeResdom(resdomno,
-                                                                         var->vartype,
-                                                                         var->vartypmod,
+                                                                         vartype,
+                                                                         vartypmod,
                                                                          NULL,
                                                                          false),
                                                   (Expr *) var);
index b3ef61c..8ad6042 100644 (file)
@@ -36,7 +36,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *       $PostgreSQL: pgsql/src/backend/utils/cache/typcache.c,v 1.6 2004/05/26 04:41:40 neilc Exp $
+ *       $PostgreSQL: pgsql/src/backend/utils/cache/typcache.c,v 1.7 2004/06/05 01:55:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -385,6 +385,19 @@ lookup_default_opclass(Oid type_id, Oid am_id)
 TupleDesc
 lookup_rowtype_tupdesc(Oid type_id, int32 typmod)
 {
+       return lookup_rowtype_tupdesc_noerror(type_id, typmod, false);
+}
+
+/*
+ * lookup_rowtype_tupdesc_noerror
+ *
+ * As above, but if the type is not a known composite type and noError
+ * is true, returns NULL instead of ereport'ing.  (Note that if a bogus
+ * type_id is passed, you'll get an ereport anyway.)
+ */
+TupleDesc
+lookup_rowtype_tupdesc_noerror(Oid type_id, int32 typmod, bool noError)
+{
        if (type_id != RECORDOID)
        {
                /*
@@ -393,8 +406,7 @@ lookup_rowtype_tupdesc(Oid type_id, int32 typmod)
                TypeCacheEntry *typentry;
 
                typentry = lookup_type_cache(type_id, TYPECACHE_TUPDESC);
-               /* this should not happen unless caller messed up: */
-               if (typentry->tupDesc == NULL)
+               if (typentry->tupDesc == NULL && !noError)
                        ereport(ERROR,
                                        (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                                         errmsg("type %u is not composite",
@@ -408,9 +420,11 @@ lookup_rowtype_tupdesc(Oid type_id, int32 typmod)
                 */
                if (typmod < 0 || typmod >= NextRecordTypmod)
                {
-                       ereport(ERROR,
-                                       (errcode(ERRCODE_WRONG_OBJECT_TYPE),
-                                        errmsg("record type has not been registered")));
+                       if (!noError)
+                               ereport(ERROR,
+                                               (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+                                                errmsg("record type has not been registered")));
+                       return NULL;
                }
                return RecordCacheArray[typmod];
        }
index 2e2fcc9..3332e9a 100644 (file)
@@ -6,7 +6,7 @@
  *
  * Copyright (c) 2000-2003, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/include/access/tuptoaster.h,v 1.17 2003/11/29 22:40:55 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/access/tuptoaster.h,v 1.18 2004/06/05 01:55:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -111,6 +111,18 @@ extern varattrib *heap_tuple_untoast_attr_slice(varattrib *attr,
                                                          int32 slicelength);
 
 /* ----------
+ * 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 -
  *
  *     Create a compressed version of a varlena datum, if possible
index 91114fb..a75ddb8 100644 (file)
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/nodes/relation.h,v 1.95 2004/06/01 03:03:05 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/nodes/relation.h,v 1.96 2004/06/05 01:55:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -91,6 +91,7 @@ typedef struct QualCost
  *                             appropriate projections have been done (ie, output width)
  *             reltargetlist - List of Var nodes for the attributes we need to
  *                                             output from this relation (in no particular order)
+ *                                             NOTE: in a child relation, may contain RowExprs
  *             pathlist - List of Path nodes, one for each potentially useful
  *                                method of generating the relation
  *             cheapest_startup_path - the pathlist member with lowest startup cost
index 8b85f51..0856fdd 100644 (file)
@@ -9,7 +9,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/utils/typcache.h,v 1.3 2004/04/01 21:28:46 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/utils/typcache.h,v 1.4 2004/06/05 01:55:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -76,6 +76,9 @@ extern TypeCacheEntry *lookup_type_cache(Oid type_id, int flags);
 
 extern TupleDesc lookup_rowtype_tupdesc(Oid type_id, int32 typmod);
 
+extern TupleDesc lookup_rowtype_tupdesc_noerror(Oid type_id, int32 typmod,
+                                                                                               bool noError);
+
 extern void assign_record_type_typmod(TupleDesc tupDesc);
 
 extern void flush_rowtype_cache(Oid type_id);