src/backend/executor/execCurrent.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * execCurrent.c
   4  *        executor support for WHERE CURRENT OF cursor
   5  *
   6  * Portions Copyright (c) 1996-2008, 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         ScanState  *scanstate;
  50         bool            lisnull;
  51         Oid                     tuple_tableoid;
  52         ItemPointer tuple_tid;
  53
  54         /* Get the cursor name --- may have to look up a parameter reference */
  55         if (cexpr->cursor_name)
  56                 cursor_name = cexpr->cursor_name;
  57         else
  58                 cursor_name = fetch_param_value(econtext, cexpr->cursor_param);
  59
  60         /* Fetch table name for possible use in error messages */
  61         table_name = get_rel_name(table_oid);
  62         if (table_name == NULL)
  63                 elog(ERROR, "cache lookup failed for relation %u", table_oid);
  64
  65         /* Find the cursor's portal */
  66         portal = GetPortalByName(cursor_name);
  67         if (!PortalIsValid(portal))
  68                 ereport(ERROR,
  69                                 (errcode(ERRCODE_UNDEFINED_CURSOR),
  70                                  errmsg("cursor \"%s\" does not exist", cursor_name)));
  71
  72         /*
  73          * We have to watch out for non-SELECT queries as well as held cursors,
  74          * both of which may have null queryDesc.
  75          */
  76         if (portal->strategy != PORTAL_ONE_SELECT)
  77                 ereport(ERROR,
  78                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  79                                  errmsg("cursor \"%s\" is not a SELECT query",
  80                                                 cursor_name)));
  81         queryDesc = PortalGetQueryDesc(portal);
  82         if (queryDesc == NULL)
  83                 ereport(ERROR,
  84                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  85                                  errmsg("cursor \"%s\" is held from a previous transaction",
  86                                                 cursor_name)));
  87
  88         /*
  89          * Dig through the cursor's plan to find the scan node.  Fail if it's not
  90          * there or buried underneath aggregation.
  91          */
  92         scanstate = search_plan_tree(ExecGetActivePlanTree(queryDesc),
  93                                                                  table_oid);
  94         if (!scanstate)
  95                 ereport(ERROR,
  96                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  97                 errmsg("cursor \"%s\" is not a simply updatable scan of table \"%s\"",
  98                            cursor_name, table_name)));
  99
 100         /*
 101          * The cursor must have a current result row: per the SQL spec, it's an
 102          * error if not.  We test this at the top level, rather than at the scan
 103          * node level, because in inheritance cases any one table scan could
 104          * easily not be on a row.      We want to return false, not raise error, if
 105          * the passed-in table OID is for one of the inactive scans.
 106          */
 107         if (portal->atStart || portal->atEnd)
 108                 ereport(ERROR,
 109                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
 110                                  errmsg("cursor \"%s\" is not positioned on a row",
 111                                                 cursor_name)));
 112
 113         /* Now OK to return false if we found an inactive scan */
 114         if (TupIsNull(scanstate->ss_ScanTupleSlot))
 115                 return false;
 116
 117         /* Use slot_getattr to catch any possible mistakes */
 118         tuple_tableoid = DatumGetObjectId(slot_getattr(scanstate->ss_ScanTupleSlot,
 119                                                                                                    TableOidAttributeNumber,
 120                                                                                                    &lisnull));
 121         Assert(!lisnull);
 122         tuple_tid = (ItemPointer)
 123                 DatumGetPointer(slot_getattr(scanstate->ss_ScanTupleSlot,
 124                                                                          SelfItemPointerAttributeNumber,
 125                                                                          &lisnull));
 126         Assert(!lisnull);
 127
 128         Assert(tuple_tableoid == table_oid);
 129
 130         *current_tid = *tuple_tid;
 131
 132         return true;
 133 }
 134
 135 /*
 136  * fetch_param_value
 137  *
 138  * Fetch the string value of a param, verifying it is of type REFCURSOR.
 139  */
 140 static char *
 141 fetch_param_value(ExprContext *econtext, int paramId)
 142 {
 143         ParamListInfo paramInfo = econtext->ecxt_param_list_info;
 144
 145         if (paramInfo &&
 146                 paramId > 0 && paramId <= paramInfo->numParams)
 147         {
 148                 ParamExternData *prm = &paramInfo->params[paramId - 1];
 149
 150                 if (OidIsValid(prm->ptype) && !prm->isnull)
 151                 {
 152                         Assert(prm->ptype == REFCURSOROID);
 153                         /* We know that refcursor uses text's I/O routines */
 154                         return TextDatumGetCString(prm->value);
 155                 }
 156         }
 157
 158         ereport(ERROR,
 159                         (errcode(ERRCODE_UNDEFINED_OBJECT),
 160                          errmsg("no value found for parameter %d", paramId)));
 161         return NULL;
 162 }
 163
 164 /*
 165  * search_plan_tree
 166  *
 167  * Search through a PlanState tree for a scan node on the specified table.
 168  * Return NULL if not found or multiple candidates.
 169  */
 170 static ScanState *
 171 search_plan_tree(PlanState *node, Oid table_oid)
 172 {
 173         if (node == NULL)
 174                 return NULL;
 175         switch (nodeTag(node))
 176         {
 177                         /*
 178                          * scan nodes can all be treated alike
 179                          */
 180                 case T_SeqScanState:
 181                 case T_IndexScanState:
 182                 case T_BitmapHeapScanState:
 183                 case T_TidScanState:
 184                         {
 185                                 ScanState  *sstate = (ScanState *) node;
 186
 187                                 if (RelationGetRelid(sstate->ss_currentRelation) == table_oid)
 188                                         return sstate;
 189                                 break;
 190                         }
 191
 192                         /*
 193                          * For Append, we must look through the members; watch out for
 194                          * multiple matches (possible if it was from UNION ALL)
 195                          */
 196                 case T_AppendState:
 197                         {
 198                                 AppendState *astate = (AppendState *) node;
 199                                 ScanState  *result = NULL;
 200                                 int                     i;
 201
 202                                 for (i = 0; i < astate->as_nplans; i++)
 203                                 {
 204                                         ScanState  *elem = search_plan_tree(astate->appendplans[i],
 205                                                                                                                 table_oid);
 206
 207                                         if (!elem)
 208                                                 continue;
 209                                         if (result)
 210                                                 return NULL;    /* multiple matches */
 211                                         result = elem;
 212                                 }
 213                                 return result;
 214                         }
 215
 216                         /*
 217                          * Result and Limit can be descended through (these are safe
 218                          * because they always return their input's current row)
 219                          */
 220                 case T_ResultState:
 221                 case T_LimitState:
 222                         return search_plan_tree(node->lefttree, table_oid);
 223
 224                         /*
 225                          * SubqueryScan too, but it keeps the child in a different place
 226                          */
 227                 case T_SubqueryScanState:
 228                         return search_plan_tree(((SubqueryScanState *) node)->subplan,
 229                                                                         table_oid);
 230
 231                 default:
 232                         /* Otherwise, assume we can't descend through it */
 233                         break;
 234         }
 235         return NULL;
 236 }