granicus.if.org Git - postgresql/blob - src/backend/optimizer/prep/preptlist.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * preptlist.c
   4  *        Routines to preprocess the parse tree target list
   5  *
   6  * For INSERT and UPDATE queries, the targetlist must contain an entry for
   7  * each attribute of the target relation in the correct order.  For all query
   8  * types, we may need to add junk tlist entries for Vars used in the RETURNING
   9  * list and row ID information needed for EvalPlanQual checking.
  10  *
  11  * NOTE: the rewriter's rewriteTargetListIU and rewriteTargetListUD
  12  * routines also do preprocessing of the targetlist.  The division of labor
  13  * between here and there is a bit arbitrary and historical.
  14  *
  15  *
  16  * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
  17  * Portions Copyright (c) 1994, Regents of the University of California
  18  *
  19  * IDENTIFICATION
  20  *        src/backend/optimizer/prep/preptlist.c
  21  *
  22  *-------------------------------------------------------------------------
  23  */
  24
  25 #include "postgres.h"
  26
  27 #include "access/heapam.h"
  28 #include "access/sysattr.h"
  29 #include "catalog/pg_type.h"
  30 #include "nodes/makefuncs.h"
  31 #include "optimizer/prep.h"
  32 #include "optimizer/subselect.h"
  33 #include "optimizer/tlist.h"
  34 #include "optimizer/var.h"
  35 #include "parser/parsetree.h"
  36 #include "parser/parse_coerce.h"
  37 #include "utils/rel.h"
  38
  39
  40 static List *expand_targetlist(List *tlist, int command_type,
  41                                   Index result_relation, List *range_table);
  42
  43
  44 /*
  45  * preprocess_targetlist
  46  *        Driver for preprocessing the parse tree targetlist.
  47  *
  48  *        Returns the new targetlist.
  49  */
  50 List *
  51 preprocess_targetlist(PlannerInfo *root, List *tlist)
  52 {
  53         Query      *parse = root->parse;
  54         int                     result_relation = parse->resultRelation;
  55         List       *range_table = parse->rtable;
  56         CmdType         command_type = parse->commandType;
  57         ListCell   *lc;
  58
  59         /*
  60          * Sanity check: if there is a result relation, it'd better be a real
  61          * relation not a subquery.  Else parser or rewriter messed up.
  62          */
  63         if (result_relation)
  64         {
  65                 RangeTblEntry *rte = rt_fetch(result_relation, range_table);
  66
  67                 if (rte->subquery != NULL || rte->relid == InvalidOid)
  68                         elog(ERROR, "subquery cannot be result relation");
  69         }
  70
  71         /*
  72          * for heap_form_tuple to work, the targetlist must match the exact order
  73          * of the attributes. We also need to fill in any missing attributes. -ay
  74          * 10/94
  75          */
  76         if (command_type == CMD_INSERT || command_type == CMD_UPDATE)
  77                 tlist = expand_targetlist(tlist, command_type,
  78                                                                   result_relation, range_table);
  79
  80         /*
  81          * Add necessary junk columns for rowmarked rels.  These values are needed
  82          * for locking of rels selected FOR UPDATE/SHARE, and to do EvalPlanQual
  83          * rechecking.  While we are at it, store these junk attnos in the
  84          * PlanRowMark list so that we don't have to redetermine them at runtime.
  85          */
  86         foreach(lc, root->rowMarks)
  87         {
  88                 PlanRowMark *rc = (PlanRowMark *) lfirst(lc);
  89                 Var                *var;
  90                 char            resname[32];
  91                 TargetEntry *tle;
  92
  93                 /* child rels should just use the same junk attrs as their parents */
  94                 if (rc->rti != rc->prti)
  95                 {
  96                         PlanRowMark *prc = get_plan_rowmark(root->rowMarks, rc->prti);
  97
  98                         /* parent should have appeared earlier in list */
  99                         if (prc == NULL || prc->toidAttNo == InvalidAttrNumber)
 100                                 elog(ERROR, "parent PlanRowMark not processed yet");
 101                         rc->ctidAttNo = prc->ctidAttNo;
 102                         rc->toidAttNo = prc->toidAttNo;
 103                         continue;
 104                 }
 105
 106                 if (rc->markType != ROW_MARK_COPY)
 107                 {
 108                         /* It's a regular table, so fetch its TID */
 109                         var = makeVar(rc->rti,
 110                                                   SelfItemPointerAttributeNumber,
 111                                                   TIDOID,
 112                                                   -1,
 113                                                   0);
 114                         snprintf(resname, sizeof(resname), "ctid%u", rc->rti);
 115                         tle = makeTargetEntry((Expr *) var,
 116                                                                   list_length(tlist) + 1,
 117                                                                   pstrdup(resname),
 118                                                                   true);
 119                         tlist = lappend(tlist, tle);
 120                         rc->ctidAttNo = tle->resno;
 121
 122                         /* if parent of inheritance tree, need the tableoid too */
 123                         if (rc->isParent)
 124                         {
 125                                 var = makeVar(rc->rti,
 126                                                           TableOidAttributeNumber,
 127                                                           OIDOID,
 128                                                           -1,
 129                                                           0);
 130                                 snprintf(resname, sizeof(resname), "tableoid%u", rc->rti);
 131                                 tle = makeTargetEntry((Expr *) var,
 132                                                                           list_length(tlist) + 1,
 133                                                                           pstrdup(resname),
 134                                                                           true);
 135                                 tlist = lappend(tlist, tle);
 136                                 rc->toidAttNo = tle->resno;
 137                         }
 138                 }
 139                 else
 140                 {
 141                         /* Not a table, so we need the whole row as a junk var */
 142                         var = makeWholeRowVar(rt_fetch(rc->rti, range_table),
 143                                                                   rc->rti,
 144                                                                   0);
 145                         snprintf(resname, sizeof(resname), "wholerow%u", rc->rti);
 146                         tle = makeTargetEntry((Expr *) var,
 147                                                                   list_length(tlist) + 1,
 148                                                                   pstrdup(resname),
 149                                                                   true);
 150                         tlist = lappend(tlist, tle);
 151                         rc->wholeAttNo = tle->resno;
 152                 }
 153         }
 154
 155         /*
 156          * If the query has a RETURNING list, add resjunk entries for any Vars
 157          * used in RETURNING that belong to other relations.  We need to do this
 158          * to make these Vars available for the RETURNING calculation.  Vars that
 159          * belong to the result rel don't need to be added, because they will be
 160          * made to refer to the actual heap tuple.
 161          */
 162         if (parse->returningList && list_length(parse->rtable) > 1)
 163         {
 164                 List       *vars;
 165                 ListCell   *l;
 166
 167                 vars = pull_var_clause((Node *) parse->returningList,
 168                                                            PVC_INCLUDE_PLACEHOLDERS);
 169                 foreach(l, vars)
 170                 {
 171                         Var                *var = (Var *) lfirst(l);
 172                         TargetEntry *tle;
 173
 174                         if (IsA(var, Var) &&
 175                                 var->varno == result_relation)
 176                                 continue;               /* don't need it */
 177
 178                         if (tlist_member((Node *) var, tlist))
 179                                 continue;               /* already got it */
 180
 181                         tle = makeTargetEntry((Expr *) var,
 182                                                                   list_length(tlist) + 1,
 183                                                                   NULL,
 184                                                                   true);
 185
 186                         tlist = lappend(tlist, tle);
 187                 }
 188                 list_free(vars);
 189         }
 190
 191         return tlist;
 192 }
 193
 194 /*****************************************************************************
 195  *
 196  *              TARGETLIST EXPANSION
 197  *
 198  *****************************************************************************/
 199
 200 /*
 201  * expand_targetlist
 202  *        Given a target list as generated by the parser and a result relation,
 203  *        add targetlist entries for any missing attributes, and ensure the
 204  *        non-junk attributes appear in proper field order.
 205  */
 206 static List *
 207 expand_targetlist(List *tlist, int command_type,
 208                                   Index result_relation, List *range_table)
 209 {
 210         List       *new_tlist = NIL;
 211         ListCell   *tlist_item;
 212         Relation        rel;
 213         int                     attrno,
 214                                 numattrs;
 215
 216         tlist_item = list_head(tlist);
 217
 218         /*
 219          * The rewriter should have already ensured that the TLEs are in correct
 220          * order; but we have to insert TLEs for any missing attributes.
 221          *
 222          * Scan the tuple description in the relation's relcache entry to make
 223          * sure we have all the user attributes in the right order.  We assume
 224          * that the rewriter already acquired at least AccessShareLock on the
 225          * relation, so we need no lock here.
 226          */
 227         rel = heap_open(getrelid(result_relation, range_table), NoLock);
 228
 229         numattrs = RelationGetNumberOfAttributes(rel);
 230
 231         for (attrno = 1; attrno <= numattrs; attrno++)
 232         {
 233                 Form_pg_attribute att_tup = rel->rd_att->attrs[attrno - 1];
 234                 TargetEntry *new_tle = NULL;
 235
 236                 if (tlist_item != NULL)
 237                 {
 238                         TargetEntry *old_tle = (TargetEntry *) lfirst(tlist_item);
 239
 240                         if (!old_tle->resjunk && old_tle->resno == attrno)
 241                         {
 242                                 new_tle = old_tle;
 243                                 tlist_item = lnext(tlist_item);
 244                         }
 245                 }
 246
 247                 if (new_tle == NULL)
 248                 {
 249                         /*
 250                          * Didn't find a matching tlist entry, so make one.
 251                          *
 252                          * For INSERT, generate a NULL constant.  (We assume the rewriter
 253                          * would have inserted any available default value.) Also, if the
 254                          * column isn't dropped, apply any domain constraints that might
 255                          * exist --- this is to catch domain NOT NULL.
 256                          *
 257                          * For UPDATE, generate a Var reference to the existing value of
 258                          * the attribute, so that it gets copied to the new tuple. But
 259                          * generate a NULL for dropped columns (we want to drop any old
 260                          * values).
 261                          *
 262                          * When generating a NULL constant for a dropped column, we label
 263                          * it INT4 (any other guaranteed-to-exist datatype would do as
 264                          * well). We can't label it with the dropped column's datatype
 265                          * since that might not exist anymore.  It does not really matter
 266                          * what we claim the type is, since NULL is NULL --- its
 267                          * representation is datatype-independent.      This could perhaps
 268                          * confuse code comparing the finished plan to the target
 269                          * relation, however.
 270                          */
 271                         Oid                     atttype = att_tup->atttypid;
 272                         int32           atttypmod = att_tup->atttypmod;
 273                         Node       *new_expr;
 274
 275                         switch (command_type)
 276                         {
 277                                 case CMD_INSERT:
 278                                         if (!att_tup->attisdropped)
 279                                         {
 280                                                 new_expr = (Node *) makeConst(atttype,
 281                                                                                                           -1,
 282                                                                                                           att_tup->attlen,
 283                                                                                                           (Datum) 0,
 284                                                                                                           true,         /* isnull */
 285                                                                                                           att_tup->attbyval);
 286                                                 new_expr = coerce_to_domain(new_expr,
 287                                                                                                         InvalidOid, -1,
 288                                                                                                         atttype,
 289                                                                                                         COERCE_IMPLICIT_CAST,
 290                                                                                                         -1,
 291                                                                                                         false,
 292                                                                                                         false);
 293                                         }
 294                                         else
 295                                         {
 296                                                 /* Insert NULL for dropped column */
 297                                                 new_expr = (Node *) makeConst(INT4OID,
 298                                                                                                           -1,
 299                                                                                                           sizeof(int32),
 300                                                                                                           (Datum) 0,
 301                                                                                                           true,         /* isnull */
 302                                                                                                           true /* byval */ );
 303                                         }
 304                                         break;
 305                                 case CMD_UPDATE:
 306                                         if (!att_tup->attisdropped)
 307                                         {
 308                                                 new_expr = (Node *) makeVar(result_relation,
 309                                                                                                         attrno,
 310                                                                                                         atttype,
 311                                                                                                         atttypmod,
 312                                                                                                         0);
 313                                         }
 314                                         else
 315                                         {
 316                                                 /* Insert NULL for dropped column */
 317                                                 new_expr = (Node *) makeConst(INT4OID,
 318                                                                                                           -1,
 319                                                                                                           sizeof(int32),
 320                                                                                                           (Datum) 0,
 321                                                                                                           true,         /* isnull */
 322                                                                                                           true /* byval */ );
 323                                         }
 324                                         break;
 325                                 default:
 326                                         elog(ERROR, "unrecognized command_type: %d",
 327                                                  (int) command_type);
 328                                         new_expr = NULL;        /* keep compiler quiet */
 329                                         break;
 330                         }
 331
 332                         new_tle = makeTargetEntry((Expr *) new_expr,
 333                                                                           attrno,
 334                                                                           pstrdup(NameStr(att_tup->attname)),
 335                                                                           false);
 336                 }
 337
 338                 new_tlist = lappend(new_tlist, new_tle);
 339         }
 340
 341         /*
 342          * The remaining tlist entries should be resjunk; append them all to the
 343          * end of the new tlist, making sure they have resnos higher than the last
 344          * real attribute.      (Note: although the rewriter already did such
 345          * renumbering, we have to do it again here in case we are doing an UPDATE
 346          * in a table with dropped columns, or an inheritance child table with
 347          * extra columns.)
 348          */
 349         while (tlist_item)
 350         {
 351                 TargetEntry *old_tle = (TargetEntry *) lfirst(tlist_item);
 352
 353                 if (!old_tle->resjunk)
 354                         elog(ERROR, "targetlist is not sorted correctly");
 355                 /* Get the resno right, but don't copy unnecessarily */
 356                 if (old_tle->resno != attrno)
 357                 {
 358                         old_tle = flatCopyTargetEntry(old_tle);
 359                         old_tle->resno = attrno;
 360                 }
 361                 new_tlist = lappend(new_tlist, old_tle);
 362                 attrno++;
 363                 tlist_item = lnext(tlist_item);
 364         }
 365
 366         heap_close(rel, NoLock);
 367
 368         return new_tlist;
 369 }
 370
 371
 372 /*
 373  * Locate PlanRowMark for given RT index, or return NULL if none
 374  *
 375  * This probably ought to be elsewhere, but there's no very good place
 376  */
 377 PlanRowMark *
 378 get_plan_rowmark(List *rowmarks, Index rtindex)
 379 {
 380         ListCell   *l;
 381
 382         foreach(l, rowmarks)
 383         {
 384                 PlanRowMark *rc = (PlanRowMark *) lfirst(l);
 385
 386                 if (rc->rti == rtindex)
 387                         return rc;
 388         }
 389         return NULL;
 390 }