src/backend/executor/execCurrent.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * execCurrent.c
   4  *        executor support for WHERE CURRENT OF cursor
   5  *
   6  * Portions Copyright (c) 1996-2007, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *      $PostgreSQL: pgsql/src/backend/executor/execCurrent.c,v 1.4 2007/11/15 22:25:15 momjian Exp $
  10  *
  11  *-------------------------------------------------------------------------
  12  */
  13 #include "postgres.h"
  14
  15 #include "catalog/pg_type.h"
  16 #include "executor/executor.h"
  17 #include "utils/builtins.h"
  18 #include "utils/lsyscache.h"
  19 #include "utils/portal.h"
  20
  21
  22 static char *fetch_param_value(ExprContext *econtext, int paramId);
  23 static ScanState *search_plan_tree(PlanState *node, Oid table_oid);
  24
  25
  26 /*
  27  * execCurrentOf
  28  *
  29  * Given a CURRENT OF expression and the OID of a table, determine which row
  30  * of the table is currently being scanned by the cursor named by CURRENT OF,
  31  * and return the row's TID into *current_tid.
  32  *
  33  * Returns TRUE if a row was identified.  Returns FALSE if the cursor is valid
  34  * for the table but is not currently scanning a row of the table (this is a
  35  * legal situation in inheritance cases).  Raises error if cursor is not a
  36  * valid updatable scan of the specified table.
  37  */
  38 bool
  39 execCurrentOf(CurrentOfExpr *cexpr,
  40                           ExprContext *econtext,
  41                           Oid table_oid,
  42                           ItemPointer current_tid)
  43 {
  44         char       *cursor_name;
  45         char       *table_name;
  46         Portal          portal;
  47         QueryDesc  *queryDesc;
  48         ScanState  *scanstate;
  49         bool            lisnull;
  50         Oid                     tuple_tableoid;
  51         ItemPointer tuple_tid;
  52
  53         /* Get the cursor name --- may have to look up a parameter reference */
  54         if (cexpr->cursor_name)
  55                 cursor_name = cexpr->cursor_name;
  56         else
  57                 cursor_name = fetch_param_value(econtext, cexpr->cursor_param);
  58
  59         /* Fetch table name for possible use in error messages */
  60         table_name = get_rel_name(table_oid);
  61         if (table_name == NULL)
  62                 elog(ERROR, "cache lookup failed for relation %u", table_oid);
  63
  64         /* Find the cursor's portal */
  65         portal = GetPortalByName(cursor_name);
  66         if (!PortalIsValid(portal))
  67                 ereport(ERROR,
  68                                 (errcode(ERRCODE_UNDEFINED_CURSOR),
  69                                  errmsg("cursor \"%s\" does not exist", cursor_name)));
  70
  71         /*
  72          * We have to watch out for non-SELECT queries as well as held cursors,
  73          * both of which may have null queryDesc.
  74          */
  75         if (portal->strategy != PORTAL_ONE_SELECT)
  76                 ereport(ERROR,
  77                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  78                                  errmsg("cursor \"%s\" is not a SELECT query",
  79                                                 cursor_name)));
  80         queryDesc = PortalGetQueryDesc(portal);
  81         if (queryDesc == NULL)
  82                 ereport(ERROR,
  83                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  84                                  errmsg("cursor \"%s\" is held from a previous transaction",
  85                                                 cursor_name)));
  86
  87         /*
  88          * Dig through the cursor's plan to find the scan node.  Fail if it's not
  89          * there or buried underneath aggregation.
  90          */
  91         scanstate = search_plan_tree(ExecGetActivePlanTree(queryDesc),
  92                                                                  table_oid);
  93         if (!scanstate)
  94                 ereport(ERROR,
  95                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
  96                 errmsg("cursor \"%s\" is not a simply updatable scan of table \"%s\"",
  97                            cursor_name, table_name)));
  98
  99         /*
 100          * The cursor must have a current result row: per the SQL spec, it's an
 101          * error if not.  We test this at the top level, rather than at the scan
 102          * node level, because in inheritance cases any one table scan could
 103          * easily not be on a row.      We want to return false, not raise error, if
 104          * the passed-in table OID is for one of the inactive scans.
 105          */
 106         if (portal->atStart || portal->atEnd)
 107                 ereport(ERROR,
 108                                 (errcode(ERRCODE_INVALID_CURSOR_STATE),
 109                                  errmsg("cursor \"%s\" is not positioned on a row",
 110                                                 cursor_name)));
 111
 112         /* Now OK to return false if we found an inactive scan */
 113         if (TupIsNull(scanstate->ss_ScanTupleSlot))
 114                 return false;
 115
 116         /* Use slot_getattr to catch any possible mistakes */
 117         tuple_tableoid = DatumGetObjectId(slot_getattr(scanstate->ss_ScanTupleSlot,
 118                                                                                                    TableOidAttributeNumber,
 119                                                                                                    &lisnull));
 120         Assert(!lisnull);
 121         tuple_tid = (ItemPointer)
 122                 DatumGetPointer(slot_getattr(scanstate->ss_ScanTupleSlot,
 123                                                                          SelfItemPointerAttributeNumber,
 124                                                                          &lisnull));
 125         Assert(!lisnull);
 126
 127         Assert(tuple_tableoid == table_oid);
 128
 129         *current_tid = *tuple_tid;
 130
 131         return true;
 132 }
 133
 134 /*
 135  * fetch_param_value
 136  *
 137  * Fetch the string value of a param, verifying it is of type REFCURSOR.
 138  */
 139 static char *
 140 fetch_param_value(ExprContext *econtext, int paramId)
 141 {
 142         ParamListInfo paramInfo = econtext->ecxt_param_list_info;
 143
 144         if (paramInfo &&
 145                 paramId > 0 && paramId <= paramInfo->numParams)
 146         {
 147                 ParamExternData *prm = &paramInfo->params[paramId - 1];
 148
 149                 if (OidIsValid(prm->ptype) && !prm->isnull)
 150                 {
 151                         Assert(prm->ptype == REFCURSOROID);
 152                         /* We know that refcursor uses text's I/O routines */
 153                         return DatumGetCString(DirectFunctionCall1(textout,
 154                                                                                                            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 }