src/backend/executor/execCurrent.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * execCurrent.c
   4  *        executor support for WHERE CURRENT OF cursor
   5  *
   6  * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *      $PostgreSQL$
  10  *
  11  *-------------------------------------------------------------------------
  12  */
  13 #include "postgres.h"
  14
  15 #include "access/sysattr.h"
  16 #include "catalog/pg_type.h"
  17 #include "executor/executor.h"
  18 #include "utils/builtins.h"
  19 #include "utils/lsyscache.h"
  20 #include "utils/portal.h"
  21
  22
  23 static char *fetch_param_value(ExprContext *econtext, int paramId);
  24 static ScanState *search_plan_tree(PlanState *node, Oid table_oid);
  25
  26
  27 /*
  28  * execCurrentOf
  29  *
  30  * Given a CURRENT OF expression and the OID of a table, determine which row
  31  * of the table is currently being scanned by the cursor named by CURRENT OF,
  32  * and return the row's TID into *current_tid.
  33  *
  34  * Returns TRUE if a row was identified.  Returns FALSE if the cursor is valid
  35  * for the table but is not currently scanning a row of the table (this is a
  36  * legal situation in inheritance cases).  Raises error if cursor is not a
  37  * valid updatable scan of the specified table.
  38  */
  39 bool
  40 execCurrentOf(CurrentOfExpr *cexpr,
  41                           ExprContext *econtext,
  42                           Oid table_oid,
  43                           ItemPointer current_tid)
  44 {
  45         char       *cursor_name;
  46         char       *table_name;
  47         Portal          portal;
  48         QueryDesc  *queryDesc;
  49
  50         /* Get the cursor name --- may have to look up a parameter reference */
  51         if (cexpr->cursor_name)
  52                 cursor_name = cexpr->cursor_name;
  53         else
  54                 cursor_name = fetch_param_value(econtext, cexpr->cursor_param);
  55
  56         /* Fetch table name for possible use in error messages */
  57         table_name = get_rel_name(table_oid);
  58         if (table_name == NULL)
  59                 elog(ERROR, "cache lookup failed for relation %u", table_oid);
  60
  61         /* Find the cursor's portal */
  62         portal = GetPortalByName(cursor_name);
  63         if (!PortalIsValid(portal))
  64                 ereport(ERROR,
  65                                 (errcode(ERRCODE_UNDEFINED_CURSOR),
  66                                  errmsg("cursor \"%s\" does not exist", cursor_name)));
  67
  68         /*
  69          * We have to watch out for non-SELECT queries as well as held cursors,
  70          * both of which may have null queryDesc.
  71          */
  72         if (portal->strategy != PORTAL_ONE_SELECT)
  73                 ereport(ERROR,
  74                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  75                                  errmsg("cursor \"%s\" is not a SELECT query",
  76                                                 cursor_name)));
  77         queryDesc = PortalGetQueryDesc(portal);
  78         if (queryDesc == NULL || queryDesc->estate == NULL)
  79                 ereport(ERROR,
  80                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  81                                  errmsg("cursor \"%s\" is held from a previous transaction",
  82                                                 cursor_name)));
  83
  84         /*
  85          * We have two different strategies depending on whether the cursor uses
  86          * FOR UPDATE/SHARE or not.  The reason for supporting both is that the
  87          * FOR UPDATE code is able to identify a target table in many cases where
  88          * the other code can't, while the non-FOR-UPDATE case allows use of WHERE
  89          * CURRENT OF with an insensitive cursor.
  90          */
  91         if (queryDesc->estate->es_rowMarks)
  92         {
  93                 ExecRowMark *erm;
  94                 ListCell   *lc;
  95
  96                 /*
  97                  * Here, the query must have exactly one FOR UPDATE/SHARE reference to
  98                  * the target table, and we dig the ctid info out of that.
  99                  */
 100                 erm = NULL;
 101                 foreach(lc, queryDesc->estate->es_rowMarks)
 102                 {
 103                         ExecRowMark *thiserm = (ExecRowMark *) lfirst(lc);
 104
 105                         if (RelationGetRelid(thiserm->relation) == table_oid)
 106                         {
 107                                 if (erm)
 108                                         ereport(ERROR,
 109                                                         (errcode(ERRCODE_INVALID_CURSOR_STATE),
 110                                                          errmsg("cursor \"%s\" has multiple FOR UPDATE/SHARE references to table \"%s\"",
 111                                                                         cursor_name, table_name)));
 112                                 erm = thiserm;
 113                         }
 114                 }
 115
 116                 if (erm == NULL)
 117                         ereport(ERROR,
 118                                         (errcode(ERRCODE_INVALID_CURSOR_STATE),
 119                                          errmsg("cursor \"%s\" does not have a FOR UPDATE/SHARE reference to table \"%s\"",
 120                                                         cursor_name, table_name)));
 121
 122                 /*
 123                  * The cursor must have a current result row: per the SQL spec, it's
 124                  * an error if not.
 125                  */
 126                 if (portal->atStart || portal->atEnd)
 127                         ereport(ERROR,
 128                                         (errcode(ERRCODE_INVALID_CURSOR_STATE),
 129                                          errmsg("cursor \"%s\" is not positioned on a row",
 130                                                         cursor_name)));
 131
 132                 /* Return the currently scanned TID, if there is one */
 133                 if (ItemPointerIsValid(&(erm->curCtid)))
 134                 {
 135                         *current_tid = erm->curCtid;
 136                         return true;
 137                 }
 138
 139                 /*
 140                  * This table didn't produce the cursor's current row; some other
 141                  * inheritance child of the same parent must have.      Signal caller to
 142                  * do nothing on this table.
 143                  */
 144                 return false;
 145         }
 146         else
 147         {
 148                 ScanState  *scanstate;
 149                 bool            lisnull;
 150                 Oid                     tuple_tableoid;
 151                 ItemPointer tuple_tid;
 152
 153                 /*
 154                  * Without FOR UPDATE, we dig through the cursor's plan to find the
 155                  * scan node.  Fail if it's not there or buried underneath
 156                  * aggregation.
 157                  */
 158                 scanstate = search_plan_tree(ExecGetActivePlanTree(queryDesc),
 159                                                                          table_oid);
 160                 if (!scanstate)
 161                         ereport(ERROR,
 162                                         (errcode(ERRCODE_INVALID_CURSOR_STATE),
 163                                          errmsg("cursor \"%s\" is not a simply updatable scan of table \"%s\"",
 164                                                         cursor_name, table_name)));
 165
 166                 /*
 167                  * The cursor must have a current result row: per the SQL spec, it's
 168                  * an error if not.  We test this at the top level, rather than at the
 169                  * scan node level, because in inheritance cases any one table scan
 170                  * could easily not be on a row. We want to return false, not raise
 171                  * error, if the passed-in table OID is for one of the inactive scans.
 172                  */
 173                 if (portal->atStart || portal->atEnd)
 174                         ereport(ERROR,
 175                                         (errcode(ERRCODE_INVALID_CURSOR_STATE),
 176                                          errmsg("cursor \"%s\" is not positioned on a row",
 177                                                         cursor_name)));
 178
 179                 /* Now OK to return false if we found an inactive scan */
 180                 if (TupIsNull(scanstate->ss_ScanTupleSlot))
 181                         return false;
 182
 183                 /* Use slot_getattr to catch any possible mistakes */
 184                 tuple_tableoid =
 185                         DatumGetObjectId(slot_getattr(scanstate->ss_ScanTupleSlot,
 186                                                                                   TableOidAttributeNumber,
 187                                                                                   &lisnull));
 188                 Assert(!lisnull);
 189                 tuple_tid = (ItemPointer)
 190                         DatumGetPointer(slot_getattr(scanstate->ss_ScanTupleSlot,
 191                                                                                  SelfItemPointerAttributeNumber,
 192                                                                                  &lisnull));
 193                 Assert(!lisnull);
 194
 195                 Assert(tuple_tableoid == table_oid);
 196
 197                 *current_tid = *tuple_tid;
 198
 199                 return true;
 200         }
 201 }
 202
 203 /*
 204  * fetch_param_value
 205  *
 206  * Fetch the string value of a param, verifying it is of type REFCURSOR.
 207  */
 208 static char *
 209 fetch_param_value(ExprContext *econtext, int paramId)
 210 {
 211         ParamListInfo paramInfo = econtext->ecxt_param_list_info;
 212
 213         if (paramInfo &&
 214                 paramId > 0 && paramId <= paramInfo->numParams)
 215         {
 216                 ParamExternData *prm = &paramInfo->params[paramId - 1];
 217
 218                 if (OidIsValid(prm->ptype) && !prm->isnull)
 219                 {
 220                         Assert(prm->ptype == REFCURSOROID);
 221                         /* We know that refcursor uses text's I/O routines */
 222                         return TextDatumGetCString(prm->value);
 223                 }
 224         }
 225
 226         ereport(ERROR,
 227                         (errcode(ERRCODE_UNDEFINED_OBJECT),
 228                          errmsg("no value found for parameter %d", paramId)));
 229         return NULL;
 230 }
 231
 232 /*
 233  * search_plan_tree
 234  *
 235  * Search through a PlanState tree for a scan node on the specified table.
 236  * Return NULL if not found or multiple candidates.
 237  */
 238 static ScanState *
 239 search_plan_tree(PlanState *node, Oid table_oid)
 240 {
 241         if (node == NULL)
 242                 return NULL;
 243         switch (nodeTag(node))
 244         {
 245                         /*
 246                          * scan nodes can all be treated alike
 247                          */
 248                 case T_SeqScanState:
 249                 case T_IndexScanState:
 250                 case T_BitmapHeapScanState:
 251                 case T_TidScanState:
 252                         {
 253                                 ScanState  *sstate = (ScanState *) node;
 254
 255                                 if (RelationGetRelid(sstate->ss_currentRelation) == table_oid)
 256                                         return sstate;
 257                                 break;
 258                         }
 259
 260                         /*
 261                          * For Append, we must look through the members; watch out for
 262                          * multiple matches (possible if it was from UNION ALL)
 263                          */
 264                 case T_AppendState:
 265                         {
 266                                 AppendState *astate = (AppendState *) node;
 267                                 ScanState  *result = NULL;
 268                                 int                     i;
 269
 270                                 for (i = 0; i < astate->as_nplans; i++)
 271                                 {
 272                                         ScanState  *elem = search_plan_tree(astate->appendplans[i],
 273                                                                                                                 table_oid);
 274
 275                                         if (!elem)
 276                                                 continue;
 277                                         if (result)
 278                                                 return NULL;    /* multiple matches */
 279                                         result = elem;
 280                                 }
 281                                 return result;
 282                         }
 283
 284                         /*
 285                          * Result and Limit can be descended through (these are safe
 286                          * because they always return their input's current row)
 287                          */
 288                 case T_ResultState:
 289                 case T_LimitState:
 290                         return search_plan_tree(node->lefttree, table_oid);
 291
 292                         /*
 293                          * SubqueryScan too, but it keeps the child in a different place
 294                          */
 295                 case T_SubqueryScanState:
 296                         return search_plan_tree(((SubqueryScanState *) node)->subplan,
 297                                                                         table_oid);
 298
 299                 default:
 300                         /* Otherwise, assume we can't descend through it */
 301                         break;
 302         }
 303         return NULL;
 304 }